Package

Everything exported at the top level.

science_adk

Science ADK — an agent development kit for autonomous scientific research.

Built on Google Agent Development Kit (ADK), Science ADK provides typed scientific ports, machine-recorded provenance ledgers, deterministic gates, and reproducible experiment lifecycles.

Basic Example

Define a scientific agent and run it:

from science_adk import AlgorithmAgent, Port, Ports

class Simulate(AlgorithmAgent):
    '''Integrate the equations of motion.'''

    ports: Ports = Ports(
        inputs=[Port("config", "json")],
        outputs=[Port("trajectory", "dataframe")],
    )

    async def execute(self):
        config = await self.input("config")
        ...
        return {"trajectory": frame}

Agent

Bases: BaseAgent

Base class for scientific agents built on Google ADK.

Subclasses declare :attr:ports and implement :meth:execute. Everything else exists to make that method short, reproducible, and verifiable.

Attributes:

Name	Type	Description
ports	Ports	Typed input and output port declarations.
params	dict[str, Any]	Static parameters from the experiment workflow definition.
purpose	str	Description of what this agent does and why.
kind	str	Primitive role played by this agent (e.g., algorithm, data).

provenance property

provenance

Returns the machine-recorded provenance ledger for this agent.

logs property

logs

Returns recorded execution log entries.

execute async

execute()

Executes the scientific computation.

Returns:

Type	Description
dict[str, Any]	A dictionary mapping declared output port names to their computed values.

Raises:

Type	Description
NotImplementedError	If the subclass does not implement this method.

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

async def execute(self) -> dict[str, Any]:
    """Executes the scientific computation.

    Returns:
      A dictionary mapping declared output port names to their computed values.

    Raises:
      NotImplementedError: If the subclass does not implement this method.
    """
    raise NotImplementedError(
        f"{type(self).__name__} must implement `async def execute(self)`."
    )

bind

bind(**inputs)

Provides input values directly.

Parameters:

Name	Type	Description	Default
**inputs	Any	Keyword arguments representing input port values.	{}

Returns:

Type	Description
Agent	Self for chaining.

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

def bind(self, **inputs: Any) -> Agent:
    """Provides input values directly.

    Args:
      **inputs: Keyword arguments representing input port values.

    Returns:
      Self for chaining.
    """
    self._inputs.update(inputs)
    return self

attach

attach(services)

Attaches execution services (tools, data storage, logging).

Parameters:

Name	Type	Description	Default
services	Any	The execution services provider.	required

Returns:

Type	Description
Agent	Self for chaining.

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

def attach(self, services: Any) -> Agent:
    """Attaches execution services (tools, data storage, logging).

    Args:
      services: The execution services provider.

    Returns:
      Self for chaining.
    """
    self._services = services
    return self

input async

input(port, default=None)

Reads a declared input port.

Parameters:

Name	Type	Description	Default
port	str	The name of the input port.	required
default	Any	Default value if the port carries no value and is optional.	None

Returns:

Type	Description
Any	The hydrated value associated with the port.

Raises:

Type	Description
AgentError	If the port is not declared or a required port has no value.

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

async def input(self, port: str, default: Any = None) -> Any:
    """Reads a declared input port.

    Args:
      port: The name of the input port.
      default: Default value if the port carries no value and is optional.

    Returns:
      The hydrated value associated with the port.

    Raises:
      AgentError: If the port is not declared or a required port has no value.
    """
    declared = self.ports.input(port)
    if declared is None:
        names = ", ".join(p.name for p in self.ports.inputs) or "none declared"
        raise AgentError(
            f"{type(self).__name__} read undeclared input port {port!r}. "
            f"Declared inputs: {names}"
        )

    value = self._inputs.get(port)
    if value is not None and self._services is not None:
        value = self._services.hydrate(value)

    found = value is not None
    self._provenance["inputs_read"].append({"port": port, "found": found})

    if not found:
        if declared.required and default is None:
            raise AgentError(
                f"Required input {port!r} of {self.name!r} has no value. "
                "Ensure an edge in workflow.json feeds this port."
            )
        return default
    return value

param

param(name, default=None)

Reads a static parameter from node configuration.

Parameters:

Name	Type	Description	Default
name	str	Name of the parameter.	required
default	Any	Default fallback value if not specified.	None

Returns:

Type	Description
Any	The parameter value.

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

def param(self, name: str, default: Any = None) -> Any:
    """Reads a static parameter from node configuration.

    Args:
      name: Name of the parameter.
      default: Default fallback value if not specified.

    Returns:
      The parameter value.
    """
    return self.params.get(name, default)

call_tool async

call_tool(name, **arguments)

Calls a tool by name and records provenance.

Parameters:

Name	Type	Description	Default
name	str	Name of the tool.	required
**arguments	Any	Arguments to pass to the tool.	{}

Returns:

Type	Description
Any	The tool execution result.

Raises:

Type	Description
AgentError	If execution services are not attached.

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

async def call_tool(self, name: str, **arguments: Any) -> Any:
    """Calls a tool by name and records provenance.

    Args:
      name: Name of the tool.
      **arguments: Arguments to pass to the tool.

    Returns:
      The tool execution result.

    Raises:
      AgentError: If execution services are not attached.
    """
    if self._services is None:
        raise AgentError(
            f"{self.name}: no tools are attached. This agent is not being "
            "run within an active runtime context."
        )
    try:
        result = await self._services.call_tool(name, arguments)
    except Exception:
        self._provenance["tool_calls"].append({"tool": name, "ok": False})
        raise
    self._provenance["tool_calls"].append({"tool": name, "ok": True})
    return result

fetch async

fetch(url, timeout=60, retries=3)

Fetches a URL with bounded exponential backoff and records provenance.

Parameters:

Name	Type	Description	Default
url	str	The URL to fetch.	required
timeout	int	Timeout in seconds.	60
retries	int	Maximum number of retry attempts.	3

Returns:

Type	Description
str	The decoded response text.

Raises:

Type	Description
AgentError	If the fetch fails after all retry attempts.

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

async def fetch(self, url: str, timeout: int = 60, retries: int = 3) -> str:
    """Fetches a URL with bounded exponential backoff and records provenance.

    Args:
      url: The URL to fetch.
      timeout: Timeout in seconds.
      retries: Maximum number of retry attempts.

    Returns:
      The decoded response text.

    Raises:
      AgentError: If the fetch fails after all retry attempts.
    """
    last: Exception | None = None
    for attempt in range(retries):
        try:
            request = urllib.request.Request(
                url, headers={"User-Agent": "science-adk/0.1"}
            )
            body = await asyncio.to_thread(_read_url, request, timeout)
            self._provenance["urls_fetched"].append({"url": url, "ok": True})
            return body
        except (urllib.error.URLError, TimeoutError, OSError) as exc:
            last = exc
            if attempt < retries - 1:
                await asyncio.sleep(2**attempt)
    self._provenance["urls_fetched"].append({"url": url, "ok": False})
    raise AgentError(f"Failed to fetch {url} after {retries} attempts: {last}")

artifact

artifact(filename)

Reserves a path in this run's data directory and logs file creation.

Parameters:

Name	Type	Description	Default
filename	str	Name of the artifact file.	required

Returns:

Type	Description
Path	Path to the reserved file location.

Raises:

Type	Description
AgentError	If execution services are not attached.

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

def artifact(self, filename: str) -> Path:
    """Reserves a path in this run's data directory and logs file creation.

    Args:
      filename: Name of the artifact file.

    Returns:
      Path to the reserved file location.

    Raises:
      AgentError: If execution services are not attached.
    """
    if self._services is None:
        raise AgentError(
            f"{self.name}: no run directory attached. This agent is not "
            "being run within an active runtime context."
        )
    directory = self._services.data_dir()
    directory.mkdir(parents=True, exist_ok=True)
    self._provenance["files_written"].append({"file": filename})
    return directory / filename

log async

log(message)

Records an informational log message in the provenance trace.

Parameters:

Name	Type	Description	Default
message	str	Message text to record.	required

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

async def log(self, message: str) -> None:
    """Records an informational log message in the provenance trace.

    Args:
      message: Message text to record.
    """
    text = str(message)
    self._logs.append(text)
    if self._services is not None:
        await self._services.log(self.name, text)

validate_outputs

validate_outputs(outputs)

Validates execution output against declared output ports.

Parameters:

Name	Type	Description	Default
outputs	Any	The output dictionary returned by execute.	required

Returns:

Type	Description
dict[str, Any]	The validated outputs dictionary.

