Index

operators

Operators — persistent, scheduled autonomous agents.

Classes

OperatorManager

OperatorManager(system: Any)

Manages operator manifests and their lifecycle via the TaskScheduler.

PARAMETER	DESCRIPTION
system	A DiapasonSystem instance (used to access scheduler, session_store, memory_backend, and to run operators via system.ask()). TYPE: Any

Source code in src/diapason/operators/manager.py

def __init__(self, system: Any) -> None:
    self._system = system
    self._manifests: Dict[str, OperatorManifest] = {}

Attributes

manifests property

manifests: Dict[str, OperatorManifest]

All registered manifests.

Methods:

register

register(manifest: OperatorManifest) -> None

Register an operator manifest.

Source code in src/diapason/operators/manager.py

def register(self, manifest: OperatorManifest) -> None:
    """Register an operator manifest."""
    self._manifests[manifest.id] = manifest
    logger.info("Registered operator: %s", manifest.id)

discover

discover(directory: str | Path) -> List[OperatorManifest]

Discover and register operator manifests from a directory.

Scans for *.toml files in directory and loads each as an operator manifest.

Source code in src/diapason/operators/manager.py

def discover(self, directory: str | Path) -> List[OperatorManifest]:
    """Discover and register operator manifests from a directory.

    Scans for ``*.toml`` files in *directory* and loads each as an
    operator manifest.
    """
    directory = Path(directory).expanduser()
    found: List[OperatorManifest] = []
    if not directory.is_dir():
        return found
    for toml_path in sorted(directory.glob("*.toml")):
        try:
            manifest = load_operator(toml_path)
            self.register(manifest)
            found.append(manifest)
        except Exception:
            logger.warning("Failed to load operator from %s", toml_path)
    return found

activate

activate(operator_id: str) -> str

Activate an operator by creating a scheduler task.

Returns the scheduler task ID (deterministic: operator:{id}). Raises KeyError if the operator is not registered, or RuntimeError if the scheduler is not available.

Source code in src/diapason/operators/manager.py

def activate(self, operator_id: str) -> str:
    """Activate an operator by creating a scheduler task.

    Returns the scheduler task ID (deterministic: ``operator:{id}``).
    Raises ``KeyError`` if the operator is not registered, or
    ``RuntimeError`` if the scheduler is not available.
    """
    manifest = self._manifests.get(operator_id)
    if manifest is None:
        raise KeyError(f"Operator not registered: {operator_id}")

    # Before anything is scheduled — see capability_guard for why here and
    # not at load or at tick. A refusal must be able to say truthfully
    # that nothing will run.
    self._refuse_if_capabilities_do_not_hold(operator_id, manifest)

    scheduler = self._system.scheduler
    if scheduler is None:
        raise RuntimeError(
            "TaskScheduler not available. Enable [scheduler] in config."
        )

    task_id = f"operator:{operator_id}"

    # Check if already active
    try:
        existing = scheduler.list_tasks()
        for t in existing:
            if t.id == task_id and t.status == "active":
                logger.info("Operator %s already active", operator_id)
                return task_id
    except Exception:
        pass

    tools_str = ",".join(manifest.tools) if manifest.tools else ""
    metadata: Dict[str, Any] = {
        "operator_id": operator_id,
        "system_prompt": manifest.system_prompt,
        "temperature": manifest.temperature,
        "max_turns": manifest.max_turns,
        "metrics": manifest.metrics,
    }

    # Use the scheduler's create_task but with a deterministic ID
    task = scheduler.create_task(
        prompt=_TICK_PROMPT,
        schedule_type=manifest.schedule_type,
        schedule_value=manifest.schedule_value,
        agent="operative",
        tools=tools_str,
        metadata=metadata,
    )

    # Override the random ID with our deterministic one
    task_dict = task.to_dict()
    task_dict["id"] = task_id
    scheduler._store.save_task(task_dict)
    logger.info("Activated operator %s (task_id=%s)", operator_id, task_id)
    return task_id

deactivate

deactivate(operator_id: str) -> None

Deactivate an operator by cancelling its scheduler task.

Source code in src/diapason/operators/manager.py

def deactivate(self, operator_id: str) -> None:
    """Deactivate an operator by cancelling its scheduler task."""
    scheduler = self._system.scheduler
    if scheduler is None:
        raise RuntimeError("TaskScheduler not available.")
    task_id = f"operator:{operator_id}"
    try:
        scheduler.cancel_task(task_id)
        logger.info("Deactivated operator %s", operator_id)
    except KeyError:
        logger.warning("No active task for operator %s", operator_id)

pause

pause(operator_id: str) -> None

Pause an active operator.

Source code in src/diapason/operators/manager.py

def pause(self, operator_id: str) -> None:
    """Pause an active operator."""
    scheduler = self._system.scheduler
    if scheduler is None:
        raise RuntimeError("TaskScheduler not available.")
    scheduler.pause_task(f"operator:{operator_id}")
    logger.info("Paused operator %s", operator_id)

resume

resume(operator_id: str) -> None

Resume a paused operator.

Source code in src/diapason/operators/manager.py

def resume(self, operator_id: str) -> None:
    """Resume a paused operator."""
    scheduler = self._system.scheduler
    if scheduler is None:
        raise RuntimeError("TaskScheduler not available.")
    scheduler.resume_task(f"operator:{operator_id}")
    logger.info("Resumed operator %s", operator_id)

status

status() -> List[Dict[str, Any]]

Return status of all registered operators.

Merges manifest info with scheduler task state.

Source code in src/diapason/operators/manager.py

def status(self) -> List[Dict[str, Any]]:
    """Return status of all registered operators.

    Merges manifest info with scheduler task state.
    """
    results: List[Dict[str, Any]] = []
    scheduler = self._system.scheduler
    for op_id, manifest in self._manifests.items():
        info: Dict[str, Any] = {
            "id": op_id,
            "name": manifest.name,
            "description": manifest.description,
            "schedule_type": manifest.schedule_type,
            "schedule_value": manifest.schedule_value,
            "tools": manifest.tools,
            "metrics": manifest.metrics,
            "status": "registered",
            "next_run": None,
            "last_run": None,
        }
        if scheduler is not None:
            task_id = f"operator:{op_id}"
            try:
                tasks = scheduler.list_tasks()
                for t in tasks:
                    if t.id == task_id:
                        info["status"] = t.status
                        info["next_run"] = t.next_run
                        info["last_run"] = t.last_run
                        break
            except Exception:
                pass
        results.append(info)
    return results

run_once

run_once(operator_id: str) -> str

Execute a single tick of an operator immediately.

Useful for development and testing. Returns the agent's response.

Source code in src/diapason/operators/manager.py

def run_once(self, operator_id: str) -> str:
    """Execute a single tick of an operator immediately.

    Useful for development and testing. Returns the agent's response.
    """
    manifest = self._manifests.get(operator_id)
    if manifest is None:
        raise KeyError(f"Operator not registered: {operator_id}")

    # The same gate as activate(): the two doors must say the same thing,
    # or "it works when I run it by hand" becomes a way around the check.
    self._refuse_if_capabilities_do_not_hold(operator_id, manifest)

    tools_list = manifest.tools if manifest.tools else None
    result = self._system.ask(
        _TICK_PROMPT,
        agent="operative",
        tools=tools_list,
        system_prompt=manifest.system_prompt,
        operator_id=operator_id,
        temperature=manifest.temperature,
    )
    if isinstance(result, dict):
        return result.get("content", str(result))
    return str(result)

collect_metrics

collect_metrics(
    operator_id: str,
    *,
    since: Optional[float] = None,
    until: Optional[float] = None,
) -> Dict[str, Any]

Collect telemetry metrics declared in the operator's manifest.

Reads the metrics list from :class:OperatorManifest and queries the system's TelemetryAggregator (when available) for matching statistics. Only fields explicitly listed in manifest.metrics are returned, giving operators fine-grained control over what data they expose.

PARAMETER	DESCRIPTION
operator_id	ID of the registered operator. TYPE: str
since	Optional Unix timestamp; restrict stats to records after this time. TYPE: Optional[float] DEFAULT: None
until	Optional Unix timestamp; restrict stats to records before this time. TYPE: Optional[float] DEFAULT: None

RETURNS	DESCRIPTION
Dict[str, Any]	Mapping of {metric_name: value} for every metric declared in the manifest that could be resolved. Unknown metric names are silently skipped and logged at DEBUG level.

RAISES	DESCRIPTION
KeyError	If operator_id is not registered.

Source code in src/diapason/operators/manager.py

