Runner

DAG execution and trace recording.

runner

Executing a scientific experiment.

This module orchestrates deterministic DAG execution, dataset externalization, machine-recorded provenance collection, and dirty-node caching.

RunError

Bases: RuntimeError

Raised when an experiment cannot be run.

RunServices dataclass

RunServices(run_dir, tools, experiment_dir, on_log=None)

Execution context and services provided to agents during a run.

Attributes:

Name	Type	Description
run_dir	Path	Path to the active run directory.
tools	ToolRegistry	Tool registry for resolving and executing tool calls.
experiment_dir	Path	Path to the experiment directory containing workflows.
on_log	Callable[[str, str], None] | None	Optional callback for streaming log messages.

data_dir

data_dir()

Returns the data/ subdirectory for artifacts in this run.

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

def data_dir(self) -> Path:
    """Returns the data/ subdirectory for artifacts in this run."""
    return self.run_dir / "data"

hydrate

hydrate(value)

Rehydrates dataset reference pointers into active Python objects.

Parameters:

Name	Type	Description	Default
value	Any	Value or dataset reference to hydrate.	required

Returns:

Type	Description
Any	The hydrated data object.

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

def hydrate(self, value: Any) -> Any:
    """Rehydrates dataset reference pointers into active Python objects.

    Args:
      value: Value or dataset reference to hydrate.

    Returns:
      The hydrated data object.
    """
    return datasets.hydrate(value, self.run_dir)

call_tool async

call_tool(name, arguments)

Calls a tool asynchronously on a thread pool.

Parameters:

Name	Type	Description	Default
name	str	Name of the tool to execute.	required
arguments	dict[str, Any]	Arguments dictionary for the tool.	required

Returns:

Type	Description
Any	Tool execution output.

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

async def call_tool(self, name: str, arguments: dict[str, Any]) -> Any:
    """Calls a tool asynchronously on a thread pool.

    Args:
      name: Name of the tool to execute.
      arguments: Arguments dictionary for the tool.

    Returns:
      Tool execution output.
    """
    return await asyncio.to_thread(self.tools.call, name, arguments)

log async

log(node_id, message)

Logs an informational message for a node.

Parameters:

Name	Type	Description	Default
node_id	str	Identifier of the node.	required
message	str	Message text to log.	required

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

async def log(self, node_id: str, message: str) -> None:
    """Logs an informational message for a node.

    Args:
      node_id: Identifier of the node.
      message: Message text to log.
    """
    if self.on_log:
        self.on_log(node_id, message)

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

fingerprint

fingerprint(code, params, inputs)

Computes a SHA-256 fingerprint of node inputs, parameters, and code.

Parameters:

Name	Type	Description	Default
code	str	Source code of the agent module.	required
params	dict[str, Any]	Node parameter dictionary.	required
inputs	dict[str, Any]	Port inputs dictionary.	required

Returns:

Type	Description
str	16-character hexadecimal hash string.

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

def fingerprint(code: str, params: dict[str, Any], inputs: dict[str, Any]) -> str:
    """Computes a SHA-256 fingerprint of node inputs, parameters, and code.

    Args:
      code: Source code of the agent module.
      params: Node parameter dictionary.
      inputs: Port inputs dictionary.

    Returns:
      16-character hexadecimal hash string.
    """
    digest = hashlib.sha256()
    digest.update(code.encode())
    digest.update(json.dumps(params, sort_keys=True, default=str).encode())
    digest.update(json.dumps(inputs, sort_keys=True, default=str).encode())
    return digest.hexdigest()[:16]

run_nested async

run_nested(workflow_path, inputs, services)

Runs a nested sub-workflow for a CompositeAgent.

Parameters:

Name	Type	Description	Default
workflow_path	str	Relative path to the nested workflow.json file.	required
inputs	dict[str, Any]	Inputs dictionary mapped to the composite agent.	required
services	RunServices	Active execution services.	required

Returns:

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

Raises:

Type	Description
RunError	If the sub-workflow file is missing or invalid.

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

async def run_nested(
    workflow_path: str, inputs: dict[str, Any], services: RunServices
) -> dict[str, Any]:
    """Runs a nested sub-workflow for a CompositeAgent.

    Args:
      workflow_path: Relative path to the nested workflow.json file.
      inputs: Inputs dictionary mapped to the composite agent.
      services: Active execution services.

    Returns:
      Dictionary of terminal outputs mapped to output ports.

    Raises:
      RunError: If the sub-workflow file is missing or invalid.
    """
    from .models import Workflow as WorkflowModel
    from .workspace import read_json

    path = services.experiment_dir / workflow_path
    if not path.exists():
        raise RunError(f"Nested workflow not found: {path}")

    workflow = WorkflowModel.from_dict(read_json(path))
    report = validate_workflow(workflow)
    if not report.ok:
        raise RunError(
            f"Nested workflow {workflow_path} is not valid:\n{report.render()}"
        )

    import graphlib

    bus: dict[str, Any] = {}
    order = list(
        graphlib.TopologicalSorter(
            {n.id: set(workflow.parents(n.id)) for n in workflow.nodes}
        ).static_order()
    )

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

        node_inputs: dict[str, Any] = {}
        for edge in workflow.edges:
            if edge.target == node_id:
                node_inputs[edge.target_port] = bus[f"{edge.source}.{edge.source_port}"]
        for port in node.ports.inputs:
            if port.name not in node_inputs and port.name in inputs:
                node_inputs[port.name] = inputs[port.name]

        agent_path = path.parent / node.module
        agent = load_agent_class(agent_path)(name=node.id, params=dict(node.params))
        agent.bind(**node_inputs).attach(services)
        outputs = agent.validate_outputs(await agent.execute())
        for port, value in outputs.items():
            bus[f"{node_id}.{port}"] = value

    surfaced: dict[str, Any] = {}
    for node_id in workflow.terminal_nodes():
        node = workflow.node(node_id)
        assert node is not None
        for port in node.ports.outputs:
            surfaced[port.name] = bus[f"{node_id}.{port.name}"]
    return surfaced