Raises:

Type	Description
AgentError	If output types or keys mismatch declared output ports.

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

def validate_outputs(self, outputs: Any) -> dict[str, Any]:
    """Validates execution output against declared output ports.

    Args:
      outputs: The output dictionary returned by `execute`.

    Returns:
      The validated outputs dictionary.

    Raises:
      AgentError: If output types or keys mismatch declared output ports.
    """
    name = type(self).__name__
    if not isinstance(outputs, dict):
        raise AgentError(
            f"{name}.execute must return a dict of output ports, "
            f"got {type(outputs).__name__}."
        )
    declared = {p.name for p in self.ports.outputs}
    missing = sorted(declared - set(outputs))
    if missing:
        raise AgentError(
            f"{name}.execute did not return declared output port(s):"
            f" {', '.join(missing)}."
        )
    extra = sorted(set(outputs) - declared)
    if extra:
        raise AgentError(
            f"{name}.execute returned undeclared output(s): {', '.join(extra)}. "
            "Add them to `ports` or remove them."
        )
    for port in self.ports.outputs:
        if outputs[port.name] is None and port.required:
            raise AgentError(
                f"{name}.execute returned None for required output {port.name!r}."
            )
    return outputs

default_ports classmethod

default_ports()

Returns default ports declared on the class without instantiation.

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

@classmethod
def default_ports(cls) -> Ports:
    """Returns default ports declared on the class without instantiation."""
    return _field_default(cls, "ports") or Ports()

describe classmethod

describe()

Returns a machine-readable summary of the agent class.

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

@classmethod
def describe(cls) -> dict[str, Any]:
    """Returns a machine-readable summary of the agent class."""
    purpose = _field_default(cls, "purpose") or ""
    return {
        "class": cls.__name__,
        "kind": cls.kind,
        "purpose": purpose or (cls.__doc__ or "").strip().split("\n")[0],
        "ports": cls.default_ports().to_dict(),
    }

AgentError

Bases: RuntimeError

Raised when an agent misuses the runtime (bad port, missing input).

Edge dataclass

Edge(source, source_port, target, target_port)

A typed connection carrying an output port into an input port.

Attributes:

Name	Type	Description
source	str	Source node id.
source_port	str	Name of the output port on the source node.
target	str	Target node id.
target_port	str	Name of the input port on the target node.

from_dict classmethod

from_dict(data)

Constructs an Edge from dictionary data.

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Edge:
    """Constructs an Edge from dictionary data."""
    return cls(
        source=data["source"],
        source_port=data["source_port"],
        target=data["target"],
        target_port=data["target_port"],
    )

to_dict

to_dict()

Serializes the Edge to a dictionary.

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

def to_dict(self) -> dict[str, Any]:
    """Serializes the Edge to a dictionary."""
    return asdict(self)

Experiment dataclass

Experiment(id, hypothesis='', rationale='', parent='', generation=0, status='draft', best_run='', best_score=None, created_at=utcnow())

One testable hypothesis and its lineage.

Attributes:

Name	Type	Description
id	str	Unique experiment identifier (e.g., '001-nonlinear-oscillator').
hypothesis	str	Precise testable claim.
rationale	str	Empirical justification for why this hypothesis was chosen.
parent	str	Parent experiment ID in the hypothesis tree.
generation	int	Tree depth level of this experiment.
status	str	Current status (draft, ready, run, superseded, abandoned).
best_run	str	Run ID of the highest scoring run.
best_score	float | None	Highest score achieved.
created_at	str	ISO-8601 creation timestamp.

from_dict classmethod

from_dict(data)

Constructs an Experiment from dictionary data.

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Experiment:
    """Constructs an Experiment from dictionary data."""
    return cls(
        id=data["id"],
        hypothesis=data.get("hypothesis", ""),
        rationale=data.get("rationale", ""),
        parent=data.get("parent", ""),
        generation=int(data.get("generation", 0)),
        status=data.get("status", "draft"),
        best_run=data.get("best_run", ""),
        best_score=data.get("best_score"),
        created_at=data.get("created_at", utcnow()),
    )

to_dict

to_dict()

Serializes the Experiment to a dictionary.

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

def to_dict(self) -> dict[str, Any]:
    """Serializes the Experiment to a dictionary."""
    return _drop_none(asdict(self))

Gate dataclass

Gate(name, passed, detail='')

One deterministic pass/fail check performed by the engine.

Attributes:

Name	Type	Description
name	str	Name of the integrity check.
passed	bool	Whether the check passed.
detail	str	Explanation or supporting context.

to_dict

to_dict()

Serializes the Gate to a dictionary.

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

def to_dict(self) -> dict[str, Any]:
    """Serializes the Gate to a dictionary."""
    return asdict(self)

from_dict classmethod

from_dict(data)

Constructs a Gate from dictionary data.

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Gate:
    """Constructs a Gate from dictionary data."""
    return cls(
        name=data["name"],
        passed=bool(data.get("passed", False)),
        detail=data.get("detail", ""),
    )

Goal dataclass

Goal(question='', background='', target_metric='', target_value=None, constraints=list(), created_at=utcnow())

The overarching research question, success metric, and constraints.

Attributes:

Name	Type	Description
question	str	The scientific question under investigation.
background	str	Contextual background and motivation.
target_metric	str	Quantitative evaluation metric name.
target_value	float | None	Success threshold for the target metric.
constraints	list[str]	List of experimental or methodological constraints.
created_at	str	ISO-8601 creation timestamp.

from_dict classmethod

from_dict(data)

Constructs a Goal from dictionary data.

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Goal:
    """Constructs a Goal from dictionary data."""
    return cls(
        question=data.get("question", ""),
        background=data.get("background", ""),
        target_metric=data.get("target_metric", ""),
        target_value=data.get("target_value"),
        constraints=list(data.get("constraints", [])),
        created_at=data.get("created_at", utcnow()),
    )

to_dict

to_dict()

Serializes the Goal to a dictionary.

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

def to_dict(self) -> dict[str, Any]:
    """Serializes the Goal to a dictionary."""
    return _drop_none(asdict(self))

Learning dataclass

Learning(insight, evidence='', experiment='', run_id='', created_at=utcnow())

An empirically validated insight backed by run evidence.

Attributes:

Name	Type	Description
insight	str	The empirical finding or generalization.
evidence	str	Specific runs or measurements demonstrating the finding.
experiment	str	Experiment identifier where the finding was made.
run_id	str	Run identifier of the supporting run.
created_at	str	ISO-8601 creation timestamp.

from_dict classmethod

from_dict(data)

Constructs a Learning from dictionary data.

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Learning:
    """Constructs a Learning from dictionary data."""
    return cls(
        insight=data["insight"],
        evidence=data.get("evidence", ""),
        experiment=data.get("experiment", ""),
        run_id=data.get("run_id", ""),
        created_at=data.get("created_at", utcnow()),
    )

to_dict

to_dict()

Serializes the Learning to a dictionary.

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

def to_dict(self) -> dict[str, Any]:
    """Serializes the Learning to a dictionary."""
    return _drop_none(asdict(self))

Node dataclass

Node(id, name, kind='algorithm', purpose='', ports=Ports(), module='', tools=list(), params=dict(), workflow='')

One step of an experiment: a named agent with a typed port surface.

Attributes:

Name	Type	Description
id	str	Unique identifier of the node within the workflow DAG.
name	str	Human-readable name for the node.
kind	str	The primitive agent kind (must be one of AGENT_KINDS).
purpose	str	Description of what this node does and why.
ports	Ports	Declared input and output ports.
module	str	Path to the Python file implementing the agent.
tools	list[str]	Advisory list of tools this agent may call.
params	dict[str, Any]	Static parameters passed to the agent.
workflow	str	For composite agents, the path to the nested workflow.

from_dict classmethod

from_dict(data)

Constructs a Node from dictionary data.

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Node:
    """Constructs a Node from dictionary data."""
    return cls(
        id=data["id"],
        name=data.get("name", data["id"]),
        kind=data.get("kind", "algorithm"),
        purpose=data.get("purpose", ""),
        ports=Ports.from_dict(data.get("ports")),
        module=data.get("module", ""),
        tools=list(data.get("tools", [])),
        params=dict(data.get("params", {})),
        workflow=data.get("workflow", ""),
    )

to_dict

to_dict()

Serializes the Node instance to a dictionary.

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

