feat: Milestone 5 - workflow manifest integration and example workflows

databusclient/cli.py

@@ -583,30 +583,67 @@ def workflow():
 
 @workflow.command("run")
 @click.argument("workflow_path", type=click.Path(exists=True, dir_okay=False))
-def workflow_run(workflow_path):
+@click.option(
+    "--manifest",
+    "manifest_path",
+    default=None,
+    help="Write a unified JSON-LD manifest of the entire workflow run to PATH.",
+)
+def workflow_run(workflow_path, manifest_path):
     """
     Run a declarative workflow pipeline from a YAML file.
 
     Executes each step in order, chaining outputs between steps via
     ${steps.name.output_files}-style references, and applying each
-    step's on_error behavior (fail/continue/retry).
+    step's on_error behavior (fail/continue/retry). Prints a console
+    summary after every run. Use --manifest to also write a unified
+    JSON-LD manifest covering every step.
     """
     try:
         parsed = parse_workflow(workflow_path)
     except WorkflowParseError as e:
         raise click.ClickException(str(e))
 
-    context = StepContext()
-    engine = WorkflowEngine(context=context)
+    # CLI flag takes priority; falls back to the YAML file's own
+    # top-level 'manifest:' key if --manifest was not given on the
+    # command line.
+    if manifest_path is None:
+        manifest_path = parsed.get("manifest")
 
+    # A workflow-level manifest is always built internally (for the
+    # automatic console summary), even when no manifest path is set.
+    # It's only written to disk when a path is provided (via --manifest
+    # or the YAML file's own 'manifest:' key).
+    manifest_ctx = ManifestContext(command="workflow")
+
+    step_context = StepContext()
+    engine = WorkflowEngine(context=step_context, manifest_context=manifest_ctx)
+
+    workflow_error = None
     try:
-        results = engine.run(parsed["steps"])
+        engine.run(parsed["steps"])
     except WorkflowExecutionError as e:
-        raise click.ClickException(str(e))
+        workflow_error = e
+    finally:
+        click.echo("Workflow complete." if workflow_error is None else "Workflow failed.")
+        for result in engine.results:
+            click.echo(f"  {result.name}: {result.status}")
+
+        click.echo("")
+        click.echo(format_summary(ManifestWriter.build_manifest_dict(manifest_ctx)))
+
+        if manifest_path:
+            try:
+                actual_path = ManifestWriter.write(manifest_ctx, manifest_path)
+                click.echo(f"\nManifest written to {actual_path}")
+            except (OSError, IOError) as e:
+                click.echo(
+                    f"WARNING: Manifest could not be written to {manifest_path}: {e}",
+                    err=True,
+                )
 
-    click.echo("Workflow complete.")
-    for result in results:
-        click.echo(f"  {result.name}: {result.status}")
+    if workflow_error is not None:
+        raise click.ClickException(str(workflow_error))
 
 if __name__ == "__main__":
     app()

databusclient/manifest/context.py

@@ -156,4 +156,25 @@ def summary(self) -> dict:
             "succeeded": succeeded,
             "failed": failed,
             "total_bytes": total_bytes,
-        }
+        }
+
+    def merge_from(self, other: "ManifestContext", step_name: Optional[str] = None) -> None:
+        """Merge another context's recorded files into this one.
+
+        Used by the workflow engine: each step records into its own
+        temporary ManifestContext (so per-step failures/successes stay
+        isolated), then that context's entries are merged into the
+        workflow-level master context here, tagged with which step
+        produced them.
+
+        Args:
+            other: The ManifestContext to merge entries from.
+            step_name: If given, tags each merged file entry with
+                "step": step_name, so a multi-step workflow manifest
+                remains traceable to which step produced which file.
+        """
+        for entry in other.files:
+            merged_entry = dict(entry)
+            if step_name is not None:
+                merged_entry["step"] = step_name
+            self.files.append(merged_entry)

databusclient/manifest/replay.py

