feat: add artifact create-from-plan cli

src/wf_api/surface.py

@@ -144,6 +144,21 @@ class WorkflowArtifactSurface(Protocol):
         version: int,
     ) -> dict[str, Any]: ...
 
+    async def create_artifact_from_plan(
+        self,
+        *,
+        artifact_id: str,
+        version: int,
+        title: str,
+        plan: dict[str, Any],
+        outcomes: Sequence[str],
+        kind: ArtifactKind = "workflow",
+        description: str | None = None,
+        required_capabilities: dict[str, dict[str, Any]] | None = None,
+        source_bindings: dict[str, str] | None = None,
+        created_from_catalog_version: str | None = None,
+    ) -> dict[str, Any]: ...
+
 
 class WorkflowDeploymentSurface(Protocol):
     """Deployment methods exposed by workflow frontends."""

src/wf_cli/commands/artifacts.py

@@ -1,12 +1,13 @@
 from __future__ import annotations
 
+from pathlib import Path
 from typing import Annotated, Literal
 
 import typer
 
 from wf_cli.context import load_cli_context_from_typer as load_cli_context
 from wf_cli.formats import ListOutputFormat, emit_list_payload
-from wf_cli.io import emit_json
+from wf_cli.io import CliInputError, emit_json, parse_bindings, parse_structured_file
 from wf_cli.remote_errors import run_cli_operation
 
 app = typer.Typer(
@@ -98,6 +99,52 @@ def _resolve_artifact_version(
     return version_arg
 
 
+@app.command("create-from-plan")
+def create_artifact_from_plan(
+    ctx: typer.Context,
+    plan_file: Annotated[Path, typer.Argument(exists=True, dir_okay=False)],
+    artifact_id: Annotated[str, typer.Option("--artifact", help="Artifact id.")],
+    version: Annotated[int, typer.Option("--version", min=1, help="Artifact version.")],
+    title: Annotated[str, typer.Option("--title", help="Artifact title.")],
+    outcome: Annotated[
+        list[str] | None,
+        typer.Option("--outcome", help="Artifact outcome. Repeatable."),
+    ] = None,
+    kind: Annotated[
+        Literal["workflow", "wrapper"], typer.Option("--kind", help="Artifact kind.")
+    ] = "workflow",
+    description: Annotated[
+        str | None, typer.Option("--description", help="Artifact description.")
+    ] = None,
+    binding: Annotated[
+        list[str] | None,
+        typer.Option("--binding", help="Logical=concrete source binding. Repeatable."),
+    ] = None,
+) -> None:
+    """Create an artifact from a raw JSON/YAML workflow plan file."""
+    try:
+        plan = parse_structured_file(plan_file)
+        source_bindings = parse_bindings(binding or [])
+    except CliInputError as exc:
+        raise typer.BadParameter(str(exc)) from exc
+    context = load_cli_context(ctx)
+    emit_json(
+        run_cli_operation(
+            context,
+            context.handlers.create_artifact_from_plan(
+                artifact_id=artifact_id,
+                version=version,
+                title=title,
+                plan=plan,
+                outcomes=tuple(outcome or ["ok"]),
+                kind=kind,
+                description=description,
+                source_bindings=source_bindings or None,
+            ),
+        )
+    )
+
+
 @app.command("delete")
 def delete_artifact(
     ctx: typer.Context,

src/wf_cli/io.py

@@ -4,6 +4,8 @@ import json
 from pathlib import Path
 from typing import Any
 
+import yaml
+
 
 class CliInputError(ValueError):
     """Raised when CLI JSON/file input cannot be parsed safely."""
@@ -58,6 +60,21 @@ def parse_bindings(bindings: list[str]) -> dict[str, str]:
     return parsed
 
 
+def parse_structured_file(path: Path) -> dict[str, Any]:
+    """Parse one JSON/YAML object file for declarative workflow inputs."""
+    try:
+        raw = path.read_text(encoding="utf-8")
+    except OSError as exc:
+        raise CliInputError(f"could not read file {path!s}: {exc}") from exc
+    try:
+        payload = yaml.safe_load(raw)
+    except yaml.YAMLError as exc:
+        raise CliInputError(f"invalid YAML/JSON file {path!s}: {exc}") from exc
+    if not isinstance(payload, dict):
+        raise CliInputError("structured file must contain an object")
+    return payload
+
+
 def _read_input_file(path: Path | None) -> str:
     """Read a required JSON input file."""
     if path is None:

tests/wf_cli/test_artifacts.py

@@ -36,6 +36,31 @@ class _ArtifactHandlers:
             "blocked_by_deployments": [],
         }
 
+    async def create_artifact_from_plan(
+        self,
+        *,
+        artifact_id: str,
+        version: int,
+        title: str,
+        plan: dict[str, Any],
+        outcomes: tuple[str, ...],
+        kind: str = "workflow",
+        description: str | None = None,
+        required_capabilities: dict[str, dict[str, Any]] | None = None,
+        source_bindings: dict[str, str] | None = None,
+        created_from_catalog_version: str | None = None,
+    ) -> dict[str, Any]:
+        self.calls.append((artifact_id, version))
+        return {
+            "artifact_id": artifact_id,
+            "version": version,
+            "title": title,
+            "plan_name": plan["name"],
+            "outcomes": list(outcomes),
+            "kind": kind,
+            "source_bindings": source_bindings,
+        }
+
 
 class _BlockedArtifactHandlers:
     def __init__(self) -> None:
@@ -140,3 +165,78 @@ def test_artifact_delete_blocked_returns_blocker_ids(monkeypatch) -> None:
     payload = json.loads(result.output)
     assert payload["deleted"] is False
     assert payload["blocked_by_deployments"] == ["echo.default"]
+
+
+def test_artifact_create_from_plan_reads_yaml(monkeypatch, tmp_path) -> None:
+    handlers = _ArtifactHandlers()
+    monkeypatch.setattr(
+        "wf_cli.commands.artifacts.load_cli_context",
+        lambda _ctx: _Context(handlers=handlers),
+    )
+    plan_file = tmp_path / "plan.yaml"
+    plan_file.write_text(
+        """
+name: yaml_plan
+input_schema: {type: object, properties: {}}
+state_schema: {type: object, properties: {}}
+output_schema: {type: object, properties: {}}
+outcomes: [ok]
+start: __end__
+nodes: []
+edges: []
+""",
+        encoding="utf-8",
+    )
+
+    result = CliRunner().invoke(
+        app,
+        [
+            "artifact",
+            "create-from-plan",
+            str(plan_file),
+            "--artifact",
+            "yaml_artifact",
+            "--version",
+            "1",
+            "--title",
+            "YAML Artifact",
+            "--outcome",
+            "ok",
+            "--binding",
+            "local.ops=local.ops",
+        ],
+    )
+
+    assert result.exit_code == 0, result.output
+    payload = json.loads(result.output)
+    assert payload["artifact_id"] == "yaml_artifact"
+    assert payload["plan_name"] == "yaml_plan"
+    assert payload["source_bindings"] == {"local.ops": "local.ops"}
+
+
+def test_artifact_create_from_plan_rejects_non_object_yaml(monkeypatch, tmp_path) -> None:
+    handlers = _ArtifactHandlers()
+    monkeypatch.setattr(
+        "wf_cli.commands.artifacts.load_cli_context",
+        lambda _ctx: _Context(handlers=handlers),
+    )
+    plan_file = tmp_path / "plan.yaml"
+    plan_file.write_text("- not\n- object\n", encoding="utf-8")
+
+    result = CliRunner().invoke(
+        app,
+        [
+            "artifact",
+            "create-from-plan",
+            str(plan_file),
+            "--artifact",
+            "bad",
+            "--version",
+            "1",
+            "--title",
+            "Bad",
+        ],
+    )
+
+    assert result.exit_code != 0
+    assert "structured file must contain an object" in result.output.lower()