orxtra v0.13.0 /scheduler.src.orxtra.scheduler._enforcement
Updated
On this page

EnforcementMixin: cost accumulation and budget events, pre/post-check execution, and mechanical constraints — tests pass, lint clean, no new deps, no removed exports, no changed signatures.

#scheduler.src.orxtra.scheduler._enforcement

#scheduler.src.orxtra.scheduler._enforcement

#EnforcementMixin

Mixin for budget enforcement, constraints, checks, and validation.

#_send_budget_events

python
async def _send_budget_events(self, task_id: UUID) -> None

Send accumulated budget events to the Overseer.

#_run_mechanical_constraints

python
async def _run_mechanical_constraints(self, task_id: UUID) -> list[CheckResult]

Run active mechanical constraints.

Cheap constraints run after every task. Expensive constraints (tests_pass, lint_clean) run only when the completing task has subtasks (workflow completion).

#_check_constraint

python
async def _check_constraint(self, kind: ConstraintKind, task_id: UUID, constraint_text: str='') -> CheckResult

Check a single mechanical constraint.

Dispatches to individual checker methods based on the constraint kind.

#_check_tests_pass

python
async def _check_tests_pass(self) -> CheckResult

Run pytest and check exit code.

#_check_lint_clean

python
async def _check_lint_clean(self) -> CheckResult

Run ruff check and check exit code.

#_check_no_new_dependencies

python
async def _check_no_new_dependencies(self) -> CheckResult

Check if dependency files changed.

#_check_no_new_files_outside

python
async def _check_no_new_files_outside(self, constraint_text: str) -> CheckResult

Check for new files outside a directory.

The directory is extracted from the constraint text, e.g. "no_new_files_outside(src/)".

#_snapshot_exports

python
def _snapshot_exports(self, glob_pattern: str) -> dict[str, set[str]]

Extract top-level names from files matching glob.

#_snapshot_signatures

python
def _snapshot_signatures(self, glob_pattern: str) -> dict[str, dict[str, list[str]]]

Extract function signatures from files matching glob.

#_capture_pre_task_snapshots

python
def _capture_pre_task_snapshots(self, task_id: UUID) -> None

Capture export and signature snapshots before task execution.

Scans all active mechanical constraints that need snapshots and stores the data keyed by task_id.

#_check_no_removed_exports

python
def _check_no_removed_exports(self, task_id: UUID, constraint_text: str) -> CheckResult

Check that no exports were removed.

#_check_no_changed_signatures

python
def _check_no_changed_signatures(self, task_id: UUID, constraint_text: str) -> CheckResult

Check that no function signatures changed.

#_validate_output_schema

python
def _validate_output_schema(self, output: str | None, schema_str: str) -> CheckResult

Validate agent output against a JSON schema.

#_analyze_structural_advisories

python
def _analyze_structural_advisories(self, task_ids: list[UUID]) -> list[dict[str, Any]]

Analyze a set of tasks for structural improvements.

Detects read-only tasks that could be front-loaded for earlier execution. Advisories are stored for the Overseer to consume.

Search