@@ -294,6 +294,12 @@ def replay_manifest(
     if not command:
         raise ManifestReplayError("Manifest missing required field dbus:command.")
 
+    if command not in ("download", "delete", "deploy"):
+        raise ManifestReplayError(
+            f"Replay for command '{command}' is not implemented yet. "
+            "Currently supported: download, delete, deploy."
+        )
+
     replay_params = _validate_replay_params(manifest.get("dbus:replayParams"))
 
     if command == "download":
@@ -305,9 +311,4 @@ def replay_manifest(
         return _replay_delete(replay_params, overrides, confirm_fn)
 
     if command == "deploy":
-        return _replay_deploy(replay_params, overrides)
-
-    raise ManifestReplayError(
-        f"Replay for command '{command}' is not implemented yet. "
-        "Currently supported: download, delete, deploy."
-    )
+        return _replay_deploy(replay_params, overrides)

databusclient/manifest/summary.py

@@ -39,7 +39,7 @@ def format_summary(manifest: Dict[str, Any]) -> str:
 
     Args:
         manifest: A manifest dict, as produced by loading a manifest
-            JSON-LD file (e.g. via replay._load_manifest).
+            JSON-LD file (e.g. via replay.load_manifest).
 
     Returns:
         A formatted multi-line string ready to print to the console.
@@ -81,4 +81,18 @@ def format_summary(manifest: Dict[str, Any]) -> str:
         else:
             lines.append(f"Error    : {error_message}")
 
+    failed_files = [
+        f for f in manifest.get("dataid:distribution", {}).get("dataid:file", [])
+        if f.get("dbus:status") == "failed"
+    ]
+    if failed_files:
+        lines.append("")
+        lines.append("Failed files:")
+        for f in failed_files:
+            step = f.get("dbus:stepName")
+            url = f.get("dcat:downloadURL", "unknown")
+            error_message = f.get("dbus:errorMessage", "no error message recorded")
+            prefix = f"  [{step}] " if step else "  "
+            lines.append(f"{prefix}{url}: {error_message}")
+
     return "\n".join(lines)

databusclient/manifest/writer.py

@@ -30,31 +30,14 @@ class ManifestWriter:
     """Serializes a ManifestContext to a JSON-LD manifest file."""
 
     @staticmethod
-    def write(context: ManifestContext, path: str) -> str:
-        """Write the manifest to a JSON-LD file at the given path.
-
-        Creates parent directories if they do not exist.
-        If a file already exists at `path`, auto-suffixes with _1, _2, etc.
-        and prints a warning rather than silently overwriting.
-        On failure, raises OSError — callers should catch and warn.
-
-        Args:
-            context: The completed ManifestContext to serialize.
-            path: File path to write the manifest to.
-
-        Raises:
-            OSError: If the file cannot be written, or if path is a directory.
+    def build_manifest_dict(context: ManifestContext) -> dict:
+        """Build the JSON-LD manifest dict from a context, without writing
+        to disk. Extracted from write() so callers (like the workflow
+        engine's automatic console summary) can get the dict without
+        needing a file path.
         """
-        if path.endswith(("/", "\\")) or os.path.isdir(path):
-            stripped = path.rstrip("/\\")
-            raise OSError(
-                f"--manifest path '{path}' is a directory, not a file. "
-                f"Please provide a full file path, e.g. '{stripped}/manifest.jsonld'."
-            )
-
         summary = context.summary()
 
-        # Build file entries using DataID vocabulary
         file_entries = []
         for f in context.files:
             entry: dict = {
@@ -79,6 +62,8 @@ def write(context: ManifestContext, path: str) -> str:
                 entry["dbus:errorTraceback"] = f["error_traceback"]
             if f.get("retry_count"):
                 entry["dbus:retryCount"] = f["retry_count"]
+            if f.get("step"):
+                entry["dbus:stepName"] = f["step"]
             file_entries.append(entry)
 
         manifest = {
@@ -121,6 +106,33 @@ def write(context: ManifestContext, path: str) -> str:
                 "dbus:errorTraceback": context.operation_error["error_traceback"],
             }
 
+        return manifest
+
+    @staticmethod
+    def write(context: ManifestContext, path: str) -> str:
+        """Write the manifest to a JSON-LD file at the given path.
+
+        Creates parent directories if they do not exist.
+        If a file already exists at `path`, auto-suffixes with _1, _2, etc.
+        and prints a warning rather than silently overwriting.
+        On failure, raises OSError — callers should catch and warn.
+
+        Args:
+            context: The completed ManifestContext to serialize.
+            path: File path to write the manifest to.
+
+        Raises:
+            OSError: If the file cannot be written, or if path is a directory.
+        """
+        if path.endswith(("/", "\\")) or os.path.isdir(path):
+            stripped = path.rstrip("/\\")
+            raise OSError(
+                f"--manifest path '{path}' is a directory, not a file. "
+                f"Please provide a full file path, e.g. '{stripped}/manifest.jsonld'."
+            )
+
+        manifest = ManifestWriter.build_manifest_dict(context)
+
         parent = os.path.dirname(os.path.abspath(path))
         if parent:
             os.makedirs(parent, exist_ok=True)

databusclient/workflow/engine.py

@@ -10,8 +10,9 @@
 from __future__ import annotations
 
 import time
-from typing import Any, Dict, List
+from typing import Any, Dict, List, Optional
 
+from databusclient.manifest.context import ManifestContext
 from databusclient.workflow.context import StepContext
 from databusclient.workflow.steps import STEP_REGISTRY
 
@@ -32,10 +33,25 @@ def __init__(self, name: str, status: str, error: Exception | None = None,
 
 
 class WorkflowEngine:
-    """Runs a parsed workflow's steps in order, handling errors per step."""
-
-    def __init__(self, context: StepContext | None = None) -> None:
+    """Runs a parsed workflow's steps in order, handling errors per step.
+
+    If manifest_context is given, one unified manifest is built for the
+    entire workflow run: each step gets its own temporary ManifestContext
+    (isolating its recorded files/errors), which is merged into
+    manifest_context afterward, tagged with the step's name. This lets a
+    single workflow manifest remain traceable to which step produced or
+    failed on which file, without touching download.py/deploy.py/delete.py
+    at all -- those already accept manifest_context=None as a no-op, and
+    here they simply receive a real (temporary, per-step) one instead.
+    """
+
+    def __init__(
+        self,
+        context: StepContext | None = None,
+        manifest_context: Optional[ManifestContext] = None,
+    ) -> None:
         self.context = context or StepContext()
+        self.manifest_context = manifest_context
         self.results: List[StepResult] = []
 
     def run(self, steps: List[Dict[str, Any]]) -> List[StepResult]:
@@ -78,32 +94,84 @@ def _run_step_with_error_handling(self, step_config: Dict[str, Any]) -> StepResu
         if on_error == "retry":
             return self._run_with_retry(name, step, step_config)
 
+        step_manifest_ctx = self._start_step_manifest(command)
         try:
             step.run(step_config, self.context)
+            self._finish_step_manifest(step_manifest_ctx, name)
             return StepResult(name, "success")
         except Exception as exc:
+            self._finish_step_manifest(step_manifest_ctx, name, error=exc)
             if on_error == "continue":
                 print(f"WARNING: step '{name}' failed and on_error is 'continue': {exc}")
                 return StepResult(name, "skipped_error", error=exc)
             # on_error == "fail" (or missing/defaulted to fail)
             return StepResult(name, "failed", error=exc)
 
+    def _start_step_manifest(self, command: str) -> Optional[ManifestContext]:
+        """If a workflow-level manifest is active, give this step its own
+        temporary ManifestContext to record into. Returns None if no
+        workflow manifest was requested -- in that case self.context's
+        manifest_context is left as whatever it already was (e.g. a step's
+        own throwaway context, like DownloadStep uses for output_urls).
+        """
+        if self.manifest_context is None:
+            return None
+        step_ctx = ManifestContext(command=command)
+        self.context.manifest_context = step_ctx
+        return step_ctx
+
+    def _finish_step_manifest(
+        self,
+        step_manifest_ctx: Optional[ManifestContext],
+        step_name: str,
+        error: Optional[Exception] = None,
+    ) -> None:
+        """Merge a completed step's temporary manifest entries into the
+        workflow-level master manifest, tagged with the step name. If the
+        step failed, also record a synthetic entry so the failure is
+        visible in the manifest even if the step recorded no per-file
+        entries before failing. The synthetic entry is tagged with the
+        same "step" field merge_from() uses, so format_summary()'s
+        [stepname] prefix mechanism works consistently for BOTH per-file
+        failures (merged from a step's own context) and whole-step
+        failures (no file-level detail available at all) -- previously
+        only the merged case was tagged, so whole-step failures (like an
+        auth error before any file work happens) showed up without the
+        [stepname] prefix, relying on the step name being embedded in a
+        fake url string instead.
+        """
+        if self.manifest_context is None or step_manifest_ctx is None:
+            return
+        self.manifest_context.merge_from(step_manifest_ctx, step_name=step_name)
+        if error is not None:
+            self.manifest_context.record_file(
+                url="(no file-level detail -- step failed before producing one)",
+                status="failed",
+                error_message=str(error),
+            )
+            self.manifest_context.files[-1]["step"] = step_name
+
     def _run_with_retry(self, name: str, step: Any, step_config: Dict[str, Any]) -> StepResult:
         retry_config = step_config["retry"]
         max_attempts = retry_config["max_attempts"]
         delay_seconds = retry_config["delay_seconds"]
+        command = step_config["command"]
 
         last_error: Exception | None = None
         for attempt in range(1, max_attempts + 1):
+            step_manifest_ctx = self._start_step_manifest(command)
             try:
                 step.run(step_config, self.context)
+                self._finish_step_manifest(step_manifest_ctx, name)
                 return StepResult(name, "success", attempts=attempt)
             except Exception as exc:
                 last_error = exc
                 print(
                     f"WARNING: step '{name}' attempt {attempt}/{max_attempts} "
                     f"failed: {exc}"
                 )
+                if attempt == max_attempts:
+                    self._finish_step_manifest(step_manifest_ctx, name, error=exc)
                 if attempt < max_attempts:
                     time.sleep(delay_seconds)
 

databusclient/workflow/steps.py

@@ -153,6 +153,14 @@ def run(self, step_config: Dict[str, Any], context: StepContext) -> None:
         context.set_output(name, "output_files", output_files)
         context.set_output(name, "version_id", resolved["version_id"])
 
+        # deploy()/deploy_from_metadata() do not accept manifest_context
+        # (unlike download()/delete()) -- manifest recording for deploy is
+        # always done manually by the caller. This mirrors exactly what
+        # cli.py's own `deploy` command does after a successful deploy.
+        if context.manifest_context is not None:
+            for url in output_files:
+                context.manifest_context.record_file(url=url, status="success")
+
     def _run_classic_mode(self, resolved: Dict[str, Any], name: str) -> list:
         files = resolved.get("files")
         if not files: