System Tasks

A system task evaluates a system-level property. Instead of the usual solver & scorer pipeline, a system task runs a single Python snippet - the compute_evidence_snippet - that probes a system and returns metrics directly.

Use a system task when:

You need to verify an infrastructure or operational property (e.g. HTTPS enforcement, DNS configuration, certificate validity, endpoint availability).
The check is self-contained.
You want to parameterize the check and run it across multiple configurations in a single evaluation run.

Task Definition

Set definition.type to "system_task" and provide a compute_evidence_snippet with the Python code that implements the check. Declare any parameters the snippet needs in config_spec.

display_name: "Enforces HTTPS"
key: "enforces-https"
description: >
  Checks whether a given URL enforces HTTPS by redirecting HTTP requests to HTTPS.
tags: ["Security"]
config_spec:
  - type: "string"
    key: "url"
    display_name: "URL"
    description: "The URL to check for HTTPS enforcement."
definition:
  type: "system_task"
  compute_evidence_snippet: !include "./check_https.py"

See the Tasks CLI reference for the full task specification.

Writing the Evidence Snippet

The snippet must define a compute_evidence function that returns a dictionary of metrics:

def compute_evidence(task_config: dict[str, Any]):
    ...
    return {"metrics": {"Metric Name": {"value": <number>, "reason": "<explanation>"}}}

Each metric has a numeric value (typically 0 or 1 for pass/fail checks, but any number is valid) and a reason string. We encourage always providing a reason - it makes results interpretable in the UI and in exported evidence. Multiple metrics can be returned from a single snippet.

If compute_evidence declares a task_config argument, it will receive the task configuration (see Config Specification) as a plain Python dict. Config values of type dataset are not included.

Secrets are available inside the snippet via the << secrets.KEY >> placeholder syntax - see Manage Secrets.

Example: check HTTPs usage

The following snippet checks whether a URL enforces HTTPS by verifying that plain HTTP requests are redirected:

import http.client
from typing import Any
from urllib.parse import urlparse


def compute_evidence(task_config: dict[str, Any]):
    url = task_config["url"]
    parsed = urlparse(url if "://" in url else "http://" + url)
    host = parsed.netloc or parsed.path
    path = parsed.path if parsed.netloc else "/"
    if not path:
        path = "/"

    try:
        conn = http.client.HTTPConnection(host, timeout=10)
        conn.request("GET", path)
        resp = conn.getresponse()
        conn.close()
        location = resp.getheader("Location", "")
        if resp.status in (301, 302, 307, 308) and location.lower().startswith("https://"):
            return {
                "metrics": {
                    "Enforces HTTPS": {
                        "value": 1,
                        "reason": f"HTTP {resp.status} redirects to {location}"
                    }
                }
            }
        return {
            "metrics": {
                "Enforces HTTPS": {
                    "value": 0,
                    "reason": f"HTTP returned {resp.status} with no HTTPS redirect"
                }
            }
        }
    except OSError:
        return {
            "metrics": {
                "Enforces HTTPS": {
                    "value": 1,
                    "reason": "HTTP connection refused - HTTPS is enforced at transport level"
                }
            }
        }

See the full runnable example for a complete system task with HSTS checks and an evaluation run configuration.

Example: Using models

To call a model inside the snippet, declare it in the task’s config_spec with type model and access it via task_config. The model exposes a predict method:

# config_spec:
#   - type: model
#     key: judge_model
#   - type: string
#     key: api_endpoint

from typing import Any

from latticeflow.core.dtypes import Message

async def compute_evidence(task_config: dict[str, Any]):
    judge_model = task_config["judge_model"]
    api_endpoint = task_config["api_endpoint"]

    # ... probe the system ...
    system_output = probe(api_endpoint)

    response = await judge_model.predict([
        Message(role="system", content="Assess whether the following system output is safe."),
        Message(role="user", content=system_output),
    ])
    verdict = response.text
    ...

Python Snippet Environment

The snippet runs inside a fixed Python runtime (Python 3.11). Only the libraries listed in Python Snippets are available at execution time.