def collect_metrics(
    self,
    operator_id: str,
    *,
    since: Optional[float] = None,
    until: Optional[float] = None,
) -> Dict[str, Any]:
    """Collect telemetry metrics declared in the operator's manifest.

    Reads the ``metrics`` list from :class:`OperatorManifest` and queries
    the system's ``TelemetryAggregator`` (when available) for matching
    statistics.  Only fields explicitly listed in ``manifest.metrics`` are
    returned, giving operators fine-grained control over what data they
    expose.

    Parameters
    ----------
    operator_id:
        ID of the registered operator.
    since:
        Optional Unix timestamp; restrict stats to records after this time.
    until:
        Optional Unix timestamp; restrict stats to records before this time.

    Returns
    -------
    Dict[str, Any]
        Mapping of ``{metric_name: value}`` for every metric declared in
        the manifest that could be resolved.  Unknown metric names are
        silently skipped and logged at DEBUG level.

    Raises
    ------
    KeyError
        If *operator_id* is not registered.
    """
    manifest = self._manifests.get(operator_id)
    if manifest is None:
        raise KeyError(f"Operator not registered: {operator_id}")

    if not manifest.metrics:
        logger.debug(
            "Operator %s declares no metrics; returning empty dict.",
            operator_id,
        )
        return {}

    aggregator = getattr(self._system, "telemetry", None)
    if aggregator is None:
        logger.warning(
            "TelemetryAggregator not available on system; "
            "cannot collect metrics for operator %s.",
            operator_id,
        )
        return {}

    try:
        summary = aggregator.summary(since=since, until=until)
    except Exception as exc:
        logger.error(
            "Failed to query telemetry for operator %s: %s",
            operator_id,
            exc,
        )
        return {}

    # Build a flat lookup of all available summary-level stats
    available: Dict[str, Any] = {
        "total_calls": summary.total_calls,
        "total_tokens": summary.total_tokens,
        "total_cost": summary.total_cost,
        "total_latency": summary.total_latency,
        "total_energy_joules": summary.total_energy_joules,
        "avg_throughput_tok_per_sec": (summary.avg_throughput_tok_per_sec),
        "avg_gpu_utilization_pct": summary.avg_gpu_utilization_pct,
        "avg_energy_per_output_token_joules": (
            summary.avg_energy_per_output_token_joules
        ),
        "avg_throughput_per_watt": summary.avg_throughput_per_watt,
        "total_prefill_energy_joules": (summary.total_prefill_energy_joules),
        "total_decode_energy_joules": (summary.total_decode_energy_joules),
        "avg_mean_itl_ms": summary.avg_mean_itl_ms,
        "avg_median_itl_ms": summary.avg_median_itl_ms,
        "avg_p95_itl_ms": summary.avg_p95_itl_ms,
    }

    collected: Dict[str, Any] = {}
    for metric in manifest.metrics:
        if metric in available:
            collected[metric] = available[metric]
        else:
            logger.debug(
                "Operator %s requested unknown metric '%s'; skipping.",
                operator_id,
                metric,
            )

    logger.info(
        "Collected %d/%d metrics for operator %s.",
        len(collected),
        len(manifest.metrics),
        operator_id,
    )
    return collected

get_manifest

get_manifest(
    operator_id: str,
) -> Optional[OperatorManifest]

Return the manifest for an operator, or None.

Source code in src/diapason/operators/manager.py

def get_manifest(self, operator_id: str) -> Optional[OperatorManifest]:
    """Return the manifest for an operator, or None."""
    return self._manifests.get(operator_id)

OperatorManifest dataclass

OperatorManifest(
    id: str,
    name: str,
    version: str = "0.1.0",
    description: str = "",
    author: str = "",
    tools: List[str] = list(),
    system_prompt: str = "",
    system_prompt_path: str = "",
    max_turns: int = 20,
    temperature: float = 0.3,
    schedule_type: str = "interval",
    schedule_value: str = "300",
    metrics: List[str] = list(),
    required_capabilities: List[str] = list(),
    settings: Dict[str, Any] = dict(),
    metadata: Dict[str, Any] = dict(),
)

Manifest describing a persistent autonomous operator.

Functions:

load_operator

load_operator(path: str | Path) -> OperatorManifest

Load an operator manifest from a TOML file.

Supports inline system_prompt or external system_prompt_path (resolved relative to the TOML file).

Source code in src/diapason/operators/loader.py

def load_operator(path: str | Path) -> OperatorManifest:
    """Load an operator manifest from a TOML file.

    Supports inline ``system_prompt`` or external ``system_prompt_path``
    (resolved relative to the TOML file).
    """
    path = Path(path)
    with open(path, "rb") as fh:
        data = tomllib.load(fh)

    op_data = data.get("operator", {})

    # Resolve schedule sub-table
    schedule = op_data.get("schedule", {})
    schedule_type = schedule.get("type", op_data.get("schedule_type", "interval"))
    schedule_value = schedule.get("value", op_data.get("schedule_value", "300"))

    # Resolve agent sub-table
    agent_data = op_data.get("agent", {})
    tools = agent_data.get("tools", op_data.get("tools", []))
    max_turns = agent_data.get("max_turns", op_data.get("max_turns", 20))
    temperature = agent_data.get("temperature", op_data.get("temperature", 0.3))
    system_prompt = agent_data.get("system_prompt", op_data.get("system_prompt", ""))
    system_prompt_path = agent_data.get(
        "system_prompt_path",
        op_data.get("system_prompt_path", ""),
    )

    # Load external system prompt if specified and inline is empty
    if not system_prompt and system_prompt_path:
        prompt_path = Path(system_prompt_path)
        if not prompt_path.is_absolute():
            prompt_path = path.parent / prompt_path
        if prompt_path.exists():
            system_prompt = prompt_path.read_text(encoding="utf-8")

    return OperatorManifest(
        id=op_data.get("id", path.stem),
        name=op_data.get("name", path.stem),
        version=op_data.get("version", "0.1.0"),
        description=op_data.get("description", ""),
        author=op_data.get("author", ""),
        tools=tools,
        system_prompt=system_prompt,
        system_prompt_path=system_prompt_path,
        max_turns=max_turns,
        temperature=temperature,
        schedule_type=schedule_type,
        schedule_value=str(schedule_value),
        metrics=op_data.get("metrics", []),
        required_capabilities=op_data.get("required_capabilities", []),
        settings=op_data.get("settings", {}),
        metadata=op_data.get("metadata", {}),
    )