Fix schema-2 result storage provenance and provide safe repair

README.md

@@ -57,6 +57,76 @@ and attachment links remain valid. Directory-backed Zarr stores are available
 through result registration or the bulk results ZIP; individual file
 attachments skip those directories and report the skipped count.
 
+## Storage provenance and repair
+
+Shallow-result provenance uses the shared schema-2 manifest parser from
+`biomero-schema>=0.3.0b1,<0.4` (also required by the matching BIOMERO core).
+Plate annotations collect the source identities from the manifest's image
+bindings; Image annotations select their own image node. Both v2 and v3
+interchange profiles are read without converting the store. Unsupported
+manifest versions, inconsistent bindings or ambiguous references produce a
+metadata warning rather than guessed provenance.
+
+The import records storage evidence before rendering workflow annotations.
+If this step or annotation persistence fails, the script's `Message` reports
+`WARNING` with the complete/reduced/incomplete counts and points to the activity
+log. The imported data remains available; a completed import does not imply
+that all its metadata was written.
+
+For an existing result whose storage evidence is missing, an administrator can
+use `backfill_import_storage_provenance()` from the installed import script.
+Run this in the worker's Python environment, with an authenticated administrator
+`conn` in the result's group and the normal workflow event-store configuration:
+
+```python
+import json
+import runpy
+from pathlib import Path
+from types import SimpleNamespace
+from biomero import WorkflowTracker
+
+scripts_root = Path("/opt/omero/server/OMERO.server/lib/scripts/biomero")
+adapter = runpy.run_path(str(scripts_root / "_data/SLURM_Import_Results.py"))
+backfill = adapter["backfill_import_storage_provenance"]
+maintenance = runpy.run_path(str(scripts_root / "admin/SLURM_Init_environment.py"))
+refresh = maintenance["refresh_workflow_metadata"]
+# Select the existing result and the workflow that imported it.
+result_type, result_id, workflow_id = "Plate", 123, "your-workflow-uuid"
+
+with WorkflowTracker() as tracker:
+    client = SimpleNamespace(workflowTracker=tracker)
+    preview = backfill(conn, client, result_type, result_id, workflow_id)
+    print(json.dumps(preview, indent=2))
+    # After reviewing the preview, append its evidence:
+    # backfill(conn, client, result_type, result_id, workflow_id, dry_run=False)
+
+    # After recording, preview the corresponding OMERO annotation changes:
+    # plan = refresh(conn, tracker, result_type, result_id, workflow_id,
+    #                refresh_storage=True)
+    # print(json.dumps(plan, indent=2))
+    # Apply with a new backup filename after reviewing that plan:
+    # refresh(conn, tracker, result_type, result_id, workflow_id,
+    #         refresh_storage=True, dry_run=False,
+    #         backup_path="/data/biomero-metadata-backups/result-before-repair.json")
+```
+
+The default is a read-only preview. Apply validates the matching workflow
+annotation, managed storage evidence and unique import task, then appends the
+observed evidence to the event store. Identical evidence is not recorded twice.
+Missing files, mismatched workflows, conflicting recorded evidence and ambiguous
+results must be resolved before applying. Source and result pixels are never read, rewritten or
+renormalized by this operation.
+
+The explicit `refresh_storage=True` option requires a coordinated core build
+whose `plan_metadata_refresh()` supports that option. It advances only the
+import-task snapshot needed to expose newly recorded storage evidence, including
+its paired job maps. Other task snapshots remain pinned. Normal metadata
+refresh, including the Slurm Init UI, preserves the old snapshots and therefore
+does not expose later evidence by itself. See
+[metadata administration](https://nl-bioimaging.github.io/NL-BIOMERO/master/sysadmin/metadata-refresh.html)
+for backup and refresh controls. Recording evidence and refreshing maps are two
+separate steps. If refresh fails, retry it; repeating the backfill is also safe.
+
 ## Development
 
 Tests live on the separate

_data/SLURM_Import_Results.py

@@ -3217,9 +3217,10 @@ def add_metadata_to_imported_images(
                 message = f"Added workflow metadata to {metadata_added}/{len(imported_images)} imported images"
             else:
                 message = (
-                    f"Workflow metadata for {len(imported_images)} imported images: "
+                    f"WARNING: Workflow metadata for {len(imported_images)} imported images: "
                     f"{metadata_added} complete, {metadata_reduced} reduced, "
-                    f"{len(imported_images) - metadata_added - metadata_reduced} incomplete")
+                    f"{len(imported_images) - metadata_added - metadata_reduced} incomplete. "
+                    "Imported data remains available. Details: activity log (i button).")
         else:
             message = f"Matched {len(imported_images)} imported screen images to workflow inputs"
         messages.append(message)
@@ -3228,7 +3229,9 @@ def add_metadata_to_imported_images(
         return "\n".join(messages)
 
     except Exception as e:
-        error_msg = f"Failed to add metadata to imported images: {str(e)}"
+        error_msg = (
+            f"WARNING: Could not complete metadata for imported images: {e}. "
+            "Details: activity log (i button).")
         logger.error(error_msg, exc_info=True)
         return error_msg
 
@@ -3351,11 +3354,17 @@ def add_metadata_to_imported_plates(
             f"Workflow metadata for {len(imported_plates)} imported plates: "
             f"{metadata_added} complete, {metadata_reduced} reduced, "
             f"{len(imported_plates) - metadata_added - metadata_reduced} incomplete")
+        if metadata_added != len(imported_plates):
+            message = (
+                f"WARNING: {message}. Imported data remains available. "
+                "Details: activity log (i button).")
         logger.info(message)
         return message
 
     except Exception as e:
-        error_msg = f"Failed to add metadata to imported plates: {str(e)}"
+        error_msg = (
+            f"WARNING: Could not complete metadata for imported plates: {e}. "
+            "Details: activity log (i button).")
         logger.error(error_msg, exc_info=True)
         return error_msg
 
@@ -3566,7 +3575,10 @@ def collect_import_storage_provenance(conn, object_type, object_id, wf_id):
     import hashlib
 
     target = conn.getObject(object_type, object_id)
+    if target is None:
+        raise ValueError('Imported result target not found')
     refs, paths = [], set()
+    other_workflow_reference = False
     for ann in target.listAnnotations():
         ns = ann.getNs()
         if ns not in ('biomero.zarr.shallow',
@@ -3576,6 +3588,8 @@ def collect_import_storage_provenance(conn, object_type, object_id, wf_id):
         if ns == 'biomero.zarr.shallow':
             if values.get('workflowId') == str(wf_id):
                 refs.append(values)
+            else:
+                other_workflow_reference = True
         else:
             path = values.get('Imported_from') or values.get('Filepath')
             if path:
@@ -3589,6 +3603,8 @@ def collect_import_storage_provenance(conn, object_type, object_id, wf_id):
         root = (mount / reference['relativePath']).resolve()
         root.relative_to(mount)
     else:
+        if other_workflow_reference:
+            raise ValueError('Shallow reference belongs to another workflow')
         roots = set()
         for path in paths:
             candidate = Path(path).resolve()
@@ -3626,37 +3642,97 @@ def collect_import_storage_provenance(conn, object_type, object_id, wf_id):
     elif shallow:
         evidence['location'] = 'unknown'
     if shallow:
+        from biomero_schema.zarr import ShallowManifest
+
         raw = manifest.read_bytes()
-        collection = json.loads(raw)
-        if collection.get('workflowId') != str(wf_id):
+        collection = ShallowManifest.from_dict(json.loads(raw))
+        if str(collection.workflow_id) != str(wf_id):
             raise ValueError('Shallow manifest belongs to another workflow')
-        images = collection['images']
+        images = collection.collection.images
         if reference and 'imageNodePath' in reference:
             images = [image for image in images
-                      if image['imageNodePath'] == reference['imageNodePath']]
+                      if image.node_path == reference['imageNodePath']]
+        elif object_type == 'Image':
+            # Older imports may only have biomero.import/Filepath. Select the
+            # exact image node it identifies; a Plate root cannot identify one
+            # of its fields. Never assign all Plate biocodes to one Image.
+            image_paths = {Path(path).resolve() for path in paths}
+            images = [image for image in images
+                      if (root / image.node_path).resolve() in image_paths]
+        if object_type == 'Image' and len(images) != 1:
+            raise ValueError('Cannot identify a unique image in shallow manifest')
         if not images:
             raise ValueError('Result missing from shallow manifest')
         evidence.update(
             manifest='import-mount-data/' + manifest.relative_to(mount).as_posix(),
             manifest_sha256=hashlib.sha256(raw).hexdigest(),
-            source_biocodes=sorted({image['source']['pixelIdentity']['iscc']
+            source_biocodes=sorted({collection.image_binding(image.node_id)
+                                   .source.pixel_identity.iscc_code
                                    for image in images}))
     return evidence
 
 
+def storage_provenance_import_task(tracker, wf_id):
+    """Resolve one import task from durable workflow history."""
+    workflow = tracker.repository.get(uuid.UUID(str(wf_id)))
+    tasks = [tracker.repository.get(tid) for tid in workflow.tasks]
+    imports = [task for task in tasks if task.task_name == 'SLURM_Import_Results.py']
+    if len(imports) != 1:
+        raise ValueError('Cannot identify a unique import task for storage provenance')
+    return imports[0]
+
+
 def record_import_storage_provenance(conn, slurm_client, object_type, object_id, wf_id):
     """Persist observed result facts before the import-task metadata snapshot."""
     evidence = collect_import_storage_provenance(conn, object_type, object_id, wf_id)
     if evidence is None:
         return
     tracker = slurm_client.workflowTracker
-    workflow = tracker.repository.get(uuid.UUID(str(wf_id)))
-    tasks = [tracker.repository.get(tid) for tid in workflow.tasks]
-    imports = [task for task in tasks if task.task_name == 'SLURM_Import_Results.py']
-    if len(imports) != 1:
-        raise ValueError('Cannot identify a unique import task for storage provenance')
+    task = storage_provenance_import_task(tracker, wf_id)
     tracker.record_storage_provenance(
-        imports[0].id, f'{object_type}:{object_id}', evidence)
+        task.id, f'{object_type}:{object_id}', evidence)
+
+
+def backfill_import_storage_provenance(
+        conn, slurm_client, object_type, object_id, wf_id, *, dry_run=True):
+    """Preview or append missing storage evidence for an existing result.
+
+    Administrator-only, using the target's group context. Reads managed
+    metadata files, not pixel arrays. After apply, use the administrative
+    metadata refresh with refresh_storage=True to expose the new evidence.
+    Repeating either step is safe after interruption.
+    """
+    if not conn.isAdmin():
+        raise ValueError('Storage provenance backfill requires an administrator')
+    if object_type not in ('Image', 'Plate'):
+        raise ValueError('Only Image and Plate result targets are supported')
+    wf_id = str(uuid.UUID(str(wf_id)))
+    target = conn.getObject(object_type, int(object_id))
+    if target is None:
+        raise ValueError('Imported result target not found')
+    if not any(
+            ann.getNs() == 'biomero/workflow'
+            and ['Workflow_ID', wf_id] in [list(pair) for pair in ann.getValue()]
+            for ann in target.listAnnotations()):
+        raise ValueError('Target has no workflow metadata matching this workflow')
+    evidence = collect_import_storage_provenance(
+        conn, object_type, int(object_id), wf_id)
+    if evidence is None:
+        raise ValueError('No managed storage evidence is available for this result')
+    tracker = slurm_client.workflowTracker
+    task = storage_provenance_import_task(tracker, wf_id)
+    target_key = f'{object_type}:{int(object_id)}'
+    previous = getattr(task, 'storage_provenance', {}).get(target_key)
+    if previous is not None and previous != evidence:
+        raise ValueError('Import task already contains different storage evidence')
+    unchanged = previous == evidence
+    status = 'unchanged' if unchanged else 'planned'
+    if not dry_run and not unchanged:
+        tracker.record_storage_provenance(task.id, target_key, evidence)
+        status = 'recorded'
+    return {'target_key': target_key, 'workflow_id': wf_id,
+            'task_id': str(task.id), 'dry_run': dry_run, 'status': status,
+            'evidence': evidence, 'metadata_refresh_required': True}
 
 
 def add_object_annotations(conn, slurmClient, object_type, object_id, job_id, wf_id=None):

admin/SLURM_Init_environment.py

@@ -120,7 +120,7 @@ def _read_values(pairs):
 
 def refresh_workflow_metadata(conn, tracker, object_type, object_id, workflow_id,
                               *, view_version='v0', dry_run=True, backup_path=None,
-                              backup_enabled=True):
+                              backup_enabled=True, refresh_storage=False):
     """Refresh existing maps in place; unlink obsolete internal-task maps.
 
     Default is a dry run. Applying normally saves a new backup file; disabling
@@ -129,6 +129,8 @@ def refresh_workflow_metadata(conn, tracker, object_type, object_id, workflow_id
     annotation, image data or event is modified. Shared annotations are refused.
     Writes are not one transaction: errors propagate and the backup remains.
     Quiesce metadata writers for the target during an administrative refresh.
+    Explicit refresh_storage asks core to advance only the import-task snapshot
+    when new storage evidence has been backfilled. Normal refresh stays pinned.
     """
     if object_type not in ('Image', 'Plate'):
         raise ValueError('Only Image and Plate result targets are supported')
@@ -153,9 +155,11 @@ def refresh_workflow_metadata(conn, tracker, object_type, object_id, workflow_id
             rows.append(MetadataAnnotation(ns, values))
             records.append({'annotation_id': ann.getId(), 'namespace': ns,
                             'pairs': [list(p) for p in pairs]})
+        repair_options = {'refresh_storage': True} if refresh_storage else {}
         plan = plan_metadata_refresh(tracker, workflow_id, rows,
                                      view_version=view_version,
-                                     target_key=f'{object_type}:{object_id}')
+                                     target_key=f'{object_type}:{object_id}',
+                                     **repair_options)
         actions = []
         for record, change in zip(records, plan):
             pairs = metadata_pairs(change.after.values) if change.after else None
@@ -172,6 +176,8 @@ def refresh_workflow_metadata(conn, tracker, object_type, object_id, workflow_id
                                     'before_pairs': a['pairs'],
                                     'after_pairs': a['new_pairs'] or []}
                                    for a in actions]}
+        if refresh_storage:
+            summary['refresh_storage'] = True
         if dry_run:
             return summary
         if all(action['action'] == 'unchanged' for action in actions):