diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8edc2a5..4cb7089 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -39,8 +39,8 @@ jobs: cache: pip - run: python -m pip install --upgrade pip - run: python -m pip install -e ".[dev]" - - run: python -m black --check src tests examples benchmarks - - run: python -m ruff check src tests examples benchmarks + - run: python -m black --check src tests examples benchmarks scripts + - run: python -m ruff check src tests examples benchmarks scripts - run: python -m mypy package: @@ -57,5 +57,4 @@ jobs: - run: python -m twine check --strict dist/* - run: python -m venv .smoke - run: .smoke/bin/python -m pip install --no-deps dist/*.whl - - run: .smoke/bin/python -c "from samsarix_core import MCPServer, ToolPolicyContext, ToolPolicyDecision, ToolRateLimit, ToolRuntime, serve_stdio, __version__; assert ToolRateLimit(calls=1, period_seconds=1).burst_capacity == 1; print(__version__)" - - run: .smoke/bin/python -c "from helix_core import MCPServer, ToolPolicyContext, ToolPolicyDecision, ToolRateLimit, serve_stdio, __version__ as legacy_version; from samsarix_core import __version__; assert legacy_version == __version__, f'{legacy_version} != {__version__}'" + - run: .smoke/bin/python scripts/smoke_check.py diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index e8a29b7..b717756 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -43,8 +43,7 @@ jobs: run: | python -m venv .release-smoke .release-smoke/bin/python -m pip install --no-deps dist/*.whl - .release-smoke/bin/python -c "from samsarix_core import MCPServer, ToolPolicyContext, ToolPolicyDecision, ToolRateLimit, ToolRuntime, serve_stdio, __version__; assert ToolRateLimit(calls=1, period_seconds=1).burst_capacity == 1; print(__version__)" - .release-smoke/bin/python -c "from helix_core import MCPServer, ToolPolicyContext, ToolPolicyDecision, ToolRateLimit, serve_stdio, __version__ as legacy_version; from samsarix_core import __version__; assert legacy_version == __version__, f'{legacy_version} != {__version__}'" + .release-smoke/bin/python scripts/smoke_check.py - name: Require the tag to match the package version if: startsWith(github.ref, 'refs/tags/v') env: diff --git a/CHANGELOG.md b/CHANGELOG.md index ad7988b..12f279e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,13 @@ ## Unreleased +### Added + +- opt-in process-local per-tool consecutive-failure circuit breakers with safe + fail-fast results, one half-open recovery probe, queued-permit invalidation, + manual inspection/reset, content-free metrics/lifecycle events, and consistent + direct, batch, MCP, and task behavior. + ### Changed - recorded immutable `v2.0.0a6` release checksums, provenance, and clean installed-wheel diff --git a/MANIFEST.in b/MANIFEST.in index 8034ef3..ea9e115 100644 --- a/MANIFEST.in +++ b/MANIFEST.in @@ -10,5 +10,6 @@ include TRADEMARKS.md recursive-include docs *.md recursive-include examples *.py recursive-include benchmarks *.py +recursive-include scripts *.py prune legacy global-exclude __pycache__ *.py[cod] diff --git a/README.md b/README.md index 3afec48..d84d7a7 100644 --- a/README.md +++ b/README.md @@ -18,11 +18,11 @@ persistence, or an untrusted-code sandbox. - turns annotated sync or async functions into inspectable tool contracts; - emits JSON Schema Draft 2020-12 input and output schemas; - validates arguments and outputs without surprising scalar coercion; -- returns structured success, validation, policy-denial, overload, rate-limit, timeout, - missing-tool, and failure results; +- returns structured success, validation, policy-denial, overload, rate-limit, + open-circuit, timeout, missing-tool, and failure results; - bounds pending invocations, registry growth, batches, value size/complexity, - global and per-tool concurrent work, sustained per-tool request rate, and - thread-pool use; + global and per-tool concurrent work, sustained per-tool request rate, repeated + calls to failing dependencies, and thread-pool use; - supports ordered batch invocation and cooperative async cancellation; - optionally requires a bounded host-owned policy decision after validation and before any tool code executes; @@ -111,6 +111,8 @@ snapshot and returns `ToolPolicyDecision.ALLOW` or `DENY`; it is not an authenti service or a durable human-approval workflow. For a quota-constrained dependency with safe retry handling, run `python examples/rate_limited_api.py`. +For a failing dependency with fail-fast recovery probing, run +`python examples/circuit_breaker_api.py`. ## Connect an MCP client @@ -192,6 +194,7 @@ exception. It returns a `ToolResult` with one of these states: - `denied` - `busy` - `rate_limited` +- `circuit_open` - `timed_out` - `failed` - `runtime_closed` @@ -217,13 +220,17 @@ them to the host's actual workload. Hosts can isolate a slow or quota-constrained dependency when registering its tool: ```python -from samsarix_core import ToolRateLimit +from samsarix_core import ToolCircuitBreaker, ToolRateLimit runtime = ToolRuntime(max_concurrency=8) runtime.register( query_warehouse, max_concurrency=2, rate_limit=ToolRateLimit(calls=60, period_seconds=60, burst=5), + circuit_breaker=ToolCircuitBreaker( + failure_threshold=3, + recovery_timeout_seconds=30, + ), ) runtime.register(health_check) ``` @@ -240,6 +247,14 @@ warehouse calls to execute concurrently. An empty bucket returns retryable statu process-local deployment policy, not tenant fairness or a distributed quota service. See [per-tool rate limits](docs/RATE_LIMITS.md). +The circuit breaker counts consecutive tool/output failures and caller-visible +timeouts. Once open, it rejects before capacity or tokens are consumed, then admits one +half-open recovery probe after 30 seconds. Invalid input, host policy outcomes, +admission/rate rejections, progress-handler failure, and caller cancellation do not +trip it. Core does not automatically retry calls. Breaker state is process-local +deployment policy and can be inspected or manually reset by the host. See +[per-tool circuit breakers](docs/CIRCUIT_BREAKERS.md). + Hosts can attach a synchronous `lifecycle_handler` to receive paired, immutable start and terminal events across direct, batch, MCP, and task calls. Events contain only invocation ID, requested tool name, status, UTC time, and terminal duration; arguments, @@ -272,7 +287,7 @@ for delivery semantics, privacy/cardinality cautions, and a content-free OpenTel See [Getting started](docs/GETTING_STARTED.md), the [API reference](docs/API_REFERENCE.md), [architecture](docs/ARCHITECTURE.md), [MCP bridge](docs/MCP.md), [lifecycle observability](docs/OBSERVABILITY.md), [best practices](docs/BEST_PRACTICES.md), -[per-tool rate limits](docs/RATE_LIMITS.md), +[per-tool rate limits](docs/RATE_LIMITS.md), [per-tool circuit breakers](docs/CIRCUIT_BREAKERS.md), [benchmark guide](docs/BENCHMARKS.md), the [adoption record](docs/ADOPTION.md), and the [productization record](docs/PRODUCTIZATION.md). diff --git a/ROADMAP.md b/ROADMAP.md index d2bf693..4e98a64 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -36,6 +36,10 @@ remain separate decisions. - Tool registrations can also apply process-local token buckets immediately before execution, protecting sustained downstream request quotas with safe retry hints without treating the runtime as a distributed or per-tenant quota service. +- Tool registrations can opt into process-local consecutive-failure circuit breakers + that fail fast before capacity and rate tokens, invalidate queued stale permits, + allow one half-open recovery probe, and expose safe results plus host inspection/reset + without automatic retry or distributed-health claims. - Strict `TypedDict` input and output contracts now preserve named nested fields, descriptions, and required/optional key semantics in JSON Schema and runtime validation. @@ -84,6 +88,8 @@ remain separate decisions. evidence. - [x] Publish immutable GitHub prerelease `v2.0.0a6` with per-tool rate limiting, independent consumer, clean-install, checksum, and SLSA provenance evidence. +- [ ] Prove independent consumer adoption and publish an immutable prerelease for the + per-tool circuit-breaker contract after exact-head review and clean-wheel evidence. ## Samsarix adoption diff --git a/docs/API_REFERENCE.md b/docs/API_REFERENCE.md index 31fbd2a..7cedd56 100644 --- a/docs/API_REFERENCE.md +++ b/docs/API_REFERENCE.md @@ -72,10 +72,12 @@ ToolRuntime( ) ``` -- `register(function, *, replace=False, max_concurrency=None, rate_limit=None) -> ToolSpec` +- `register(function, *, replace=False, max_concurrency=None, rate_limit=None, circuit_breaker=None) -> ToolSpec` - `await invoke(name, arguments=None, *, timeout=None, progress_handler=None) -> ToolResult` - `await invoke_many(calls) -> list[ToolResult]` - `metrics() -> RuntimeMetrics` +- `circuit_state(name) -> ToolCircuitState | None` +- `reset_circuit(name) -> bool` - `pending_sync_calls -> int` - `await wait_for_sync(*, timeout=None) -> bool` - `await aclose(*, wait_for_sync=False, timeout=None) -> bool` @@ -122,6 +124,17 @@ per-tenant quota. Token counts are positive integers no larger than `2**53 - 1`; period and derived refill/retry magnitudes must be positive and finite. See [Per-tool rate limits](RATE_LIMITS.md). +`register(..., circuit_breaker=ToolCircuitBreaker(failure_threshold=N, +recovery_timeout_seconds=S))` adds a process-local consecutive-execution-failure +breaker to the exact registration. It fails fast with retryable status `circuit_open` +and safe code `tool_circuit_open`, admits one half-open probe after the recovery +interval, and closes after a successful validated output. Invalid input, policy and +admission outcomes, rate rejection, progress-handler failure, and caller cancellation +do not count; tool/output failures and caller-visible timeouts do. The breaker is +checked before capacity and again before execution so a queued stale permit cannot run +after another call opens it. `circuit_state()` inspects configured state and +`reset_circuit()` forces it closed. See [Per-tool circuit breakers](CIRCUIT_BREAKERS.md). + Argument and output sizes are the UTF-8 byte length of compact JSON. The root is depth zero; every child increments depth. Each container and scalar is one node, while object keys are covered by the byte limit but do not count as nodes. Cyclic, @@ -255,6 +268,9 @@ All public models are frozen, slotted dataclasses. - `ToolRateLimit`: positive token allocation, finite refill period, and optional burst capacity. `burst_capacity` resolves the default and `to_dict()` returns normalized deployment-local configuration. +- `ToolCircuitBreaker`: positive consecutive-failure threshold and finite positive + recovery interval for one process-local exact registration. +- `ToolCircuitState`: `closed`, `open`, or `half_open`. - `ToolPolicyContext`: invocation ID plus detached validated arguments and tool spec; unlike result models, it intentionally has no serialization helper. - `ToolPolicyDecision`: explicit `allow` or `deny` policy outcome. @@ -269,13 +285,13 @@ All public models are frozen, slotted dataclasses. - `ToolError`: code, safe message, optional exception type/details, and retryable flag. - `ToolProgress`: numeric progress, optional total, and optional human-readable message. - `RuntimeMetrics`: content-free counters only, including policy denials, aggregate - process-wide rate-limit rejections, and runtime saturation. + process-wide rate/circuit rejections and breaker trips, and runtime saturation. - `ToolStatus`: `success`, `not_found`, `invalid_arguments`, `denied`, `busy`, - `rate_limited`, `timed_out`, `failed`, and `runtime_closed`. + `rate_limited`, `circuit_open`, `timed_out`, `failed`, and `runtime_closed`. - `TaskSupport`: the `"forbidden" | "optional" | "required"` public type alias. -`ToolSpec`, `ToolRateLimit`, `ToolResult`, `ToolError`, `ToolLifecycleEvent`, and -`RuntimeMetrics` provide `to_dict()`. +`ToolSpec`, `ToolRateLimit`, `ToolCircuitBreaker`, `ToolResult`, `ToolError`, +`ToolLifecycleEvent`, and `RuntimeMetrics` provide `to_dict()`. ## Supported annotations diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 4c5e7ef..4295db8 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -16,7 +16,9 @@ invocation: incoming call -> iterative argument resource preflight -> ToolRuntime argument validation -> optional bounded host policy decision + -> optional per-tool circuit permit -> optional per-tool execution bulkhead + -> queued circuit-permit revalidation -> optional per-tool token-bucket start check -> bounded async or thread-pool execution -> optional invocation-scoped progress handler @@ -33,9 +35,9 @@ invocation: incoming call iterative resource checks, strict input validation, and JSON normalization. - `registry.py` stores a capped set of compiled callable contracts behind a small thread-safe map. -- `runtime.py` owns concurrency, process-local per-tool rate controls, the sync thread - pool, timeouts, cancellation, exception redaction, batch workers, lifecycle, and - content-free counters. +- `runtime.py` owns concurrency, process-local per-tool circuit/rate controls, the sync + thread pool, timeouts, cancellation, exception redaction, batch workers, lifecycle, + and content-free counters. - `progress.py` owns invocation-scoped async progress validation, ordering, resource caps, and lifecycle closure. - `_mcp_tasks.py` owns finite in-memory task retention, secure identifiers, TTL @@ -97,6 +99,14 @@ global and per-tool semaphores still bound execution. This prevents workers queu one constrained tool from head-of-line blocking unrelated calls that fit within the batch's available pending capacity. +An optional circuit breaker is also bound to the exact registration and checked before +capacity is acquired. A generation-bound permit is revalidated at the execution +boundary; opening or manually resetting a circuit invalidates queued permits and late +outcomes. Breaker outcome is committed before async or thread-backed capacity is +released. Open state admits no work until its monotonic recovery interval elapses, then +reserves exactly one half-open probe. The runtime performs no automatic retries and no +background health checks. + An exact registration may also own an event-loop-local token bucket. After validation, policy, and concurrency acquisition, the runtime checks one token immediately before starting tool code. It never waits for a token. Rejection releases the acquired slots, diff --git a/docs/BEST_PRACTICES.md b/docs/BEST_PRACTICES.md index 6d03d86..5322f2f 100644 --- a/docs/BEST_PRACTICES.md +++ b/docs/BEST_PRACTICES.md @@ -56,8 +56,8 @@ this tool-specific slot before a global execution slot, preserving unrelated too availability while the constrained tool queues. The per-tool concurrency limit is an in-process execution bulkhead, not a request-rate -limit, tenant quota, circuit breaker, or process sandbox. When the dependency also has a -sustained request quota, add `rate_limit=ToolRateLimit(...)`. The token bucket is checked +limit, tenant quota, or process sandbox. When the dependency also has a sustained +request quota, add `rate_limit=ToolRateLimit(...)`. The token bucket is checked immediately before tool start and returns a safe retry delay instead of waiting. Combine both controls: concurrency protects simultaneous downstream work while rate protects starts over time. Keep total admission finite and set downstream I/O deadlines. @@ -66,6 +66,14 @@ for the reliability trade-offs and complementary controls. See [per-tool rate limits](RATE_LIMITS.md) for token accounting, safe retry behavior, primary references, and process-local limitations. +When repeatedly calling a known-failing dependency would waste capacity or amplify an +incident, add `circuit_breaker=ToolCircuitBreaker(...)`. Choose the consecutive-failure +threshold and recovery interval from observed failure modes and recovery time. Monitor +both trips and open-circuit rejections. Keep dependency connect/read/write deadlines: +the breaker reacts to observed outcomes and does not interrupt a hung operation by +itself. It also does not retry. See [per-tool circuit breakers](CIRCUIT_BREAKERS.md) for +counting, half-open probes, manual reset, primary references, and process-local limits. + Choose the runtime-wide `max_concurrency` from aggregate downstream capacity, not CPU count alone. Set `max_pending_invocations` to the total policy/execution work one process can safely hold; monitor `busy` and `peak_pending_invocations` to find sustained saturation. @@ -105,8 +113,11 @@ Branch on `ToolStatus`; do not infer success from a truthy output. A `busy` resu retryable because no tool or policy code ran, but use capped exponential backoff with jitter rather than retrying immediately. A `rate_limited` result also means tool code did not run; wait at least `error.details["retry_after_ms"]`, add bounded jitter for -competing callers, and remember that the hint does not reserve the next token. Other -failures remain non-retryable because a timed-out sync function may still finish and +competing callers, and remember that the hint does not reserve the next token. A +`circuit_open` result also means this call did not run; its retry delay may be absent +while another recovery probe is active. Retry only after bounded backoff and only when +the tool's side-effect semantics allow it. Other failures remain non-retryable because +a timed-out sync function may still finish and cause its side effect. Apply any broader retry policy only when the tool's semantics make that safe. diff --git a/docs/CIRCUIT_BREAKERS.md b/docs/CIRCUIT_BREAKERS.md new file mode 100644 index 0000000..a3ddaee --- /dev/null +++ b/docs/CIRCUIT_BREAKERS.md @@ -0,0 +1,137 @@ +# Per-tool circuit breakers + +Samsarix Core can attach an opt-in circuit breaker to one exact tool registration. It +protects a host from repeatedly calling a dependency that is already failing, while +leaving unrelated tools available. + +```python +from samsarix_core import ToolCircuitBreaker, ToolRuntime + +runtime = ToolRuntime(max_concurrency=8) +runtime.register( + query_vendor_api, + max_concurrency=2, + circuit_breaker=ToolCircuitBreaker( + failure_threshold=3, + recovery_timeout_seconds=30, + ), +) +``` + +The breaker starts closed. Three consecutive execution failures open it for 30 seconds. +Calls rejected while open do not wait for concurrency, consume a rate-limit token, or +run tool code. Once the recovery interval has elapsed, one caller becomes the half-open +probe. A successful probe closes the breaker; a failed probe opens it for another full +recovery interval. Other callers fail fast while that probe is active. + +## Result contract + +An open circuit returns a safe structured result: + +```json +{ + "status": "circuit_open", + "error": { + "code": "tool_circuit_open", + "message": "Tool dependency circuit is temporarily open", + "retryable": true, + "details": {"retry_after_ms": 30000} + } +} +``` + +`details.retry_after_ms` is present while the configured recovery interval remains. It +is a hint, not a reservation. A rejection caused by another active half-open probe has +no retry delay because the probe's completion time is unknown. Clients should use +bounded jitter and retry a side-effecting tool only when application semantics make the +retry safe. + +The error contains no arguments, output, exception text, policy context, or progress +message. `RuntimeMetrics.circuit_open` counts rejections and +`RuntimeMetrics.circuit_breaker_trips` counts transitions caused by a threshold or a +failed probe. Lifecycle events use terminal status `circuit_open`. MCP serializes the +same result as a tool-origin error with `isError: true`; a task-augmented call reaches +task status `failed` and retains only that safe terminal result. + +## What counts as a failure + +The consecutive-failure counter changes only after the host policy has allowed an +invocation and the protected execution has started or reached its caller-visible +deadline: + +- a tool exception counts; +- output validation or resource-limit failure counts; +- a caller-visible timeout counts, including a timed-out synchronous worker that may + still be stopping; +- a successful validated output resets the consecutive-failure count; +- missing tools, invalid arguments, policy denial or policy failure, runtime admission + rejection, rate-limit rejection, progress-handler failure, and caller cancellation do + not count. + +Core does not automatically retry a failed call. Retries and circuit breaking solve +different problems: an application may add bounded retry behavior around a transient +operation, but should account for idempotency, its total deadline, and the extra load +that retrying creates. + +## Ordering and concurrency + +One invocation proceeds in this order: + +1. bounded runtime admission; +2. tool lookup plus argument resource and schema validation; +3. optional host policy; +4. circuit permit; +5. optional per-tool and global concurrency acquisition; +6. queued-permit revalidation; +7. optional rate-token consumption; +8. tool execution and output validation. + +The permit is checked again at the execution boundary. If one failure opens the circuit +while another call is waiting for capacity, the queued call is rejected instead of +reaching the dependency. Breaker outcome is recorded before capacity is released, for +both coroutine and thread-backed tools. This preserves the fail-fast boundary under +concurrency. Waiting before execution remains covered by the ordinary invocation +timeout and the runtime-wide pending-invocation cap. + +## Inspection, reset, and replacement + +```python +from samsarix_core import ToolCircuitState + +state = runtime.circuit_state("query_vendor_api") +if state is ToolCircuitState.OPEN: + alerted = True + +runtime.reset_circuit("query_vendor_api") +``` + +`circuit_state(name)` returns `closed`, `open`, or `half_open` for a configured breaker, +and `None` for a registered tool without one. It reports the explicit current state; +observing an elapsed recovery interval does not reserve the single probe. +`reset_circuit(name)` returns `True` when it reset a configured breaker and `False` for +an unprotected registered tool. Both methods raise `ToolNotFoundError` for an unknown +name. A manual reset invalidates outstanding permits, so a late result from pre-reset +work cannot mutate the new state. + +Replacing a tool discards its old breaker. Omitting `circuit_breaker` leaves the +replacement unprotected apart from other runtime controls. Direct registry mutation +does not add a breaker; use `ToolRuntime.register` when this control is required. + +## Scope and limits + +The breaker is dependency-free, process-local, monotonic-clock-based, and content-free. +It coordinates direct, batch, ordinary MCP, and task-augmented MCP calls through one +runtime. It does not coordinate multiple processes or machines, persist across restart, +identify tenants, provide a statistical sliding window, probe dependency health out of +band, or replace dependency-level connect/read/write deadlines. Deployments needing a +shared breaker should place it at a shared authenticated gateway or service boundary. + +The closed/open/half-open model and fail-fast behavior follow current vendor-neutral +guidance, while the deliberately small consecutive-failure policy keeps the public +contract auditable: + +- [Azure Architecture Center: Circuit Breaker pattern](https://learn.microsoft.com/en-us/azure/architecture/patterns/circuit-breaker) +- [AWS Prescriptive Guidance: Circuit breaker pattern](https://docs.aws.amazon.com/prescriptive-guidance/latest/cloud-design-patterns/circuit-breaker.html) +- [Polly circuit-breaker strategy](https://www.pollydocs.org/strategies/circuit-breaker) +- [Resilience4j circuit breaker](https://resilience4j.readme.io/docs/circuitbreaker) +- [MCP schema: tool execution errors](https://modelcontextprotocol.io/specification/2025-11-25/schema) diff --git a/docs/GETTING_STARTED.md b/docs/GETTING_STARTED.md index c33beed..2f37313 100644 --- a/docs/GETTING_STARTED.md +++ b/docs/GETTING_STARTED.md @@ -85,19 +85,26 @@ When one dependency has both a concurrency ceiling and a sustained request quota configure the exact registration rather than slowing unrelated tools: ```python -from samsarix_core import ToolRateLimit +from samsarix_core import ToolCircuitBreaker, ToolRateLimit async with ToolRuntime(max_concurrency=8) as quota_runtime: quota_runtime.register( greet, max_concurrency=2, rate_limit=ToolRateLimit(calls=30, period_seconds=60, burst=3), + circuit_breaker=ToolCircuitBreaker( + failure_threshold=3, + recovery_timeout_seconds=30, + ), ) result = await quota_runtime.invoke("greet", {"name": "Ada"}) ``` An empty bucket returns the retryable `rate_limited` result without running the tool. The bucket is local to this runtime process; see [per-tool rate limits](RATE_LIMITS.md). +Three consecutive execution failures open the independent circuit for 30 seconds; +open calls return a retryable `circuit_open` result without running the tool. One +half-open probe tests recovery. See [per-tool circuit breakers](CIRCUIT_BREAKERS.md). ## 4. Invoke a bounded batch diff --git a/docs/MCP.md b/docs/MCP.md index 25276ea..b2289c3 100644 --- a/docs/MCP.md +++ b/docs/MCP.md @@ -11,7 +11,7 @@ The supported server surface is intentionally narrow: - `ping`; - `tools/list` with JSON Schema Draft 2020-12 input and output contracts; - `tools/call` with Samsarix validation, timeouts, concurrency and opt-in per-tool - rate limits, and safe structured errors, including optional host-policy denial; + circuit/rate controls, and safe structured errors, including optional host-policy denial; - requested `notifications/progress` updates from cooperative asynchronous tools; - opt-in `logging/setLevel` and content-free `notifications/message` operational events; @@ -100,6 +100,16 @@ not identify clients or coordinate multiple server processes. A network adapter needs authenticated per-principal and distributed controls. See [per-tool rate limits](RATE_LIMITS.md) for configuration and token-accounting semantics. +## Fail fast around an unhealthy dependency + +`ToolRuntime.register(..., circuit_breaker=ToolCircuitBreaker(...))` applies the same +process-local breaker to direct, ordinary MCP, and task-augmented calls. An open circuit +returns `isError: true`, Samsarix status `circuit_open`, safe code +`tool_circuit_open`, and a retry delay when the recovery interval has time remaining. +The failure is a tool execution result rather than a JSON-RPC protocol error. A task +reaches `failed` and retains that exact safe result; call arguments and the triggering +exception are not reflected. See [per-tool circuit breakers](CIRCUIT_BREAKERS.md). + ## Structured output MCP requires an object at the root of `outputSchema`. Object-returning Samsarix diff --git a/docs/OBSERVABILITY.md b/docs/OBSERVABILITY.md index 1b41993..6d90e62 100644 --- a/docs/OBSERVABILITY.md +++ b/docs/OBSERVABILITY.md @@ -132,6 +132,7 @@ Primary references: | Need | Core surface | Cardinality/content | | --- | --- | --- | | Aggregate runtime health | `RuntimeMetrics` | Content-free counters; no tool names | +| Circuit health | `circuit_open`, `circuit_breaker_trips`, `circuit_state(name)` | Aggregate counters plus host-requested per-tool state | | Per-invocation traces or host logs | `lifecycle_handler` | Tool name and invocation ID; no call content | | MCP client diagnostics | opt-in MCP logging | Terminal tool name, ID, status, and duration | | User-facing work progress | `progress_handler` | Application text may cross the trust boundary | @@ -139,3 +140,10 @@ Primary references: Avoid enabling two terminal logging paths into the same backend unless duplicate events are intentional. Lifecycle signals operate at the direct runtime boundary and therefore cover direct, batch, MCP, and task-augmented calls uniformly. + +Alert on sustained `circuit_open` growth and breaker trips rather than on a single +expected rejection. `circuit_state(name)` is an explicit host diagnostic for a known +registration; unlike aggregate metrics it identifies which dependency is protected. +Do not use caller-controlled tool names as unbounded metric labels. Manual +`reset_circuit(name)` should be an operator action with its own application-owned audit +trail because Core deliberately emits no content-bearing reset event. diff --git a/docs/RATE_LIMITS.md b/docs/RATE_LIMITS.md index ccf310d..26b07f2 100644 --- a/docs/RATE_LIMITS.md +++ b/docs/RATE_LIMITS.md @@ -59,9 +59,11 @@ One invocation proceeds in this order: 1. bounded runtime admission; 2. tool lookup plus argument resource and schema validation; 3. optional host policy; -4. optional per-tool and global concurrency acquisition; -5. token check and consumption; -6. tool execution. +4. optional circuit permit; +5. optional per-tool and global concurrency acquisition; +6. queued circuit-permit revalidation; +7. token check and consumption; +8. tool execution. Missing tools, invalid calls, explicit policy denials, policy failures, and calls that time out or are cancelled before the execution controls are acquired do not spend a @@ -75,6 +77,11 @@ and omitting `rate_limit` leaves the replacement unrestricted apart from other r controls. Direct registry mutation does not add a rate policy; use `ToolRuntime.register` when this control is required. +An open circuit rejects before the bucket is checked, so it does not spend a token. A +rate rejection while holding the single half-open probe does not count as a dependency +failure or leave the breaker stuck; the next eligible caller can probe after a token is +available. See [per-tool circuit breakers](CIRCUIT_BREAKERS.md). + ## Scope and limits The bucket belongs to one `ToolRuntime` process and one exact registration. It is diff --git a/examples/circuit_breaker_api.py b/examples/circuit_breaker_api.py new file mode 100644 index 0000000..591456f --- /dev/null +++ b/examples/circuit_breaker_api.py @@ -0,0 +1,79 @@ +# Copyright 2026 Samsarix LLC +# SPDX-License-Identifier: MPL-2.0 + +"""Fail fast around an unhealthy dependency, then prove one recovery probe.""" + +from __future__ import annotations + +import asyncio +from typing import TypedDict + +from samsarix_core import ( + ToolCircuitBreaker, + ToolCircuitState, + ToolRuntime, + ToolStatus, + samsarix_tool, +) + + +class VendorStatus(TypedDict): + service: str + available: bool + + +dependency_available = False + + +@samsarix_tool(read_only=True, title="Get vendor status") +async def get_vendor_status(service: str) -> VendorStatus: + """Return a stand-in for one application-owned dependency request.""" + + await asyncio.sleep(0) + if not dependency_available: + raise ConnectionError("vendor is unavailable") + return {"service": service, "available": True} + + +async def main() -> None: + global dependency_available + + async with ToolRuntime(max_concurrency=4) as runtime: + runtime.register( + get_vendor_status, + max_concurrency=1, + circuit_breaker=ToolCircuitBreaker( + failure_threshold=1, + recovery_timeout_seconds=0.05, + ), + ) + + failed = await runtime.invoke("get_vendor_status", {"service": "catalog"}) + blocked = await runtime.invoke("get_vendor_status", {"service": "catalog"}) + print(failed.to_dict()) + print(blocked.to_dict()) + + if failed.status is not ToolStatus.FAILED: + raise RuntimeError("Expected the dependency call to fail") + if blocked.status is not ToolStatus.CIRCUIT_OPEN or blocked.error is None: + raise RuntimeError("Expected the open circuit to reject the next call") + if runtime.circuit_state("get_vendor_status") is not ToolCircuitState.OPEN: + raise RuntimeError("Expected an observable open circuit") + + details = blocked.error.details + retry_after_ms = details.get("retry_after_ms") if details is not None else None + if isinstance(retry_after_ms, bool) or not isinstance(retry_after_ms, int): + raise RuntimeError("Open-circuit result did not include a retry delay") + + dependency_available = True + await asyncio.sleep(retry_after_ms / 1_000 + 0.01) + recovered = await runtime.invoke("get_vendor_status", {"service": "catalog"}) + print(recovered.to_dict()) + if not recovered.success: + raise RuntimeError("Half-open recovery probe did not succeed") + if runtime.circuit_state("get_vendor_status") is not ToolCircuitState.CLOSED: + raise RuntimeError("Successful recovery probe did not close the circuit") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/pyproject.toml b/pyproject.toml index 3b0d18f..dccd804 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -76,7 +76,7 @@ select = ["E", "F", "I", "UP", "B", "ASYNC"] [tool.mypy] python_version = "3.10" strict = true -files = ["src/samsarix_core", "src/helix_core", "examples", "benchmarks"] +files = ["src/samsarix_core", "src/helix_core", "examples", "benchmarks", "scripts"] [tool.pytest.ini_options] addopts = "--strict-config --strict-markers --cov=samsarix_core --cov-branch --cov-report=term-missing --cov-fail-under=90" diff --git a/scripts/smoke_check.py b/scripts/smoke_check.py new file mode 100644 index 0000000..0d3aab2 --- /dev/null +++ b/scripts/smoke_check.py @@ -0,0 +1,70 @@ +# Copyright 2026 Samsarix LLC +# SPDX-License-Identifier: MPL-2.0 + +"""Verify the installed wheel's public canonical and compatibility exports.""" + +from __future__ import annotations + +import helix_core +import samsarix_core + + +def _require(condition: bool, message: str) -> None: + """Fail the smoke check even when Python assertions are optimized away.""" + + if not condition: + raise RuntimeError(message) + + +def main() -> None: + """Assert the release-critical import and model surface.""" + + canonical_exports = ( + samsarix_core.MCPServer, + samsarix_core.ToolPolicyContext, + samsarix_core.ToolPolicyDecision, + samsarix_core.ToolRuntime, + samsarix_core.serve_stdio, + ) + legacy_exports = ( + helix_core.MCPServer, + helix_core.ToolPolicyContext, + helix_core.ToolPolicyDecision, + helix_core.ToolRuntime, + helix_core.serve_stdio, + ) + _require(legacy_exports == canonical_exports, "legacy callable exports differ") + _require( + helix_core.ToolRateLimit is samsarix_core.ToolRateLimit, + "legacy rate-limit export differs", + ) + _require( + helix_core.ToolCircuitBreaker is samsarix_core.ToolCircuitBreaker, + "legacy circuit-policy export differs", + ) + _require( + helix_core.ToolCircuitState is samsarix_core.ToolCircuitState, + "legacy circuit-state export differs", + ) + _require( + helix_core.__version__ == samsarix_core.__version__, + "legacy package version differs", + ) + _require( + samsarix_core.ToolRateLimit(calls=1, period_seconds=1).burst_capacity == 1, + "rate-limit model behavior differs", + ) + circuit = samsarix_core.ToolCircuitBreaker( + failure_threshold=1, + recovery_timeout_seconds=1, + ) + _require(circuit.recovery_timeout_seconds == 1.0, "circuit model behavior differs") + _require( + samsarix_core.ToolCircuitState.CLOSED.value == "closed", + "circuit state value differs", + ) + print(samsarix_core.__version__) + + +if __name__ == "__main__": + main() diff --git a/src/helix_core/__init__.py b/src/helix_core/__init__.py index 09c41c8..b641c9a 100644 --- a/src/helix_core/__init__.py +++ b/src/helix_core/__init__.py @@ -16,6 +16,8 @@ SamsarixError, TaskSupport, ToolCall, + ToolCircuitBreaker, + ToolCircuitState, ToolDefinitionError, ToolError, ToolLifecycleEvent, @@ -49,6 +51,8 @@ "SamsarixError", "TaskSupport", "ToolCall", + "ToolCircuitBreaker", + "ToolCircuitState", "ToolDefinitionError", "ToolError", "ToolLifecycleEvent", diff --git a/src/samsarix_core/__init__.py b/src/samsarix_core/__init__.py index 42c1e3a..f47cc8f 100644 --- a/src/samsarix_core/__init__.py +++ b/src/samsarix_core/__init__.py @@ -19,6 +19,8 @@ RuntimeMetrics, TaskSupport, ToolCall, + ToolCircuitBreaker, + ToolCircuitState, ToolError, ToolLifecycleEvent, ToolLifecycleHandler, @@ -46,6 +48,8 @@ "SamsarixError", "TaskSupport", "ToolCall", + "ToolCircuitBreaker", + "ToolCircuitState", "ToolDefinitionError", "ToolError", "ToolLifecycleEvent", diff --git a/src/samsarix_core/models.py b/src/samsarix_core/models.py index a59b14f..1df0608 100644 --- a/src/samsarix_core/models.py +++ b/src/samsarix_core/models.py @@ -27,6 +27,7 @@ class ToolStatus(str, Enum): DENIED = "denied" BUSY = "busy" RATE_LIMITED = "rate_limited" + CIRCUIT_OPEN = "circuit_open" TIMED_OUT = "timed_out" FAILED = "failed" RUNTIME_CLOSED = "runtime_closed" @@ -42,6 +43,7 @@ class ToolLifecycleStatus(str, Enum): DENIED = "denied" BUSY = "busy" RATE_LIMITED = "rate_limited" + CIRCUIT_OPEN = "circuit_open" TIMED_OUT = "timed_out" FAILED = "failed" RUNTIME_CLOSED = "runtime_closed" @@ -49,6 +51,14 @@ class ToolLifecycleStatus(str, Enum): ABORTED = "aborted" +class ToolCircuitState(str, Enum): + """Observable state of one configured process-local circuit breaker.""" + + CLOSED = "closed" + OPEN = "open" + HALF_OPEN = "half_open" + + @dataclass(frozen=True, slots=True) class ToolError: """A safe error suitable for serialization across an application boundary.""" @@ -181,6 +191,46 @@ def to_dict(self) -> dict[str, JSONValue]: } +@dataclass(frozen=True, slots=True, kw_only=True) +class ToolCircuitBreaker: + """Consecutive-failure circuit policy for one exact tool registration.""" + + failure_threshold: int + recovery_timeout_seconds: float + + def __post_init__(self) -> None: + if isinstance(self.failure_threshold, bool) or not isinstance(self.failure_threshold, int): + raise TypeError("failure_threshold must be an integer") + if self.failure_threshold <= 0: + raise ValueError("failure_threshold must be positive") + if self.failure_threshold > _MAX_EXACT_TOKEN_COUNT: + raise ValueError("failure_threshold exceeds the exact count limit") + if isinstance(self.recovery_timeout_seconds, bool) or not isinstance( + self.recovery_timeout_seconds, (int, float) + ): + raise TypeError("recovery_timeout_seconds must be a number") + try: + recovery_timeout_seconds = float(self.recovery_timeout_seconds) + except OverflowError as exc: + raise ValueError("recovery_timeout_seconds must be finite and positive") from exc + retry_after_ms = recovery_timeout_seconds * 1_000 + if ( + recovery_timeout_seconds <= 0 + or not isfinite(recovery_timeout_seconds) + or not isfinite(retry_after_ms) + ): + raise ValueError("recovery_timeout_seconds must be finite and positive") + object.__setattr__(self, "recovery_timeout_seconds", recovery_timeout_seconds) + + def to_dict(self) -> dict[str, JSONValue]: + """Return the normalized deployment-local configuration.""" + + return { + "failure_threshold": self.failure_threshold, + "recovery_timeout_seconds": self.recovery_timeout_seconds, + } + + class ToolPolicyDecision(str, Enum): """A host policy's explicit decision for one validated invocation.""" @@ -277,6 +327,8 @@ class RuntimeMetrics: peak_pending_invocations: int = 0 lifecycle_handler_failures: int = 0 rate_limited: int = 0 + circuit_open: int = 0 + circuit_breaker_trips: int = 0 def to_dict(self) -> dict[str, int]: """Return the counters as a plain mapping.""" @@ -289,6 +341,7 @@ def to_dict(self) -> dict[str, int]: "denied": self.denied, "busy": self.busy, "rate_limited": self.rate_limited, + "circuit_open": self.circuit_open, "timed_out": self.timed_out, "failed": self.failed, "runtime_closed": self.runtime_closed, @@ -298,4 +351,5 @@ def to_dict(self) -> dict[str, int]: "in_flight": self.in_flight, "peak_in_flight": self.peak_in_flight, "lifecycle_handler_failures": self.lifecycle_handler_failures, + "circuit_breaker_trips": self.circuit_breaker_trips, } diff --git a/src/samsarix_core/runtime.py b/src/samsarix_core/runtime.py index b5a4258..aefa4cc 100644 --- a/src/samsarix_core/runtime.py +++ b/src/samsarix_core/runtime.py @@ -12,7 +12,7 @@ from concurrent.futures import Future, ThreadPoolExecutor from contextlib import suppress from copy import deepcopy -from dataclasses import dataclass +from dataclasses import dataclass, field from datetime import datetime, timezone from functools import partial from math import ceil @@ -25,6 +25,8 @@ JSONValue, RuntimeMetrics, ToolCall, + ToolCircuitBreaker, + ToolCircuitState, ToolError, ToolLifecycleEvent, ToolLifecycleHandler, @@ -64,6 +66,14 @@ def __init__(self, retry_after_ms: int) -> None: self.retry_after_ms = retry_after_ms +class _ToolCircuitOpen(Exception): + """Signal that one tool registration is not admitting dependency calls.""" + + def __init__(self, retry_after_ms: int | None) -> None: + super().__init__(retry_after_ms) + self.retry_after_ms = retry_after_ms + + @dataclass(slots=True) class _TokenBucket: """Mutable event-loop-local token bucket for one exact registration.""" @@ -100,6 +110,173 @@ def try_acquire(self, *, now: float) -> int | None: return max(1, ceil((1 - self.tokens) * self.milliseconds_per_token)) +@dataclass(frozen=True, slots=True) +class _CircuitPermit: + """Generation-bound permission to attempt one protected execution.""" + + generation: int + probe: bool + + +class _CircuitBreaker: + """Thread-safe consecutive-failure breaker for one exact registration.""" + + def __init__(self, policy: ToolCircuitBreaker, *, now: float) -> None: + self.failure_threshold = policy.failure_threshold + self.recovery_timeout_seconds = policy.recovery_timeout_seconds + self._state = ToolCircuitState.CLOSED + self._consecutive_failures = 0 + self._opened_until = 0.0 + self._generation = 0 + self._last_observed = now + self._lock = Lock() + + def acquire(self, *, now: float) -> _CircuitPermit: + """Return one permit or raise a retryable internal open-circuit signal.""" + + with self._lock: + current = self._observe(now) + if self._state is ToolCircuitState.CLOSED: + return _CircuitPermit(self._generation, probe=False) + if self._state is ToolCircuitState.OPEN: + if current < self._opened_until: + raise _ToolCircuitOpen(self._retry_after_ms(current)) + self._state = ToolCircuitState.HALF_OPEN + self._generation += 1 + return _CircuitPermit(self._generation, probe=True) + raise _ToolCircuitOpen(None) + + def validate(self, permit: _CircuitPermit, *, now: float) -> None: + """Reject a queued permit when another invocation changed the circuit.""" + + with self._lock: + current = self._observe(now) + if self._permit_is_current(permit): + return + retry_after_ms = ( + self._retry_after_ms(current) + if self._state is ToolCircuitState.OPEN and current < self._opened_until + else None + ) + raise _ToolCircuitOpen(retry_after_ms) + + def succeed(self, permit: _CircuitPermit, *, now: float) -> None: + """Record a successful execution if its permit is still current.""" + + with self._lock: + self._observe(now) + if not self._permit_is_current(permit): + return + if permit.probe: + self._state = ToolCircuitState.CLOSED + self._generation += 1 + self._consecutive_failures = 0 + + def fail(self, permit: _CircuitPermit, *, now: float) -> bool: + """Record one execution failure and report whether the circuit opened.""" + + with self._lock: + current = self._observe(now) + if not self._permit_is_current(permit): + return False + if permit.probe: + self._open(current) + return True + self._consecutive_failures += 1 + if self._consecutive_failures < self.failure_threshold: + return False + self._open(current) + return True + + def abandon(self, permit: _CircuitPermit, *, now: float) -> None: + """Release a non-executed half-open probe without counting a failure.""" + + with self._lock: + current = self._observe(now) + if self._permit_is_current(permit) and permit.probe: + self._state = ToolCircuitState.OPEN + self._opened_until = current + self._generation += 1 + + def reset(self, *, now: float) -> None: + """Force the circuit closed and invalidate outstanding permits.""" + + with self._lock: + self._observe(now) + self._state = ToolCircuitState.CLOSED + self._consecutive_failures = 0 + self._opened_until = 0.0 + self._generation += 1 + + @property + def state(self) -> ToolCircuitState: + """Return the current explicit state without reserving a probe.""" + + with self._lock: + return self._state + + def _observe(self, now: float) -> float: + current = max(now, self._last_observed) + self._last_observed = current + return current + + def _permit_is_current(self, permit: _CircuitPermit) -> bool: + expected_state = ToolCircuitState.HALF_OPEN if permit.probe else ToolCircuitState.CLOSED + return permit.generation == self._generation and self._state is expected_state + + def _open(self, now: float) -> None: + self._state = ToolCircuitState.OPEN + self._opened_until = now + self.recovery_timeout_seconds + self._generation += 1 + + def _retry_after_ms(self, now: float) -> int: + return max(1, ceil((self._opened_until - now) * 1_000)) + + +@dataclass(slots=True) +class _CircuitAttempt: + """Invocation-local holder that records exactly one breaker outcome.""" + + breaker: _CircuitBreaker | None = None + permit: _CircuitPermit | None = None + _lock: Lock = field(default_factory=Lock, repr=False) + + def acquire(self, breaker: _CircuitBreaker | None, *, now: float) -> None: + if breaker is None: + return + permit = breaker.acquire(now=now) + with self._lock: + self.breaker = breaker + self.permit = permit + + def validate(self, *, now: float) -> None: + with self._lock: + breaker, permit = self.breaker, self.permit + if breaker is not None and permit is not None: + breaker.validate(permit, now=now) + + def succeed(self, *, now: float) -> None: + breaker, permit = self._take() + if breaker is not None and permit is not None: + breaker.succeed(permit, now=now) + + def fail(self, *, now: float) -> bool: + breaker, permit = self._take() + return breaker is not None and permit is not None and breaker.fail(permit, now=now) + + def abandon(self, *, now: float) -> None: + breaker, permit = self._take() + if breaker is not None and permit is not None: + breaker.abandon(permit, now=now) + + def _take(self) -> tuple[_CircuitBreaker | None, _CircuitPermit | None]: + with self._lock: + breaker, permit = self.breaker, self.permit + self.breaker = None + self.permit = None + return breaker, permit + + def _is_async_callable(value: object) -> bool: """Recognize async functions and objects with an async ``__call__``.""" @@ -178,7 +355,9 @@ def __init__( self._policy_semaphore = asyncio.Semaphore(max_concurrency) self._tool_semaphores: dict[int, tuple[RegisteredTool, asyncio.Semaphore]] = {} self._tool_rate_limiters: dict[int, tuple[RegisteredTool, _TokenBucket]] = {} + self._tool_circuit_breakers: dict[int, tuple[RegisteredTool, _CircuitBreaker]] = {} self._rate_limit_clock: Callable[[], float] = time.monotonic + self._circuit_clock: Callable[[], float] = time.monotonic self._executor = ThreadPoolExecutor( max_workers=max_concurrency, thread_name_prefix="samsarix-tool" ) @@ -196,6 +375,8 @@ def __init__( "denied": 0, "busy": 0, "rate_limited": 0, + "circuit_open": 0, + "circuit_breaker_trips": 0, "timed_out": 0, "failed": 0, "runtime_closed": 0, @@ -214,8 +395,9 @@ def register( replace: bool = False, max_concurrency: int | None = None, rate_limit: ToolRateLimit | None = None, + circuit_breaker: ToolCircuitBreaker | None = None, ) -> ToolSpec: - """Register a callable with optional concurrency and sustained-rate controls.""" + """Register a callable with optional deployment-local resilience controls.""" if max_concurrency is not None: if isinstance(max_concurrency, bool) or not isinstance(max_concurrency, int): @@ -224,6 +406,8 @@ def register( raise ValueError("max_concurrency must be positive") if rate_limit is not None and not isinstance(rate_limit, ToolRateLimit): raise TypeError("rate_limit must be a ToolRateLimit or None") + if circuit_breaker is not None and not isinstance(circuit_breaker, ToolCircuitBreaker): + raise TypeError("circuit_breaker must be a ToolCircuitBreaker or None") # Keep the callable registration and its deployment-local policy atomic # with respect to both direct registry mutation and invocation resolution. @@ -252,6 +436,17 @@ def register( _TokenBucket.from_limit(rate_limit, now=self._rate_limit_clock()), ) self._tool_rate_limiters = tool_rate_limiters + tool_circuit_breakers = { + key: entry + for key, entry in self._tool_circuit_breakers.items() + if entry[0].spec.name != spec.name + } + if circuit_breaker is not None: + tool_circuit_breakers[id(registered)] = ( + registered, + _CircuitBreaker(circuit_breaker, now=self._circuit_clock()), + ) + self._tool_circuit_breakers = tool_circuit_breakers return spec async def invoke( @@ -389,6 +584,12 @@ async def _invoke_admitted( if tool_rate_limit is not None and tool_rate_limit[0] is registered else None ) + tool_circuit = self._tool_circuit_breakers.get(id(registered)) + circuit_breaker = ( + tool_circuit[1] + if tool_circuit is not None and tool_circuit[0] is registered + else None + ) except ToolNotFoundError: self._increment("not_found") return self._result( @@ -431,6 +632,7 @@ async def _invoke_admitted( max_updates=self.max_progress_updates, max_message_bytes=self.max_progress_message_bytes, ) + circuit_attempt = _CircuitAttempt() execution = asyncio.create_task( self._authorize_and_execute( registered, @@ -438,6 +640,8 @@ async def _invoke_admitted( invocation_id=invocation_id, tool_semaphore=tool_semaphore, rate_limiter=rate_limiter, + circuit_breaker=circuit_breaker, + circuit_attempt=circuit_attempt, ) ) self._active.add(execution) @@ -449,6 +653,7 @@ async def _invoke_admitted( ) except asyncio.TimeoutError: _stop_progress(progress_scope) + self._record_circuit_failure(circuit_attempt) execution.cancel() with suppress(asyncio.CancelledError): await execution @@ -469,9 +674,11 @@ async def _invoke_admitted( execution.cancel() with suppress(asyncio.CancelledError): await execution + circuit_attempt.abandon(now=self._circuit_clock()) self._increment("cancelled") raise except _ToolPolicyDenied: + circuit_attempt.abandon(now=self._circuit_clock()) self._increment("denied") return self._result( invocation_id, @@ -485,6 +692,7 @@ async def _invoke_admitted( ), ) except _ToolPolicyFailed: + circuit_attempt.abandon(now=self._circuit_clock()) self._increment("failed") return self._result( invocation_id, @@ -497,7 +705,29 @@ async def _invoke_admitted( "Tool invocation policy failed", ), ) + except _ToolCircuitOpen as exc: + circuit_attempt.abandon(now=self._circuit_clock()) + self._increment("circuit_open") + details = ( + None + if exc.retry_after_ms is None + else {"retry_after_ms": cast(JSONValue, exc.retry_after_ms)} + ) + return self._result( + invocation_id, + name, + ToolStatus.CIRCUIT_OPEN, + started_at, + started, + error=ToolError( + "tool_circuit_open", + "Tool dependency circuit is temporarily open", + retryable=True, + details=details, + ), + ) except _ToolRateLimited as exc: + circuit_attempt.abandon(now=self._circuit_clock()) self._increment("rate_limited") return self._result( invocation_id, @@ -513,6 +743,7 @@ async def _invoke_admitted( ), ) except ToolOutputError as exc: + self._record_circuit_failure(circuit_attempt) self._increment("failed") return self._result( invocation_id, @@ -527,8 +758,10 @@ async def _invoke_admitted( ), ) except ProgressHandlerError: + circuit_attempt.abandon(now=self._circuit_clock()) raise except Exception as exc: + self._record_circuit_failure(circuit_attempt) self._increment("failed") message = str(exc) if self.expose_exceptions else "Tool execution failed" return self._result( @@ -540,6 +773,7 @@ async def _invoke_admitted( error=ToolError("tool_failed", message, type=type(exc).__name__), ) else: + circuit_attempt.succeed(now=self._circuit_clock()) self._increment("succeeded") return self._result( invocation_id, @@ -584,6 +818,27 @@ def metrics(self) -> RuntimeMetrics: with self._metrics_lock: return RuntimeMetrics(**self._counters) + def circuit_state(self, name: str) -> ToolCircuitState | None: + """Return one configured breaker's state, or ``None`` when it is disabled.""" + + with self.registry._lock: + registered = self.registry._resolve(name) + entry = self._tool_circuit_breakers.get(id(registered)) + breaker = entry[1] if entry is not None and entry[0] is registered else None + return None if breaker is None else breaker.state + + def reset_circuit(self, name: str) -> bool: + """Force one configured breaker closed and report whether it existed.""" + + with self.registry._lock: + registered = self.registry._resolve(name) + entry = self._tool_circuit_breakers.get(id(registered)) + breaker = entry[1] if entry is not None and entry[0] is registered else None + if breaker is None: + return False + breaker.reset(now=self._circuit_clock()) + return True + @property def pending_sync_calls(self) -> int: """Return the number of submitted sync calls that have not actually stopped.""" @@ -650,18 +905,32 @@ async def _execute( arguments: dict[str, Any], tool_semaphore: asyncio.Semaphore | None, rate_limiter: _TokenBucket | None, + circuit_breaker: _CircuitBreaker | None, + circuit_attempt: _CircuitAttempt, ) -> JSONValue: + circuit_attempt.acquire(circuit_breaker, now=self._circuit_clock()) if registered.spec.is_async: if tool_semaphore is not None: await tool_semaphore.acquire() try: async with self._semaphore: + circuit_attempt.validate(now=self._circuit_clock()) self._require_rate_token(rate_limiter) self._begin_execution() try: - awaitable = cast(Awaitable[Any], registered.function(**arguments)) - raw_output = await awaitable - return self._normalize_output(registered, raw_output) + try: + awaitable = cast(Awaitable[Any], registered.function(**arguments)) + raw_output = await awaitable + output = self._normalize_output(registered, raw_output) + except ProgressHandlerError: + circuit_attempt.abandon(now=self._circuit_clock()) + raise + except Exception: + self._record_circuit_failure(circuit_attempt) + raise + else: + circuit_attempt.succeed(now=self._circuit_clock()) + return output finally: self._end_execution() finally: @@ -677,6 +946,7 @@ async def _execute( tool_semaphore.release() raise try: + circuit_attempt.validate(now=self._circuit_clock()) self._require_rate_token(rate_limiter) except BaseException: self._semaphore.release() @@ -685,7 +955,7 @@ async def _execute( raise loop = asyncio.get_running_loop() try: - sync_future = self._executor.submit(self._run_sync, registered.function, arguments) + sync_future = self._executor.submit(self._run_sync, registered, arguments) except BaseException: self._semaphore.release() if tool_semaphore is not None: @@ -693,6 +963,9 @@ async def _execute( raise with self._sync_futures_lock: self._sync_futures.add(sync_future) + sync_future.add_done_callback( + partial(self._sync_circuit_finished, circuit_attempt=circuit_attempt) + ) sync_future.add_done_callback( partial( self._sync_finished, @@ -700,8 +973,7 @@ async def _execute( tool_semaphore=tool_semaphore, ) ) - raw_output = await self._wrap_sync_future(sync_future) - return self._normalize_output(registered, raw_output) + return cast(JSONValue, await self._wrap_sync_future(sync_future)) async def _authorize_and_execute( self, @@ -711,6 +983,8 @@ async def _authorize_and_execute( invocation_id: str, tool_semaphore: asyncio.Semaphore | None, rate_limiter: _TokenBucket | None, + circuit_breaker: _CircuitBreaker | None, + circuit_attempt: _CircuitAttempt, ) -> JSONValue: """Fail closed on one bounded policy decision before tool execution.""" @@ -731,7 +1005,20 @@ async def _authorize_and_execute( raise _ToolPolicyFailed if decision is ToolPolicyDecision.DENY: raise _ToolPolicyDenied - return await self._execute(registered, arguments, tool_semaphore, rate_limiter) + return await self._execute( + registered, + arguments, + tool_semaphore, + rate_limiter, + circuit_breaker, + circuit_attempt, + ) + + def _record_circuit_failure(self, circuit_attempt: _CircuitAttempt) -> None: + """Record one protected execution failure and any resulting trip.""" + + if circuit_attempt.fail(now=self._circuit_clock()): + self._increment("circuit_breaker_trips") def _require_rate_token(self, rate_limiter: _TokenBucket | None) -> None: """Consume one start token or raise a safe internal throttling signal.""" @@ -742,15 +1029,35 @@ def _require_rate_token(self, rate_limiter: _TokenBucket | None) -> None: if retry_after_ms is not None: raise _ToolRateLimited(retry_after_ms) - def _run_sync(self, function: Callable[..., Any], arguments: dict[str, Any]) -> Any: - """Run one sync callable while tracking its real thread lifetime.""" + def _run_sync(self, registered: RegisteredTool, arguments: dict[str, Any]) -> JSONValue: + """Run and normalize one sync callable while tracking its real lifetime.""" self._begin_execution() try: - return function(**arguments) + raw_output = registered.function(**arguments) + return self._normalize_output(registered, raw_output) finally: self._end_execution() + def _sync_circuit_finished( + self, + future: Future[JSONValue], + *, + circuit_attempt: _CircuitAttempt, + ) -> None: + """Record the sync outcome before releasing capacity to queued calls.""" + + try: + future.result() + except ProgressHandlerError: + circuit_attempt.abandon(now=self._circuit_clock()) + except Exception: + self._record_circuit_failure(circuit_attempt) + except BaseException: + circuit_attempt.abandon(now=self._circuit_clock()) + else: + circuit_attempt.succeed(now=self._circuit_clock()) + def _sync_finished( self, future: Future[Any], diff --git a/tests/test_definitions.py b/tests/test_definitions.py index 39e6702..114cb17 100644 --- a/tests/test_definitions.py +++ b/tests/test_definitions.py @@ -186,6 +186,8 @@ def test_samsarix_names_are_canonical_and_legacy_names_remain_compatible() -> No assert helix_core.ToolRuntime is samsarix_core.ToolRuntime assert helix_core.ToolLifecycleEvent is samsarix_core.ToolLifecycleEvent assert helix_core.ToolLifecycleStatus is samsarix_core.ToolLifecycleStatus + assert helix_core.ToolCircuitBreaker is samsarix_core.ToolCircuitBreaker + assert helix_core.ToolCircuitState is samsarix_core.ToolCircuitState assert helix_core.ToolRateLimit is samsarix_core.ToolRateLimit assert helix_core.__version__ == samsarix_core.__version__ diff --git a/tests/test_mcp.py b/tests/test_mcp.py index f4b137e..729b6ef 100644 --- a/tests/test_mcp.py +++ b/tests/test_mcp.py @@ -14,6 +14,7 @@ from samsarix_core import ( MCPServer, ProgressHandlerError, + ToolCircuitBreaker, ToolLifecycleEvent, ToolLifecycleStatus, ToolPolicyContext, @@ -592,6 +593,68 @@ async def quota_target(value: str) -> str: assert "private-rate-limit-value" not in json.dumps(limited) +@pytest.mark.asyncio +async def test_mcp_exposes_safe_retryable_open_circuit() -> None: + executions: list[str] = [] + + @samsarix_tool + async def unstable_dependency(value: str) -> str: + """Represent one failing call to a circuit-protected dependency.""" + + executions.append(value) + raise RuntimeError("private dependency failure") + + runtime = ToolRuntime() + runtime._circuit_clock = lambda: 40.0 + runtime.register( + unstable_dependency, + circuit_breaker=ToolCircuitBreaker( + failure_threshold=1, + recovery_timeout_seconds=30, + ), + ) + server = MCPServer(runtime) + try: + await initialize(server) + failed = await server.handle( + { + "jsonrpc": "2.0", + "id": "failing-circuit-call", + "method": "tools/call", + "params": {"name": "unstable_dependency", "arguments": {"value": "first"}}, + } + ) + blocked = await server.handle( + { + "jsonrpc": "2.0", + "id": "blocked-circuit-call", + "method": "tools/call", + "params": { + "name": "unstable_dependency", + "arguments": {"value": "private-open-circuit-value"}, + }, + } + ) + finally: + await server.aclose() + + assert failed is not None and failed["result"]["isError"] is True + assert blocked is not None and blocked["result"]["isError"] is True + assert blocked["result"]["_meta"]["com.samsarix/status"] == "circuit_open" + assert json.loads(blocked["result"]["content"][0]["text"]) == { + "error": { + "code": "tool_circuit_open", + "message": "Tool dependency circuit is temporarily open", + "retryable": True, + "details": {"retry_after_ms": 30000}, + } + } + assert executions == ["first"] + assert runtime.metrics().circuit_breaker_trips == 1 + assert runtime.metrics().circuit_open == 1 + assert "private-open-circuit-value" not in json.dumps(blocked) + + @pytest.mark.asyncio async def test_mcp_calls_share_runtime_tool_bulkheads_without_cross_tool_starvation() -> None: started = asyncio.Event() diff --git a/tests/test_mcp_tasks.py b/tests/test_mcp_tasks.py index 1eb1662..d492713 100644 --- a/tests/test_mcp_tasks.py +++ b/tests/test_mcp_tasks.py @@ -11,6 +11,7 @@ from samsarix_core import ( MCPServer, + ToolCircuitBreaker, ToolPolicyContext, ToolPolicyDecision, ToolRateLimit, @@ -621,6 +622,85 @@ async def quota_job(value: str) -> str: ) +@pytest.mark.asyncio +async def test_task_open_circuit_retains_retryable_safe_terminal_result() -> None: + executions: list[str] = [] + + @samsarix_tool(task_support="optional") + async def unstable_job(value: str) -> str: + """Represent a task-wrapped call to a circuit-protected dependency.""" + + executions.append(value) + raise RuntimeError("private task dependency failure") + + runtime = ToolRuntime() + runtime._circuit_clock = lambda: 20.0 + runtime.register( + unstable_job, + circuit_breaker=ToolCircuitBreaker( + failure_threshold=1, + recovery_timeout_seconds=25, + ), + ) + server = MCPServer(runtime, enable_tasks=True) + try: + await initialize(server) + failed_create = await request( + server, + "create-failing-circuit", + "tools/call", + {"name": "unstable_job", "arguments": {"value": "first"}, "task": {}}, + ) + failed_result = await request( + server, + "result-failing-circuit", + "tasks/result", + {"taskId": task_id(failed_create)}, + ) + blocked_create = await request( + server, + "create-blocked-circuit", + "tools/call", + { + "name": "unstable_job", + "arguments": {"value": "private-task-circuit-value"}, + "task": {}, + }, + ) + identifier = task_id(blocked_create) + blocked_result = await request( + server, + "result-blocked-circuit", + "tasks/result", + {"taskId": identifier}, + ) + blocked_state = await request( + server, + "state-blocked-circuit", + "tasks/get", + {"taskId": identifier}, + ) + finally: + await server.aclose() + + assert failed_result["result"]["isError"] is True + assert blocked_state["result"]["status"] == "failed" + assert blocked_result["result"]["isError"] is True + assert blocked_result["result"]["_meta"]["com.samsarix/status"] == "circuit_open" + assert json.loads(blocked_result["result"]["content"][0]["text"])["error"] == { + "code": "tool_circuit_open", + "message": "Tool dependency circuit is temporarily open", + "retryable": True, + "details": {"retry_after_ms": 25000}, + } + assert executions == ["first"] + assert runtime.metrics().circuit_breaker_trips == 1 + assert runtime.metrics().circuit_open == 1 + assert "private-task-circuit-value" not in json.dumps( + [blocked_create, blocked_state, blocked_result] + ) + + def test_task_configuration_and_metadata_validation() -> None: with pytest.raises(ValueError, match="task_support"): diff --git a/tests/test_runtime.py b/tests/test_runtime.py index db3574f..947cba4 100644 --- a/tests/test_runtime.py +++ b/tests/test_runtime.py @@ -16,8 +16,11 @@ from samsarix_core import ( ProgressHandlerError, ToolCall, + ToolCircuitBreaker, + ToolCircuitState, ToolLifecycleEvent, ToolLifecycleStatus, + ToolNotFoundError, ToolPolicyContext, ToolPolicyDecision, ToolProgress, @@ -746,6 +749,19 @@ def test_per_tool_rate_limit_type_is_validated_before_registration() -> None: asyncio.run(runtime.aclose()) +def test_per_tool_circuit_breaker_type_is_validated_before_registration() -> None: + runtime = ToolRuntime() + try: + with pytest.raises(TypeError, match="ToolCircuitBreaker"): + runtime.register( + add, + circuit_breaker={"failure_threshold": 1}, # type: ignore[arg-type] + ) + assert "add" not in runtime.registry + finally: + asyncio.run(runtime.aclose()) + + @pytest.mark.asyncio async def test_replacing_a_tool_does_not_inherit_its_previous_bulkhead() -> None: active = 0 @@ -825,6 +841,10 @@ async def concurrent_tool() -> int: function, max_concurrency=1, rate_limit=ToolRateLimit(calls=2, period_seconds=60, burst=2), + circuit_breaker=ToolCircuitBreaker( + failure_threshold=1, + recovery_timeout_seconds=60, + ), ) for function in functions ] @@ -840,6 +860,9 @@ async def concurrent_tool() -> int: assert [result.status for result in results].count(ToolStatus.SUCCESS) == tool_count * 2 assert peaks == [1] * tool_count + assert {runtime.circuit_state(f"concurrent_{index}") for index in range(tool_count)} == { + ToolCircuitState.CLOSED + } @pytest.mark.asyncio @@ -1116,6 +1139,528 @@ def sync_quota_target(value: str) -> str: assert "private-second" not in str(limited.to_dict()) +@pytest.mark.asyncio +async def test_per_tool_circuit_breaker_opens_recovers_and_reports_safe_state() -> None: + now = 100.0 + executions: list[str] = [] + events: list[ToolLifecycleEvent] = [] + + @helix_tool + async def fragile_dependency(value: str, fail: bool) -> str: + """Represent a dependency that can fail until it recovers.""" + + executions.append(value) + if fail: + raise RuntimeError(value) + return value + + runtime = ToolRuntime(lifecycle_handler=events.append) + runtime._circuit_clock = lambda: now + runtime.register( + fragile_dependency, + circuit_breaker=ToolCircuitBreaker( + failure_threshold=2, + recovery_timeout_seconds=10, + ), + ) + try: + assert runtime.circuit_state("fragile_dependency") is ToolCircuitState.CLOSED + first = await runtime.invoke("fragile_dependency", {"value": "private-first", "fail": True}) + assert runtime.circuit_state("fragile_dependency") is ToolCircuitState.CLOSED + second = await runtime.invoke( + "fragile_dependency", {"value": "private-second", "fail": True} + ) + assert runtime.circuit_state("fragile_dependency") is ToolCircuitState.OPEN + blocked = await runtime.invoke( + "fragile_dependency", {"value": "private-blocked", "fail": False} + ) + now += 10 + probe = await runtime.invoke( + "fragile_dependency", {"value": "recovered-probe", "fail": False} + ) + assert runtime.circuit_state("fragile_dependency") is ToolCircuitState.CLOSED + healthy = await runtime.invoke("fragile_dependency", {"value": "healthy", "fail": False}) + metrics = runtime.metrics() + finally: + await runtime.aclose() + + assert first.status is ToolStatus.FAILED + assert second.status is ToolStatus.FAILED + assert blocked.status is ToolStatus.CIRCUIT_OPEN + assert blocked.error is not None + assert blocked.error.to_dict() == { + "code": "tool_circuit_open", + "message": "Tool dependency circuit is temporarily open", + "retryable": True, + "details": {"retry_after_ms": 10000}, + } + assert probe.success and healthy.success + assert executions == ["private-first", "private-second", "recovered-probe", "healthy"] + assert "private-blocked" not in str(blocked.to_dict()) + assert metrics.calls_total == 5 + assert metrics.succeeded == 2 + assert metrics.failed == 2 + assert metrics.circuit_open == 1 + assert metrics.circuit_breaker_trips == 1 + assert metrics.to_dict()["circuit_open"] == 1 + assert metrics.to_dict()["circuit_breaker_trips"] == 1 + assert [event.status for event in events] == [ + ToolLifecycleStatus.STARTED, + ToolLifecycleStatus.FAILED, + ToolLifecycleStatus.STARTED, + ToolLifecycleStatus.FAILED, + ToolLifecycleStatus.STARTED, + ToolLifecycleStatus.CIRCUIT_OPEN, + ToolLifecycleStatus.STARTED, + ToolLifecycleStatus.SUCCESS, + ToolLifecycleStatus.STARTED, + ToolLifecycleStatus.SUCCESS, + ] + assert "private-blocked" not in str([event.to_dict() for event in events]) + + +@pytest.mark.asyncio +async def test_circuit_breaker_counts_only_post_policy_execution_failures() -> None: + executions: list[str] = [] + + @helix_tool + async def guarded_dependency(label: str) -> str: + """Succeed or fail after validation and policy admission.""" + + executions.append(label) + if label == "fail": + raise RuntimeError("dependency unavailable") + return label + + async def policy(context: ToolPolicyContext) -> ToolPolicyDecision: + return ( + ToolPolicyDecision.DENY + if context.arguments["label"] == "deny" + else ToolPolicyDecision.ALLOW + ) + + runtime = ToolRuntime(policy=policy) + runtime.register( + guarded_dependency, + circuit_breaker=ToolCircuitBreaker( + failure_threshold=2, + recovery_timeout_seconds=60, + ), + ) + try: + invalid = await runtime.invoke("guarded_dependency", {"label": 7}) + denied = await runtime.invoke("guarded_dependency", {"label": "deny"}) + first_failure = await runtime.invoke("guarded_dependency", {"label": "fail"}) + success = await runtime.invoke("guarded_dependency", {"label": "success"}) + second_failure = await runtime.invoke("guarded_dependency", {"label": "fail"}) + assert runtime.circuit_state("guarded_dependency") is ToolCircuitState.CLOSED + third_failure = await runtime.invoke("guarded_dependency", {"label": "fail"}) + metrics = runtime.metrics() + finally: + await runtime.aclose() + + assert invalid.status is ToolStatus.INVALID_ARGUMENTS + assert denied.status is ToolStatus.DENIED + assert first_failure.status is ToolStatus.FAILED + assert success.status is ToolStatus.SUCCESS + assert second_failure.status is ToolStatus.FAILED + assert third_failure.status is ToolStatus.FAILED + assert runtime.circuit_state("guarded_dependency") is ToolCircuitState.OPEN + assert executions == ["fail", "success", "fail", "fail"] + assert metrics.circuit_breaker_trips == 1 + assert metrics.circuit_open == 0 + + +@pytest.mark.asyncio +async def test_sync_output_validation_failure_trips_before_queued_execution() -> None: + executions = 0 + + @helix_tool + def malformed_dependency() -> int: + """Return a private malformed value from a protected sync dependency.""" + + nonlocal executions + executions += 1 + return "private-malformed-output" # type: ignore[return-value] + + runtime = ToolRuntime() + runtime.register( + malformed_dependency, + circuit_breaker=ToolCircuitBreaker( + failure_threshold=1, + recovery_timeout_seconds=60, + ), + ) + try: + failed = await runtime.invoke("malformed_dependency") + blocked = await runtime.invoke("malformed_dependency") + finally: + await runtime.aclose() + + assert failed.status is ToolStatus.FAILED + assert failed.error is not None + assert failed.error.code == "invalid_output" + assert blocked.status is ToolStatus.CIRCUIT_OPEN + assert executions == 1 + assert "private-malformed-output" not in str(failed.to_dict()) + assert runtime.metrics().circuit_breaker_trips == 1 + + +@pytest.mark.asyncio +async def test_half_open_rate_rejection_releases_probe_without_changing_trip_count() -> None: + circuit_now = 10.0 + rate_now = 20.0 + available = False + executions = 0 + + @helix_tool + async def circuit_and_rate_dependency() -> str: + """Exercise independent circuit and token-bucket admission.""" + + nonlocal executions + executions += 1 + if not available: + raise RuntimeError("dependency unavailable") + return "available" + + runtime = ToolRuntime() + runtime._circuit_clock = lambda: circuit_now + runtime._rate_limit_clock = lambda: rate_now + runtime.register( + circuit_and_rate_dependency, + rate_limit=ToolRateLimit(calls=1, period_seconds=60), + circuit_breaker=ToolCircuitBreaker( + failure_threshold=1, + recovery_timeout_seconds=5, + ), + ) + try: + failed = await runtime.invoke("circuit_and_rate_dependency") + circuit_now += 5 + throttled_probe = await runtime.invoke("circuit_and_rate_dependency") + assert runtime.circuit_state("circuit_and_rate_dependency") is ToolCircuitState.OPEN + rate_now += 60 + available = True + recovered = await runtime.invoke("circuit_and_rate_dependency") + finally: + await runtime.aclose() + + assert failed.status is ToolStatus.FAILED + assert throttled_probe.status is ToolStatus.RATE_LIMITED + assert recovered.success + assert executions == 2 + assert runtime.metrics().rate_limited == 1 + assert runtime.metrics().circuit_breaker_trips == 1 + assert runtime.circuit_state("circuit_and_rate_dependency") is ToolCircuitState.CLOSED + + +@pytest.mark.asyncio +async def test_opening_circuit_invalidates_queued_closed_state_permits() -> None: + started = asyncio.Event() + release = asyncio.Event() + executions: list[str] = [] + + @helix_tool + async def serialized_dependency(label: str) -> str: + """Hold and fail the first call while another call waits for capacity.""" + + executions.append(label) + if label == "first": + started.set() + await release.wait() + raise RuntimeError("dependency unavailable") + return label + + runtime = ToolRuntime(max_concurrency=1) + runtime.register( + serialized_dependency, + circuit_breaker=ToolCircuitBreaker( + failure_threshold=1, + recovery_timeout_seconds=60, + ), + ) + first = asyncio.create_task(runtime.invoke("serialized_dependency", {"label": "first"})) + second: asyncio.Task[ToolResult] | None = None + try: + await asyncio.wait_for(started.wait(), timeout=1) + second = asyncio.create_task( + runtime.invoke("serialized_dependency", {"label": "private-queued"}) + ) + for _ in range(100): + if runtime.metrics().pending_invocations == 2: + break + await asyncio.sleep(0) + else: + pytest.fail("second circuit-protected invocation was not admitted") + await asyncio.sleep(0) + release.set() + failed, blocked = await asyncio.gather(first, second) + finally: + release.set() + first.cancel() + if second is not None: + second.cancel() + await asyncio.gather( + first, *(item for item in (second,) if item is not None), return_exceptions=True + ) + await runtime.aclose() + + assert failed.status is ToolStatus.FAILED + assert blocked.status is ToolStatus.CIRCUIT_OPEN + assert executions == ["first"] + assert "private-queued" not in str(blocked.to_dict()) + + +@pytest.mark.asyncio +async def test_circuit_breaker_allows_only_one_half_open_probe() -> None: + now = 20.0 + probe_started = asyncio.Event() + release_probe = asyncio.Event() + executions: list[str] = [] + + @helix_tool + async def recovering_dependency(label: str) -> str: + """Fail initially and hold one recovery probe.""" + + executions.append(label) + if label == "failure": + raise RuntimeError("dependency unavailable") + probe_started.set() + await release_probe.wait() + return label + + runtime = ToolRuntime(max_concurrency=2) + runtime._circuit_clock = lambda: now + runtime.register( + recovering_dependency, + circuit_breaker=ToolCircuitBreaker( + failure_threshold=1, + recovery_timeout_seconds=5, + ), + ) + probe: asyncio.Task[ToolResult] | None = None + try: + failed = await runtime.invoke("recovering_dependency", {"label": "failure"}) + assert failed.status is ToolStatus.FAILED + now += 5 + probe = asyncio.create_task(runtime.invoke("recovering_dependency", {"label": "probe"})) + await asyncio.wait_for(probe_started.wait(), timeout=1) + assert runtime.circuit_state("recovering_dependency") is ToolCircuitState.HALF_OPEN + competing = await runtime.invoke("recovering_dependency", {"label": "private-competing"}) + release_probe.set() + recovered = await probe + finally: + release_probe.set() + if probe is not None: + probe.cancel() + await asyncio.gather(probe, return_exceptions=True) + await runtime.aclose() + + assert competing.status is ToolStatus.CIRCUIT_OPEN + assert competing.error is not None + assert competing.error.retryable is True + assert competing.error.details is None + assert "private-competing" not in str(competing.to_dict()) + assert recovered.success + assert executions == ["failure", "probe"] + assert runtime.circuit_state("recovering_dependency") is ToolCircuitState.CLOSED + + +@pytest.mark.asyncio +async def test_cancelled_half_open_probe_releases_recovery_slot_without_tripping() -> None: + now = 10.0 + probe_started = asyncio.Event() + hold_probe = asyncio.Event() + executions: list[str] = [] + + @helix_tool + async def cancellable_probe(label: str) -> str: + """Fail once, then expose one cancellable recovery probe.""" + + executions.append(label) + if label == "failure": + raise RuntimeError("dependency unavailable") + if label == "cancelled-probe": + probe_started.set() + await hold_probe.wait() + return label + + runtime = ToolRuntime() + runtime._circuit_clock = lambda: now + runtime.register( + cancellable_probe, + circuit_breaker=ToolCircuitBreaker( + failure_threshold=1, + recovery_timeout_seconds=5, + ), + ) + probe: asyncio.Task[ToolResult] | None = None + try: + failed = await runtime.invoke("cancellable_probe", {"label": "failure"}) + assert failed.status is ToolStatus.FAILED + now += 5 + probe = asyncio.create_task( + runtime.invoke("cancellable_probe", {"label": "cancelled-probe"}) + ) + await asyncio.wait_for(probe_started.wait(), timeout=1) + assert runtime.circuit_state("cancellable_probe") is ToolCircuitState.HALF_OPEN + probe.cancel() + with pytest.raises(asyncio.CancelledError): + await probe + assert runtime.circuit_state("cancellable_probe") is ToolCircuitState.OPEN + recovered = await runtime.invoke("cancellable_probe", {"label": "recovery"}) + finally: + hold_probe.set() + if probe is not None: + probe.cancel() + await asyncio.gather(probe, return_exceptions=True) + await runtime.aclose() + + assert recovered.success + assert executions == ["failure", "cancelled-probe", "recovery"] + assert runtime.circuit_state("cancellable_probe") is ToolCircuitState.CLOSED + assert runtime.metrics().circuit_breaker_trips == 1 + + +@pytest.mark.asyncio +async def test_async_timeout_trips_circuit_before_releasing_capacity() -> None: + started = asyncio.Event() + release = asyncio.Event() + executions = 0 + + @helix_tool + async def slow_async_dependency(value: str) -> str: + """Remain active until the caller-visible deadline cancels execution.""" + + nonlocal executions + executions += 1 + started.set() + await release.wait() + return value + + runtime = ToolRuntime(max_concurrency=1) + runtime._circuit_clock = lambda: 45.0 + runtime.register( + slow_async_dependency, + circuit_breaker=ToolCircuitBreaker( + failure_threshold=1, + recovery_timeout_seconds=60, + ), + ) + timed_out_call: asyncio.Task[ToolResult] | None = None + try: + timed_out_call = asyncio.create_task( + runtime.invoke( + "slow_async_dependency", + {"value": "private-timeout"}, + timeout=0.2, + ) + ) + await asyncio.wait_for(started.wait(), timeout=1) + timed_out = await timed_out_call + blocked = await runtime.invoke( + "slow_async_dependency", {"value": "private-blocked"}, timeout=1 + ) + finally: + release.set() + if timed_out_call is not None: + timed_out_call.cancel() + await asyncio.gather(timed_out_call, return_exceptions=True) + await runtime.aclose() + + assert timed_out.status is ToolStatus.TIMED_OUT + assert blocked.status is ToolStatus.CIRCUIT_OPEN + assert executions == 1 + assert runtime.metrics().circuit_breaker_trips == 1 + assert "private-blocked" not in str(blocked.to_dict()) + + +@pytest.mark.asyncio +async def test_sync_timeout_trips_circuit_without_submitting_another_worker() -> None: + started = Event() + release = Event() + executions = 0 + + @helix_tool + def slow_dependency(value: str) -> str: + """Hold one real worker until the host releases it.""" + + nonlocal executions + executions += 1 + started.set() + release.wait(timeout=5) + return value + + runtime = ToolRuntime(max_concurrency=1) + runtime._circuit_clock = lambda: 50.0 + runtime.register( + slow_dependency, + circuit_breaker=ToolCircuitBreaker( + failure_threshold=1, + recovery_timeout_seconds=60, + ), + ) + try: + timed_out = await runtime.invoke( + "slow_dependency", {"value": "private-timeout"}, timeout=0.02 + ) + assert started.wait(timeout=1) + blocked = await runtime.invoke("slow_dependency", {"value": "private-blocked"}, timeout=1) + metrics = runtime.metrics() + finally: + release.set() + await runtime.aclose(wait_for_sync=True, timeout=5) + + assert timed_out.status is ToolStatus.TIMED_OUT + assert blocked.status is ToolStatus.CIRCUIT_OPEN + assert executions == 1 + assert runtime.pending_sync_calls == 0 + assert metrics.timed_out == 1 + assert metrics.circuit_open == 1 + assert metrics.circuit_breaker_trips == 1 + assert "private-blocked" not in str(blocked.to_dict()) + + +@pytest.mark.asyncio +async def test_circuit_state_reset_and_replacement_are_registration_scoped() -> None: + @helix_tool(name="replaceable_circuit") + async def failing() -> str: + """Trip the original registration's circuit.""" + + raise RuntimeError("offline") + + @helix_tool(name="replaceable_circuit") + async def replacement() -> str: + """Represent a healthy replacement registration.""" + + return "available" + + runtime = ToolRuntime() + runtime._circuit_clock = lambda: 5.0 + runtime.register( + failing, + circuit_breaker=ToolCircuitBreaker( + failure_threshold=1, + recovery_timeout_seconds=60, + ), + ) + try: + assert (await runtime.invoke("replaceable_circuit")).status is ToolStatus.FAILED + assert runtime.circuit_state("replaceable_circuit") is ToolCircuitState.OPEN + assert runtime.reset_circuit("replaceable_circuit") is True + assert runtime.circuit_state("replaceable_circuit") is ToolCircuitState.CLOSED + runtime.register(replacement, replace=True) + assert runtime.circuit_state("replaceable_circuit") is None + assert runtime.reset_circuit("replaceable_circuit") is False + assert (await runtime.invoke("replaceable_circuit")).output == "available" + with pytest.raises(ToolNotFoundError): + runtime.circuit_state("missing") + with pytest.raises(ToolNotFoundError): + runtime.reset_circuit("missing") + finally: + await runtime.aclose() + + @pytest.mark.parametrize( ("kwargs", "error"), [ @@ -1148,6 +1693,39 @@ def test_tool_rate_limit_normalizes_its_default_burst() -> None: assert limit.to_dict() == {"calls": 3, "period_seconds": 2.0, "burst": 3} +@pytest.mark.parametrize( + ("kwargs", "error"), + [ + ({"failure_threshold": True, "recovery_timeout_seconds": 1}, TypeError), + ({"failure_threshold": 0, "recovery_timeout_seconds": 1}, ValueError), + ({"failure_threshold": 10**1_000, "recovery_timeout_seconds": 1}, ValueError), + ({"failure_threshold": 1, "recovery_timeout_seconds": True}, TypeError), + ({"failure_threshold": 1, "recovery_timeout_seconds": 0}, ValueError), + ({"failure_threshold": 1, "recovery_timeout_seconds": math.inf}, ValueError), + ({"failure_threshold": 1, "recovery_timeout_seconds": math.nan}, ValueError), + ({"failure_threshold": 1, "recovery_timeout_seconds": 1e308}, ValueError), + ({"failure_threshold": 1, "recovery_timeout_seconds": -1}, ValueError), + ({"failure_threshold": 1, "recovery_timeout_seconds": "1"}, TypeError), + ({"failure_threshold": "1", "recovery_timeout_seconds": 1}, TypeError), + ({"failure_threshold": 1, "recovery_timeout_seconds": 10**1_000}, ValueError), + ], +) +def test_tool_circuit_breaker_configuration_is_validated( + kwargs: dict[str, object], error: type[Exception] +) -> None: + with pytest.raises(error): + ToolCircuitBreaker(**kwargs) # type: ignore[arg-type] + + +def test_tool_circuit_breaker_normalizes_its_configuration() -> None: + policy = ToolCircuitBreaker(failure_threshold=3, recovery_timeout_seconds=2) + + assert policy.to_dict() == { + "failure_threshold": 3, + "recovery_timeout_seconds": 2.0, + } + + @pytest.mark.asyncio async def test_policy_receives_detached_validated_calls_and_fails_closed_on_denial() -> None: executions: list[tuple[list[int], str]] = []