def to_dict(self) -> dict[str, Any]:
    """Serializes the Node instance to a dictionary."""
    return _drop_none(
        {
            "id": self.id,
            "name": self.name,
            "kind": self.kind,
            "purpose": self.purpose,
            "ports": self.ports.to_dict(),
            "module": self.module or None,
            "tools": self.tools or None,
            "params": self.params or None,
            "workflow": self.workflow or None,
        }
    )

NodeRun dataclass

NodeRun(node_id, state='pending', started_at='', finished_at='', duration_s=0.0, outputs=dict(), logs=list(), error='', traceback='', provenance=dict(), fingerprint='')

The execution record of one node.

Attributes:

Name	Type	Description
node_id	str	Identifier of the executing node.
state	str	Execution status (pending, running, passed, failed, cached).
started_at	str	ISO-8601 execution start timestamp.
finished_at	str	ISO-8601 execution finish timestamp.
duration_s	float	Total run duration in seconds.
outputs	dict[str, Any]	Map of output port names to values or dataset references.
logs	list[str]	Informational log messages recorded during execution.
error	str	Error message string if execution failed.
traceback	str	Exception traceback if execution failed.
provenance	dict[str, list[dict[str, Any]]]	Machine-recorded ingress and tool-call evidence.
fingerprint	str	Execution cache hash (code + inputs + parameters).

from_dict classmethod

from_dict(data)

Constructs a NodeRun from dictionary data.

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> NodeRun:
    """Constructs a NodeRun from dictionary data."""
    return cls(
        node_id=data["node_id"],
        state=data.get("state", "pending"),
        started_at=data.get("started_at", ""),
        finished_at=data.get("finished_at", ""),
        duration_s=float(data.get("duration_s", 0.0)),
        outputs=dict(data.get("outputs", {})),
        logs=list(data.get("logs", [])),
        error=data.get("error", ""),
        traceback=data.get("traceback", ""),
        provenance=dict(data.get("provenance", {})),
        fingerprint=data.get("fingerprint", ""),
    )

to_dict

to_dict()

Serializes the NodeRun to a dictionary.

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

def to_dict(self) -> dict[str, Any]:
    """Serializes the NodeRun to a dictionary."""
    return asdict(self)

Port dataclass

Port(name, type='any', description='', required=True)

A single typed input or output port of an agent.

Attributes:

Name	Type	Description
name	str	The unique name of the port on this agent.
type	str	Data type of the port (must be one of PORT_TYPES).
description	str	Explanation of what this port carries.
required	bool	Whether a value is strictly required on this port.

from_dict classmethod

from_dict(data)

Constructs a Port from a dictionary or string shorthand.

Parameters:

Name	Type	Description	Default
data	dict[str, Any] | str	Dictionary containing port specification or string port name.	required

Returns:

Type	Description
Port	A new Port instance.

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

@classmethod
def from_dict(cls, data: dict[str, Any] | str) -> Port:
    """Constructs a Port from a dictionary or string shorthand.

    Args:
      data: Dictionary containing port specification or string port name.

    Returns:
      A new Port instance.
    """
    if isinstance(data, str):
        return cls(name=data)
    return cls(
        name=data["name"],
        type=data.get("type", "any"),
        description=data.get("description", ""),
        required=bool(data.get("required", True)),
    )

to_dict

to_dict()

Serializes the Port to a dictionary.

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

def to_dict(self) -> dict[str, Any]:
    """Serializes the Port to a dictionary."""
    return asdict(self)

Ports dataclass

Ports(inputs=list(), outputs=list())

The complete input and output port surface of an agent.

Attributes:

Name	Type	Description
inputs	list[Port]	List of declared input ports.
outputs	list[Port]	List of declared output ports.

required_inputs property

required_inputs

Returns all input ports marked as required.

input

input(name)

Retrieves an input port by name, or None if not found.

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

def input(self, name: str) -> Port | None:
    """Retrieves an input port by name, or None if not found."""
    return next((p for p in self.inputs if p.name == name), None)

output

output(name)

Retrieves an output port by name, or None if not found.

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

def output(self, name: str) -> Port | None:
    """Retrieves an output port by name, or None if not found."""
    return next((p for p in self.outputs if p.name == name), None)

from_dict classmethod

from_dict(data)

Constructs Ports from dictionary data.

Parameters:

Name	Type	Description	Default
data	dict[str, Any] | None	Dictionary with 'inputs' and 'outputs' lists.	required

Returns:

Type	Description
Ports	A new Ports instance.

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

@classmethod
def from_dict(cls, data: dict[str, Any] | None) -> Ports:
    """Constructs Ports from dictionary data.

    Args:
      data: Dictionary with 'inputs' and 'outputs' lists.

    Returns:
      A new Ports instance.
    """
    data = data or {}
    return cls(
        inputs=[Port.from_dict(p) for p in data.get("inputs", [])],
        outputs=[Port.from_dict(p) for p in data.get("outputs", [])],
    )

to_dict

to_dict()

Serializes the Ports instance to a dictionary.

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

def to_dict(self) -> dict[str, Any]:
    """Serializes the Ports instance to a dictionary."""
    return {
        "inputs": [p.to_dict() for p in self.inputs],
        "outputs": [p.to_dict() for p in self.outputs],
    }

Score dataclass

Score(run_id, gates=list(), pillars=dict(), rationale='', target_metric='', target_value=None, observed_value=None, audited_at='')

The evaluation verdict on one run.

Attributes:

Name	Type	Description
run_id	str	Unique run identifier.
gates	list[Gate]	List of deterministic Gate results.
pillars	dict[str, float]	Judgement pillar scores (in [0, 1]).
rationale	str	Written audit rationale.
target_metric	str	Name of the target metric.
target_value	float | None	Goal threshold target.
observed_value	float | None	Empirically measured value.
audited_at	str	ISO-8601 audit timestamp.

gates_passed property

gates_passed

Returns True if all deterministic gates passed.

failed_gates property

failed_gates

Returns all failed gates.

audited property

audited

Returns True if an audit verdict has been recorded.

value property

value

Calculates overall score in [0.0, 1.0]. Returns 0.0 if gates fail.

meets_target property

meets_target

Returns True if observed value meets or exceeds target value.

from_dict classmethod

from_dict(data)

Constructs a Score from dictionary data.

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Score:
    """Constructs a Score from dictionary data."""
    return cls(
        run_id=data["run_id"],
        gates=[Gate.from_dict(g) for g in data.get("gates", [])],
        pillars={k: float(v) for k, v in (data.get("pillars") or {}).items()},
        rationale=data.get("rationale", ""),
        target_metric=data.get("target_metric", ""),
        target_value=data.get("target_value"),
        observed_value=data.get("observed_value"),
        audited_at=data.get("audited_at", ""),
    )

to_dict

to_dict()

Serializes the Score to a dictionary.

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

def to_dict(self) -> dict[str, Any]:
    """Serializes the Score to a dictionary."""
    return _drop_none(
        {
            "run_id": self.run_id,
            "value": self.value,
            "gates_passed": self.gates_passed,
            "audited": self.audited,
            "gates": [g.to_dict() for g in self.gates],
            "pillars": self.pillars or None,
            "rationale": self.rationale or None,
            "target_metric": self.target_metric or None,
            "target_value": self.target_value,
            "observed_value": self.observed_value,
            "audited_at": self.audited_at or None,
        }
    )

Trace dataclass

Trace(run_id, experiment='', state='pending', started_at=utcnow(), finished_at='', duration_s=0.0, nodes=list(), error='')

The complete record of one workflow execution run.

Attributes:

Name	Type	Description
run_id	str	Unique run identifier.
experiment	str	Experiment identifier.
state	str	Overall workflow state.
started_at	str	ISO-8601 start timestamp.
finished_at	str	ISO-8601 finish timestamp.
duration_s	float	Total execution duration in seconds.
nodes	list[NodeRun]	List of NodeRun records.
error	str	Top-level error message if run aborted.

failed_nodes property

failed_nodes

Returns all NodeRun instances that failed.

ok property

ok

Returns True if the workflow completed successfully.

node

node(node_id)

Retrieves a NodeRun by node ID.

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

def node(self, node_id: str) -> NodeRun | None:
    """Retrieves a NodeRun by node ID."""
    return next((n for n in self.nodes if n.node_id == node_id), None)

from_dict classmethod

from_dict(data)

Constructs a Trace from dictionary data.

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Trace:
    """Constructs a Trace from dictionary data."""
    return cls(
        run_id=data["run_id"],
        experiment=data.get("experiment", ""),
        state=data.get("state", "pending"),
        started_at=data.get("started_at", utcnow()),
        finished_at=data.get("finished_at", ""),
        duration_s=float(data.get("duration_s", 0.0)),
        nodes=[NodeRun.from_dict(n) for n in data.get("nodes", [])],
        error=data.get("error", ""),
    )

to_dict

to_dict()

Serializes the Trace to a dictionary.

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

def to_dict(self) -> dict[str, Any]:
    """Serializes the Trace to a dictionary."""
    return {
        "run_id": self.run_id,
        "experiment": self.experiment,
        "state": self.state,
        "started_at": self.started_at,
        "finished_at": self.finished_at,
        "duration_s": self.duration_s,
        "error": self.error,
        "nodes": [n.to_dict() for n in self.nodes],
    }

Workflow dataclass

Workflow(name='', description='', nodes=list(), edges=list(), created_at=utcnow())

The experiment DAG: what runs, in what order, carrying what data.

Attributes:

Name	Type	Description
name	str	Descriptive name of the workflow.
description	str	Summary of the experiment pipeline.
nodes	list[Node]	List of nodes in the DAG.
edges	list[Edge]	List of edges connecting nodes.
created_at	str	ISO-8601 creation timestamp.

node

node(node_id)

Finds a node by its ID.

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

def node(self, node_id: str) -> Node | None:
    """Finds a node by its ID."""
    return next((n for n in self.nodes if n.id == node_id), None)

parents

parents(node_id)

Returns sorted list of parent node IDs that precede node_id.

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

def parents(self, node_id: str) -> list[str]:
    """Returns sorted list of parent node IDs that precede node_id."""
    return sorted({e.source for e in self.edges if e.target == node_id})

children

children(node_id)

Returns sorted list of child node IDs that consume output from node_id.

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

def children(self, node_id: str) -> list[str]:
    """Returns sorted list of child node IDs that consume output from node_id."""
    return sorted({e.target for e in self.edges if e.source == node_id})

descendants

descendants(node_id)

Returns all node IDs transitively downstream of node_id.

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

def descendants(self, node_id: str) -> set[str]:
    """Returns all node IDs transitively downstream of node_id."""
    seen: set[str] = set()
    stack = list(self.children(node_id))
    while stack:
        current = stack.pop()
        if current in seen:
            continue
        seen.add(current)
        stack.extend(self.children(current))
    return seen

terminal_nodes

terminal_nodes()

Returns node IDs that produce terminal experiment outputs.

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

def terminal_nodes(self) -> list[str]:
    """Returns node IDs that produce terminal experiment outputs."""
    return [n.id for n in self.nodes if not self.children(n.id)]

from_dict classmethod

from_dict(data)

Constructs a Workflow from dictionary data.

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

@classmethod
def from_dict(cls, data: dict[str, Any]) -> Workflow:
    """Constructs a Workflow from dictionary data."""
    return cls(
        name=data.get("name", ""),
        description=data.get("description", ""),
        nodes=[Node.from_dict(n) for n in data.get("nodes", [])],
        edges=[Edge.from_dict(e) for e in data.get("edges", [])],
        created_at=data.get("created_at", utcnow()),
    )

to_dict

to_dict()

Serializes the Workflow to a dictionary.

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

def to_dict(self) -> dict[str, Any]:
    """Serializes the Workflow to a dictionary."""
    return {
        "name": self.name,
        "description": self.description,
        "created_at": self.created_at,
        "nodes": [n.to_dict() for n in self.nodes],
        "edges": [e.to_dict() for e in self.edges],
    }

AlgorithmAgent

Bases: Agent

Base class for algorithm and model implementation agents.

Attributes:

Name	Type	Description
assumptions	list[str]	ClassVar list of approximations and model assumptions made by this method.

assumption_summary classmethod

assumption_summary()

Returns a Markdown summary of declared assumptions.

Source code in python/src/science_adk/primitives/algorithm.py

@classmethod
def assumption_summary(cls) -> str:
    """Returns a Markdown summary of declared assumptions."""
    if not cls.assumptions:
        return "No assumptions declared."
    return "\n".join(f"- {a}" for a in cls.assumptions)

CompositeAgent

Bases: Agent

A node that executes an entire nested workflow DAG.

Attributes:

Name	Type	Description
workflow	str	Relative path to the nested workflow.json file.

execute async

execute()

Executes the nested workflow and surfaces its terminal outputs.

Returns:

Type	Description
dict[str, Any]	Dictionary of outputs mapped to declared output ports.

Raises:

Type	Description
AgentError	If the workflow path is missing, not in runtime, or outputs mismatch.

Source code in python/src/science_adk/primitives/composite.py

async def execute(self) -> dict[str, Any]:
    """Executes the nested workflow and surfaces its terminal outputs.

    Returns:
      Dictionary of outputs mapped to declared output ports.

    Raises:
      AgentError: If the workflow path is missing, not in runtime, or
        outputs mismatch.
    """
    from ..runner import run_nested

    if not self.workflow:
        raise AgentError(
            f"{self.name}: composite node has no `workflow` path. "
            "Set it in workflow.json or on the class."
        )
    if self._services is None:
        raise AgentError(
            f"{self.name}: composite nodes can only run within the runtime."
        )

    inputs = {port.name: await self.input(port.name) for port in self.ports.inputs}
    await self.log(f"entering nested workflow {self.workflow}")

    outputs = await run_nested(self.workflow, inputs, self._services)

    declared = {p.name for p in self.ports.outputs}
    missing = sorted(declared - set(outputs))
    if missing:
        raise AgentError(
            f"{self.name}: nested workflow did not produce declared"
            f" output(s): {', '.join(missing)}. Check terminal nodes of"
            f" {self.workflow}."
        )
    return {name: outputs[name] for name in declared}

ConfigAgent

Bases: Agent

Emits configuration settings, hyperparameters, and physical constants.

settings

settings(**defaults)

Merges default values with workflow params overrides.

Parameters:

Name	Type	Description	Default
**defaults	Any	Default configuration key-value pairs.	{}

Returns:

Type	Description
dict[str, Any]	Combined settings dictionary.

Source code in python/src/science_adk/primitives/config.py

def settings(self, **defaults: Any) -> dict[str, Any]:
    """Merges default values with workflow params overrides.

    Args:
      **defaults: Default configuration key-value pairs.

    Returns:
      Combined settings dictionary.
    """
    return {**defaults, **self.params}

DataAgent

Bases: Agent

Produces or ingests dataset payloads for experiment workflows.

Attributes:

Name	Type	Description
source_kind	str	ClassVar string indicating data provenance ('measured', 'simulated', 'synthetic').

check_frame

check_frame(frame, columns, min_rows=1)

Validates that a DataFrame has required columns and minimum row count.

Parameters:

Name	Type	Description	Default
frame	Any	The DataFrame object to validate.	required
columns	list[str]	List of required column names.	required
min_rows	int	Minimum required number of rows.	1

Returns:

Type	Description
Any	The validated DataFrame.

Raises:

Type	Description
AgentError	If columns are missing or row count is insufficient.

Source code in python/src/science_adk/primitives/data.py

def check_frame(self, frame: Any, columns: list[str], min_rows: int = 1) -> Any:
    """Validates that a DataFrame has required columns and minimum row count.

    Args:
      frame: The DataFrame object to validate.
      columns: List of required column names.
      min_rows: Minimum required number of rows.

    Returns:
      The validated DataFrame.

    Raises:
      AgentError: If columns are missing or row count is insufficient.
    """
    actual = list(getattr(frame, "columns", []))
    missing = [c for c in columns if c not in actual]
    if missing:
        raise AgentError(
            f"{self.name}: produced data is missing column(s) "
            f"{', '.join(missing)}. Present:"
            f" {', '.join(map(str, actual)) or 'none'}"
        )
    rows = len(frame)
    if rows < min_rows:
        raise AgentError(
            f"{self.name}: produced only {rows} row(s), expected at least"
            f" {min_rows}."
        )
    return frame

EvaluationAgent

Bases: Agent

Measures the experiment's quantitative outcome.

report

report(metric, value, detail=None, higher_is_better=True)

Constructs standardized evaluation output port dictionary.

Parameters:

Name	Type	Description	Default
metric	str	Name of the evaluated metric.	required
value	float	Quantitative numeric result.	required
detail	dict[str, Any] | None	Optional supporting dictionary of details.	None
higher_is_better	bool	Optimization direction indicator.	True

Returns:

Type	Description
dict[str, Any]	Output port dictionary for execute().

Raises:

