feat: add draft lifecycle cli commands

src/wf_cli/commands/draft_options.py

@@ -100,6 +100,14 @@ def parse_json_file(path: Path, *, option_name: str) -> Any:
         ) from exc
 
 
+def parse_json_object_file(path: Path, *, option_name: str) -> dict[str, Any]:
+    """Read a JSON object file used by a workflow schema option."""
+    value = parse_json_file(path, option_name=option_name)
+    if not isinstance(value, dict):
+        raise typer.BadParameter(f"{option_name}: expected a JSON object")
+    return value
+
+
 def route_source(from_step: str | None, from_outcome: str | None) -> RouteSource | None:
     if from_step is None:
         if from_outcome is not None:

src/wf_cli/commands/drafts.py

@@ -11,6 +11,7 @@ from wf_cli.commands.draft_options import (
     _parse_map_flags,
     _parse_route_flags,
     _parse_step_input_map_flags,
+    parse_json_object_file,
 )
 from wf_cli.context import load_cli_context_from_typer as load_cli_context
 from wf_cli.formats import ListOutputFormat, emit_list_payload
@@ -25,6 +26,18 @@ app = typer.Typer(
 app.add_typer(draft_add.app, name="add")
 
 
+def _validate_outcomes(values: list[str] | None) -> tuple[str, ...] | None:
+    """Validate repeated outcome flags before loading local or remote context."""
+    if values is None:
+        return None
+    outcomes = tuple(value.strip() for value in values)
+    if any(not outcome for outcome in outcomes):
+        raise typer.BadParameter("outcomes must not be blank")
+    if len(set(outcomes)) != len(outcomes):
+        raise typer.BadParameter("outcomes must be unique")
+    return outcomes
+
+
 @app.command("list")
 def list_drafts(
     ctx: typer.Context,
@@ -66,34 +79,106 @@ def inspect_draft(
 
 
 @app.command("create")
-def create_from_capability(
+def create_draft(
     ctx: typer.Context,
     workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
     capability_name: Annotated[
-        str,
+        str | None,
         typer.Option(
             "--capability",
-            help="Qualified capability name used to bootstrap the draft.",
+            help="Optional qualified capability used to bootstrap the draft.",
         ),
-    ],
+    ] = None,
     name: Annotated[
         str | None, typer.Option("--name", help="Draft workflow name.")
     ] = None,
     title: Annotated[
         str | None, typer.Option("--title", help="Workspace title.")
     ] = None,
+    input_schema_file: Annotated[
+        Path | None,
+        typer.Option(
+            "--input-schema-file", help="Path to an input JSON Schema object."
+        ),
+    ] = None,
+    state_schema_file: Annotated[
+        Path | None,
+        typer.Option("--state-schema-file", help="Path to a state JSON Schema object."),
+    ] = None,
+    output_schema_file: Annotated[
+        Path | None,
+        typer.Option(
+            "--output-schema-file", help="Path to an output JSON Schema object."
+        ),
+    ] = None,
+    outcome: Annotated[
+        list[str] | None,
+        typer.Option(
+            "--outcome", help="Workflow outcome. Repeat for multiple outcomes."
+        ),
+    ] = None,
 ) -> None:
-    """Create a patchable draft workspace from one capability."""
+    """Create an empty draft or bootstrap one from a capability."""
+    contract_options = (
+        input_schema_file,
+        state_schema_file,
+        output_schema_file,
+        outcome,
+    )
+    if capability_name is not None and any(
+        value is not None for value in contract_options
+    ):
+        raise typer.BadParameter(
+            "only valid without --capability: schema and outcome options"
+        )
+    if capability_name is None and name is None:
+        raise typer.BadParameter("--name is required without --capability")
+
+    outcomes = _validate_outcomes(outcome)
+    input_schema = (
+        None
+        if input_schema_file is None
+        else parse_json_object_file(
+            input_schema_file, option_name="--input-schema-file"
+        )
+    )
+    state_schema = (
+        None
+        if state_schema_file is None
+        else parse_json_object_file(
+            state_schema_file, option_name="--state-schema-file"
+        )
+    )
+    output_schema = (
+        None
+        if output_schema_file is None
+        else parse_json_object_file(
+            output_schema_file, option_name="--output-schema-file"
+        )
+    )
     context = load_cli_context(ctx)
+    if capability_name is not None:
+        operation = context.handlers.create_draft_workspace_from_capability(
+            workspace_id=workspace_id,
+            capability_name=capability_name,
+            name=name,
+            title=title,
+        )
+    else:
+        assert name is not None  # Validated before context loading above.
+        operation = context.handlers.create_empty_draft_workspace(
+            workspace_id=workspace_id,
+            name=name,
+            title=title,
+            input_schema=input_schema,
+            state_schema=state_schema,
+            output_schema=output_schema,
+            outcomes=("ok",) if outcomes is None else outcomes,
+        )
     emit_json(
         run_cli_operation(
             context,
-            context.handlers.create_draft_workspace_from_capability(
+            operation,
-                workspace_id=workspace_id,
-                capability_name=capability_name,
-                name=name,
-                title=title,
-            ),
         )
     )
 
@@ -155,6 +240,109 @@ def set_draft_name(
     )
 
 
+@app.command("set-start")
+def set_draft_start(
+    ctx: typer.Context,
+    workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
+    revision: Annotated[
+        int, typer.Option("--revision", min=1, help="Expected workspace revision.")
+    ],
+    step_id: Annotated[str, typer.Option("--step", help="New start step id.")],
+) -> None:
+    """Set the draft workflow's start step."""
+    context = load_cli_context(ctx)
+    emit_json(
+        run_cli_operation(
+            context,
+            context.handlers.set_draft_start(
+                workspace_id=workspace_id,
+                revision=revision,
+                step_id=step_id,
+            ),
+        )
+    )
+
+
+@app.command("set-contract")
+def set_draft_contract(
+    ctx: typer.Context,
+    workspace_id: Annotated[str, typer.Argument(help="Draft workspace id.")],
+    revision: Annotated[
+        int, typer.Option("--revision", min=1, help="Expected workspace revision.")
+    ],
+    input_schema_file: Annotated[
+        Path | None,
+        typer.Option(
+            "--input-schema-file", help="Path to an input JSON Schema object."
+        ),
+    ] = None,
+    state_schema_file: Annotated[
+        Path | None,
+        typer.Option("--state-schema-file", help="Path to a state JSON Schema object."),
+    ] = None,
+    output_schema_file: Annotated[
+        Path | None,
+        typer.Option(
+            "--output-schema-file", help="Path to an output JSON Schema object."
+        ),
+    ] = None,
+    outcome: Annotated[
+        list[str] | None,
+        typer.Option(
+            "--outcome", help="Workflow outcome. Repeat for multiple outcomes."
+        ),
+    ] = None,
+) -> None:
+    """Replace selected top-level workflow contract fields."""
+    if all(
+        value is None
+        for value in (
+            input_schema_file,
+            state_schema_file,
+            output_schema_file,
+            outcome,
+        )
+    ):
+        raise typer.BadParameter("set-contract requires at least one contract field")
+
+    outcomes = _validate_outcomes(outcome)
+    input_schema = (
+        None
+        if input_schema_file is None
+        else parse_json_object_file(
+            input_schema_file, option_name="--input-schema-file"
+        )
+    )
+    state_schema = (
+        None
+        if state_schema_file is None
+        else parse_json_object_file(
+            state_schema_file, option_name="--state-schema-file"
+        )
+    )
+    output_schema = (
+        None
+        if output_schema_file is None
+        else parse_json_object_file(
+            output_schema_file, option_name="--output-schema-file"
+        )
+    )
+    context = load_cli_context(ctx)
+    emit_json(
+        run_cli_operation(
+            context,
+            context.handlers.set_draft_contract(
+                workspace_id=workspace_id,
+                revision=revision,
+                input_schema=input_schema,
+                state_schema=state_schema,
+                output_schema=output_schema,
+                outcomes=outcomes,
+            ),
+        )
+    )
+
+
 @app.command("set-route")
 def set_draft_route(
     ctx: typer.Context,

tests/wf_cli/test_app.py

@@ -151,7 +151,322 @@ def test_wf_draft_create_help_accepts_capability_option() -> None:
     assert result.exit_code == 0
     output = " ".join(result.output.split())
     assert "--capability" in output
+    assert "--name" in output
     assert "--title" in output
+    assert "--input-schema-file" in output
+    assert "--state-schema-file" in output
+    assert "--output-schema-file" in output
+    assert "--outcome" in output
+
+
+def test_wf_draft_lifecycle_command_help() -> None:
+    set_start = runner.invoke(app, ["draft", "set-start", "--help"])
+    set_contract = runner.invoke(app, ["draft", "set-contract", "--help"])
+
+    assert set_start.exit_code == 0
+    assert "--revision" in set_start.output
+    assert "--step" in set_start.output
+    assert set_contract.exit_code == 0
+    assert "--revision" in set_contract.output
+    assert "--input-schema-file" in set_contract.output
+    assert "--state-schema-file" in set_contract.output
+    assert "--output-schema-file" in set_contract.output
+    assert "--outcome" in set_contract.output
+
+
+def test_wf_draft_lifecycle_commands_dispatch_exact_fields(
+    monkeypatch,
+    tmp_path,
+) -> None:
+    calls: list[tuple[str, dict[str, Any]]] = []
+
+    class FakeHandlers:
+        async def create_empty_draft_workspace(self, **kwargs: Any) -> dict[str, Any]:
+            calls.append(("create_empty", kwargs))
+            return {"revision": 1, "status": "invalid"}
+
+        async def create_draft_workspace_from_capability(
+            self, **kwargs: Any
+        ) -> dict[str, Any]:
+            calls.append(("create_capability", kwargs))
+            return {"revision": 1, "status": "valid"}
+
+        async def set_draft_start(self, **kwargs: Any) -> dict[str, Any]:
+            calls.append(("set_start", kwargs))
+            return {"revision": 2, "status": "invalid"}
+
+        async def set_draft_contract(self, **kwargs: Any) -> dict[str, Any]:
+            calls.append(("set_contract", kwargs))
+            return {"revision": 3, "status": "invalid"}
+
+    context = SimpleNamespace(handlers=FakeHandlers(), verbose=False)
+    monkeypatch.setattr(
+        "wf_cli.commands.drafts.load_cli_context",
+        lambda _ctx: context,
+    )
+    input_schema = tmp_path / "input.json"
+    state_schema = tmp_path / "state.json"
+    output_schema = tmp_path / "output.json"
+    input_schema.write_text('{"type":"object","properties":{}}', encoding="utf-8")
+    state_schema.write_text(
+        '{"type":"object","properties":{"items":{"type":"array",'
+        '"reducer":"wf.std.append"}}}',
+        encoding="utf-8",
+    )
+    output_schema.write_text(
+        '{"type":"object","properties":{"result":{"type":"string"}}}',
+        encoding="utf-8",
+    )
+
+    empty = runner.invoke(
+        app,
+        [
+            "draft",
+            "create",
+            "control_ws",
+            "--name",
+            "control",
+            "--title",
+            "Control",
+            "--input-schema-file",
+            str(input_schema),
+            "--state-schema-file",
+            str(state_schema),
+            "--output-schema-file",
+            str(output_schema),
+            "--outcome",
+            "submitted",
+            "--outcome",
+            "cancelled",
+        ],
+    )
+    capability = runner.invoke(
+        app,
+        [
+            "draft",
+            "create",
+            "capability_ws",
+            "--capability",
+            "wf.std.constant",
+            "--name",
+            "constant",
+        ],
+    )
+    started = runner.invoke(
+        app,
+        [
+            "draft",
+            "set-start",
+            "control_ws",
+            "--revision",
+            "1",
+            "--step",
+            "gate",
+        ],
+    )
+    contracted = runner.invoke(
+        app,
+        [
+            "draft",
+            "set-contract",
+            "control_ws",
+            "--revision",
+            "2",
+            "--state-schema-file",
+            str(state_schema),
+            "--outcome",
+            "submitted",
+            "--outcome",
+            "cancelled",
+        ],
+    )
+
+    assert empty.exit_code == 0, empty.output
+    assert capability.exit_code == 0, capability.output
+    assert started.exit_code == 0, started.output
+    assert contracted.exit_code == 0, contracted.output
+    assert calls == [
+        (
+            "create_empty",
+            {
+                "workspace_id": "control_ws",
+                "name": "control",
+                "title": "Control",
+                "input_schema": {"type": "object", "properties": {}},
+                "state_schema": {
+                    "type": "object",
+                    "properties": {
+                        "items": {
+                            "type": "array",
+                            "reducer": "wf.std.append",
+                        }
+                    },
+                },
+                "output_schema": {
+                    "type": "object",
+                    "properties": {"result": {"type": "string"}},
+                },
+                "outcomes": ("submitted", "cancelled"),
+            },
+        ),
+        (
+            "create_capability",
+            {
+                "workspace_id": "capability_ws",
+                "capability_name": "wf.std.constant",
+                "name": "constant",
+                "title": None,
+            },
+        ),
+        (
+            "set_start",
+            {"workspace_id": "control_ws", "revision": 1, "step_id": "gate"},
+        ),
+        (
+            "set_contract",
+            {
+                "workspace_id": "control_ws",
+                "revision": 2,
+                "input_schema": None,
+                "state_schema": {
+                    "type": "object",
+                    "properties": {
+                        "items": {
+                            "type": "array",
+                            "reducer": "wf.std.append",
+                        }
+                    },
+                },
+                "output_schema": None,
+                "outcomes": ("submitted", "cancelled"),
+            },
+        ),
+    ]
+
+
+@pytest.mark.parametrize(
+    ("arguments", "message"),
+    [
+        (["draft", "create", "ws"], "--name is required"),
+        (
+            [
+                "draft",
+                "create",
+                "ws",
+                "--capability",
+                "wf.std.constant",
+                "--outcome",
+                "ok",
+            ],
+            "only valid without --capability",
+        ),
+        (
+            ["draft", "set-contract", "ws", "--revision", "1"],
+            "requires at least one contract field",
+        ),
+        (
+            [
+                "draft",
+                "create",
+                "ws",
+                "--name",
+                "ws",
+                "--outcome",
+                "ok",
+                "--outcome",
+                "ok",
+            ],
+            "outcomes must be unique",
+        ),
+        (
+            [
+                "draft",
+                "set-contract",
+                "ws",
+                "--revision",
+                "1",
+                "--outcome",
+                "",
+            ],
+            "outcomes must not be blank",
+        ),
+    ],
+)
+def test_wf_draft_lifecycle_rejects_invalid_options_before_loading_context(
+    monkeypatch,
+    arguments: list[str],
+    message: str,
+) -> None:
+    monkeypatch.setattr(
+        "wf_cli.commands.drafts.load_cli_context",
+        lambda _ctx: (_ for _ in ()).throw(AssertionError("context loaded")),
+    )
+
+    result = runner.invoke(app, arguments)
+
+    assert result.exit_code != 0
+    assert message in result.output
+    assert "context loaded" not in result.output
+
+
+@pytest.mark.parametrize("value", ["[1]", '"schema"', "null"])
+def test_wf_draft_create_rejects_non_object_schema_files(
+    monkeypatch,
+    tmp_path,
+    value: str,
+) -> None:
+    schema = tmp_path / "schema.json"
+    schema.write_text(value, encoding="utf-8")
+    monkeypatch.setattr(
+        "wf_cli.commands.drafts.load_cli_context",
+        lambda _ctx: (_ for _ in ()).throw(AssertionError("context loaded")),
+    )
+
+    result = runner.invoke(
+        app,
+        [
+            "draft",
+            "create",
+            "ws",
+            "--name",
+            "ws",
+            "--input-schema-file",
+            str(schema),
+        ],
+    )
+
+    assert result.exit_code != 0
+    assert "--input-schema-file: expected a JSON object" in result.output
+    assert "context loaded" not in result.output
+
+
+def test_wf_draft_create_rejects_malformed_schema_file(
+    monkeypatch,
+    tmp_path,
+) -> None:
+    schema = tmp_path / "schema.json"
+    schema.write_text("{", encoding="utf-8")
+    monkeypatch.setattr(
+        "wf_cli.commands.drafts.load_cli_context",
+        lambda _ctx: (_ for _ in ()).throw(AssertionError("context loaded")),
+    )
+
+    result = runner.invoke(
+        app,
+        [
+            "draft",
+            "create",
+            "ws",
+            "--name",
+            "ws",
+            "--input-schema-file",
+            str(schema),
+        ],
+    )
+
+    assert result.exit_code != 0
+    assert "--input-schema-file: invalid JSON" in result.output
+    assert "context loaded" not in result.output
 
 
 def test_wf_draft_help_does_not_list_old_create_from_capability() -> None:

tests/wf_cli/test_remote_target.py

@@ -748,6 +748,77 @@ def test_wf_remote_draft_artifact_deploy_lifecycle(monkeypatch, tmp_path) -> Non
     assert '"status": "runnable"' in validated_deployment.output
 
 
+def test_wf_remote_capability_free_draft_lifecycle(monkeypatch, tmp_path) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    _patch_rpc_client_to_server(monkeypatch, server)
+    config_path = tmp_path / "wf.json"
+    config_path.write_text('{"version": 1}', encoding="utf-8")
+    runner = CliRunner()
+    base_args = ["--config", str(config_path), "--url", "http://test/rpc"]
+    commands = [
+        ["draft", "create", "control_ws", "--name", "control"],
+        [
+            "draft",
+            "add",
+            "join",
+            "control_ws",
+            "--revision",
+            "1",
+            "--step",
+            "gate",
+            "--route",
+            "done=finish",
+        ],
+        [
+            "draft",
+            "set-start",
+            "control_ws",
+            "--revision",
+            "2",
+            "--step",
+            "gate",
+        ],
+        [
+            "draft",
+            "add",
+            "end",
+            "control_ws",
+            "--revision",
+            "3",
+            "--step",
+            "finish",
+            "--outcome",
+            "error",
+        ],
+        [
+            "draft",
+            "set-contract",
+            "control_ws",
+            "--revision",
+            "4",
+            "--outcome",
+            "error",
+        ],
+        ["draft", "validate", "control_ws"],
+    ]
+
+    results = [runner.invoke(app, [*base_args, *command]) for command in commands]
+    inspected = runner.invoke(
+        app,
+        [*base_args, "draft", "inspect", "control_ws", "--include-draft"],
+    )
+
+    for result in results:
+        assert result.exit_code == 0, result.output
+    assert '"status": "valid"' in results[-1].output
+    assert inspected.exit_code == 0, inspected.output
+    payload = json.loads(inspected.output)
+    assert payload["revision"] == 5
+    assert payload["draft"]["start"] == "gate"
+    assert payload["draft"]["outcomes"] == ["error"]
+    assert set(payload["draft"]["steps"]) == {"gate", "finish"}
+
+
 def test_wf_remote_run_resume_interrupted_deployment(monkeypatch, tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     asyncio.run(