Validation

Static workflow and source checks.

validate

Static checks on a workflow and its agents before execution.

Deterministic AST and graph checks identify cycles, dangling edges, port-type mismatches, missing methods, or integrity violations prior to execution.

Issue dataclass

Issue(level, where, message, fix='')

One validation finding.

Attributes:

Name	Type	Description
level	str	Severity level ('error' blocks execution; 'warning' does not).
where	str	Location string (e.g. file, node, or line number).
message	str	Explanation of the issue.
fix	str	Actionable guidance on how to fix the issue.

Report dataclass

Report(issues=list())

The outcome of validation checks.

Attributes:

Name	Type	Description
issues	list[Issue]	List of recorded Issue objects.

errors property

errors

Returns all error-level issues.

warnings property

warnings

Returns all warning-level issues.

ok property

ok

Returns True when no errors block execution.

error

error(where, message, fix='')

Appends an error-level issue.

Source code in python/src/science_adk/validate.py

def error(self, where: str, message: str, fix: str = "") -> None:
    """Appends an error-level issue."""
    self.issues.append(Issue("error", where, message, fix))

warn

warn(where, message, fix='')

Appends a warning-level issue.

Source code in python/src/science_adk/validate.py

def warn(self, where: str, message: str, fix: str = "") -> None:
    """Appends a warning-level issue."""
    self.issues.append(Issue("warning", where, message, fix))

extend

extend(other)

Merges issues from another Report.

Source code in python/src/science_adk/validate.py

def extend(self, other: Report) -> None:
    """Merges issues from another Report."""
    self.issues.extend(other.issues)

to_dict

to_dict()

Serializes the Report to a dictionary.

Source code in python/src/science_adk/validate.py

def to_dict(self) -> dict[str, Any]:
    """Serializes the Report to a dictionary."""
    return {
        "ok": self.ok,
        "errors": len(self.errors),
        "warnings": len(self.warnings),
        "issues": [
            {
                "level": i.level,
                "where": i.where,
                "message": i.message,
                "fix": i.fix,
            }
            for i in self.issues
        ],
    }

render

render()

Formats all issues into a readable multi-line string.

Source code in python/src/science_adk/validate.py

def render(self) -> str:
    """Formats all issues into a readable multi-line string."""
    if not self.issues:
        return "ok: no issues found"
    return "\n".join(str(i) for i in self.issues)

validate_workflow

validate_workflow(workflow)

Verifies that a workflow DAG is acyclic, connected, and typed correctly.

Parameters:

Name	Type	Description	Default
workflow	Workflow	The Workflow object to validate.	required

Returns:

Type	Description
Report	A validation Report.

Source code in python/src/science_adk/validate.py

def validate_workflow(workflow: Workflow) -> Report:
    """Verifies that a workflow DAG is acyclic, connected, and typed correctly.

    Args:
      workflow: The Workflow object to validate.

    Returns:
      A validation Report.
    """
    report = Report()

    if not workflow.nodes:
        report.error("workflow", "has no nodes", "Add at least one node.")
        return report

    ids = [n.id for n in workflow.nodes]
    duplicates = sorted({i for i in ids if ids.count(i) > 1})
    for dup in duplicates:
        report.error(
            "workflow", f"duplicate node id {dup!r}", "Node ids must be unique."
        )

    known = set(ids)

    for edge in workflow.edges:
        where = f"edge {edge}"
        if edge.source not in known:
            report.error(where, f"unknown source node {edge.source!r}")
            continue
        if edge.target not in known:
            report.error(where, f"unknown target node {edge.target!r}")
            continue

        source_node = workflow.node(edge.source)
        target_node = workflow.node(edge.target)
        assert source_node and target_node

        out_port = source_node.ports.output(edge.source_port)
        in_port = target_node.ports.input(edge.target_port)

        if out_port is None:
            available = ", ".join(p.name for p in source_node.ports.outputs) or "none"
            report.error(
                where,
                f"{edge.source!r} has no output port {edge.source_port!r}",
                f"Available outputs: {available}",
            )
        if in_port is None:
            available = ", ".join(p.name for p in target_node.ports.inputs) or "none"
            report.error(
                where,
                f"{edge.target!r} has no input port {edge.target_port!r}",
                f"Available inputs: {available}",
            )
        if out_port and in_port and not types_compatible(out_port.type, in_port.type):
            report.error(
                where,
                f"type mismatch: {out_port.type} -> {in_port.type}",
                "Change one port's type, or insert a node that converts.",
            )

    seen: dict[tuple[str, str], str] = {}
    for edge in workflow.edges:
        key = (edge.target, edge.target_port)
        if key in seen:
            report.error(
                f"edge {edge}",
                f"input {edge.target}.{edge.target_port} is already fed by {seen[key]}",
                "An input port accepts exactly one edge.",
            )
        else:
            seen[key] = f"{edge.source}.{edge.source_port}"

    graph = {n.id: set(workflow.parents(n.id)) for n in workflow.nodes}
    try:
        graphlib.TopologicalSorter(graph).prepare()
    except graphlib.CycleError as exc:
        cycle = " -> ".join(str(part) for part in exc.args[1])
        report.error(
            "workflow",
            f"cycle detected: {cycle}",
            "Remove an edge to break the cycle.",
        )

    for node in workflow.nodes:
        for port in node.ports.required_inputs:
            if (node.id, port.name) not in seen:
                report.error(
                    f"node {node.id}",
                    f"required input {port.name!r} is not connected",
                    f"Add an edge into {node.id}.{port.name}, or mark the port"
                    " optional.",
                )

    if len(workflow.nodes) > 1:
        for node in workflow.nodes:
            if not workflow.parents(node.id) and not workflow.children(node.id):
                report.warn(
                    f"node {node.id}",
                    "is not connected to anything",
                    "Connect it, or remove it from the workflow.",
                )

    if not any(n.kind == "evaluation" for n in workflow.nodes):
        report.warn(
            "workflow",
            "has no evaluation node",
            "Add a node of kind 'evaluation' so the result can be measured"
            " against the goal.",
        )

    return report