Type	Description
AgentError	If the value is non-numeric or NaN.

Source code in python/src/science_adk/primitives/evaluation.py

def report(
    self,
    metric: str,
    value: float,
    detail: dict[str, Any] | None = None,
    higher_is_better: bool = True,
) -> dict[str, Any]:
    """Constructs standardized evaluation output port dictionary.

    Args:
      metric: Name of the evaluated metric.
      value: Quantitative numeric result.
      detail: Optional supporting dictionary of details.
      higher_is_better: Optimization direction indicator.

    Returns:
      Output port dictionary for execute().

    Raises:
      AgentError: If the value is non-numeric or NaN.
    """
    try:
        numeric = float(value)
    except (TypeError, ValueError):
        raise AgentError(
            f"{self.name}: metric {metric!r} must be a number, got {value!r}."
        ) from None

    if numeric != numeric:
        raise AgentError(
            f"{self.name}: metric {metric!r} evaluated to NaN. A NaN result"
            " is a failed computation, not a score."
        )

    payload = dict(detail or {})
    payload["higher_is_better"] = higher_is_better
    return {"metric": metric, "value": numeric, "detail": payload}

TrainingAgent

Bases: Agent

Fits models to data and manages reproducible train/test splits.

Attributes:

Name	Type	Description
seed	int | None	Seed for reproducible random splitting and initialization.

split

split(data, test_fraction=0.2, shuffle=True)

Splits dataset into (train, test) subsets deterministically under seed.

Parameters:

Name	Type	Description	Default
data	Any	Data sequence or DataFrame.	required
test_fraction	float	Fraction of rows allocated to the test set (0 < frac < 1).	0.2
shuffle	bool	Whether to shuffle before splitting.	True

Returns:

Type	Description
tuple[Any, Any]	Tuple of (train_data, test_data).

Raises:

Type	Description
AgentError	If test_fraction is invalid or data has fewer than 2 elements.

Source code in python/src/science_adk/primitives/training.py

def split(
    self,
    data: Any,
    test_fraction: float = 0.2,
    shuffle: bool = True,
) -> tuple[Any, Any]:
    """Splits dataset into (train, test) subsets deterministically under seed.

    Args:
      data: Data sequence or DataFrame.
      test_fraction: Fraction of rows allocated to the test set (0 < frac < 1).
      shuffle: Whether to shuffle before splitting.

    Returns:
      Tuple of (train_data, test_data).

    Raises:
      AgentError: If test_fraction is invalid or data has fewer than 2 elements.
    """
    if not 0.0 < test_fraction < 1.0:
        raise AgentError(
            f"test_fraction must be between 0 and 1, got {test_fraction}"
        )

    total = len(data)
    if total < 2:
        raise AgentError(
            f"{self.name}: need at least 2 rows to split, got {total}."
        )

    indices = list(range(total))
    if shuffle:
        random.Random(self.seed).shuffle(indices)

    cut = max(1, int(round(total * (1.0 - test_fraction))))
    train_idx, test_idx = indices[:cut], indices[cut:]

    if hasattr(data, "iloc"):
        return data.iloc[train_idx], data.iloc[test_idx]
    return [data[i] for i in train_idx], [data[i] for i in test_idx]

seed_everything

seed_everything()

Sets random seed across Python standard library, NumPy, and PyTorch.

Source code in python/src/science_adk/primitives/training.py

def seed_everything(self) -> None:
    """Sets random seed across Python standard library, NumPy, and PyTorch."""
    if self.seed is None:
        return
    random.seed(self.seed)
    try:
        import numpy

        numpy.random.seed(self.seed)
    except ImportError:
        pass
    try:
        import torch

        torch.manual_seed(self.seed)
    except ImportError:
        pass

VisualizationAgent

Bases: Agent

Renders visual plots and figures from results produced upstream.

figure_path

figure_path(name, extension='png')

Reserves a path for a figure file inside this run's data directory.

Parameters:

Name	Type	Description	Default
name	str	Base figure name.	required
extension	str	File extension without leading dot.	'png'

Returns:

Type	Description
Path	Path to the reserved figure file.

Source code in python/src/science_adk/primitives/visualization.py

def figure_path(self, name: str, extension: str = "png") -> Path:
    """Reserves a path for a figure file inside this run's data directory.

    Args:
      name: Base figure name.
      extension: File extension without leading dot.

    Returns:
      Path to the reserved figure file.
    """
    filename = name if name.endswith(f".{extension}") else f"{name}.{extension}"
    return self.artifact(filename)

save_figure

save_figure(figure, name, dpi=150)

Saves a matplotlib figure and returns its run-relative path.

Parameters:

Name	Type	Description	Default
figure	Any	Matplotlib Figure object.	required
name	str	Name of the figure.	required
dpi	int	Resolution in dots per inch.	150

Returns:

Type	Description
str	Run-relative path string.

Source code in python/src/science_adk/primitives/visualization.py

def save_figure(self, figure: Any, name: str, dpi: int = 150) -> str:
    """Saves a matplotlib figure and returns its run-relative path.

    Args:
      figure: Matplotlib Figure object.
      name: Name of the figure.
      dpi: Resolution in dots per inch.

    Returns:
      Run-relative path string.
    """
    path = self.figure_path(name)
    figure.savefig(path, dpi=dpi, bbox_inches="tight")
    try:
        import matplotlib.pyplot

        matplotlib.pyplot.close(figure)
    except ImportError:
        pass
    return f"data/{path.name}"

RunError

Bases: RuntimeError

Raised when an experiment cannot be run.

Runner dataclass

Runner(project, experiment_id, tools=None, verbose=True)

Orchestrates experiment execution and persists provenance traces.

Attributes:

Name	Type	Description
project	Project	The active scientific research Project workspace.
experiment_id	str	Target experiment identifier.
tools	ToolRegistry | None	Tool registry for tools available to the experiment.
verbose	bool	Whether to log execution progress to stdout.

run

run(only=None, reuse=None)

Synchronously executes the experiment workflow.

Parameters:

Name	Type	Description	Default
only	list[str] | None	Optional list of node IDs to restrict execution to.	None
reuse	str | None	Optional prior run ID to reuse cached node outputs from.	None

Returns:

Type	Description
Trace	Completed Trace object.

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

def run(
    self,
    only: list[str] | None = None,
    reuse: str | None = None,
) -> Trace:
    """Synchronously executes the experiment workflow.

    Args:
      only: Optional list of node IDs to restrict execution to.
      reuse: Optional prior run ID to reuse cached node outputs from.

    Returns:
      Completed Trace object.
    """
    return asyncio.run(self.run_async(only=only, reuse=reuse))

run_async async

run_async(only=None, reuse=None)

Asynchronously executes the experiment workflow.

Parameters:

Name	Type	Description	Default
only	list[str] | None	Optional list of node IDs to restrict execution to.	None
reuse	str | None	Optional prior run ID to reuse cached node outputs from.	None

Returns:

Type	Description
Trace	Completed Trace object.

Raises:

Type	Description
RunError	If the workflow fails validation.

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

async def run_async(
    self,
    only: list[str] | None = None,
    reuse: str | None = None,
) -> Trace:
    """Asynchronously executes the experiment workflow.

    Args:
      only: Optional list of node IDs to restrict execution to.
      reuse: Optional prior run ID to reuse cached node outputs from.

    Returns:
      Completed Trace object.

    Raises:
      RunError: If the workflow fails validation.
    """
    workflow = self.project.read_workflow(self.experiment_id)

    report = validate_workflow(workflow)
    if not report.ok:
        raise RunError(
            "Workflow is not valid, so nothing was run:\n" + report.render()
        )

    run_id = self.project.new_run_id(self.experiment_id)
    run_dir = self.project.run_dir(self.experiment_id, run_id)
    run_dir.mkdir(parents=True, exist_ok=True)

    services = RunServices(
        run_dir=run_dir,
        tools=self.tools,
        experiment_dir=self.project.experiment_dir(self.experiment_id),
        on_log=self._print if self.verbose else None,
    )

    trace = Trace(run_id=run_id, experiment=self.experiment_id, state="running")
    self.project.write_trace(self.experiment_id, trace)

    cached = self._load_cache(reuse) if reuse else {}
    targets = self._targets(workflow, only)
    started = time.monotonic()

    for node_id in self._execution_order(workflow):
        node = workflow.node(node_id)
        assert node is not None

        if node_id not in targets:
            record = cached.get(node_id)
            if record is not None:
                self._adopt(record, node_id, trace, workflow)
                continue

        record = await self._run_node(node, workflow, services, cached)
        trace.nodes.append(record)
        self.project.write_trace(self.experiment_id, trace)

        if record.state == "failed":
            trace.state = "failed"
            trace.error = f"Node {node_id!r} failed: {record.error}"
            break
    else:
        trace.state = "passed"

    trace.finished_at = utcnow()
    trace.duration_s = round(time.monotonic() - started, 3)
    self.project.write_trace(self.experiment_id, trace)
    self.tools.close()
    return trace

