From 0082d9a4579d821e132e62ae00b9fa45993bda5c Mon Sep 17 00:00:00 2001
From: Nathan Thorpe <nthorpe@cirro.bio>
Date: Wed, 28 Oct 2026 16:46:15 -0700
Subject: [PATCH 01/10] init

---
 cirro/models/cirro_uri.py          |  86 +++++++++++
 cirro/models/form_specification.py | 237 ++++++++++++++++++++++++++++-
 cirro/sdk/dataset.py               |  53 +++----
 cirro/sdk/process.py               |  69 +++++++--
 cirro/services/execution.py        |  11 +-
 tests/test_structured_inputs.py    | 181 ++++++++++++++++++++++
 6 files changed, 583 insertions(+), 54 deletions(-)
 create mode 100644 cirro/models/cirro_uri.py
 create mode 100644 tests/test_structured_inputs.py

diff --git a/cirro/models/cirro_uri.py b/cirro/models/cirro_uri.py
new file mode 100644
index 00000000..6c7dd70a
--- /dev/null
+++ b/cirro/models/cirro_uri.py
@@ -0,0 +1,86 @@
+from typing import Any, List, NamedTuple, Optional
+
+_SCHEME = 'cirro:'
+_DATASET_TYPE = 'dataset'
+# The type segment the backend emitted before it adopted entity type names
+_LEGACY_DATASET_TYPE = 'data'
+
+
+class CirroUri(NamedTuple):
+    """
+    A parsed `cirro:<tenant_id>:<project_id>:<type>:<id>/<path>` reference
+    """
+    tenant_id: str
+    project_id: str
+    type: str
+    id: str
+    path: Optional[str]
+    """Relative to the dataset's `data/` directory, `''` for the whole directory, or None for the dataset itself"""
+
+
+def format_cirro_uri(project_id: str, dataset_id: str, path: Optional[str] = None) -> str:
+    """
+    Builds a Cirro URI referencing a dataset, or a file or folder within it.
+
+    This is the value file and folder columns of a structured input
+    (and path parameters) expect, rather than a local or S3 path.
+
+    Args:
+        project_id (str): ID of the project the dataset is in
+        dataset_id (str): ID of the dataset
+        path (str): Path of a file or folder, relative to the dataset's `data/` directory.
+         Use `''` for the dataset's whole data directory, or omit it to reference the dataset itself.
+
+    ```python
+    format_cirro_uri("project-id", "dataset-id", "sample1_R1.fastq.gz")
+    # 'cirro::project-id:dataset:dataset-id/sample1_R1.fastq.gz'
+    ```
+    """
+    head = f'{_SCHEME}:{project_id}:{_DATASET_TYPE}:{dataset_id}'
+    if path is None:
+        return head
+    return f'{head}/{path.removeprefix("data/")}'
+
+
+def parse_cirro_uri(value: Any) -> Optional[CirroUri]:
+    """
+    Parses a Cirro URI referencing a dataset, or a file or folder within it,
+    returning None for anything else.
+    """
+    if not isinstance(value, str) or not value.startswith(_SCHEME):
+        return None
+    head, slash, path = value[len(_SCHEME):].partition('/')
+    parts = head.split(':')
+    if len(parts) != 4:
+        return None
+    tenant_id, project_id, uri_type, resource_id = parts
+    if uri_type not in (_DATASET_TYPE, _LEGACY_DATASET_TYPE) or not resource_id:
+        return None
+    return CirroUri(tenant_id, project_id, uri_type, resource_id, path if slash else None)
+
+
+def find_dataset_ids(value: Any, project_id: Optional[str] = None) -> List[str]:
+    """
+    The datasets referenced by any Cirro URI within a value (e.g. analysis parameters,
+    or structured input rows), in the order they first appear.
+
+    Args:
+        value: String, list, or dict to search
+        project_id (str): If given, only count URIs from this project
+    """
+    ids = {}
+
+    def collect(item: Any):
+        if isinstance(item, str):
+            uri = parse_cirro_uri(item.strip())
+            if uri and (project_id is None or uri.project_id == project_id):
+                ids[uri.id] = None
+        elif isinstance(item, dict):
+            for v in item.values():
+                collect(v)
+        elif isinstance(item, (list, tuple)):
+            for v in item:
+                collect(v)
+
+    collect(value)
+    return list(ids)
diff --git a/cirro/models/form_specification.py b/cirro/models/form_specification.py
index dd16bdbb..5e8348b9 100644
--- a/cirro/models/form_specification.py
+++ b/cirro/models/form_specification.py
@@ -1,7 +1,21 @@
-from typing import Dict, Any, List
+import json
+from typing import TYPE_CHECKING, Dict, Any, List, Mapping, Optional, Tuple, Union
 
 import jsonschema
-from cirro_api_client.v1.models import FormSchema
+from cirro_api_client.v1.models import FormSchema, ProcessParameterSchema, StructuredInputDefinition, \
+    FileFormat, StructuredInput, StructuredInputDataItem, RunAnalysisRequestStructuredFileInputs
+from cirro_api_client.v1.types import Unset
+
+from cirro.models.cirro_uri import parse_cirro_uri
+
+if TYPE_CHECKING:
+    import pandas
+
+# Rows for a structured input, as a list of dicts (one per row) or a pandas DataFrame
+StructuredInputRows = Union[List[Dict[str, Any]], 'pandas.DataFrame']
+
+# JSON schema formats whose values are references to a file or folder in Cirro
+_PATH_FORMATS = {'file-path', 'file-path-pattern', 'directory-path', 'path'}
 
 
 def _get_fields_in_schema(schema: Dict, parent_path='') -> List['Parameter']:
@@ -19,15 +33,176 @@ def _get_fields_in_schema(schema: Dict, parent_path='') -> List['Parameter']:
     return fields
 
 
