Add tolid broker

README.md

@@ -20,6 +20,7 @@ A dedicated broker workflow enables integration with external submission pipelin
   - Claim drafts and obtain a lease: `/api/v1/broker/organisms/{taxon_id}/claim`
   - ENA broker contract endpoints: `/api/v1/broker/claims/ready`, `/api/v1/broker/claims/entity`, `/api/v1/broker/validation`, `/api/v1/broker/reports/{attempt_id}`
   - Renew lease, finalise, and report results: `/api/v1/broker/attempts/{attempt_id}/...`
+  - Lightweight ToLID persistence/reporting endpoints: `/api/v1/broker/tolids/...`
   - Attempt listing and summaries for dashboard views
 - Bulk import endpoints for organisms, samples, and experiments
 - XML export endpoints for downstream systems
@@ -238,6 +239,7 @@ pyproject.toml, uv.lock, docker-compose.yml, Dockerfile, schema.sql, scripts/, d
     ```
 
   For the flat ENA broker contract used by Canopy, see [docs/ena_broker_contract.md](docs/ena_broker_contract.md).
+  For the lightweight ToLID persistence/reporting flow, see [docs/tolid_broker_api.md](docs/tolid_broker_api.md).
   For a deeper overview of attempt leasing and statuses, see the `broker` endpoints in `app/api/v1/endpoints/broker.py` and the interactive docs.
 
   ## Bulk Import API

alembic/versions/0005_add_tolid_requests.py

@@ -0,0 +1,101 @@
+"""Add durable ToLID request state.
+
+Revision ID: 0005_add_tolid_requests
+Revises: 0004_qc_reads_assembly_refs
+Create Date: 2026-06-16
+"""
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+from alembic import op
+
+revision = "0005_add_tolid_requests"
+down_revision = "0004_qc_reads_assembly_refs"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    tolid_status = postgresql.ENUM(
+        "not_requested",
+        "pending",
+        "assigned",
+        "failed",
+        name="tolid_request_status",
+    )
+    tolid_status.create(op.get_bind(), checkfirst=True)
+
+    op.create_table(
+        "tolid_request",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, nullable=False),
+        sa.Column(
+            "sample_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("sample.id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("tolid_external_id", sa.Text(), nullable=False),
+        sa.Column(
+            "taxon_id",
+            sa.Integer(),
+            sa.ForeignKey("organism.taxon_id", ondelete="CASCADE"),
+            nullable=False,
+        ),
+        sa.Column("scientific_name", sa.Text(), nullable=True),
+        sa.Column("tolid", sa.Text(), nullable=True),
+        sa.Column("request_id", sa.Text(), nullable=True),
+        sa.Column(
+            "status",
+            postgresql.ENUM(
+                "not_requested",
+                "pending",
+                "assigned",
+                "failed",
+                name="tolid_request_status",
+                create_type=False,
+            ),
+            nullable=False,
+            server_default="not_requested",
+        ),
+        sa.Column("last_requested_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("error_message", sa.Text(), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            nullable=False,
+            server_default=sa.func.now(),
+        ),
+    )
+
+    op.create_index("uq_tolid_request_sample_id", "tolid_request", ["sample_id"], unique=True)
+    op.create_index("idx_tolid_request_status", "tolid_request", ["status"])
+    op.create_index(
+        "idx_tolid_request_status_last_requested_at",
+        "tolid_request",
+        ["status", "last_requested_at"],
+    )
+    op.create_index("idx_tolid_request_request_id", "tolid_request", ["request_id"])
+
+
+def downgrade() -> None:
+    op.drop_index("idx_tolid_request_request_id", table_name="tolid_request")
+    op.drop_index("idx_tolid_request_status_last_requested_at", table_name="tolid_request")
+    op.drop_index("idx_tolid_request_status", table_name="tolid_request")
+    op.drop_index("uq_tolid_request_sample_id", table_name="tolid_request")
+    op.drop_table("tolid_request")
+
+    tolid_status = postgresql.ENUM(
+        "not_requested",
+        "pending",
+        "assigned",
+        "failed",
+        name="tolid_request_status",
+    )
+    tolid_status.drop(op.get_bind(), checkfirst=True)

app/api/v1/endpoints/broker.py

@@ -38,6 +38,8 @@
     BrokerValidationRequest,
     BrokerValidationResponse,
 )
+from app.schemas.tolid import TolidRequestBrokerView, TolidRequestReport, TolidRequestStatus
+from app.services.tolid_service import tolid_request_service
 
 router = APIRouter()
 CLAIMABLE_SUBMISSION_STATES = ("draft", "ready")
@@ -2229,7 +2231,6 @@ def report_results(
             # On conflict by (authority, accession) or (authority, entity_type, entity_id), do nothing
             stmt = stmt.on_conflict_do_nothing(index_elements=[AccessionRegistry.accession])
             db.execute(stmt)
-
         # Clear lease on finalise (anything other than submitting)
         if item.status != "submitting":
             sub.attempt_id = None
@@ -2504,6 +2505,73 @@ def report_results(
     )
 
 
+@router.get("/tolids/by-specimen-accession/{specimen_id}", response_model=TolidRequestBrokerView)
+@policy("broker:read")
+def get_tolid_by_specimen_accession(
+    *,
+    specimen_id: str,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_active_user),
+) -> TolidRequestBrokerView:
+    """Resolve a specimen sample by ENA accession and return current or virtual ToLID state."""
+    return tolid_request_service.get_by_specimen_accession(db, specimen_id)
+
+
+@router.get("/tolids/pending", response_model=List[TolidRequestBrokerView])
+@policy("broker:read")
+def get_pending_tolids(
+    *,
+    db: Session = Depends(get_db),
+    taxon_id: Optional[int] = Query(None, description="Filter by organism taxon_id"),
+    sample_id: Optional[UUID] = Query(None, description="Filter by sample_id"),
+    sample_ids: Optional[List[UUID]] = Query(None, description="Filter by sample_ids"),
+    requested_before: Optional[datetime] = Query(
+        None, description="Return only rows requested before this timestamp"
+    ),
+    skip: int = Query(0, ge=0),
+    limit: int = Query(100, ge=1, le=1000),
+    current_user: User = Depends(get_current_active_user),
+) -> List[TolidRequestBrokerView]:
+    """Return specimen ToLID rows still waiting on remote assignment."""
+    return tolid_request_service.list_rows(
+        db,
+        row_status=TolidRequestStatus.PENDING,
+        taxon_id=taxon_id,
+        sample_id=sample_id,
+        sample_ids=sample_ids,
+        requested_before=requested_before,
+        skip=skip,
+        limit=limit,
+    )
+
+
+@router.get("/tolids/{sample_id}", response_model=TolidRequestBrokerView)
+@policy("broker:read")
+def get_tolid_by_sample(
+    *,
+    sample_id: UUID,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_active_user),
+) -> TolidRequestBrokerView:
+    """Return the durable ToLID state for one specimen sample."""
+    return tolid_request_service.get_row_for_sample(db, sample_id)
+
+
+@router.post("/tolids/{sample_id}/report", response_model=TolidRequestBrokerView)
+@policy("broker:claim")
+def report_tolid_result(
+    *,
+    sample_id: UUID,
+    payload: TolidRequestReport = Body(...),
+    db: Session = Depends(get_db),
+    current_user: User = Depends(get_current_active_user),
+) -> TolidRequestBrokerView:
+    """Persist the broker-reported ToLID result for one specimen sample."""
+    row = tolid_request_service.report_result(db, sample_id=sample_id, report=payload)
+    db.commit()
+    return row
+
+
 # ---------- Dashboard Helpers ----------
 def expire_stale_leases(db: Session) -> Dict[str, int]:
     """Expire all submissions with expired leases by resetting them to 'draft' status.

app/api/v1/endpoints/samples.py

@@ -518,14 +518,14 @@ def _create_sample_with_submission(
 def bulk_import_specimen_samples(
     *,
     db: Session = Depends(get_db),
-    samples_data: Dict[str, Dict[str, Any]],
+    samples_data: Dict[str, Dict[str, Dict[str, Any]]],
     current_user: User = Depends(get_current_active_user),
 ) -> Any:
     """
     Bulk import specimen samples (kind='specimen').
 
-    Expected format: Dictionary keyed by sample_key (a concat of taxon_id and specimen_id).
-    Each sample must have taxon_id and specimen_id.
+    Expected format: Dictionary keyed by taxon_id, then specimen_id.
+    The nested keys supply taxon_id and specimen_id for each sample.
     Enforces uniqueness constraint: one specimen per (taxon_id, specimen_id).
     """
     # Load the ENA-ATOL mapping file
@@ -541,69 +541,97 @@ def bulk_import_specimen_samples(
     skipped_count = 0
     errors = []
 
-    for sample_key, sample_data in samples_data.items():
+    for taxon_key, specimen_map in samples_data.items():
+        if not isinstance(specimen_map, dict):
+            errors.append(f"{taxon_key}: Expected an object keyed by specimen_id")
+            skipped_count += 1
+            continue
+
         try:
-            # Get organism reference
-            taxon_id = sample_data.get("taxon_id")
-            if taxon_id is None:
-                errors.append(f"{sample_key}: Missing taxon_id")
-                skipped_count += 1
-                continue
+            taxon_id = int(taxon_key)
+        except (TypeError, ValueError):
+            errors.append(f"{taxon_key}: Invalid taxon_id key")
+            skipped_count += len(specimen_map) or 1
+            continue
 
-            organism = db.query(Organism).filter(Organism.taxon_id == int(taxon_id)).first()
-            if not organism:
-                errors.append(f"{sample_key}: Organism not found with taxon_id '{taxon_id}'")
-                skipped_count += 1
-                continue
+        organism = db.query(Organism).filter(Organism.taxon_id == taxon_id).first()
+        if not organism:
+            errors.append(f"{taxon_key}: Organism not found with taxon_id '{taxon_id}'")
+            skipped_count += len(specimen_map) or 1
+            continue
 
-            organism_taxon_id = _organism_taxon_id(organism)
+        organism_taxon_id = _organism_taxon_id(organism)
 
-            # Validate specimen_id is present
-            specimen_id = sample_data.get("specimen_id")
-            if not specimen_id:
-                errors.append(f"{sample_key}: specimen_id is required for specimen samples")
-                skipped_count += 1
-                continue
+        for specimen_key, raw_sample_data in specimen_map.items():
+            sample_key = f"{taxon_key}/{specimen_key}"
+            try:
+                if not isinstance(raw_sample_data, dict):
+                    errors.append(f"{sample_key}: Expected sample payload object")
+                    skipped_count += 1
+                    continue
 
-            # Check for duplicate specimen
-            existing_specimen = (
-                db.query(Sample)
-                .filter(
-                    Sample.taxon_id == organism_taxon_id,
-                    Sample.specimen_id == specimen_id,
-                    Sample.kind == SampleKind.SPECIMEN,
-                )
-                .first()
-            )
-            if existing_specimen:
-                errors.append(
-                    f"{sample_key}: Specimen already exists for taxon_id '{organism_taxon_id}' "
-                    f"and specimen_id '{specimen_id}'"
+                sample_data = dict(raw_sample_data)
+                specimen_id = specimen_key
+                if not specimen_id:
+                    errors.append(f"{sample_key}: specimen_id key is required for specimen samples")
+                    skipped_count += 1
+                    continue
+
+                if sample_data.get("taxon_id") not in (None, organism_taxon_id):
+                    errors.append(
+                        f"{sample_key}: taxon_id in payload does not match outer key '{taxon_key}'"
+                    )
+                    skipped_count += 1
+                    continue
+
+                if sample_data.get("specimen_id") not in (None, specimen_id):
+                    errors.append(
+                        f"{sample_key}: specimen_id in payload does not match nested key '{specimen_id}'"
+                    )
+                    skipped_count += 1
+                    continue
+
+                sample_data["taxon_id"] = organism_taxon_id
+                sample_data["specimen_id"] = specimen_id
+
+                existing_specimen = (
+                    db.query(Sample)
+                    .filter(
+                        Sample.taxon_id == organism_taxon_id,
+                        Sample.specimen_id == specimen_id,
+                        Sample.kind == SampleKind.SPECIMEN,
+                    )
+                    .first()
                 )
-                skipped_count += 1
-                continue
-            sample_data["organism_part"] = "WHOLE ORGANISM"
+                if existing_specimen:
+                    errors.append(
+                        f"{sample_key}: Specimen already exists for taxon_id '{organism_taxon_id}' "
+                        f"and specimen_id '{specimen_id}'"
+                    )
+                    skipped_count += 1
+                    continue
 
-            # Create specimen sample (bpa_sample_id is optional for specimens)
-            sample, sample_submission = _create_sample_with_submission(
-                db=db,
-                bpa_sample_id=sample_data.get("bpa_sample_id"),  # Optional for specimens
-                sample_data=sample_data,
-                taxon_id=organism_taxon_id,
-                kind=SampleKind.SPECIMEN,
-                derived_from_sample_id=None,
-                ena_atol_map=ena_atol_map,
-            )
+                sample_data["organism_part"] = "WHOLE ORGANISM"
 
-            db.add(sample)
-            db.add(sample_submission)
-            db.commit()
-            created_count += 1
+                sample, sample_submission = _create_sample_with_submission(
+                    db=db,
+                    bpa_sample_id=sample_data.get("bpa_sample_id"),
+                    sample_data=sample_data,
+                    taxon_id=organism_taxon_id,
+                    kind=SampleKind.SPECIMEN,
+                    derived_from_sample_id=None,
+                    ena_atol_map=ena_atol_map,
+                )
 
-        except Exception as e:
-            errors.append(f"{sample_key}: {str(e)}")
-            db.rollback()
-            skipped_count += 1
+                db.add(sample)
+                db.add(sample_submission)
+                db.commit()
+                created_count += 1
+
+            except Exception as e:
+                errors.append(f"{sample_key}: {str(e)}")
+                db.rollback()
+                skipped_count += 1
 
     message = f"Specimen import complete. Created: {created_count}, Skipped: {skipped_count}"
 

app/models/__init__.py

@@ -11,4 +11,5 @@
 from app.models.sample import Sample, SampleSubmission
 from app.models.taxonomy_info import TaxonomyInfo
 from app.models.token import RefreshToken
+from app.models.tolid_request import TolidRequest
 from app.models.user import User

app/models/project.py

@@ -84,6 +84,7 @@ class ProjectSubmission(Base):
     response_payload = Column(JSONB, nullable=True)
 
     accession = Column(Text, nullable=True)
+    submitted_at = Column(DateTime(timezone=True), nullable=True)
 
     created_at = Column(DateTime(timezone=True), nullable=False, server_default=func.now())
     updated_at = Column(