ToolError

Bases: RuntimeError

Raised when a tool cannot be resolved or fails during execution.

ToolRegistry

ToolRegistry(providers, root)

Aggregates and resolves tools across all configured providers.

Source code in python/src/science_adk/tools/registry.py

def __init__(self, providers: list[ToolProvider], root: Path):
    self.root = Path(root)
    self._backends: dict[str, Any] = {}
    self._index: dict[str, ToolSpec] | None = None
    for provider in providers:
        if provider.enabled:
            self._backends[provider.name] = self._build(provider)

from_project classmethod

from_project(project)

Constructs a ToolRegistry from a Project instance.

Source code in python/src/science_adk/tools/registry.py

@classmethod
def from_project(cls, project) -> ToolRegistry:
    """Constructs a ToolRegistry from a Project instance."""
    return cls(project.config().providers, project.root)

specs

specs(refresh=False)

Returns all ToolSpecs from all active providers.

Source code in python/src/science_adk/tools/registry.py

def specs(self, refresh: bool = False) -> list[ToolSpec]:
    """Returns all ToolSpecs from all active providers."""
    if self._index is None or refresh:
        index: dict[str, ToolSpec] = {}
        for name, backend in self._backends.items():
            try:
                for spec in backend.list_tools():
                    index[spec.qualified] = spec
            except ToolError as exc:
                print(
                    f"warning: tool provider {name!r} unavailable: {exc}",
                    file=sys.stderr,
                )
        self._index = index
    return list(self._index.values())

resolve

resolve(name)

Resolves a tool by qualified or unqualified name.

Parameters:

Name	Type	Description	Default
name	str	Qualified ('provider.tool') or bare tool name.	required

Returns:

Type	Description
ToolSpec	Matching ToolSpec.

Raises:

Type	Description
ToolError	If tool is missing or name is ambiguous.

Source code in python/src/science_adk/tools/registry.py

def resolve(self, name: str) -> ToolSpec:
    """Resolves a tool by qualified or unqualified name.

    Args:
      name: Qualified ('provider.tool') or bare tool name.

    Returns:
      Matching ToolSpec.

    Raises:
      ToolError: If tool is missing or name is ambiguous.
    """
    specs = self.specs()
    if "." in name:
        for spec in specs:
            if spec.qualified == name:
                return spec
    matches = [s for s in specs if s.name == name]
    if len(matches) == 1:
        return matches[0]
    if len(matches) > 1:
        options = ", ".join(sorted(m.qualified for m in matches))
        raise ToolError(
            f"Tool name {name!r} is ambiguous. Qualify it as one of: {options}"
        )
    available = sorted(s.name for s in specs)
    hint = ", ".join(available[:12]) or "none configured"
    raise ToolError(f"Unknown tool {name!r}. Available: {hint}")

call

call(name, arguments)

Invokes a tool by name with arguments.

Source code in python/src/science_adk/tools/registry.py

def call(self, name: str, arguments: dict[str, Any]) -> Any:
    """Invokes a tool by name with arguments."""
    spec = self.resolve(name)
    backend = self._backends[spec.provider]
    return backend.call(spec.name, arguments)

close

close()

Closes all active provider backends.

Source code in python/src/science_adk/tools/registry.py

def close(self) -> None:
    """Closes all active provider backends."""
    for backend in self._backends.values():
        close_func = getattr(backend, "close", None)
        if close_func:
            try:
                close_func()
            except Exception:
                pass

ToolSpec dataclass

ToolSpec(name, provider, description='', parameters=dict())

Specification of a tool callable by agents.

Attributes:

Name	Type	Description
name	str	Tool name.
provider	str	Name of the provider providing the tool.
description	str	Explanation of what the tool does.
parameters	dict[str, Any]	JSON schema dictionary describing input arguments.

qualified property

qualified

Returns the fully qualified name (provider.name).

signature

signature()

Returns a Python-like signature representation for prompting.

Source code in python/src/science_adk/tools/registry.py

def signature(self) -> str:
    """Returns a Python-like signature representation for prompting."""
    props = (self.parameters or {}).get("properties", {}) or {}
    required = set((self.parameters or {}).get("required", []) or [])
    args = []
    for pname, schema in props.items():
        ptype = schema.get("type", "Any")
        hint = {
            "string": "str",
            "integer": "int",
            "number": "float",
            "boolean": "bool",
            "array": "list",
            "object": "dict",
        }.get(ptype, "Any")
        args.append(pname if pname in required else f"{pname}: {hint} = ...")
    return f"{self.name}({', '.join(args)})"

to_dict

to_dict()

Serializes the ToolSpec to a dictionary.

Source code in python/src/science_adk/tools/registry.py

def to_dict(self) -> dict[str, Any]:
    """Serializes the ToolSpec to a dictionary."""
    return {
        "name": self.name,
        "provider": self.provider,
        "description": self.description,
        "parameters": self.parameters,
    }

Project

Project(root)

Manages files and directories in a Science ADK research workspace.

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

def __init__(self, root: Path | str):
    self.root = Path(root).resolve()

exists property

exists

Returns True if science.toml exists in this project root.

config_path property

config_path

Path to science.toml.

goal_path property

goal_path

Path to GOAL.md.

learnings_path property

learnings_path

Path to LEARNINGS.md.

campaign_path property

campaign_path

Path to CAMPAIGN.md.

research_dir property

research_dir

Path to research/ directory.

tools_dir property

tools_dir

Path to tools/ directory.

find classmethod

find(start=None)

Discovers the root Project by walking up parent directories.

Parameters:

Name	Type	Description	Default
start	Path | str | None	Directory to start searching from (defaults to cwd).	None

Returns:

Type	Description
Project	Discovered Project instance.

Raises:

Type	Description
ProjectError	If no science.toml is found.

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

@classmethod
def find(cls, start: Path | str | None = None) -> Project:
    """Discovers the root Project by walking up parent directories.

    Args:
      start: Directory to start searching from (defaults to cwd).

    Returns:
      Discovered Project instance.

    Raises:
      ProjectError: If no science.toml is found.
    """
    current = Path(start or Path.cwd()).resolve()
    for candidate in [current, *current.parents]:
        if (candidate / CONFIG_FILE).exists():
            return cls(candidate)
    raise ProjectError(
        f"No {CONFIG_FILE} found in {current} or any parent directory. "
        "Run `science-adk init` to create a research project here."
    )

experiment_dir

experiment_dir(experiment_id)

Returns the directory for a specific experiment ID.

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

def experiment_dir(self, experiment_id: str) -> Path:
    """Returns the directory for a specific experiment ID."""
    return self.research_dir / experiment_id

runs_dir

runs_dir(experiment_id)

Returns the runs/ directory for a specific experiment ID.

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

def runs_dir(self, experiment_id: str) -> Path:
    """Returns the runs/ directory for a specific experiment ID."""
    return self.experiment_dir(experiment_id) / "runs"

run_dir

run_dir(experiment_id, run_id)

Returns the specific run directory.

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

def run_dir(self, experiment_id: str, run_id: str) -> Path:
    """Returns the specific run directory."""
    return self.runs_dir(experiment_id) / run_id

rel

rel(path)

Formats path relative to the project root for display.

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

def rel(self, path: Path) -> str:
    """Formats path relative to the project root for display."""
    try:
        return str(path.relative_to(self.root))
    except ValueError:
        return str(path)

config

config()

Loads and returns the Project configuration.

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

def config(self) -> Config:
    """Loads and returns the Project configuration."""
    return Config.load(self.config_path)

read_goal

read_goal()

Reads and parses GOAL.md.

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

def read_goal(self) -> Goal:
    """Reads and parses GOAL.md."""
    if not self.goal_path.exists():
        return Goal()
    meta, body = parse_frontmatter(self.goal_path.read_text(encoding="utf-8"))
    constraints = [
        line.lstrip("-* ").strip()
        for line in _section(body, "Constraints").splitlines()
        if line.strip().startswith(("-", "*"))
    ]
    return Goal(
        question=_text(meta.get("question")) or _section(body, "Question"),
        background=_section(body, "Background"),
        target_metric=_text(meta.get("target_metric")),
        target_value=meta.get("target_value"),
        constraints=constraints,
    )