+def _is_structured_input_field(field_spec: Any) -> bool:
+    """
+    A form field is filled by a structured input when it names one
+    (by its key in `structured_inputs`) in its own `schema` property
+    """
+    return isinstance(field_spec, dict) and 'schema' in field_spec
+
+
+def _get_structured_input_fields(schema: Dict, parent_path='') -> List[Tuple[str, str, bool]]:
+    """
+    Every form field filled by a structured input, as (field path, structured input name, required)
+    """
+    fields = []
+    required = set(schema.get('required') or [])
+    for field_key, field_spec in (schema.get('properties') or {}).items():
+        field_path = field_key if parent_path == '' else f'{parent_path}.{field_key}'
+        if _is_structured_input_field(field_spec):
+            fields.append((field_path, field_spec['schema'], field_key in required))
+        elif isinstance(field_spec, dict):
+            fields.extend(_get_structured_input_fields(field_spec, field_path))
+    return fields
+
+
+def _without_structured_input_fields(schema: Dict) -> Dict:
+    """
+    A copy of the form's JSON schema without the fields filled by structured inputs,
+    whose rows are passed separately from the parameters and validated against their own schema
+    """
+    properties = schema.get('properties')
+    if not isinstance(properties, dict):
+        return schema
+    removed = {key for key, spec in properties.items() if _is_structured_input_field(spec)}
+    stripped = dict(schema)
+    stripped['properties'] = {
+        key: _without_structured_input_fields(spec) if isinstance(spec, dict) else spec
+        for key, spec in properties.items()
+        if key not in removed
+    }
+    if 'required' in schema:
+        stripped['required'] = [key for key in schema['required'] if key not in removed]
+    return stripped
+
+
+def _get_by_path(params: Mapping, path: str) -> Any:
+    value = params
+    for key in path.split('.'):
+        if not isinstance(value, Mapping) or key not in value:
+            return None
+        value = value[key]
+    return value
+
+
+def _to_rows(rows: StructuredInputRows) -> List[Dict[str, Any]]:
+    """
+    Normalizes rows to a list of plain dicts, leaving out empty (None/NaN) cells
+    """
+    if hasattr(rows, 'to_json'):
+        # A pandas DataFrame: round-trip through JSON for plain Python values with NaN as None
+        rows = json.loads(rows.to_json(orient='records', date_format='iso'))
+    if not isinstance(rows, list):
+        raise TypeError(f'Structured input rows must be a list of dicts or a DataFrame, got {type(rows).__name__}')
+    return [
+        {key: value for key, value in dict(row).items() if value is not None}
+        for row in rows
+    ]
+
+
+class StructuredInputSpecification:
+    """
+    Describes a structured input: a table of rows (e.g., a samplesheet) that a process
+    takes as an input file. The rows are validated against `schema` and written out
+    as `file_format` for the pipeline.
+    """
+    def __init__(self, name: str, definition: StructuredInputDefinition,
+                 field_path: Optional[str] = None, required: bool = False):
+        self.name = name
+        """Key of the structured input, used in `run_analysis(structured_inputs=...)`"""
+        self.file_format: FileFormat = definition.file_format
+        """Format of the file passed to the pipeline (CSV, TSV, or JSON)"""
+        self.schema: Dict = definition.schema.to_dict()
+        """JSON schema of the rows (an array of objects)"""
+        self.examples: List[Dict] = [e.to_dict() for e in definition.examples] \
+            if isinstance(definition.examples, list) else []
+        """Example rows conforming to the schema"""
+        self.field_path = field_path
+        """The form field this structured input fills, if any"""
+        self.required = required
+        """Whether rows must be provided to run the process"""
+
+        items = self.schema.get('items') or {}
+        required_columns = set(items.get('required') or [])
+        self.columns: List[Parameter] = [
+            Parameter(key, spec, key, required=key in required_columns)
+            for key, spec in (items.get('properties') or {}).items()
+        ]
+        """Columns of each row"""
+
+    def validate_rows(self, rows: StructuredInputRows) -> List[Dict[str, Any]]:
+        """
+        Validates that the given rows conform to the specification
+
+        Args:
+            rows: List of dicts (one per row) or a pandas DataFrame
+
+        Returns:
+            The rows, as a list of dicts with empty cells left out
+        """
+        rows = _to_rows(rows)
+        try:
+            jsonschema.validate(instance=rows, schema=self.schema)
+        except jsonschema.ValidationError as e:
+            raise RuntimeError(f'Structured input {self.name} at {e.json_path} error: {e.message}') from e
+
+        # A file or folder cell must say which dataset it's in, which a bare path can't
+        path_columns = [c.key for c in self.columns if c.spec.get('format') in _PATH_FORMATS]
+        for index, row in enumerate(rows):
+            for column in path_columns:
+                value = row.get(column)
+                if value not in (None, '') and parse_cirro_uri(value) is None:
+                    raise RuntimeError(
+                        f'Structured input {self.name} at $[{index}].{column} error: {value!r} is not a Cirro URI. '
+                        'Reference files and folders with cirro.models.cirro_uri.format_cirro_uri'
+                    )
+        return rows
+
+    def to_request(self, rows: StructuredInputRows) -> StructuredInput:
+        """
+        Validates the rows and builds them into the structured input of a run analysis request
+        """
+        return StructuredInput(
+            data=[StructuredInputDataItem.from_dict(row) for row in self.validate_rows(rows)],
+            file_format=self.file_format,
+            import_source_uris=[]
+        )
+
+    def print(self):
+        """
+        Prints out a user-friendly view of the structured input's columns
+        """
+        required = 'required' if self.required else 'optional'
+        print(f"{self.name} ({self.file_format.value}, {required}):")
+        for column in self.columns:
+            print(f'\t{column}')
+
+
 class ParameterSpecification:
     """
     Used to describe parameters used in a process (uses JSONSchema)
     """
-    def __init__(self, form_schema: FormSchema):
+    def __init__(self, form_schema: Union[ProcessParameterSchema, FormSchema]):
         self._form_spec_raw: Dict = form_schema.form.additional_properties
         self._form_spec_ui: Dict = form_schema.ui.additional_properties
         self.form_spec = _get_fields_in_schema(self._form_spec_raw.get('properties') or {})
 
+        definitions = getattr(form_schema, 'structured_inputs', None)
+        if definitions is None or isinstance(definitions, Unset):
+            definitions = {}
+        else:
+            definitions = definitions.additional_properties
+        fields = {name: (path, required)
+                  for path, name, required in _get_structured_input_fields(self._form_spec_raw)}
+        self.structured_inputs: Dict[str, StructuredInputSpecification] = {
+            name: StructuredInputSpecification(name, definition, *fields.get(name, (None, False)))
+            for name, definition in definitions.items()
+        }
+        """
+        Structured inputs (e.g., samplesheets) the process takes, keyed by name.
+        Pass their rows to `run_analysis` with `structured_inputs`, rather than in `params`.
+        """
+
     @property
     def form_spec_json(self) -> dict:
         """
@@ -35,14 +210,49 @@ def form_spec_json(self) -> dict:
         """
         return self._form_spec_raw
 
-    def validate_params(self, params: Dict):
+    def validate_params(self, params: Dict, structured_inputs: Dict[str, StructuredInputRows] = None):
         """
-        Validates that the given parameters conforms to the specification
+        Validates that the given parameters (and structured inputs) conform to the specification
+
+        Args:
+            params (dict): Analysis parameters
+            structured_inputs (dict): Rows for each structured input, keyed by its name
         """
+        for spec in self.structured_inputs.values():
+            if spec.field_path and isinstance(_get_by_path(params, spec.field_path), list):
+                raise RuntimeError(f'Parameter {spec.field_path} is a structured input, '
+                                   f'pass its rows in structured_inputs["{spec.name}"] instead of params')
         try:
-            jsonschema.validate(instance=params, schema=self._form_spec_raw)
+            jsonschema.validate(instance=params, schema=_without_structured_input_fields(self._form_spec_raw))
         except jsonschema.ValidationError as e:
             raise RuntimeError(f'Parameter at {e.json_path} error: {e.message}') from e
+        self.build_structured_inputs(structured_inputs)
+
+    def build_structured_inputs(
+            self,
+            structured_inputs: Dict[str, StructuredInputRows] = None
+    ) -> RunAnalysisRequestStructuredFileInputs:
+        """
+        Validates the rows of each structured input and builds them into the
+        `structured_inputs` of a run analysis request
+
+        Args:
+            structured_inputs (dict): Rows for each structured input, keyed by its name
+        """
+        structured_inputs = structured_inputs or {}
+        unknown = [name for name in structured_inputs if name not in self.structured_inputs]
+        if unknown:
+            accepted = ', '.join(self.structured_inputs) or 'none'
+            raise RuntimeError(f'Unknown structured input(s): {", ".join(unknown)} (accepted: {accepted})')
+        missing = [name for name, spec in self.structured_inputs.items()
+                   if spec.required and name not in structured_inputs]
+        if missing:
+            raise RuntimeError(f'Structured input(s) required: {", ".join(missing)}')
+
+        request = RunAnalysisRequestStructuredFileInputs()
+        for name, rows in structured_inputs.items():
+            request[name] = self.structured_inputs[name].to_request(rows)
+        return request
 
     def print(self):
         """
@@ -52,13 +262,18 @@ def print(self):
         for field in self.form_spec:
             tab_prefix = '\t' * (field.path.count('.') + 1)
             print(tab_prefix + str(field))
+        if self.structured_inputs:
+            print("Structured inputs:")
+            for spec in self.structured_inputs.values():
+                spec.print()
 
 
 class Parameter:
-    def __init__(self, key: str, spec: Dict[str, Any], path: str):
+    def __init__(self, key: str, spec: Dict[str, Any], path: str, required: bool = False):
         self.spec = spec
         self.key = key
         self.path = path
+        self.required = required
 
     @property
     def is_group(self):
@@ -74,7 +289,13 @@ def __str__(self):
         else:
             display_value += self.key
 
-        for prop in ['default', 'type', 'enum', 'description']:
+        if self.required:
+            additional_data += 'required, '
+
+        if structured_input := self.spec.get('schema'):
+            additional_data += f'structured_input={structured_input}, '
+
+        for prop in ['default', 'type', 'format', 'enum', 'description']:
             if prop_value := self.spec.get(prop):
                 additional_data += f'{prop}={prop_value}, '
 
diff --git a/cirro/sdk/dataset.py b/cirro/sdk/dataset.py
index 8bcb7532..b524612c 100644
--- a/cirro/sdk/dataset.py
+++ b/cirro/sdk/dataset.py
@@ -2,12 +2,12 @@
 from functools import cached_property
 import re
 from pathlib import Path
-from typing import Union, List, Optional, Any
+from typing import Union, List, Optional, Any, Dict
 
 from cirro_api_client.v1.api.processes import validate_file_requirements
 from cirro_api_client.v1.errors import CirroException, UnexpectedStatus
-from cirro_api_client.v1.models import Dataset, DatasetDetail, RunAnalysisRequest, ProcessDetail, \
-    Status, RunAnalysisRequestParams, Tag, ArtifactType, NamedItem, ValidateFileRequirementsRequest, \
+from cirro_api_client.v1.models import Dataset, DatasetDetail, ProcessDetail, \
+    Status, Tag, ArtifactType, NamedItem, ValidateFileRequirementsRequest, \
     CostResponse
 
 from cirro.cirro_client import CirroApi
@@ -15,6 +15,7 @@
 from cirro.file_utils import bytes_to_human_readable, filter_files_by_pattern
 from cirro.models.assets import DatasetAssets
 from cirro.models.file import PathLike
+from cirro.models.form_specification import StructuredInputRows
 from cirro.sdk.asset import DataPortalAssets, DataPortalAsset
 from cirro.sdk.task import DataPortalTask
 from cirro.sdk.exceptions import DataPortalAssetNotFound
@@ -684,7 +685,8 @@ def run_analysis(
             notifications_emails: List[str] = None,
             compute_environment: str = None,
             resume_dataset_id: str = None,
-            source_sample_ids: List[str] = None
+            source_sample_ids: List[str] = None,
+            structured_inputs: Dict[str, StructuredInputRows] = None
     ) -> str:
         """
         Runs an analysis on a dataset, returns the ID of the newly created dataset.
@@ -703,6 +705,7 @@ def run_analysis(
         spec = process.get_parameter_spec()
         spec.print()               # human-readable listing of every parameter
         spec.validate_params(params)   # raises if params do not fit the schema
+        spec.structured_inputs     # tables of rows (e.g., a samplesheet) the process takes
         ```
 
         To follow the analysis, re-fetch the dataset each time round the loop.
@@ -732,6 +735,8 @@ def run_analysis(
             resume_dataset_id (str): ID of dataset to resume from, used for caching task execution.
              It will attempt to re-use the previous output to minimize duplicate work
             source_sample_ids (List[str]): List of sample IDs to use as input for the analysis.
+            structured_inputs (Dict[str, List[dict] or DataFrame]): Rows for each structured input
+             the process takes, keyed by its name. See `cirro.sdk.process.DataPortalProcess.run_analysis`.
 
         Returns:
             dataset_id (str): ID of newly created dataset
@@ -740,42 +745,22 @@ def run_analysis(
             raise DataPortalInputError("Must specify 'name' for run_analysis")
         if process is None:
             raise DataPortalInputError("Must specify 'process' for run_analysis")
-        if notifications_emails is None:
-            notifications_emails = []
-        if params is None:
-            params = {}
 
         # If the process is a string, try to parse it as a process name or ID
         process = parse_process_name_or_id(process, self._client)
 
-        if compute_environment:
-            compute_environment_name = compute_environment
-            compute_environments = self._client.compute_environments.list_environments_for_project(
-                project_id=self.project_id
-            )
-            compute_environment = next(
-                (env for env in compute_environments
-                 if env.name == compute_environment or env.id == compute_environment),
-                None
-            )
-            if compute_environment is None:
-                raise DataPortalInputError(f"Compute environment '{compute_environment_name}' not found")
-
-        resp = self._client.execution.run_analysis(
+        return process.run_analysis(
+            name=name,
             project_id=self.project_id,
-            request=RunAnalysisRequest(
-                name=name,
-                description=description,
-                process_id=process.id,
-                source_dataset_ids=[self.id],
-                params=RunAnalysisRequestParams.from_dict(params),
-                notification_emails=notifications_emails,
-                resume_dataset_id=resume_dataset_id,
-                source_sample_ids=source_sample_ids,
-                compute_environment_id=compute_environment.id if compute_environment else None
-            )
+            datasets=[self.id],
+            description=description,
+            params=params,
+            notifications_emails=notifications_emails,
+            compute_environment=compute_environment,
+            resume_dataset_id=resume_dataset_id,
+            source_sample_ids=source_sample_ids,
+            structured_inputs=structured_inputs
         )
-        return resp.id
 
     def update_samplesheet(self,
                            contents: str = None,
diff --git a/cirro/sdk/process.py b/cirro/sdk/process.py
index c99d4640..139ac88e 100644
--- a/cirro/sdk/process.py
+++ b/cirro/sdk/process.py
@@ -1,10 +1,12 @@
-from typing import List, Union
+from typing import Dict, List, Union
 
 from cirro_api_client.v1.models import Process, Executor, ProcessDetail, CustomPipelineSettings, PipelineCode, \
     RunAnalysisRequest, RunAnalysisRequestParams
+from cirro_api_client.v1.types import UNSET
 
 from cirro.cirro_client import CirroApi
-from cirro.models.form_specification import ParameterSpecification
+from cirro.models.cirro_uri import find_dataset_ids
+from cirro.models.form_specification import ParameterSpecification, StructuredInputRows
 from cirro.sdk.asset import DataPortalAssets, DataPortalAsset
 from cirro.sdk.exceptions import DataPortalInputError
 
@@ -100,6 +102,18 @@ def get_parameter_spec(self) -> ParameterSpecification:
         Call `print()` on the result for a readable listing, or
         `validate_params()` to check a `params` dict before submitting it.
 
+        Some processes take structured inputs (e.g., a samplesheet): tables of rows
+        validated against their own schema, listed in `structured_inputs` on the result.
+        Their rows are passed to `run_analysis` with `structured_inputs` rather than `params`.
+
+        ```python
+        spec = process.get_parameter_spec()
+        spec.print()
+        samplesheet = spec.structured_inputs["samplesheet"]
+        samplesheet.print()         # the columns each row takes
+        samplesheet.examples        # example rows
+        ```
+
         Returns:
             `cirro.models.form_specification.ParameterSpecification`
         """
@@ -115,7 +129,8 @@ def run_analysis(
             notifications_emails: List[str] = None,
             compute_environment: str = None,
             resume_dataset_id: str = None,
-            source_sample_ids: List[str] = None
+            source_sample_ids: List[str] = None,
+            structured_inputs: Dict[str, StructuredInputRows] = None
     ) -> str:
         """
         Runs this process on one or more input datasets, returns the ID of the newly created dataset.
@@ -130,7 +145,9 @@ def run_analysis(
             name (str): Name of newly created dataset
             project_id (str): ID of the project to run the analysis in
             datasets (List[DataPortalDataset or str]): One or more input datasets
-             (as DataPortalDataset objects or dataset ID strings)
+             (as DataPortalDataset objects or dataset ID strings).
+             Datasets referenced by Cirro URIs in `params` or `structured_inputs` are added automatically,
+             so this may be omitted when those reference all the inputs.
             description (str): Description of newly created dataset
             params (dict): Analysis parameters
             notifications_emails (List[str]): Notification email address(es)
@@ -141,26 +158,55 @@ def run_analysis(
             Note that Nextflow does not require this parameter, as it will automatically resume
              from any previous attempts using a global cache.
             source_sample_ids (List[str]): List of sample IDs to use as input for the analysis.
+            structured_inputs (Dict[str, List[dict] or DataFrame]): Rows for each structured input
+             the process takes (see `get_parameter_spec`), keyed by its name.
+             Files and folders are referenced with Cirro URIs (see `cirro.models.cirro_uri.format_cirro_uri`).
 
         Returns:
             dataset_id (str): ID of newly created dataset
+
+        ```python
+        from cirro.models.cirro_uri import format_cirro_uri
+
+        process.run_analysis(
+            name="Output",
+            project_id=project_id,
+            structured_inputs={
+                "samplesheet": [
+                    {
+                        "sample": "sample1",
+                        "fastq_1": format_cirro_uri(project_id, dataset_id, "sample1_R1.fastq.gz"),
+                        "fastq_2": format_cirro_uri(project_id, dataset_id, "sample1_R2.fastq.gz"),
+                    },
+                ]
+            }
+        )
+        ```
         """
         if name is None:
             raise DataPortalInputError("Must specify 'name' for run_analysis")
         if project_id is None:
             raise DataPortalInputError("Must specify 'project_id' for run_analysis")
-        if not datasets:
-            raise DataPortalInputError("Must specify 'datasets' for run_analysis")
         if notifications_emails is None:
             notifications_emails = []
         if params is None:
             params = {}
 
-        # Accept DataPortalDataset objects or raw ID strings
-        source_dataset_ids = [
-            ds if isinstance(ds, str) else ds.id
-            for ds in datasets
-        ]
+        spec = self.get_parameter_spec()
+        try:
+            request_structured_inputs = spec.build_structured_inputs(structured_inputs)
+        except RuntimeError as e:
+            raise DataPortalInputError(str(e)) from e
+
+        # Accept DataPortalDataset objects or raw ID strings, along with
+        # any dataset the parameters or structured inputs reference
+        source_dataset_ids = list(dict.fromkeys([
+            *(ds if isinstance(ds, str) else ds.id for ds in datasets or []),
+            *find_dataset_ids(params, project_id),
+            *find_dataset_ids(request_structured_inputs.to_dict(), project_id)
+        ]))
+        if not source_dataset_ids:
+            raise DataPortalInputError("Must specify 'datasets' for run_analysis")
 
         if compute_environment:
             compute_environment_name = compute_environment
@@ -186,6 +232,7 @@ def run_analysis(
                 notification_emails=notifications_emails,
                 resume_dataset_id=resume_dataset_id,
                 source_sample_ids=source_sample_ids,
+                structured_inputs=request_structured_inputs if structured_inputs else UNSET,
                 compute_environment_id=compute_environment.id if compute_environment else None
             )
         )
diff --git a/cirro/services/execution.py b/cirro/services/execution.py
index 5586797e..89b61c8b 100644
--- a/cirro/services/execution.py
+++ b/cirro/services/execution.py
@@ -50,6 +50,10 @@ def run_analysis(self, project_id: str, request: RunAnalysisRequest) -> CreateRe
         )
         cirro.execution.run_analysis("project-id", request)
         ```
+
+        Processes which take structured inputs (e.g., a samplesheet) accept their rows in
+        `structured_inputs`, built with
+        `cirro.models.form_specification.ParameterSpecification.build_structured_inputs`.
         """
 
         form_spec = get_process_parameters.sync(
@@ -57,10 +61,15 @@ def run_analysis(self, project_id: str, request: RunAnalysisRequest) -> CreateRe
             client=self._api_client
         )
 
+        structured_inputs = request.structured_inputs
         ParameterSpecification(
             form_spec
         ).validate_params(
-            request.params.to_dict() if request.params else {}
+            request.params.to_dict() if request.params else {},
+            {
+                name: [row.to_dict() for row in structured_input.data]
+                for name, structured_input in structured_inputs.additional_properties.items()
+            } if structured_inputs else None
         )
 
         return run_analysis.sync(
diff --git a/tests/test_structured_inputs.py b/tests/test_structured_inputs.py
new file mode 100644
index 00000000..badd7903
--- /dev/null
+++ b/tests/test_structured_inputs.py
@@ -0,0 +1,181 @@
+import unittest
+from unittest.mock import Mock
+
+import pandas as pd
+from cirro_api_client.v1.models import ProcessParameterSchema, FileFormat, CreateResponse
+
+from cirro.models.cirro_uri import format_cirro_uri, parse_cirro_uri, find_dataset_ids
+from cirro.models.form_specification import ParameterSpecification
+from cirro.sdk.exceptions import DataPortalInputError
+from cirro.sdk.process import DataPortalProcess
+
+PROJECT_ID = 'project-1'
+
+PARAMETER_SCHEMA = {
+    'form': {
+        'type': 'object',
+        'required': ['input', 'genome'],
+        'properties': {
+            'input': {
+                'type': 'string',
+                'format': 'file-path',
+                'schema': 'samplesheet',
+                'title': 'Samplesheet'
+            },
+            'genome': {'type': 'string', 'enum': ['GRCh38', 'GRCm39']},
+            'options': {
+                'type': 'object',
+                'properties': {
+                    'contrasts': {'type': 'string', 'schema': 'contrasts'}
+                }
+            }
+        }
+    },
+    'ui': {},
+    'structuredInputs': {
+        'samplesheet': {
+            'fileFormat': 'CSV',
+            'schema': {
+                'type': 'array',
+                'items': {
+                    'type': 'object',
+                    'required': ['sample', 'fastq_1'],
+                    'properties': {
+                        'sample': {'type': 'string'},
+                        'fastq_1': {'type': 'string', 'format': 'file-path'},
+                        'fastq_2': {'type': 'string', 'format': 'file-path'},
+                        'replicate': {'type': 'integer'}
+                    }
+                }
+            },
+            'examples': [{'sample': 'a', 'fastq_1': 'cirro::p:dataset:d/a_R1.fastq.gz'}]
+        },
+        'contrasts': {
+            'fileFormat': 'TSV',
+            'schema': {
+                'type': 'array',
+                'items': {'type': 'object', 'properties': {'id': {'type': 'string'}}}
+            }
+        }
+    }
+}
+
+
+def _spec() -> ParameterSpecification:
+    return ParameterSpecification(ProcessParameterSchema.from_dict(PARAMETER_SCHEMA))
+
+
+def _row(sample='s1', dataset_id='dataset-1', **cells):
+    return {'sample': sample, 'fastq_1': format_cirro_uri(PROJECT_ID, dataset_id, f'{sample}_R1.fastq.gz'), **cells}
+
+
+class TestCirroUri(unittest.TestCase):
+    def test_format_and_parse(self):
+        uri = format_cirro_uri(PROJECT_ID, 'dataset-1', 'data/folder/a.fastq.gz')
+        self.assertEqual(uri, 'cirro::project-1:dataset:dataset-1/folder/a.fastq.gz')
+        parsed = parse_cirro_uri(uri)
+        self.assertEqual((parsed.project_id, parsed.id, parsed.path), (PROJECT_ID, 'dataset-1', 'folder/a.fastq.gz'))
+        self.assertIsNone(parse_cirro_uri(format_cirro_uri(PROJECT_ID, 'dataset-1')).path)
+        self.assertIsNone(parse_cirro_uri('s3://bucket/datasets/dataset-1/data/a.fastq.gz'))
+
+    def test_find_dataset_ids(self):
+        value = {'a': [_row(dataset_id='d1'), _row(dataset_id='d2'), _row(dataset_id='d1')],
+                 'b': format_cirro_uri('other-project', 'd3', '')}
+        self.assertEqual(find_dataset_ids(value), ['d1', 'd2', 'd3'])
+        self.assertEqual(find_dataset_ids(value, PROJECT_ID), ['d1', 'd2'])
+
+
+class TestStructuredInputSpecification(unittest.TestCase):
+    def test_parses_structured_inputs(self):
+        spec = _spec()
+        samplesheet = spec.structured_inputs['samplesheet']
+        self.assertEqual(samplesheet.file_format, FileFormat.CSV)
+        self.assertEqual(samplesheet.field_path, 'input')
+        self.assertTrue(samplesheet.required)
+        self.assertEqual([c.key for c in samplesheet.columns], ['sample', 'fastq_1', 'fastq_2', 'replicate'])
+        self.assertEqual([c.required for c in samplesheet.columns], [True, True, False, False])
+        self.assertEqual(len(samplesheet.examples), 1)
+
+        contrasts = spec.structured_inputs['contrasts']
+        self.assertEqual(contrasts.field_path, 'options.contrasts')
+        self.assertFalse(contrasts.required)
+
+    def test_validate_params_skips_structured_input_fields(self):
+        spec = _spec()
+        spec.validate_params({'genome': 'GRCh38'}, {'samplesheet': [_row()]})
+        with self.assertRaisesRegex(RuntimeError, 'genome'):
+            spec.validate_params({'genome': 'hg19'}, {'samplesheet': [_row()]})
+
+    def test_validate_params_requires_structured_inputs(self):
+        with self.assertRaisesRegex(RuntimeError, 'required: samplesheet'):
+            _spec().validate_params({'genome': 'GRCh38'})
+
+    def test_validate_params_rejects_rows_in_params(self):
+        with self.assertRaisesRegex(RuntimeError, r'structured_inputs\["samplesheet"\]'):
+            _spec().validate_params({'genome': 'GRCh38', 'input': [_row()]})
+
+    def test_rejects_unknown_structured_input(self):
+        with self.assertRaisesRegex(RuntimeError, 'Unknown structured input'):
+            _spec().build_structured_inputs({'samplesheet': [_row()], 'other': []})
+
+    def test_validates_rows_against_schema(self):
+        samplesheet = _spec().structured_inputs['samplesheet']
+        with self.assertRaisesRegex(RuntimeError, r"\$\[1\].*'fastq_1' is a required property"):
+            samplesheet.validate_rows([_row(), {'sample': 's2'}])
+        with self.assertRaisesRegex(RuntimeError, 'is not of type'):
+            samplesheet.validate_rows([_row(replicate='one')])
+
+    def test_requires_cirro_uris_for_paths(self):
+        samplesheet = _spec().structured_inputs['samplesheet']
+        with self.assertRaisesRegex(RuntimeError, r'\$\[0\]\.fastq_1.*not a Cirro URI'):
+            samplesheet.validate_rows([{'sample': 's1', 'fastq_1': 's3://bucket/a.fastq.gz'}])
+
+    def test_accepts_dataframe(self):
+        samplesheet = _spec().structured_inputs['samplesheet']
+        df = pd.DataFrame([_row('s1', replicate=1), _row('s2')])
+        rows = samplesheet.validate_rows(df)
+        # Empty cells are left out, rather than sent as null
+        self.assertEqual(rows, [_row('s1', replicate=1), _row('s2')])
+
+    def test_build_structured_inputs(self):
+        request = _spec().build_structured_inputs({'samplesheet': [_row()]})
+        self.assertEqual(request.to_dict(), {
+            'samplesheet': {'data': [_row()], 'fileFormat': 'CSV', 'importSourceUris': []}
+        })
+
+
+class TestRunAnalysisStructuredInputs(unittest.TestCase):
+    def _process(self):
+        client = Mock()
+        client.processes.get_parameter_spec.return_value = _spec()
+        client.execution.run_analysis.return_value = CreateResponse(id='new-dataset', message='')
+        process = Mock(id='process-1')
+        return DataPortalProcess(process, client), client
+
+    def test_sends_structured_inputs_and_their_datasets(self):
+        process, client = self._process()
+        dataset_id = process.run_analysis(
+            name='Output',
+            project_id=PROJECT_ID,
+            params={'genome': 'GRCh38'},
+            structured_inputs={'samplesheet': [_row('s1', 'd1'), _row('s2', 'd2')]}
+        )
+        self.assertEqual(dataset_id, 'new-dataset')
+        request = client.execution.run_analysis.call_args.kwargs['request']
+        self.assertEqual(request.source_dataset_ids, ['d1', 'd2'])
+        self.assertEqual(request.to_dict()['structuredInputs']['samplesheet']['data'],
+                         [_row('s1', 'd1'), _row('s2', 'd2')])
+
+    def test_requires_datasets_without_references(self):
+        process, client = self._process()
+        client.processes.get_parameter_spec.return_value = ParameterSpecification(
+            ProcessParameterSchema.from_dict({'form': {}, 'ui': {}})
+        )
+        with self.assertRaisesRegex(DataPortalInputError, 'datasets'):
+            process.run_analysis(name='Output', project_id=PROJECT_ID)
+
+    def test_invalid_rows_raise_input_error(self):
+        process, _ = self._process()
+        with self.assertRaisesRegex(DataPortalInputError, 'fastq_1'):
+            process.run_analysis(name='Output', project_id=PROJECT_ID, params={'genome': 'GRCh38'},
+                                 structured_inputs={'samplesheet': [{'sample': 's1'}]})

From 90bbc2ed8ab1fd1aaeaeabb343dc97d0a6de63c9 Mon Sep 17 00:00:00 2001
From: Nathan Thorpe <nthorpe@cirro.bio>
Date: Thu, 29 Oct 2026 06:29:04 -0700
Subject: [PATCH 02/10] clean

---
 cirro/models/form_specification.py | 8 ++------
 1 file changed, 2 insertions(+), 6 deletions(-)

diff --git a/cirro/models/form_specification.py b/cirro/models/form_specification.py
index 5e8348b9..cdac268d 100644
--- a/cirro/models/form_specification.py
+++ b/cirro/models/form_specification.py
@@ -185,18 +185,14 @@ class ParameterSpecification:
     def __init__(self, form_schema: Union[ProcessParameterSchema, FormSchema]):
         self._form_spec_raw: Dict = form_schema.form.additional_properties
         self._form_spec_ui: Dict = form_schema.ui.additional_properties
+        self._structured_input_raw: Dict = form_schema.structured_inputs.additional_properties
         self.form_spec = _get_fields_in_schema(self._form_spec_raw.get('properties') or {})
 
-        definitions = getattr(form_schema, 'structured_inputs', None)
-        if definitions is None or isinstance(definitions, Unset):
-            definitions = {}
-        else:
-            definitions = definitions.additional_properties
         fields = {name: (path, required)
                   for path, name, required in _get_structured_input_fields(self._form_spec_raw)}
         self.structured_inputs: Dict[str, StructuredInputSpecification] = {
             name: StructuredInputSpecification(name, definition, *fields.get(name, (None, False)))
-            for name, definition in definitions.items()
+            for name, definition in self._structured_input_raw.items()
         }
         """
         Structured inputs (e.g., samplesheets) the process takes, keyed by name.

From 9b686f4d1cff496bc1f6ca0a11cb6170ee0fb2a8 Mon Sep 17 00:00:00 2001
From: Nathan Thorpe <nthorpe@cirro.bio>
Date: Thu, 29 Oct 2026 06:57:16 -0700
Subject: [PATCH 03/10] clean and add example with sheet

---
 cirro/models/cirro_uri.py |  4 +---
 cirro/sdk/dataset.py      |  2 +-
 cirro/sdk/file.py         | 19 +++++++++++++++++--
 cirro/sdk/process.py      | 26 +++++++++++++++++++++-----
 4 files changed, 40 insertions(+), 11 deletions(-)

diff --git a/cirro/models/cirro_uri.py b/cirro/models/cirro_uri.py
index 6c7dd70a..d705f9c7 100644
--- a/cirro/models/cirro_uri.py
+++ b/cirro/models/cirro_uri.py
@@ -2,8 +2,6 @@
 
 _SCHEME = 'cirro:'
 _DATASET_TYPE = 'dataset'
-# The type segment the backend emitted before it adopted entity type names
-_LEGACY_DATASET_TYPE = 'data'
 
 
 class CirroUri(NamedTuple):
@@ -54,7 +52,7 @@ def parse_cirro_uri(value: Any) -> Optional[CirroUri]:
     if len(parts) != 4:
         return None
     tenant_id, project_id, uri_type, resource_id = parts
-    if uri_type not in (_DATASET_TYPE, _LEGACY_DATASET_TYPE) or not resource_id:
+    if uri_type not in _DATASET_TYPE or not resource_id:
         return None
     return CirroUri(tenant_id, project_id, uri_type, resource_id, path if slash else None)
 
diff --git a/cirro/sdk/dataset.py b/cirro/sdk/dataset.py
index b524612c..98e1323a 100644
--- a/cirro/sdk/dataset.py
+++ b/cirro/sdk/dataset.py
@@ -452,7 +452,7 @@ def list_files(self, file_limit: int = 100000) -> DataPortalFiles:
 
         return DataPortalFiles(
             [
-                DataPortalFile(file=file, client=self._client)
+                DataPortalFile(file=file, client=self._client, dataset_id=self.id)
                 for file in files
             ]
         )
diff --git a/cirro/sdk/file.py b/cirro/sdk/file.py
index 36608ee1..2cb62994 100644
--- a/cirro/sdk/file.py
+++ b/cirro/sdk/file.py
@@ -1,8 +1,9 @@
 from pathlib import Path
-from typing import List
+from typing import List, Optional
 
 from cirro.cirro_client import CirroApi
 from cirro.config import Constants
+from cirro.models.cirro_uri import format_cirro_uri
 from cirro.models.file import File, PathLike
 from cirro.sdk.asset import DataPortalAssets, DataPortalAsset
 from cirro.sdk.exceptions import DataPortalInputError
@@ -15,7 +16,7 @@ class DataPortalFile(DataPortalAsset, FileReadMixin):
     Datasets are made up of a collection of File objects in the Data Portal.
     """
 
-    def __init__(self, file: File, client: CirroApi):
+    def __init__(self, file: File, client: CirroApi, dataset_id: Optional[str] = None):
         """
         Instantiate by listing files from a dataset.
 
@@ -32,6 +33,7 @@ def __init__(self, file: File, client: CirroApi):
         # Attach the file object
         self._file = file
         self._client = client
+        self._dataset_id = dataset_id
 
     # Note that the 'name' and 'id' attributes are set to the relative path
     # The purpose of this is to support the DataPortalAssets class functions
@@ -60,6 +62,19 @@ def absolute_path(self) -> str:
         """Fully URI to file object in AWS S3"""
         return self._file.absolute_path
 
+    @property
+    def cirro_uri(self) -> str:
+        """
+        Cirro URI referencing this file, for use in analysis parameters
+        and structured inputs (see `cirro.models.cirro_uri.format_cirro_uri`).
+
+        Only available for files under the `data/` directory of a dataset
+        listed with `cirro.sdk.dataset.DataPortalDataset.list_files`.
+        """
+        if self._dataset_id is None or not self.relative_path.startswith("data/"):
+            raise DataPortalInputError(f"{self.relative_path} is not a dataset data file")
+        return format_cirro_uri(self._file.access_context.project_id, self._dataset_id, self.relative_path)
+
     @property
     def metadata(self) -> dict:
         """File metadata"""
diff --git a/cirro/sdk/process.py b/cirro/sdk/process.py
index 139ac88e..71de2b6d 100644
--- a/cirro/sdk/process.py
+++ b/cirro/sdk/process.py
@@ -160,28 +160,44 @@ def run_analysis(
             source_sample_ids (List[str]): List of sample IDs to use as input for the analysis.
             structured_inputs (Dict[str, List[dict] or DataFrame]): Rows for each structured input
              the process takes (see `get_parameter_spec`), keyed by its name.
-             Files and folders are referenced with Cirro URIs (see `cirro.models.cirro_uri.format_cirro_uri`).
+             Files and folders are referenced with Cirro URIs (see `cirro.sdk.file.DataPortalFile.cirro_uri`
+             and `cirro.models.cirro_uri.format_cirro_uri`).
 
         Returns:
             dataset_id (str): ID of newly created dataset
 
         ```python
-        from cirro.models.cirro_uri import format_cirro_uri
+        files = dataset.list_files()
 
         process.run_analysis(
             name="Output",
-            project_id=project_id,
+            project_id=dataset.project_id,
             structured_inputs={
                 "samplesheet": [
                     {
                         "sample": "sample1",
-                        "fastq_1": format_cirro_uri(project_id, dataset_id, "sample1_R1.fastq.gz"),
-                        "fastq_2": format_cirro_uri(project_id, dataset_id, "sample1_R2.fastq.gz"),
+                        "fastq_1": files.get_by_id("data/sample1_R1.fastq.gz").cirro_uri,
+                        "fastq_2": files.get_by_id("data/sample1_R2.fastq.gz").cirro_uri,
                     },
                 ]
             }
         )
         ```
+
+        Rows can also come from a sheet, whose file columns hold Cirro URIs:
+
+        ```python
+        samplesheet = project.get_sheet_by_name("Samplesheet").to_dataframe()
+
+        process.run_analysis(
+            name="Output",
+            project_id=project.id,
+            structured_inputs={
+                # Drop the sheet's row identifier, which isn't a samplesheet column
+                "samplesheet": samplesheet.drop(columns="_row_id")
+            }
+        )
+        ```
         """
         if name is None:
             raise DataPortalInputError("Must specify 'name' for run_analysis")

From 800b38cae22b5c666c5aae14daab37b284c6d66b Mon Sep 17 00:00:00 2001
From: Nathan Thorpe <nthorpe@cirro.bio>
Date: Thu, 29 Oct 2026 16:15:43 -0700
Subject: [PATCH 04/10] index df by _row_id

---
 cirro/sdk/sheet.py                |  8 ++++----
 samples/Working_with_sheets.ipynb |  4 ++--
 tests/test_sheets.py              | 10 ++++++----
 3 files changed, 12 insertions(+), 10 deletions(-)

diff --git a/cirro/sdk/sheet.py b/cirro/sdk/sheet.py
index 4bb86c4b..9917adab 100644
--- a/cirro/sdk/sheet.py
+++ b/cirro/sdk/sheet.py
@@ -157,8 +157,8 @@ def to_dataframe(self,
         Read the sheet's rows into a Pandas DataFrame.
 
         Pages through the whole sheet, or stops early at `max_rows` if one is
-        given. The `_row_id` column identifies each row for `update_rows` and
-        `delete_rows`.
+        given. The frame is indexed by `_row_id`, which identifies each row for
+        `update_rows` and `delete_rows`.
 
         Args:
             max_rows (int): Stop after this many rows; reads the whole sheet
@@ -167,7 +167,7 @@ def to_dataframe(self,
             order (SqlSortOrder): Sort direction
 
         Returns:
-            `pandas.DataFrame`
+            `pandas.DataFrame` indexed by `_row_id`
         """
         sort = SheetSort(order_by=order_by if order_by is not None else UNSET,
                          order=order if order is not None else UNSET)
@@ -178,7 +178,7 @@ def to_dataframe(self,
                 request=SheetDataRequest(limit=limit, page=page, sort=sort)
             ),
             max_rows=max_rows
-        )
+        ).set_index("_row_id")
 
     def insert_rows(self, rows: List[dict]) -> int:
         """
diff --git a/samples/Working_with_sheets.ipynb b/samples/Working_with_sheets.ipynb
index e3dabc82..1ffc7caf 100644
--- a/samples/Working_with_sheets.ipynb
+++ b/samples/Working_with_sheets.ipynb
@@ -117,7 +117,7 @@
    "source": [
     "#### Reading the sheet\n",
     "\n",
-    "`to_dataframe` reads the whole sheet. Its first column, `_row_id`, identifies each row, and is how the updates and deletes below pick out the rows to change."
+    "`to_dataframe` reads the whole sheet. Its index, `_row_id`, identifies each row, and is how the updates and deletes below pick out the rows to change."
    ]
   },
   {
@@ -234,7 +234,7 @@
    ],
    "source": [
     "df = sheet.to_dataframe()\n",
-    "row_ids = {sample_id: int(row_id) for sample_id, row_id in zip(df[\"sample_id\"], df[\"_row_id\"])}\n",
+    "row_ids = {sample_id: int(row_id) for sample_id, row_id in zip(df[\"sample_id\"], df.index)}\n",
     "df"
    ]
   },
diff --git a/tests/test_sheets.py b/tests/test_sheets.py
index 5b87ecd5..0e2fffd4 100644
--- a/tests/test_sheets.py
+++ b/tests/test_sheets.py
@@ -80,7 +80,8 @@ def test_single_partial_page(self):
         df = _make_sheet(client).to_dataframe()
 
         self.assertEqual(250, len(df))
-        self.assertEqual(["_row_id", "icd_code"], list(df.columns))
+        self.assertEqual(["icd_code"], list(df.columns))
+        self.assertEqual("_row_id", df.index.name)
         self.assertEqual([(1, ROWS_PER_PAGE)], requests)
 
     def test_full_page_does_not_fetch_another(self):
@@ -96,7 +97,7 @@ def test_reads_every_page(self):
         df = _make_sheet(client).to_dataframe()
 
         self.assertEqual(2500, len(df))
-        self.assertEqual(list(range(2500)), list(df["_row_id"]))
+        self.assertEqual(list(range(2500)), list(df.index))
         self.assertEqual([(1, ROWS_PER_PAGE), (2, ROWS_PER_PAGE), (3, ROWS_PER_PAGE)], requests)
 
     def test_reads_the_whole_sheet_when_uncapped(self):
@@ -120,7 +121,7 @@ def test_max_rows_truncates_and_stops_early(self):
         df = _make_sheet(client).to_dataframe(max_rows=1500)
 
         self.assertEqual(1500, len(df))
-        self.assertEqual(list(range(1500)), list(df["_row_id"]))
+        self.assertEqual(list(range(1500)), list(df.index))
         self.assertEqual(2, len(requests))
 
     def test_max_rows_below_page_size_shrinks_the_page(self):
@@ -135,7 +136,8 @@ def test_empty_sheet_keeps_its_columns(self):
         df = _make_sheet(client).to_dataframe()
 
         self.assertTrue(df.empty)
-        self.assertEqual(["_row_id", "icd_code"], list(df.columns))
+        self.assertEqual(["icd_code"], list(df.columns))
+        self.assertEqual("_row_id", df.index.name)
         self.assertEqual(1, len(requests))
 
     def test_sort_is_passed_through(self):

From 881f00233e703885e9d1f977ccdde025f701d824 Mon Sep 17 00:00:00 2001
From: Nathan Thorpe <nthorpe@cirro.bio>
Date: Thu, 29 Oct 2026 16:17:32 -0700
Subject: [PATCH 05/10] example of running with structured input

---
 cirro/models/cirro_uri.py |  2 +-
 cirro/sdk/file.py         |  4 ++--
 cirro/sdk/process.py      | 21 ++++++++++++++-------
 3 files changed, 17 insertions(+), 10 deletions(-)

diff --git a/cirro/models/cirro_uri.py b/cirro/models/cirro_uri.py
index d705f9c7..5240b3cd 100644
--- a/cirro/models/cirro_uri.py
+++ b/cirro/models/cirro_uri.py
@@ -16,7 +16,7 @@ class CirroUri(NamedTuple):
     """Relative to the dataset's `data/` directory, `''` for the whole directory, or None for the dataset itself"""
 
 
