mirror of
https://github.com/usestrix/strix.git
synced 2026-08-22 11:02:08 +02:00
Add configurable token / cost usage limits (#576)
* fix: resolve pre-commit check failures - Change RuntimeError to TypeError for type validation in report/writer.py - Update pyupgrade to v3.21.2 for Python 3.14 compatibility * feat(cli): add --max-budget-usd flag Raises BudgetExceededError in ReportUsageHooks after each LLM call when accumulated cost reaches the limit, with clean "stopped" status and child-agent cancellation in non-interactive mode. * test: add budget enforcement unit tests 7 tests covering no-budget, under-budget, at-limit, over-limit, error message content, None report state, and exception hierarchy. Also adds pytest/pytest-asyncio to dev deps and a mypy override for tests. * fix(budget): validate positive budget and check the live cost ledger Two hardening fixes for --max-budget-usd enforcement: - Reject non-positive budgets. ReportUsageHooks now raises ValueError for max_budget_usd <= 0, and the CLI validates the flag via a custom argparse type so '--max-budget-usd 0' fails fast with a friendly message instead of silently killing the scan on the first model response. - Read the live cost. The budget check now reads ReportState.get_total_llm_cost() (the live ledger) instead of the persisted run-record snapshot, so it stays accurate even when a usage save fails after a model call. * fix(budget): stop the entire scan deterministically when the limit is hit Previously a BudgetExceededError was handled per-agent: it was swallowed in interactive mode (the loop kept waiting), a child's error escaped its detached task as an unretrieved-exception warning, the parent was never released from wait_for_message, and the stop was logged at ERROR with a traceback as if the agent had failed. Replace that with a single scan-wide signal on the coordinator: - AgentCoordinator.trigger_budget_stop() sets a flag and wakes every parked agent; wait_for_message returns as soon as the flag is set. - The run loops check coordinator.budget_stopped and raise to exit cleanly, marking themselves 'stopped'. The root's exception reaches run_strix_scan's handler, which cancels descendants and tears the scan down once; child exceptions are swallowed in their detached task. - The budget stop is logged at INFO, not as a failure. This is deterministic regardless of tree depth or which agent first sees the limit, fixing the interactive/TUI hang where a deep agent's stop never reached a parked root. Also re-raises BudgetExceededError explicitly in the stream handler so it can't be mistaken for the LiteLLM 'after shutdown' race. * fix(budget): treat a budget stop as a clean stop in the TUI Add an explicit BudgetExceededError handler in the TUI scan thread so that, if the error ever reaches it, the budget stop is logged as a graceful stop rather than surfaced as a red scan error by the broad 'except Exception'. The runner normally absorbs the error and returns cleanly, so this is defensive depth for a money-spending feature. * docs(cli): document --max-budget-usd behavior and limitations Clarify that the budget is cumulative across all agents, checked after each model response, that the scan stops cleanly (not as a failure), that the value must be > 0, and that spend can slightly overshoot due to in-flight calls and best-effort cost estimation. * Apply suggestions from code review ---------
This commit is contained in:
@@ -183,6 +183,7 @@ async def run_cli(args: Any) -> None: # noqa: PLR0915
|
||||
image=_resolve_sandbox_image(),
|
||||
local_sources=getattr(args, "local_sources", None) or [],
|
||||
interactive=bool(getattr(args, "interactive", False)),
|
||||
max_budget_usd=getattr(args, "max_budget_usd", None),
|
||||
)
|
||||
finally:
|
||||
stop_updates.set()
|
||||
|
||||
@@ -301,6 +301,17 @@ def get_version() -> str:
|
||||
return "unknown"
|
||||
|
||||
|
||||
def _positive_budget(value: str) -> float:
|
||||
try:
|
||||
budget = float(value)
|
||||
except ValueError as exc:
|
||||
raise argparse.ArgumentTypeError(f"invalid float value: {value!r}") from exc
|
||||
import math
|
||||
if not math.isfinite(budget) or budget <= 0:
|
||||
raise argparse.ArgumentTypeError("must be a finite number greater than 0")
|
||||
return budget
|
||||
|
||||
|
||||
def parse_arguments() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Strix Multi-Agent Cybersecurity Penetration Testing Tool",
|
||||
@@ -424,6 +435,13 @@ Examples:
|
||||
help="Path to a custom config file (JSON) to use instead of ~/.strix/cli-config.json",
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--max-budget-usd",
|
||||
type=_positive_budget,
|
||||
default=None,
|
||||
help="Maximum LLM cost in USD (> 0). The scan stops cleanly when this limit is reached.",
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--resume",
|
||||
type=str,
|
||||
|
||||
@@ -31,6 +31,7 @@ from textual.widgets import Button, Label, Static, TextArea, Tree
|
||||
from textual.widgets.tree import TreeNode
|
||||
|
||||
from strix.config import load_settings
|
||||
from strix.core.hooks import BudgetExceededError
|
||||
from strix.core.runner import run_strix_scan
|
||||
from strix.interface.tui.live_view import TuiLiveView
|
||||
from strix.interface.tui.messages import send_user_message_to_agent
|
||||
@@ -1369,12 +1370,18 @@ class StrixTUIApp(App): # type: ignore[misc]
|
||||
local_sources=getattr(self.args, "local_sources", None) or [],
|
||||
coordinator=self.coordinator,
|
||||
interactive=True,
|
||||
max_budget_usd=getattr(self.args, "max_budget_usd", None),
|
||||
event_sink=self._capture_sdk_event,
|
||||
),
|
||||
)
|
||||
|
||||
except (KeyboardInterrupt, asyncio.CancelledError):
|
||||
logger.info("Scan interrupted by user")
|
||||
except BudgetExceededError:
|
||||
# Defensive: the runner stops the scan cleanly on budget and
|
||||
# returns, so this normally never propagates. Treat it as a
|
||||
# graceful stop, not a scan error, if it ever does.
|
||||
logger.info("Scan stopped: --max-budget-usd limit reached")
|
||||
except (ConnectionError, TimeoutError) as e:
|
||||
logging.exception("Network error during scan")
|
||||
self._scan_error = e
|
||||
|
||||
Reference in New Issue
Block a user