write_goal

write_goal(goal)

Serializes and writes Goal data to GOAL.md.

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

def write_goal(self, goal: Goal) -> Path:
    """Serializes and writes Goal data to GOAL.md."""
    meta = render_frontmatter(
        {
            "question": goal.question,
            "target_metric": goal.target_metric,
            "target_value": goal.target_value,
        }
    )
    parts = [
        meta,
        "",
        "# Research Goal",
        "",
        "## Question",
        "",
        goal.question or "_Not set._",
    ]
    if goal.background:
        parts += ["", "## Background", "", goal.background]
    if goal.constraints:
        parts += ["", "## Constraints", ""]
        parts += [f"- {c}" for c in goal.constraints]
    write_text(self.goal_path, "\n".join(parts))
    return self.goal_path

experiment_ids

experiment_ids()

Returns sorted list of experiment directory IDs.

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

def experiment_ids(self) -> list[str]:
    """Returns sorted list of experiment directory IDs."""
    if not self.research_dir.exists():
        return []
    return sorted(
        d.name
        for d in self.research_dir.iterdir()
        if d.is_dir() and (d / "HYPOTHESIS.md").exists()
    )

experiments

experiments()

Returns all Experiment objects in the project.

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

def experiments(self) -> list[Experiment]:
    """Returns all Experiment objects in the project."""
    return [self.read_experiment(eid) for eid in self.experiment_ids()]

next_experiment_id

next_experiment_id(hypothesis)

Generates a sequential experiment ID slug.

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

def next_experiment_id(self, hypothesis: str) -> str:
    """Generates a sequential experiment ID slug."""
    existing = self.experiment_ids()
    number = 1
    for eid in existing:
        head = eid.split("-", 1)[0]
        if head.isdigit():
            number = max(number, int(head) + 1)
    return f"{number:03d}-{slugify(hypothesis, 40)}"

read_experiment

read_experiment(experiment_id)

Reads and parses an experiment's HYPOTHESIS.md file.

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

def read_experiment(self, experiment_id: str) -> Experiment:
    """Reads and parses an experiment's HYPOTHESIS.md file."""
    path = self.experiment_dir(experiment_id) / "HYPOTHESIS.md"
    if not path.exists():
        raise ProjectError(
            f"Unknown experiment {experiment_id!r} (expected {self.rel(path)})."
        )
    meta, body = parse_frontmatter(path.read_text(encoding="utf-8"))
    return Experiment(
        id=experiment_id,
        hypothesis=(
            str(meta.get("hypothesis", "")) or _section(body, "Hypothesis")
        ),
        rationale=_section(body, "Rationale"),
        parent=str(meta.get("parent", "") or ""),
        generation=int(meta.get("generation", 0) or 0),
        status=str(meta.get("status", "draft")),
        best_run=str(meta.get("best_run", "") or ""),
        best_score=meta.get("best_score"),
        created_at=str(meta.get("created_at", utcnow())),
    )

write_experiment

write_experiment(experiment)

Writes an Experiment model to HYPOTHESIS.md.

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

def write_experiment(self, experiment: Experiment) -> Path:
    """Writes an Experiment model to HYPOTHESIS.md."""
    path = self.experiment_dir(experiment.id) / "HYPOTHESIS.md"
    meta = render_frontmatter(
        {
            "hypothesis": experiment.hypothesis,
            "parent": experiment.parent,
            "generation": experiment.generation,
            "status": experiment.status,
            "best_run": experiment.best_run,
            "best_score": experiment.best_score,
            "created_at": experiment.created_at,
        }
    )
    parts = [
        meta,
        "",
        f"# {experiment.id}",
        "",
        "## Hypothesis",
        "",
        experiment.hypothesis or "_Not set._",
    ]
    if experiment.rationale:
        parts += ["", "## Rationale", "", experiment.rationale]
    if experiment.parent:
        parts += [
            "",
            "## Lineage",
            "",
            f"Descends from `{experiment.parent}` (generation"
            f" {experiment.generation}).",
        ]
    write_text(path, "\n".join(parts))
    return path

delete_experiment

delete_experiment(experiment_id)

Deletes an experiment directory from disk.

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

def delete_experiment(self, experiment_id: str) -> None:
    """Deletes an experiment directory from disk."""
    shutil.rmtree(self.experiment_dir(experiment_id), ignore_errors=True)

workflow_path

workflow_path(experiment_id)

Returns path to workflow.json for an experiment.

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

def workflow_path(self, experiment_id: str) -> Path:
    """Returns path to workflow.json for an experiment."""
    return self.experiment_dir(experiment_id) / "workflow.json"

read_workflow

read_workflow(experiment_id)

Reads workflow.json for an experiment.

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

def read_workflow(self, experiment_id: str) -> Workflow:
    """Reads workflow.json for an experiment."""
    return Workflow.from_dict(read_json(self.workflow_path(experiment_id)))

write_workflow

write_workflow(experiment_id, workflow)

Writes workflow.json for an experiment.

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

def write_workflow(self, experiment_id: str, workflow: Workflow) -> Path:
    """Writes workflow.json for an experiment."""
    path = self.workflow_path(experiment_id)
    write_json(path, workflow.to_dict())
    return path

agent_path

agent_path(experiment_id, module)

Returns the path to an agent Python source file.

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

def agent_path(self, experiment_id: str, module: str) -> Path:
    """Returns the path to an agent Python source file."""
    return self.experiment_dir(experiment_id) / module

new_run_id

new_run_id(experiment_id)

Allocates a unique chronological run ID.

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

def new_run_id(self, experiment_id: str) -> str:
    """Allocates a unique chronological run ID."""
    stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
    run_id, suffix = stamp, 1
    while self.run_dir(experiment_id, run_id).exists():
        suffix += 1
        run_id = f"{stamp}-{suffix}"
    return run_id

run_ids

run_ids(experiment_id)

Returns sorted list of run IDs for an experiment.

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

def run_ids(self, experiment_id: str) -> list[str]:
    """Returns sorted list of run IDs for an experiment."""
    runs = self.runs_dir(experiment_id)
    if not runs.exists():
        return []
    return sorted(d.name for d in runs.iterdir() if d.is_dir())

latest_run_id

latest_run_id(experiment_id)

Returns the most recent run ID for an experiment.

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

def latest_run_id(self, experiment_id: str) -> str:
    """Returns the most recent run ID for an experiment."""
    ids = self.run_ids(experiment_id)
    if not ids:
        raise ProjectError(
            f"Experiment {experiment_id!r} has no runs yet. Run"
            " `science-adk run` first."
        )
    return ids[-1]

read_trace

read_trace(experiment_id, run_id)

Reads trace.json for a specific run.

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

def read_trace(self, experiment_id: str, run_id: str) -> Trace:
    """Reads trace.json for a specific run."""
    return Trace.from_dict(
        read_json(self.run_dir(experiment_id, run_id) / "trace.json")
    )

write_trace

write_trace(experiment_id, trace)

Writes trace.json for a specific run.

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

def write_trace(self, experiment_id: str, trace: Trace) -> Path:
    """Writes trace.json for a specific run."""
    path = self.run_dir(experiment_id, trace.run_id) / "trace.json"
    write_json(path, trace.to_dict())
    return path

score_path

score_path(experiment_id, run_id)

Returns path to score.json for a run.

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

def score_path(self, experiment_id: str, run_id: str) -> Path:
    """Returns path to score.json for a run."""
    return self.run_dir(experiment_id, run_id) / "score.json"

read_score

read_score(experiment_id, run_id)

Reads score.json for a run.

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

def read_score(self, experiment_id: str, run_id: str) -> Score:
    """Reads score.json for a run."""
    return Score.from_dict(read_json(self.score_path(experiment_id, run_id)))

write_score

write_score(experiment_id, score)

Writes score.json for a run.

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

def write_score(self, experiment_id: str, score: Score) -> Path:
    """Writes score.json for a run."""
    path = self.score_path(experiment_id, score.run_id)
    write_json(path, score.to_dict())
    return path

data_dir

data_dir(experiment_id, run_id)

Returns the data/ directory path for a run.

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

def data_dir(self, experiment_id: str, run_id: str) -> Path:
    """Returns the data/ directory path for a run."""
    return self.run_dir(experiment_id, run_id) / "data"