-def format_cirro_uri(project_id: str, dataset_id: str, path: Optional[str] = None) -> str:
+def format_dataset_uri(project_id: str, dataset_id: str, path: Optional[str] = None) -> str:
     """
     Builds a Cirro URI referencing a dataset, or a file or folder within it.
 
diff --git a/cirro/sdk/file.py b/cirro/sdk/file.py
index 2cb62994..2d33b249 100644
--- a/cirro/sdk/file.py
+++ b/cirro/sdk/file.py
@@ -3,7 +3,7 @@
 
 from cirro.cirro_client import CirroApi
 from cirro.config import Constants
-from cirro.models.cirro_uri import format_cirro_uri
+from cirro.models.cirro_uri import format_dataset_uri
 from cirro.models.file import File, PathLike
 from cirro.sdk.asset import DataPortalAssets, DataPortalAsset
 from cirro.sdk.exceptions import DataPortalInputError
@@ -73,7 +73,7 @@ def cirro_uri(self) -> str:
         """
         if self._dataset_id is None or not self.relative_path.startswith("data/"):
             raise DataPortalInputError(f"{self.relative_path} is not a dataset data file")
-        return format_cirro_uri(self._file.access_context.project_id, self._dataset_id, self.relative_path)
+        return format_dataset_uri(self._file.access_context.project_id, self._dataset_id, self.relative_path)
 
     @property
     def metadata(self) -> dict:
diff --git a/cirro/sdk/process.py b/cirro/sdk/process.py
index 71de2b6d..57a293af 100644
--- a/cirro/sdk/process.py
+++ b/cirro/sdk/process.py
@@ -111,7 +111,6 @@ def get_parameter_spec(self) -> ParameterSpecification:
         spec.print()
         samplesheet = spec.structured_inputs["samplesheet"]
         samplesheet.print()         # the columns each row takes
-        samplesheet.examples        # example rows
         ```
 
         Returns:
@@ -184,18 +183,26 @@ def run_analysis(
         )
         ```
 
-        Rows can also come from a sheet, whose file columns hold Cirro URIs:
+        A sheet's columns rarely match the structured input's exactly. Rename the
+        ones that differ and keep only the columns the structured input defines
+        if needed, or just pass sheet.to_dataframe() directly.
 
         ```python
-        samplesheet = project.get_sheet_by_name("Samplesheet").to_dataframe()
+        spec = process.get_parameter_spec().structured_inputs["samplesheet"]
+        sheet = project.get_sheet_by_name("Samples").to_dataframe()
+
+        samplesheet = (
+            sheet
+            # Sheet column -> structured input column
+            .rename(columns={"sample_id": "sample", "read1": "fastq_1", "read2": "fastq_2"})
+            # Drop sheet columns the structured input doesn't define (e.g., notes, tissue);
+            .filter(items=[column.key for column in spec.columns])
+        )
 
         process.run_analysis(
             name="Output",
             project_id=project.id,
-            structured_inputs={
-                # Drop the sheet's row identifier, which isn't a samplesheet column
-                "samplesheet": samplesheet.drop(columns="_row_id")
-            }
+            structured_inputs={"samplesheet": samplesheet}
         )
         ```
         """

From 7432ea31699cc8384097a769d10471ca2c1770d1 Mon Sep 17 00:00:00 2001
From: Nathan Thorpe <nthorpe@cirro.bio>
Date: Thu, 29 Oct 2026 16:17:50 -0700
Subject: [PATCH 06/10] update

---
 tests/test_structured_inputs.py | 10 +++++-----
 1 file changed, 5 insertions(+), 5 deletions(-)

diff --git a/tests/test_structured_inputs.py b/tests/test_structured_inputs.py
index badd7903..47e580c9 100644
--- a/tests/test_structured_inputs.py
+++ b/tests/test_structured_inputs.py
@@ -4,7 +4,7 @@
 import pandas as pd
 from cirro_api_client.v1.models import ProcessParameterSchema, FileFormat, CreateResponse
 
-from cirro.models.cirro_uri import format_cirro_uri, parse_cirro_uri, find_dataset_ids
+from cirro.models.cirro_uri import format_dataset_uri, parse_cirro_uri, find_dataset_ids
 from cirro.models.form_specification import ParameterSpecification
 from cirro.sdk.exceptions import DataPortalInputError
 from cirro.sdk.process import DataPortalProcess
@@ -66,21 +66,21 @@ def _spec() -> ParameterSpecification:
 
 
 def _row(sample='s1', dataset_id='dataset-1', **cells):
-    return {'sample': sample, 'fastq_1': format_cirro_uri(PROJECT_ID, dataset_id, f'{sample}_R1.fastq.gz'), **cells}
+    return {'sample': sample, 'fastq_1': format_dataset_uri(PROJECT_ID, dataset_id, f'{sample}_R1.fastq.gz'), **cells}
 
 
 class TestCirroUri(unittest.TestCase):
     def test_format_and_parse(self):
-        uri = format_cirro_uri(PROJECT_ID, 'dataset-1', 'data/folder/a.fastq.gz')
+        uri = format_dataset_uri(PROJECT_ID, 'dataset-1', 'data/folder/a.fastq.gz')
         self.assertEqual(uri, 'cirro::project-1:dataset:dataset-1/folder/a.fastq.gz')
         parsed = parse_cirro_uri(uri)
         self.assertEqual((parsed.project_id, parsed.id, parsed.path), (PROJECT_ID, 'dataset-1', 'folder/a.fastq.gz'))
-        self.assertIsNone(parse_cirro_uri(format_cirro_uri(PROJECT_ID, 'dataset-1')).path)
+        self.assertIsNone(parse_cirro_uri(format_dataset_uri(PROJECT_ID, 'dataset-1')).path)
         self.assertIsNone(parse_cirro_uri('s3://bucket/datasets/dataset-1/data/a.fastq.gz'))
 
     def test_find_dataset_ids(self):
         value = {'a': [_row(dataset_id='d1'), _row(dataset_id='d2'), _row(dataset_id='d1')],
-                 'b': format_cirro_uri('other-project', 'd3', '')}
+                 'b': format_dataset_uri('other-project', 'd3', '')}
         self.assertEqual(find_dataset_ids(value), ['d1', 'd2', 'd3'])
         self.assertEqual(find_dataset_ids(value, PROJECT_ID), ['d1', 'd2'])
 

From 8307bf0726acc0a683b0f9bfc1c2ca889b445ba7 Mon Sep 17 00:00:00 2001
From: Nathan Thorpe <nthorpe@cirro.bio>
Date: Thu, 29 Oct 2026 16:41:41 -0700
Subject: [PATCH 07/10] cleanup

---
 cirro/models/form_specification.py | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/cirro/models/form_specification.py b/cirro/models/form_specification.py
index cdac268d..e622b9a3 100644
--- a/cirro/models/form_specification.py
+++ b/cirro/models/form_specification.py
@@ -4,12 +4,11 @@
 import jsonschema
 from cirro_api_client.v1.models import FormSchema, ProcessParameterSchema, StructuredInputDefinition, \
     FileFormat, StructuredInput, StructuredInputDataItem, RunAnalysisRequestStructuredFileInputs
-from cirro_api_client.v1.types import Unset
 
 from cirro.models.cirro_uri import parse_cirro_uri
 
 if TYPE_CHECKING:
-    import pandas
+    pass
 
 # Rows for a structured input, as a list of dicts (one per row) or a pandas DataFrame
 StructuredInputRows = Union[List[Dict[str, Any]], 'pandas.DataFrame']

From 19d078793c98dbef8f88f84f2c9b0710d100cb94 Mon Sep 17 00:00:00 2001
From: Nathan Thorpe <nthorpe@cirro.bio>
Date: Thu, 29 Oct 2026 16:44:46 -0700
Subject: [PATCH 08/10] remove sys columns in to_dataframe

---
 cirro/sdk/sheet.py   | 14 ++++++++++++--
 tests/test_sheets.py | 24 ++++++++++++++++++++++++
 2 files changed, 36 insertions(+), 2 deletions(-)

diff --git a/cirro/sdk/sheet.py b/cirro/sdk/sheet.py
index 9917adab..1a4e0cd9 100644
--- a/cirro/sdk/sheet.py
+++ b/cirro/sdk/sheet.py
@@ -16,6 +16,9 @@
 # much wider than this can come back as a 502.
 ROWS_PER_PAGE = 1000
 
+# Columns the platform maintains on every sheet (e.g., sys_created_at)
+SYSTEM_COLUMN_PREFIX = "sys_"
+
 
 def query_to_dataframe(fetch_page: Callable[[int, int], SheetQueryResponse],
                        max_rows: int = None) -> 'DataFrame':
@@ -152,7 +155,8 @@ def __str__(self):
     def to_dataframe(self,
                      max_rows: int = None,
                      order_by: str = None,
-                     order: SqlSortOrder = None) -> 'DataFrame':
+                     order: SqlSortOrder = None,
+                     include_system_columns: bool = False) -> 'DataFrame':
         """
         Read the sheet's rows into a Pandas DataFrame.
 