validate_agent_source

validate_agent_source(path, node=None)

Statically validates an agent Python source file against its node contract.

Parameters:

Name	Type	Description	Default
path	Path	Path to the agent Python file.	required
node	Node | None	Optional Node specification from workflow.json to verify consistency against.	None

Returns:

Type	Description
Report	A validation Report.

Source code in python/src/science_adk/validate.py

def validate_agent_source(path: Path, node: Node | None = None) -> Report:
    """Statically validates an agent Python source file against its node contract.

    Args:
      path: Path to the agent Python file.
      node: Optional Node specification from workflow.json to verify
        consistency against.

    Returns:
      A validation Report.
    """
    report = Report()
    where = path.name

    if not path.exists():
        report.error(where, "file does not exist", f"Write the agent at {path}.")
        return report

    source = path.read_text(encoding="utf-8")
    try:
        tree = ast.parse(source, filename=str(path))
    except SyntaxError as exc:
        report.error(where, f"syntax error on line {exc.lineno}: {exc.msg}")
        return report

    classes = [n for n in tree.body if isinstance(n, ast.ClassDef)]
    agent_classes = [c for c in classes if _inherits_agent(c)]

    if not agent_classes:
        report.error(
            where,
            "defines no Agent subclass",
            "An agent file contains exactly one class inheriting from Agent "
            "(or a primitive such as AlgorithmAgent).",
        )
        return report
    if len(agent_classes) > 1:
        names = ", ".join(c.name for c in agent_classes)
        report.error(
            where,
            f"defines {len(agent_classes)} agent classes ({names})",
            "Split them into separate files: one agent per file.",
        )

    cls = agent_classes[0]

    methods = {
        n.name: n
        for n in cls.body
        if isinstance(n, (ast.AsyncFunctionDef, ast.FunctionDef))
    }

    if "run" in methods:
        report.error(
            f"{where}:{cls.name}",
            "defines `run`, which belongs to the ADK runner protocol",
            "Rename it to `execute`: `async def execute(self) -> dict:`.",
        )

    entry = methods.get("execute")
    if entry is None and "run" not in methods:
        report.error(
            f"{where}:{cls.name}",
            "has no `execute` method",
            "Implement `async def execute(self) -> dict:`.",
        )
    elif isinstance(entry, ast.FunctionDef):
        report.error(
            f"{where}:{cls.name}",
            "`execute` must be async",
            "Change `def execute` to `async def execute`.",
        )

    for name in _unannotated_attributes(cls):
        if name == "ports":
            report.error(
                f"{where}:{cls.name}",
                "declares `ports` without a type annotation",
                "Write `ports: Ports = Ports(...)`, not `ports = Ports(...)`.",
            )
        else:
            report.error(
                f"{where}:{cls.name}",
                f"class attribute `{name}` has no type annotation",
                "Agents are pydantic models. For a constant write "
                f"`{name}: ClassVar[...] = ...` (import ClassVar from typing); "
                "for configuration, read it from `params` instead.",
            )

    declared = _declared_ports(cls)
    if declared is None:
        report.error(
            f"{where}:{cls.name}",
            "declares no `ports`",
            "Add `ports: Ports = Ports(inputs=[...], outputs=[...])`.",
        )
    elif node is not None:
        expected_in = {p.name for p in node.ports.inputs}
        expected_out = {p.name for p in node.ports.outputs}
        actual_in, actual_out = declared
        if actual_in != expected_in:
            report.error(
                f"{where}:{cls.name}",
                f"input ports {sorted(actual_in)} do not match workflow.json"
                f" {sorted(expected_in)}",
                "The class and the DAG must agree. Update whichever is wrong.",
            )
        if actual_out != expected_out:
            report.error(
                f"{where}:{cls.name}",
                f"output ports {sorted(actual_out)} do not match workflow.json"
                f" {sorted(expected_out)}",
                "The class and the DAG must agree. Update whichever is wrong.",
            )

    report.extend(_check_integrity(tree, where))
    return report

validate_experiment

validate_experiment(project, experiment_id)

Validates an entire experiment including workflow DAG and agent files.

Parameters:

Name	Type	Description	Default
project	Project	The Project workspace instance.	required
experiment_id	str	Identifier of the experiment to validate.	required

Returns:

Type	Description
Report	A combined validation Report.

Source code in python/src/science_adk/validate.py

def validate_experiment(project: Project, experiment_id: str) -> Report:
    """Validates an entire experiment including workflow DAG and agent files.

    Args:
      project: The Project workspace instance.
      experiment_id: Identifier of the experiment to validate.

    Returns:
      A combined validation Report.
    """
    report = Report()

    workflow_path = project.workflow_path(experiment_id)
    if not workflow_path.exists():
        report.error(
            f"{experiment_id}",
            "has no workflow.json",
            "Design the experiment DAG first.",
        )
        return report

    workflow = project.read_workflow(experiment_id)
    report.extend(validate_workflow(workflow))

    for node in workflow.nodes:
        if node.kind == "composite":
            nested = project.experiment_dir(experiment_id) / node.workflow
            if not nested.exists():
                report.error(
                    f"node {node.id}",
                    f"nested workflow {node.workflow} does not exist",
                )
            continue
        report.extend(
            validate_agent_source(project.agent_path(experiment_id, node.module), node)
        )

    return report