iter_runs

iter_runs(experiment_id)

Yields (run_id, trace) tuples for an experiment.

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

def iter_runs(self, experiment_id: str) -> Iterator[tuple[str, Trace]]:
    """Yields (run_id, trace) tuples for an experiment."""
    for run_id in self.run_ids(experiment_id):
        try:
            yield run_id, self.read_trace(experiment_id, run_id)
        except ProjectError:
            continue

best_score

best_score(experiment_id)

Finds the highest-scoring audited run for an experiment.

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

def best_score(self, experiment_id: str) -> tuple[str, Score] | None:
    """Finds the highest-scoring audited run for an experiment."""
    best: tuple[str, Score] | None = None
    for run_id in self.run_ids(experiment_id):
        path = self.score_path(experiment_id, run_id)
        if not path.exists():
            continue
        score = Score.from_dict(read_json(path))
        if best is None or score.value > best[1].value:
            best = (run_id, score)
    return best

append_learning

append_learning(learning)

Appends an empirical insight to LEARNINGS.md.

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

def append_learning(self, learning: Learning) -> Path:
    """Appends an empirical insight to LEARNINGS.md."""
    path = self.learnings_path
    if not path.exists():
        write_text(
            path,
            "# Learnings\n\n"
            "Empirical insights from this project. Append-only: each entry "
            "cites the run that produced it.\n",
        )
    cite = " ".join(
        part
        for part in (
            f"`{learning.experiment}`" if learning.experiment else "",
            f"run `{learning.run_id}`" if learning.run_id else "",
        )
        if part
    )
    entry = [f"\n## {learning.created_at}", ""]
    entry.append(learning.insight)
    if learning.evidence:
        entry += ["", f"**Evidence:** {learning.evidence}"]
    if cite:
        entry += ["", f"**Source:** {cite}"]
    with path.open("a", encoding="utf-8") as handle:
        handle.write("\n".join(entry) + "\n")
    return path

read_learnings

read_learnings()

Parses LEARNINGS.md into Learning objects.

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

def read_learnings(self) -> list[Learning]:
    """Parses LEARNINGS.md into Learning objects."""
    if not self.learnings_path.exists():
        return []
    text = self.learnings_path.read_text(encoding="utf-8")
    out: list[Learning] = []
    for block in re.split(r"^##\s+", text, flags=re.MULTILINE)[1:]:
        lines = block.strip().splitlines()
        if not lines:
            continue
        created = lines[0].strip()
        body = "\n".join(lines[1:]).strip()
        insight = body.split("\n**")[0].strip()
        evidence = ""
        match = re.search(r"\*\*Evidence:\*\*\s*(.+)", body)
        if match:
            evidence = match.group(1).strip()
        out.append(Learning(insight=insight, evidence=evidence, created_at=created))
    return out

ProjectError

Bases: RuntimeError

Raised when the project workspace is missing, invalid, or corrupted.

load_agent_class

load_agent_class(path)

Imports a single-class agent file and returns its Agent subclass.

Parameters:

Name	Type	Description	Default
path	Path	Path to the agent Python file.	required

Returns:

Type	Description
type[Agent]	The single Agent subclass defined in the file.

Raises:

Type	Description
AgentError	If the file does not exist, fails to import, or defines zero or multiple Agent subclasses.

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

def load_agent_class(path: Path) -> type[Agent]:
    """Imports a single-class agent file and returns its Agent subclass.

    Args:
      path: Path to the agent Python file.

    Returns:
      The single Agent subclass defined in the file.

    Raises:
      AgentError: If the file does not exist, fails to import, or defines zero or
        multiple Agent subclasses.
    """
    if not path.exists():
        raise AgentError(f"Agent file not found: {path}")

    spec = importlib.util.spec_from_file_location(
        f"_science_adk_agent_{path.stem}", path
    )
    if spec is None or spec.loader is None:
        raise AgentError(f"Cannot load agent file: {path}")
    module = importlib.util.module_from_spec(spec)
    try:
        spec.loader.exec_module(module)
    except Exception as exc:
        raise AgentError(
            f"Error importing {path.name}: {type(exc).__name__}: {exc}"
        ) from exc

    candidates = [
        obj
        for obj in vars(module).values()
        if isinstance(obj, type)
        and issubclass(obj, Agent)
        and obj is not Agent
        and obj.__module__ == module.__name__
    ]
    leaves = [
        c
        for c in candidates
        if not any(other is not c and issubclass(other, c) for other in candidates)
    ]

    if not leaves:
        raise AgentError(
            f"{path.name} defines no Agent subclass. "
            "An agent file must contain exactly one class inheriting from Agent."
        )
    if len(leaves) > 1:
        names = ", ".join(sorted(c.__name__ for c in leaves))
        raise AgentError(
            f"{path.name} defines {len(leaves)} Agent subclasses ({names}). "
            "An agent file must contain exactly one."
        )
    return leaves[0]

compute_gates

compute_gates(trace, workflow)

Computes all deterministic integrity gates against a completed run trace.

Parameters:

Name	Type	Description	Default
trace	Trace	The completed Trace record.	required
workflow	Workflow	The Workflow specification that was executed.	required

Returns:

Type	Description
list[Gate]	A list of Gate results.

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

def compute_gates(trace: Trace, workflow: Workflow) -> list[Gate]:
    """Computes all deterministic integrity gates against a completed run trace.

    Args:
      trace: The completed Trace record.
      workflow: The Workflow specification that was executed.

    Returns:
      A list of Gate results.
    """
    return [
        _gate_completed(trace),
        _gate_all_nodes_ran(trace, workflow),
        _gate_measured(trace, workflow),
        _gate_evaluation_is_numeric(trace, workflow),
        _gate_produced_artifacts(trace),
        _gate_tool_claims_hold(trace, workflow),
    ]

record_audit

record_audit(score, pillars, rationale)

Records audit judgements and rationale onto a Score object.

Parameters:

Name	Type	Description	Default
score	Score	The target Score object to augment.	required
pillars	dict[str, float]	Dictionary mapping SCORE_PILLARS to scores in [0.0, 1.0].	required
rationale	str	Written explanation supporting the evaluation.	required

Returns:

Type	Description
Score	The updated Score instance.

Raises:

Type	Description
ValueError	If pillars are unknown, missing, or rationale is empty.

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

def record_audit(
    score: Score,
    pillars: dict[str, float],
    rationale: str,
) -> Score:
    """Records audit judgements and rationale onto a Score object.

    Args:
      score: The target Score object to augment.
      pillars: Dictionary mapping SCORE_PILLARS to scores in [0.0, 1.0].
      rationale: Written explanation supporting the evaluation.

    Returns:
      The updated Score instance.

    Raises:
      ValueError: If pillars are unknown, missing, or rationale is empty.
    """
    unknown = sorted(set(pillars) - set(SCORE_PILLARS))
    if unknown:
        raise ValueError(
            f"Unknown pillar(s): {', '.join(unknown)}. Expected:"
            f" {', '.join(SCORE_PILLARS)}"
        )
    missing = sorted(set(SCORE_PILLARS) - set(pillars))
    if missing:
        raise ValueError(f"Missing rating for pillar(s): {', '.join(missing)}")
    if not rationale.strip():
        raise ValueError("An audit needs a rationale explaining the ratings.")

    score.pillars = {k: max(0.0, min(1.0, float(v))) for k, v in pillars.items()}
    score.rationale = rationale.strip()
    score.audited_at = utcnow()
    return score

score_run

score_run(trace, workflow, target_metric='', target_value=None)

Constructs the deterministic score structure for a completed run.

Parameters:

Name	Type	Description	Default
trace	Trace	The completed Trace.	required
workflow	Workflow	The Workflow specification.	required
target_metric	str	The metric name being targeted.	''
target_value	float | None	The target value threshold.	None

Returns:

Type	Description
Score	A populated Score instance.

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

def score_run(
    trace: Trace,
    workflow: Workflow,
    target_metric: str = "",
    target_value: float | None = None,
) -> Score:
    """Constructs the deterministic score structure for a completed run.

    Args:
      trace: The completed Trace.
      workflow: The Workflow specification.
      target_metric: The metric name being targeted.
      target_value: The target value threshold.

    Returns:
      A populated Score instance.
    """
    observed = _observed_value(trace, workflow)
    return Score(
        run_id=trace.run_id,
        gates=compute_gates(trace, workflow),
        target_metric=target_metric,
        target_value=target_value,
        observed_value=observed,
    )

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

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