Skip to content

Sandbox

sandbox

Own executor setup and compose sandbox provisioning below the agent roster.

check_setup(cfg=None, *, live=False)

Check owned routes and compose public sandbox checks without changing state.

Source code in src/terok_executor/sandbox.py
def check_setup(cfg: SandboxConfig | None = None, *, live: bool = False) -> tuple[SetupCheck, ...]:
    """Check owned routes and compose public sandbox checks without changing state."""
    cfg = cfg or SandboxConfig()
    try:
        expected = AgentRoster.load().generate_routes_json()
        routes = _check_routes(cfg, expected)
    except (OSError, ValueError) as exc:
        expected = ""
        routes = SetupCheck(_OWNER, "roster", SetupStatus.INVALID, str(exc))
    return (
        _receipt(cfg, expected).check(),
        routes,
        *check_sandbox_setup(cfg, live=live),
    )

ensure_sandbox_ready(*, cfg=None, no_vault=False, **aggregator_kwargs)

Preflight the full closure, generate routes, then provision sandbox services.

Routes must be current before the supervisor starts. Skipping route generation does not certify absent routes or incomplete lower setup. Successful child receipts survive failure in executor's final verification or receipt write.

Source code in src/terok_executor/sandbox.py
@setup_lock()
def ensure_sandbox_ready(
    *,
    cfg: SandboxConfig | None = None,
    no_vault: bool = False,
    **aggregator_kwargs: Any,
) -> None:
    """Preflight the full closure, generate routes, then provision sandbox services.

    Routes must be current before the supervisor starts. Skipping route generation
    does not certify absent routes or incomplete lower setup. Successful child
    receipts survive failure in executor's final verification or receipt write.
    """
    from terok_executor.integrations.sandbox import _handle_sandbox_setup, stage_line

    cfg = cfg or SandboxConfig()
    require_no_downgrade(check_setup(cfg))
    roster = AgentRoster.load()
    expected = roster.generate_routes_json()
    receipt = _receipt(cfg, expected)
    if not no_vault:
        receipt.clear()
        with stage_line("Vault routes") as stage:
            roster.ensure_vault_routes(cfg=cfg)
            stage.ok("regenerated")
    _handle_sandbox_setup(cfg=cfg, no_vault=no_vault, **aggregator_kwargs)
    require_setup((_check_routes(cfg, expected), *check_sandbox_setup(cfg)))
    receipt.write()