feat: add draft lifecycle authoring operations

src/wf_api/drafts.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping, Sequence
+from copy import deepcopy
 from typing import Any
 
 from wf_artifacts import (
@@ -47,6 +48,30 @@ from .operation_context import WorkflowOperationContext
 from .schema_projection import project_property_to_schema_path
 
 
+def _empty_object_schema() -> dict[str, Any]:
+    """Return one fresh unconstrained object schema for an empty draft."""
+    return {"type": "object", "properties": {}}
+
+
+def _validated_schema_object(value: object, *, field_name: str) -> dict[str, Any]:
+    """Return an isolated schema object after validating the public envelope."""
+    if not isinstance(value, dict):
+        raise ValueError(f"{field_name} must be a JSON object")
+    return deepcopy(value)
+
+
+def _validated_workflow_outcomes(outcomes: Sequence[str]) -> list[str]:
+    """Return ordered public outcomes after rejecting unusable contracts."""
+    values = list(outcomes)
+    if not values:
+        raise ValueError("workflow outcomes must contain at least one value")
+    if any(not isinstance(value, str) or not value.strip() for value in values):
+        raise ValueError("workflow outcomes must not contain blank values")
+    if len(set(values)) != len(values):
+        raise ValueError("workflow outcomes must be unique")
+    return values
+
+
 class WorkflowDraftApi:
     """Draft validation and workspace editing operations.
 
@@ -148,6 +173,60 @@ class WorkflowDraftApi:
             title=title,
         )
 
+    async def create_empty_draft_workspace(
+        self,
+        *,
+        workspace_id: str,
+        name: str,
+        title: str | None = None,
+        input_schema: dict[str, Any] | None = None,
+        state_schema: dict[str, Any] | None = None,
+        output_schema: dict[str, Any] | None = None,
+        outcomes: Sequence[str] = ("ok",),
+    ) -> dict[str, Any]:
+        """Create an intentionally invalid, capability-free draft workspace.
+
+        The empty entry point is persisted so callers can assemble the graph in
+        later revisions while retaining normal workspace diagnostics.
+        """
+        draft = {
+            "name": name,
+            "input_schema": (
+                _empty_object_schema()
+                if input_schema is None
+                else _validated_schema_object(
+                    input_schema,
+                    field_name="input_schema",
+                )
+            ),
+            "state_schema": (
+                _empty_object_schema()
+                if state_schema is None
+                else _validated_schema_object(
+                    state_schema,
+                    field_name="state_schema",
+                )
+            ),
+            "output_schema": (
+                _empty_object_schema()
+                if output_schema is None
+                else _validated_schema_object(
+                    output_schema,
+                    field_name="output_schema",
+                )
+            ),
+            "outcomes": _validated_workflow_outcomes(outcomes),
+            "output": [],
+            "start": "",
+            "steps": {},
+            "routes": {},
+        }
+        return await self.create_draft_workspace(
+            workspace_id=workspace_id,
+            draft=draft,
+            title=title,
+        )
+
     async def get_draft_workspace(
         self,
         *,
@@ -223,6 +302,70 @@ class WorkflowDraftApi:
             patch=[{"op": "replace", "path": "/name", "value": name}],
         )
 
+    async def set_draft_start(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+    ) -> dict[str, Any]:
+        """Select an entry point, including a forward-referenced step id."""
+        if not isinstance(step_id, str) or not step_id.strip():
+            raise ValueError("draft start step id must not be blank")
+        return await self.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=[{"op": "replace", "path": "/start", "value": step_id}],
+        )
+
+    async def set_draft_contract(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        input_schema: dict[str, Any] | None = None,
+        state_schema: dict[str, Any] | None = None,
+        output_schema: dict[str, Any] | None = None,
+        outcomes: Sequence[str] | None = None,
+    ) -> dict[str, Any]:
+        """Replace supplied top-level contract fields in one draft revision.
+
+        Complete schema replacement is intentional: deep merging JSON Schema
+        would make reducer metadata and required-field removal ambiguous.
+        """
+        patch: list[dict[str, Any]] = []
+        for field_name, schema in (
+            ("input_schema", input_schema),
+            ("state_schema", state_schema),
+            ("output_schema", output_schema),
+        ):
+            if schema is not None:
+                patch.append(
+                    {
+                        "op": "replace",
+                        "path": f"/{field_name}",
+                        "value": _validated_schema_object(
+                            schema,
+                            field_name=field_name,
+                        ),
+                    }
+                )
+        if outcomes is not None:
+            patch.append(
+                {
+                    "op": "replace",
+                    "path": "/outcomes",
+                    "value": _validated_workflow_outcomes(outcomes),
+                }
+            )
+        if not patch:
+            raise ValueError("set_draft_contract requires at least one contract field")
+        return await self.patch_draft_workspace(
+            workspace_id=workspace_id,
+            revision=revision,
+            patch=patch,
+        )
+
     async def set_draft_route(
         self,
         *,

src/wf_api/service.py

@@ -263,6 +263,27 @@ class WorkflowApi:
             title=title,
         )
 
+    async def create_empty_draft_workspace(
+        self,
+        *,
+        workspace_id: str,
+        name: str,
+        title: str | None = None,
+        input_schema: dict[str, Any] | None = None,
+        state_schema: dict[str, Any] | None = None,
+        output_schema: dict[str, Any] | None = None,
+        outcomes: Sequence[str] = ("ok",),
+    ) -> dict[str, Any]:
+        return await self.drafts.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=outcomes,
+        )
+
     async def get_draft_workspace(
         self,
         *,
@@ -321,6 +342,38 @@ class WorkflowApi:
             name=name,
         )
 
+    async def set_draft_start(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        step_id: str,
+    ) -> dict[str, Any]:
+        return await self.drafts.set_draft_start(
+            workspace_id=workspace_id,
+            revision=revision,
+            step_id=step_id,
+        )
+
+    async def set_draft_contract(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        input_schema: dict[str, Any] | None = None,
+        state_schema: dict[str, Any] | None = None,
+        output_schema: dict[str, Any] | None = None,
+        outcomes: Sequence[str] | None = None,
+    ) -> dict[str, Any]:
+        return await self.drafts.set_draft_contract(
+            workspace_id=workspace_id,
+            revision=revision,
+            input_schema=input_schema,
+            state_schema=state_schema,
+            output_schema=output_schema,
+            outcomes=outcomes,
+        )
+
     async def set_draft_route(
         self,
         *,

tests/wf_api/test_drafts_service.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
 import pytest
 from pydantic import BaseModel, TypeAdapter
@@ -140,6 +140,332 @@ async def test_create_draft_workspace_creates_workspace(tmp_path: Path) -> None:
     assert fetched["draft"]["steps"]["echo"]["use"] == "demo.personal.echo_tool"
 
 
+@pytest.mark.asyncio
+async def test_create_empty_draft_workspace_persists_invalid_skeleton(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_create_empty")
+    _drafts, _service, authoring = _draft_api(artifact_store)
+    facade = WorkflowApi(authoring.context)
+
+    created = await facade.create_empty_draft_workspace(
+        workspace_id="control_first",
+        name="control_first",
+        title="Control First",
+    )
+    stored = await facade.get_draft_workspace(
+        workspace_id="control_first",
+        include_draft=True,
+    )
+
+    assert created["revision"] == 1
+    assert created["status"] == "invalid"
+    assert created["diagnostics"]
+    assert stored["title"] == "Control First"
+    assert stored["draft"] == {
+        "name": "control_first",
+        "input_schema": {"type": "object", "properties": {}},
+        "state_schema": {"type": "object", "properties": {}},
+        "output_schema": {"type": "object", "properties": {}},
+        "outcomes": ["ok"],
+        "output": [],
+        "start": "",
+        "steps": {},
+        "routes": {},
+    }
+
+
+@pytest.mark.asyncio
+async def test_create_empty_draft_workspace_preserves_custom_contract(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_create_contract")
+    _drafts, _service, authoring = _draft_api(artifact_store)
+    facade = WorkflowApi(authoring.context)
+    input_schema = {
+        "type": "object",
+        "properties": {"topic": {"type": "string"}},
+    }
+    state_schema = {
+        "type": "object",
+        "properties": {
+            "issues": {
+                "type": "array",
+                "items": {"type": "string"},
+                "x-reducer": "append",
+            }
+        },
+    }
+    output_schema = {
+        "type": "object",
+        "properties": {"report": {"type": "string"}},
+    }
+
+    await facade.create_empty_draft_workspace(
+        workspace_id="custom_contract",
+        name="custom_contract",
+        title="Custom Contract",
+        input_schema=input_schema,
+        state_schema=state_schema,
+        output_schema=output_schema,
+        outcomes=("submitted", "cancelled"),
+    )
+    input_schema["properties"]["late"] = {"type": "boolean"}
+    stored = await facade.get_draft_workspace(
+        workspace_id="custom_contract",
+        include_draft=True,
+    )
+
+    assert stored["title"] == "Custom Contract"
+    assert stored["draft"]["input_schema"] == {
+        "type": "object",
+        "properties": {"topic": {"type": "string"}},
+    }
+    assert stored["draft"]["state_schema"] == state_schema
+    assert stored["draft"]["output_schema"] == output_schema
+    assert stored["draft"]["outcomes"] == ["submitted", "cancelled"]
+
+
+@pytest.mark.asyncio
+async def test_create_empty_draft_workspace_isolates_default_schemas(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_schema_isolation")
+    _drafts, _service, authoring = _draft_api(artifact_store)
+    facade = WorkflowApi(authoring.context)
+    input_schema = {
+        "type": "object",
+        "properties": {"topic": {"type": "string"}},
+    }
+
+    await facade.create_empty_draft_workspace(
+        workspace_id="isolated",
+        name="isolated",
+        input_schema=input_schema,
+    )
+    input_schema["properties"]["late"] = {"type": "boolean"}
+    stored = await facade.get_draft_workspace(
+        workspace_id="isolated",
+        include_draft=True,
+    )
+    stored["draft"]["state_schema"]["properties"]["state_only"] = {"type": "string"}
+
+    assert stored["draft"]["input_schema"] == {
+        "type": "object",
+        "properties": {"topic": {"type": "string"}},
+    }
+    assert stored["draft"]["output_schema"] == {
+        "type": "object",
+        "properties": {},
+    }
+
+
+@pytest.mark.asyncio
+async def test_create_empty_draft_workspace_reports_duplicate_conflict(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_create_conflict")
+    _drafts, _service, authoring = _draft_api(artifact_store)
+    facade = WorkflowApi(authoring.context)
+    await facade.create_empty_draft_workspace(
+        workspace_id="control_first",
+        name="control_first",
+    )
+
+    duplicate = await facade.create_empty_draft_workspace(
+        workspace_id="control_first",
+        name="replacement",
+    )
+
+    assert duplicate["status"] == "conflict"
+    assert duplicate["diagnostics"][0]["code"] == "workspace_exists"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    "contract",
+    [
+        {"input_schema": cast(Any, [])},
+        {"outcomes": ()},
+        {"outcomes": cast(Any, (1,))},
+        {"outcomes": ("ok", " ")},
+        {"outcomes": ("ok", "ok")},
+    ],
+)
+async def test_create_empty_draft_workspace_rejects_invalid_contract_before_mutation(
+    tmp_path: Path,
+    contract: dict[str, Any],
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_create_rejected")
+    _drafts, _service, authoring = _draft_api(artifact_store)
+    facade = WorkflowApi(authoring.context)
+
+    with pytest.raises(ValueError):
+        await facade.create_empty_draft_workspace(
+            workspace_id="rejected",
+            name="rejected",
+            **contract,
+        )
+
+    assert await facade.list_draft_workspaces() == {"workspaces": []}
+
+
+@pytest.mark.asyncio
+async def test_set_draft_start_and_contract_replace_top_level_fields_atomically(
+    tmp_path: Path,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(tmp_path / "drafts_set_lifecycle")
+    _drafts, _service, authoring = _draft_api(artifact_store)
+    facade = WorkflowApi(authoring.context)
+    await facade.create_empty_draft_workspace(
+        workspace_id="control_first",
+        name="control_first",
+        input_schema={
+            "type": "object",
+            "properties": {"topic": {"type": "string"}},
+        },
+    )
+    state_schema = {
+        "type": "object",
+        "properties": {
+            "issues": {
+                "type": "array",
+                "items": {"type": "string"},
+                "x-reducer": "append",
+            }
+        },
+    }
+    output_schema = {
+        "type": "object",
+        "properties": {"report": {"type": "string"}},
+    }
+
+    forward = await facade.set_draft_start(
+        workspace_id="control_first",
+        revision=1,
+        step_id="gate",
+    )
+    contract = await facade.set_draft_contract(
+        workspace_id="control_first",
+        revision=2,
+        state_schema=state_schema,
+        output_schema=output_schema,
+        outcomes=("submitted", "cancelled"),
+    )
+    stored = await facade.get_draft_workspace(
+        workspace_id="control_first",
+        include_draft=True,
+    )
+
+    assert forward["revision"] == 2
+    assert forward["status"] == "invalid"
+    assert contract["revision"] == 3
+    assert stored["draft"]["start"] == "gate"
+    assert stored["draft"]["input_schema"] == {
+        "type": "object",
+        "properties": {"topic": {"type": "string"}},
+    }
+    assert stored["draft"]["state_schema"] == state_schema
+    assert stored["draft"]["output_schema"] == output_schema
+    assert stored["draft"]["outcomes"] == ["submitted", "cancelled"]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("operation", "arguments"),
+    [
+        ("start", {"step_id": " "}),
+        ("start", {"step_id": cast(Any, 1)}),
+        ("contract", {}),
+        ("contract", {"state_schema": cast(Any, [])}),
+        ("contract", {"outcomes": ()}),
+        ("contract", {"outcomes": cast(Any, (1,))}),
+        ("contract", {"outcomes": ("ok", " ")}),
+        ("contract", {"outcomes": ("ok", "ok")}),
+    ],
+)
+async def test_lifecycle_edits_reject_invalid_envelopes_without_mutation(
+    tmp_path: Path,
+    operation: str,
+    arguments: dict[str, Any],
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        tmp_path / f"drafts_lifecycle_rejected_{operation}"
+    )
+    _drafts, _service, authoring = _draft_api(artifact_store)
+    facade = WorkflowApi(authoring.context)
+    await facade.create_empty_draft_workspace(
+        workspace_id="control_first",
+        name="control_first",
+    )
+    before = await facade.get_draft_workspace(
+        workspace_id="control_first",
+        include_draft=True,
+    )
+
+    with pytest.raises(ValueError):
+        if operation == "start":
+            await facade.set_draft_start(
+                workspace_id="control_first",
+                revision=2,
+                **arguments,
+            )
+        else:
+            await facade.set_draft_contract(
+                workspace_id="control_first",
+                revision=2,
+                **arguments,
+            )
+
+    after = await facade.get_draft_workspace(
+        workspace_id="control_first",
+        include_draft=True,
+    )
+    assert after == before
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("operation", ["start", "contract"])
+async def test_lifecycle_edits_report_stale_revision_without_mutation(
+    tmp_path: Path,
+    operation: str,
+) -> None:
+    artifact_store = FileWorkflowArtifactStore(
+        tmp_path / f"drafts_lifecycle_stale_{operation}"
+    )
+    _drafts, _service, authoring = _draft_api(artifact_store)
+    facade = WorkflowApi(authoring.context)
+    await facade.create_empty_draft_workspace(
+        workspace_id="control_first",
+        name="control_first",
+    )
+    before = await facade.get_draft_workspace(
+        workspace_id="control_first",
+        include_draft=True,
+    )
+
+    if operation == "start":
+        result = await facade.set_draft_start(
+            workspace_id="control_first",
+            revision=2,
+            step_id="gate",
+        )
+    else:
+        result = await facade.set_draft_contract(
+            workspace_id="control_first",
+            revision=2,
+            outcomes=("submitted", "cancelled"),
+        )
+
+    after = await facade.get_draft_workspace(
+        workspace_id="control_first",
+        include_draft=True,
+    )
+    assert result["status"] == "conflict"
+    assert result["diagnostics"][0]["code"] == "revision_conflict"
+    assert after == before
+
+
 @pytest.mark.asyncio
 async def test_list_draft_workspaces_returns_sorted_summaries_without_drafts(
     tmp_path: Path,