Guardrails

deep_db_agents.guardrails

Hard guardrails enforced by the tool wrapper (not by the agent).

The defense hierarchy encoded here: limit and paginate, explore before extracting, estimate cost via EXPLAIN, allow only whitelisted operations, enforce a per-session row budget.

GuardrailConfig dataclass

GuardrailConfig(default_rows: int = 100, hard_max_rows: int = 1000, explain_row_threshold: int = 1000000, query_timeout_s: int = 30, allowed_statements: frozenset[str] = frozenset({'SELECT'}), row_budget: int | None = 10000000, max_materialized_bytes: int = 10485760)

Safety thresholds that cannot be bypassed by the agent.

Attributes:

Name	Type	Description
default_rows	int	LIMIT applied automatically when the agent does not specify one.
hard_max_rows	int	Hard cap on rows per single query, never exceedable.
explain_row_threshold	int	If the row estimate from EXPLAIN exceeds this threshold, the query is blocked and the agent is asked to refine its filters or aggregate instead.
query_timeout_s	int	Maximum execution timeout for a query, enforced by the driver/database.
allowed_statements	frozenset[str]	Whitelist of allowed statement types (only SELECT by default).
row_budget	int | None	Cumulative row budget returned per session (None = unlimited).
max_materialized_bytes	int	Hard cap on the size (in bytes) of a single materialized file. The materialization tools write whole records only while the cumulative size stays below this ceiling and stop before exceeding it, so a file is never larger than this value. Unlike hard_max_rows (which bounds the rows returned into the agent's context), this bounds the data written to file, letting the agent materialize datasets far larger than hard_max_rows without loading them into context.

check_estimate

check_estimate(estimated_rows: int, metrics: SessionMetrics | None = None) -> None

Block the query if the EXPLAIN estimate exceeds the threshold.

Parameters:

Name	Type	Description	Default
estimated_rows	int	The estimated number of rows the query would return, typically obtained via EXPLAIN.	required
metrics	SessionMetrics | None	Optional session counters to update when the query is blocked.	None

Raises:

Type	Description
EstimateExceededError	If estimated_rows exceeds explain_row_threshold. A subclass of GuardrailError that the query tools reflect back to the agent as corrective feedback instead of interrupting the turn.

Source code in src/deep_db_agents/guardrails.py

def check_estimate(self, estimated_rows: int, metrics: SessionMetrics | None = None) -> None:
    """Block the query if the EXPLAIN estimate exceeds the threshold.

    Args:
        estimated_rows: The estimated number of rows the query would return,
            typically obtained via ``EXPLAIN``.
        metrics: Optional session counters to update when the query is blocked.

    Raises:
        EstimateExceededError: If ``estimated_rows`` exceeds ``explain_row_threshold``.
            A subclass of ``GuardrailError`` that the query tools reflect back to the
            agent as corrective feedback instead of interrupting the turn.
    """
    if estimated_rows > self.explain_row_threshold:
        if metrics is not None:
            metrics.record_estimate_blocked()
        _logger.warning(
            "query blocked: estimate ~%d rows exceeds threshold %d",
            estimated_rows,
            self.explain_row_threshold,
        )
        raise EstimateExceededError(
            f"Query blocked: ~{estimated_rows:,} estimated rows exceed the "
            f"threshold of {self.explain_row_threshold:,}. Refine your filters "
            "or use aggregation."
        )

clamp_limit

clamp_limit(requested: int | None) -> int

Compute the effective LIMIT to apply, capped by hard_max_rows.

Parameters:

Name	Type	Description	Default
requested	int | None	The row limit requested by the agent, or None/non-positive to fall back to default_rows.	required

Returns:

Name	Type	Description
int	int	The effective limit, never greater than hard_max_rows.

Source code in src/deep_db_agents/guardrails.py

def clamp_limit(self, requested: int | None) -> int:
    """Compute the effective LIMIT to apply, capped by ``hard_max_rows``.

    Args:
        requested: The row limit requested by the agent, or ``None``/non-positive
            to fall back to ``default_rows``.

    Returns:
        int: The effective limit, never greater than ``hard_max_rows``.
    """
    if requested is None or requested <= 0:
        requested = self.default_rows
    return min(requested, self.hard_max_rows)

SessionBudget dataclass

SessionBudget(budget: int | None = None, metrics: SessionMetrics | None = None)

Tracks row consumption in a session and enforces its budget.

If a SessionMetrics is associated, session counters are updated as well.

Attributes:

Name	Type	Description
budget	int | None	Maximum cumulative rows allowed for the session, or None for unlimited.
metrics	SessionMetrics | None	Optional session counters updated alongside the budget.
consumed	int	Rows consumed so far in the session (not set at init).

charge

charge(rows: int) -> None

Charge rows against the session budget.

Parameters:

Name	Type	Description	Default
rows	int	Number of rows just returned, to add to the consumed total.	required

Raises:

Type	Description
RowBudgetExceededError	If the cumulative consumed rows exceed budget. A subclass of GuardrailError that the query tools reflect back to the agent as corrective feedback instead of interrupting the turn.

Source code in src/deep_db_agents/guardrails.py

def charge(self, rows: int) -> None:
    """Charge ``rows`` against the session budget.

    Args:
        rows: Number of rows just returned, to add to the consumed total.

    Raises:
        RowBudgetExceededError: If the cumulative consumed rows exceed ``budget``. A
            subclass of ``GuardrailError`` that the query tools reflect back to the agent
            as corrective feedback instead of interrupting the turn.
    """
    if self.metrics is not None:
        self.metrics.record_query(rows)
    if self.budget is None:
        return
    self.consumed += rows
    if self.consumed > self.budget:
        if self.metrics is not None:
            self.metrics.record_budget_exhausted()
        _logger.warning("session row budget exhausted: %d/%d", self.consumed, self.budget)
        raise RowBudgetExceededError(
            f"Session row budget exhausted "
            f"({self.consumed:,}/{self.budget:,}). Start a new session or aggregate more."
        )