@@ -165,13 +169,16 @@ def to_dataframe(self,
                 if omitted
             order_by (str): Column to sort by
             order (SqlSortOrder): Sort direction
+            include_system_columns (bool): Keep the columns the platform
+                maintains on every row (e.g., `sys_created_at`), which are
+                left out by default
 
         Returns:
             `pandas.DataFrame` indexed by `_row_id`
         """
         sort = SheetSort(order_by=order_by if order_by is not None else UNSET,
                          order=order if order is not None else UNSET)
-        return query_to_dataframe(
+        df = query_to_dataframe(
             lambda limit, page: self._client.sheets.query_data(
                 project_id=self.project_id,
                 sheet_id=self.id,
@@ -179,6 +186,9 @@ def to_dataframe(self,
             ),
             max_rows=max_rows
         ).set_index("_row_id")
+        if not include_system_columns:
+            df = df.loc[:, ~df.columns.str.startswith(SYSTEM_COLUMN_PREFIX)]
+        return df
 
     def insert_rows(self, rows: List[dict]) -> int:
         """
diff --git a/tests/test_sheets.py b/tests/test_sheets.py
index 0e2fffd4..1f77ec60 100644
--- a/tests/test_sheets.py
+++ b/tests/test_sheets.py
@@ -148,6 +148,30 @@ def test_sort_is_passed_through(self):
         self.assertEqual("icd_code", kwargs["request"].sort.order_by)
         self.assertEqual(SqlSortOrder.DESC, kwargs["request"].sort.order)
 
+    @staticmethod
+    def _client_with_system_columns():
+        columns = _COLUMNS + [
+            QueryColumn(name="sys_created_at", data_type=ColumnDataType.STRING),
+            QueryColumn(name="sys_updated_at", data_type=ColumnDataType.STRING)
+        ]
+        client = Mock()
+        client.sheets.query_data.return_value = SheetQueryResponse(
+            columns=columns,
+            rows=[[0, "G65", "2026-01-01T00:00:00Z", "2026-01-02T00:00:00Z"]],
+            total_row_count=1
+        )
+        return client
+
+    def test_system_columns_are_left_out(self):
+        df = _make_sheet(self._client_with_system_columns()).to_dataframe()
+
+        self.assertEqual(["icd_code"], list(df.columns))
+
+    def test_system_columns_can_be_kept(self):
+        df = _make_sheet(self._client_with_system_columns()).to_dataframe(include_system_columns=True)
+
+        self.assertEqual(["icd_code", "sys_created_at", "sys_updated_at"], list(df.columns))
+
 
 class NamespaceQueryTest(unittest.TestCase):
     """project.query_sheets pages a raw SQL result the same way."""

From 458e5c3adf11475f6fe20c9fe672d1b23a127c28 Mon Sep 17 00:00:00 2001
From: Nathan Thorpe <nthorpe@cirro.bio>
Date: Thu, 29 Oct 2026 16:54:35 -0700
Subject: [PATCH 09/10] fix type check

---
 cirro/models/form_specification.py | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/cirro/models/form_specification.py b/cirro/models/form_specification.py
index e622b9a3..5ffdd595 100644
--- a/cirro/models/form_specification.py
+++ b/cirro/models/form_specification.py
@@ -8,10 +8,10 @@
 from cirro.models.cirro_uri import parse_cirro_uri
 
 if TYPE_CHECKING:
-    pass
+    from pandas import DataFrame
 
 # Rows for a structured input, as a list of dicts (one per row) or a pandas DataFrame
-StructuredInputRows = Union[List[Dict[str, Any]], 'pandas.DataFrame']
+StructuredInputRows = Union[List[Dict[str, Any]], 'DataFrame']
 
 # JSON schema formats whose values are references to a file or folder in Cirro
 _PATH_FORMATS = {'file-path', 'file-path-pattern', 'directory-path', 'path'}

From e8b1ad3ef3673a67fcc85b7ea7d7107465505a76 Mon Sep 17 00:00:00 2001
From: Nathan Thorpe <nthorpe@cirro.bio>
Date: Fri, 30 Oct 2026 07:43:12 -0700
Subject: [PATCH 10/10] fix test

---
 tests/test_structured_inputs.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/tests/test_structured_inputs.py b/tests/test_structured_inputs.py
index 47e580c9..4e83ed26 100644
--- a/tests/test_structured_inputs.py
+++ b/tests/test_structured_inputs.py
@@ -169,7 +169,7 @@ def test_sends_structured_inputs_and_their_datasets(self):
     def test_requires_datasets_without_references(self):
         process, client = self._process()
         client.processes.get_parameter_spec.return_value = ParameterSpecification(
-            ProcessParameterSchema.from_dict({'form': {}, 'ui': {}})
+            ProcessParameterSchema.from_dict({'form': {}, 'ui': {}, 'structuredInputs': {}})
         )
         with self.assertRaisesRegex(DataPortalInputError, 'datasets'):
             process.run_analysis(name='Output', project_id=PROJECT_ID)