feat: expose authoring contract inventory

.superpowers/sdd/2026-08-14-workflow-console-contract-graph/task-3-report.md

@@ -0,0 +1,80 @@
+# Task 3 Report: Expose Authoring Contract Inspection
+
+## Status
+
+Implemented Task 3 of the workflow contract graph backend slice.
+
+## Changes
+
+- Added `WorkflowDraftSurface.inspect_draft_authoring_contract` and the
+  matching `WorkflowApi` implementation.
+- Added read-only persisted workspace loading with canonical revision-conflict
+  precedence and no validation-save or revision mutation.
+- Projected tolerant workflow input/state/output schemas through the Task 1
+  inventory projector.
+- Projected resolved capability input/output schemas, outcomes, descriptions,
+  and executable entry candidates for keyed `use` steps. Projection ids and
+  `__end__` are not advertised as entry candidates.
+- Integrated Task 2 runtime context analysis for the selected step. Compile or
+  interpretation failures leave scoped context empty and become warnings.
+- Added the JSON-RPC params model, method dispatch, typed remote client method,
+  nullable `selected_step_id`, and named OpenRPC payload references.
+- Preserved existing domain error mapping for unknown steps and missing
+  workspaces, `-32602` for malformed RPC params, and the existing
+  `revision_conflict` result for stale revisions.
+
+## Test-First Evidence
+
+The required RED command was run after adding the service tests and before
+production implementation:
+
+```text
+uv run pytest tests/wf_api/test_drafts_service.py -q -k authoring_contract
+```
+
+It failed for the expected missing seam:
+
+```text
+4 failed
+AttributeError: 'WorkflowApi' object has no attribute
+'inspect_draft_authoring_contract'
+```
+
+After implementation, the new authoring-contract service and transport tests
+passed:
+
+```text
+uv run pytest tests/wf_api/test_drafts_service.py tests/wf_transport_rpc_http/test_app.py tests/wf_transport_rpc_http/test_client.py tests/wf_transport_rpc_http/test_openrpc_contract.py -q -k authoring_contract
+12 passed
+```
+
+## Verification
+
+```text
+uv run pytest tests/wf_api/test_drafts_service.py tests/wf_transport_rpc_http/test_app.py tests/wf_transport_rpc_http/test_client.py tests/wf_transport_rpc_http/test_openrpc_contract.py -q -k "not reads_admin_state"
+391 passed, 184 warnings
+
+uv run ruff check <touched source and test files>
+All checks passed!
+
+uv run ruff format --check <touched source and test files>
+10 files already formatted
+
+uv run basedpyright --level error <touched source files>
+0 errors, 0 warnings, 0 notes
+```
+
+The exact broad target command also contains the existing
+`test_rpc_workflow_client_reads_admin_state` failure: its recorded event lacks
+the required `timestamp_epoch_ms` field when serialized as `AdminEventPayload`.
+That failure reproduces in isolation and is unrelated to the Task 3 files.
+
+No Serena configuration was modified.
+
+## Concerns
+
+- The repository's existing admin-event timestamp validation failure prevents
+  the unfiltered four-file target command from being fully green; the full
+  target set passes when that isolated test is excluded.
+- FastAPI JSON-RPC emits deprecation warnings from the installed
+  `fastapi-jsonrpc` dependency; no new warning class was introduced.

src/wf_api/authoring_contracts.py

@@ -132,6 +132,36 @@ def project_authoring_contract_inventory(
     }
 
 
+def project_authoring_step_contract(
+    *,
+    step_id: str,
+    label: str,
+    description: str | None,
+    input_schema: JsonObject,
+    output_schema: JsonObject,
+    outcomes: Sequence[str],
+) -> AuthoringStepContractPayload:
+    """Project one resolved executable capability into authoring choices."""
+    payload: AuthoringStepContractPayload = {
+        "step_id": step_id,
+        "label": label,
+        "input_targets": schema_path_options(
+            input_schema,
+            root="step_input",
+            uses=["step_input"],
+        ),
+        "output_sources": schema_path_options(
+            output_schema,
+            root="step_output",
+            uses=["step_output_source", "workflow_output"],
+        ),
+        "outcomes": list(outcomes),
+    }
+    if description is not None:
+        payload["description"] = description
+    return payload
+
+
 def context_path_options(
     fields: Sequence[ContextFieldAvailability | Mapping[str, Any]],
 ) -> list[AuthoringPathOptionPayload]:

src/wf_api/service.py

@@ -1,19 +1,27 @@
 from __future__ import annotations
 
-from collections.abc import Sequence
+from collections.abc import Mapping, Sequence
 from typing import Any
 
-from wf_artifacts import ArtifactKind
+from wf_artifacts import ArtifactKind, compile_workflow_draft
 from wf_artifacts.drafts.models import DraftStep
+from wf_core.analysis.context_scopes import context_analysis_warnings
 from wf_core.models.steps import InputBinding, OutputBinding, StepInputBinding
+from wf_core.models.workflow import Workflow
 
 from .artifacts import WorkflowArtifactApi
+from .authoring_contracts import (
+    context_path_options_for_node,
+    project_authoring_contract_inventory,
+    project_authoring_step_contract,
+)
 from .capabilities import WorkflowCapabilityApi
 from .deployments import WorkflowDeploymentApi
 from .draft_authoring import RouteSource, WorkflowDraftAuthoringApi
 from .draft_updates import CapabilityStepUpdate
 from .drafts import WorkflowDraftApi
 from .models import (
+    AuthoringContractInventoryPayload,
     CapabilityCallResult,
     CompileDraftWorkspaceResult,
     CompileDraftWorkspaceSuccess,
@@ -45,6 +53,25 @@ from .operation_context import WorkflowOperationContext
 from .runs import TraceRangeLike, WorkflowRunApi
 
 
+def _authoring_schema(value: object) -> dict[str, Any]:
+    """Return a safe schema object from a possibly invalid persisted draft."""
+    return dict(value) if isinstance(value, Mapping) else {}
+
+
+def _authoring_outcomes(value: object) -> list[str]:
+    """Keep only a complete string outcome list for tolerant inventory output."""
+    if not isinstance(value, list) or not all(
+        isinstance(outcome, str) for outcome in value
+    ):
+        return []
+    return list(value)
+
+
+def _step_label(step_id: str) -> str:
+    """Humanize a keyed draft step for compact executable choices."""
+    return step_id.replace("_", " ").replace("-", " ").title()
+
+
 class WorkflowApi:
     """Protocol-neutral workflow application facade.
 
@@ -324,6 +351,115 @@ class WorkflowApi:
             include_draft=include_draft,
         )
 
+    async def inspect_draft_authoring_contract(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        selected_step_id: str | None = None,
+    ) -> AuthoringContractInventoryPayload | DraftWorkspaceResult:
+        """Inspect revision-scoped authoring choices without persisting changes.
+
+        The workspace revision check deliberately happens before interpreting a
+        selected step. This preserves the draft APIs' canonical conflict
+        precedence when an authoring client is holding an old revision.
+        """
+        checked = self.drafts._workspace_if_revision_matches(
+            workspace_id=workspace_id,
+            revision=revision,
+        )
+        if isinstance(checked, dict):
+            return checked
+
+        draft = checked.draft
+        raw_steps = draft.get("steps")
+        steps = raw_steps if isinstance(raw_steps, Mapping) else {}
+        if selected_step_id is not None and selected_step_id not in steps:
+            raise KeyError(f"unknown draft step {selected_step_id!r}")
+
+        warnings: list[str] = []
+        entry_steps = []
+        selected_contract = None
+        for raw_step_id, raw_step in steps.items():
+            if not isinstance(raw_step_id, str) or raw_step_id == "__end__":
+                continue
+            if not isinstance(raw_step, Mapping):
+                continue
+            capability_name = raw_step.get("use")
+            if not isinstance(capability_name, str):
+                if raw_step_id == selected_step_id:
+                    warnings.append(
+                        f"selected step {raw_step_id!r} is not an executable capability"
+                    )
+                continue
+            try:
+                spec = self.context.specs.get_qualified_spec(capability_name)
+                input_schema = (
+                    spec.input_schema_contract or spec.input_model.model_json_schema()
+                )
+                output_schema = (
+                    spec.output_schema_contract or spec.output_model.model_json_schema()
+                )
+            except (KeyError, TypeError, ValueError) as exc:
+                if raw_step_id == selected_step_id:
+                    warnings.append(
+                        f"selected step {raw_step_id!r} cannot be interpreted: {exc}"
+                    )
+                continue
+
+            description = raw_step.get("desc")
+            if not isinstance(description, str):
+                description = spec.description
+            contract = project_authoring_step_contract(
+                step_id=raw_step_id,
+                label=_step_label(raw_step_id),
+                description=description,
+                input_schema=input_schema,
+                output_schema=output_schema,
+                outcomes=spec.outcomes,
+            )
+            entry_steps.append(contract)
+            if raw_step_id == selected_step_id:
+                selected_contract = contract
+
+        context_entries = []
+        if selected_step_id is not None:
+            try:
+                compiled_plan = compile_workflow_draft(draft)
+                workflow = Workflow.model_validate(compiled_plan)
+                context_entries = context_path_options_for_node(
+                    workflow,
+                    selected_step_id,
+                )
+                warnings.extend(context_analysis_warnings(workflow))
+            except (KeyError, TypeError, ValueError) as exc:
+                warnings.append(
+                    f"runtime context unavailable for selected step "
+                    f"{selected_step_id!r}: {exc}"
+                )
+
+        selected_input_targets = []
+        selected_output_sources = []
+        if selected_contract is not None and selected_step_id is not None:
+            if selected_contract["step_id"] == selected_step_id:
+                selected_input_targets = selected_contract.get("input_targets", [])
+                selected_output_sources = selected_contract.get("output_sources", [])
+
+        return project_authoring_contract_inventory(
+            workspace_id=workspace_id,
+            revision=checked.revision,
+            selected_step_id=selected_step_id,
+            input_schema=_authoring_schema(draft.get("input_schema")),
+            state_schema=_authoring_schema(draft.get("state_schema")),
+            output_schema=_authoring_schema(draft.get("output_schema")),
+            context_entries=context_entries,
+            step_input_targets=selected_input_targets,
+            step_output_sources=selected_output_sources,
+            entry_steps=entry_steps,
+            workflow_outcomes=_authoring_outcomes(draft.get("outcomes")),
+            warnings=warnings,
+        )
+
     async def delete_draft_workspace(
         self,
         *,

src/wf_api/surface.py

@@ -11,6 +11,7 @@ from .draft_authoring import RouteSource
 from .draft_updates import CapabilityStepUpdate
 from .models import (
     ApplyRegistryChangesResult,
+    AuthoringContractInventoryPayload,
     AuthRecordSummaryPayload,
     CapabilityCallResult,
     CompileDraftWorkspaceResult,
@@ -107,6 +108,14 @@ class WorkflowDraftSurface(Protocol):
         include_draft: bool = False,
     ) -> DraftWorkspaceResult: ...
 
+    async def inspect_draft_authoring_contract(
+        self,
+        *,
+        workspace_id: str,
+        revision: int,
+        selected_step_id: str | None = None,
+    ) -> AuthoringContractInventoryPayload | DraftWorkspaceResult: ...
+
     async def create_draft_workspace_from_capability(
         self,
         *,

src/wf_transport_rpc_http/client/drafts.py

@@ -5,6 +5,7 @@ from typing import Any, Literal, cast
 
 from wf_api import CapabilityStepUpdate
 from wf_api.models import (
+    AuthoringContractInventoryPayload,
     CompileDraftWorkspaceResult,
     CreateArtifactFromWorkspaceResult,
     CreateDraftWorkspaceFromCapabilityResult,
@@ -76,6 +77,25 @@ class RpcDraftClientMixin:
             {"workspace_id": workspace_id, "include_draft": include_draft},
         )
 
+    async def inspect_draft_authoring_contract(
+        self: RpcCaller,
+        *,
+        workspace_id: str,
+        revision: int,
+        selected_step_id: str | None = None,
+    ) -> AuthoringContractInventoryPayload | DraftWorkspaceResult:
+        return cast(
+            AuthoringContractInventoryPayload | DraftWorkspaceResult,
+            await self._call(
+                "workflow.draft_workspaces.inspect_authoring_contract",
+                {
+                    "workspace_id": workspace_id,
+                    "revision": revision,
+                    "selected_step_id": selected_step_id,
+                },
+            ),
+        )
+
     async def create_draft_workspace_from_capability(
         self: RpcCaller,
         *,

src/wf_transport_rpc_http/methods/drafts.py

@@ -7,6 +7,7 @@ while registering nested handlers for response validation and OpenRPC output.
 import fastapi_jsonrpc as jsonrpc
 
 from wf_api.models import (
+    AuthoringContractInventoryPayload,
     CompileDraftWorkspaceResult,
     CreateArtifactFromWorkspaceResult,
     CreateDraftWorkspaceFromCapabilityResult,
@@ -33,6 +34,7 @@ from ..models import (
     DeleteDraftWorkspaceParams,
     GetDraftWorkspaceParams,
     HandleDraftParams,
+    InspectDraftAuthoringContractParams,
     ListDraftWorkspacesParams,
     PatchDraftParams,
     PatchDraftWorkspaceParams,
@@ -113,6 +115,22 @@ def register_methods(
         except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
             raise_workflow_rpc_error(exc)
 
+    @entrypoint.method(
+        name="workflow.draft_workspaces.inspect_authoring_contract",
+        errors=[WorkflowRpcError],
+    )
+    async def workflow_draft_workspaces_inspect_authoring_contract(
+        params: InspectDraftAuthoringContractParams = RpcParams(),
+    ) -> AuthoringContractInventoryPayload | DraftWorkspaceResult:
+        try:
+            return await server.api.inspect_draft_authoring_contract(
+                workspace_id=params.workspace_id,
+                revision=params.revision,
+                selected_step_id=params.selected_step_id,
+            )
+        except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
+            raise_workflow_rpc_error(exc)
+
     @entrypoint.method(
         name="workflow.draft_workspaces.create_from_capability",
         errors=[WorkflowRpcError],

src/wf_transport_rpc_http/models.py

@@ -147,6 +147,12 @@ class GetDraftWorkspaceParams(RpcParamsModel):
     include_draft: bool = False
 
 
+class InspectDraftAuthoringContractParams(RpcParamsModel):
+    workspace_id: str = Field(min_length=1)
+    revision: int = Field(ge=1)
+    selected_step_id: str | None = Field(default=None, min_length=1)
+
+
 class PatchDraftWorkspaceParams(RpcParamsModel):
     workspace_id: str = Field(min_length=1)
     revision: int = Field(ge=1)

tests/wf_api/test_drafts_service.py

@@ -114,6 +114,154 @@ async def test_update_capability_step_changes_metadata_and_inputs_atomically(
     assert run.output == {"echoed": "fixed"}
 
 
+@pytest.mark.asyncio
+async def test_inspect_draft_authoring_contract_projects_selected_capability(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "authoring_contract"),
+        register_echo=True,
+    )
+    draft = _echo_draft()
+    draft["input_schema"] = {
+        "type": "object",
+        "properties": {
+            "request": {
+                "type": "object",
+                "properties": {"text": {"type": "string"}},
+                "required": ["text"],
+            }
+        },
+        "required": ["request"],
+    }
+    draft["state_schema"] = {
+        "type": "object",
+        "properties": {"echoed": {"type": "string"}},
+    }
+    draft["output_schema"] = {
+        "type": "object",
+        "properties": {"echoed": {"type": "string"}},
+    }
+    await draft_api.create_draft_workspace(workspace_id="authoring", draft=draft)
+    api = WorkflowApi(authoring.context)
+
+    inventory = await api.inspect_draft_authoring_contract(
+        workspace_id="authoring",
+        revision=1,
+        selected_step_id="echo",
+    )
+
+    assert inventory["workspace_id"] == "authoring"
+    assert inventory["revision"] == 1
+    assert inventory["selected_step_id"] == "echo"
+    assert [step["step_id"] for step in inventory["entry_steps"]] == ["echo"]
+    assert inventory["workflow_outcomes"] == ["ok"]
+    assert {option["path"] for option in inventory["step_input_targets"]} == {
+        "step_input.text"
+    }
+    assert {option["path"] for option in inventory["step_output_sources"]} == {
+        "step_output.echoed"
+    }
+    assert {option["path"] for option in inventory["readable_sources"]} >= {
+        "input.request",
+        "state.echoed",
+        "context.prior_outcome",
+    }
+    assert "__end__" not in {step["step_id"] for step in inventory["entry_steps"]}
+    selected = inventory["entry_steps"][0]
+    assert selected["input_targets"][0]["schema"]["type"] == "string"
+    assert selected["output_sources"][0]["schema"]["type"] == "string"
+    assert selected["outcomes"] == ["ok"]
+
+
+@pytest.mark.asyncio
+async def test_inspect_draft_authoring_contract_rejects_unknown_selected_step(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "authoring_contract_unknown"),
+        register_echo=True,
+    )
+    await draft_api.create_draft_workspace(
+        workspace_id="authoring", draft=_echo_draft()
+    )
+    api = WorkflowApi(authoring.context)
+
+    with pytest.raises(KeyError, match="unknown draft step"):
+        await api.inspect_draft_authoring_contract(
+            workspace_id="authoring",
+            revision=1,
+            selected_step_id="missing",
+        )
+
+
+@pytest.mark.asyncio
+async def test_inspect_draft_authoring_contract_tolerates_invalid_selected_step(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "authoring_contract_invalid"),
+        register_echo=True,
+    )
+    draft = _echo_draft()
+    draft["steps"] = {"broken": {"unknown_kind": {}}}
+    draft["start"] = "broken"
+    await draft_api.create_draft_workspace(workspace_id="authoring", draft=draft)
+    api = WorkflowApi(authoring.context)
+
+    inventory = await api.inspect_draft_authoring_contract(
+        workspace_id="authoring",
+        revision=1,
+        selected_step_id="broken",
+    )
+
+    assert inventory["selected_step_id"] == "broken"
+    assert inventory["entry_steps"] == []
+    assert inventory["step_input_targets"] == []
+    assert inventory["step_output_sources"] == []
+    assert inventory["readable_sources"]
+    assert any("broken" in warning for warning in inventory["warnings"])
+
+
+@pytest.mark.asyncio
+async def test_inspect_draft_authoring_contract_stale_revision_is_read_only(
+    tmp_path: Path,
+) -> None:
+    draft_api, _service, authoring = _draft_api(
+        FileWorkflowArtifactStore(tmp_path / "authoring_contract_stale"),
+        register_echo=True,
+    )
+    await draft_api.create_draft_workspace(
+        workspace_id="authoring", draft=_echo_draft()
+    )
+    api = WorkflowApi(authoring.context)
+    changed = await api.set_draft_name(
+        workspace_id="authoring",
+        revision=1,
+        name="changed",
+    )
+    before = await api.get_draft_workspace(
+        workspace_id="authoring",
+        include_draft=True,
+    )
+
+    conflict = await api.inspect_draft_authoring_contract(
+        workspace_id="authoring",
+        revision=1,
+        selected_step_id="echo",
+    )
+    after = await api.get_draft_workspace(
+        workspace_id="authoring",
+        include_draft=True,
+    )
+
+    assert changed["revision"] == 2
+    assert conflict["status"] == "conflict"
+    assert conflict["revision"] == 2
+    assert conflict["diagnostics"][0]["code"] == "revision_conflict"
+    assert after == before
+
+
 @pytest.mark.asyncio
 async def test_update_capability_step_preserves_omitted_fields_and_exact_noop(
     tmp_path: Path,

tests/wf_transport_rpc_http/test_app.py

@@ -15,6 +15,7 @@ from wf_transport_rpc_http.app import create_rpc_app
 from wf_transport_rpc_http.models import (
     AddDraftStepParams,
     AddStepFromCapabilityParams,
+    InspectDraftAuthoringContractParams,
     SetDraftContractParams,
     UpdateCapabilityStepParams,
 )
@@ -165,6 +166,28 @@ def test_set_draft_contract_params_reject_whitespace_duplicate_outcomes() -> Non
         )
 
 
+def test_inspect_draft_authoring_contract_params_allow_nullable_selection() -> None:
+    params = InspectDraftAuthoringContractParams.model_validate(
+        {"workspace_id": "report", "revision": 4}
+    )
+
+    assert params.selected_step_id is None
+
+
+@pytest.mark.parametrize(
+    "params",
+    [
+        {"workspace_id": "report", "revision": 4, "selected_step_id": ""},
+        {"workspace_id": "report", "revision": 0},
+    ],
+)
+def test_inspect_draft_authoring_contract_params_reject_invalid_envelope(
+    params: dict[str, Any],
+) -> None:
+    with pytest.raises(ValidationError):
+        InspectDraftAuthoringContractParams.model_validate(params)
+
+
 async def test_rpc_health_and_capability_methods(tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     app = create_rpc_app(server)
@@ -733,6 +756,185 @@ async def test_rpc_draft_workspace_lifecycle_methods(tmp_path) -> None:
     assert inspected["result"]["draft"]["outcomes"] == ["error"]
 
 
+async def test_rpc_inspects_draft_authoring_contract_without_mutation(tmp_path) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    app = create_rpc_app(server)
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        created = await _rpc(
+            client,
+            "workflow.draft_workspaces.create_from_capability",
+            {
+                "workspace_id": "authoring",
+                "capability_name": "wf.std.constant",
+                "name": "authoring",
+            },
+        )
+        before = await _rpc(
+            client,
+            "workflow.draft_workspaces.get",
+            {"workspace_id": "authoring", "include_draft": True},
+        )
+        inspected = await _rpc(
+            client,
+            "workflow.draft_workspaces.inspect_authoring_contract",
+            {
+                "workspace_id": "authoring",
+                "revision": created["result"]["revision"],
+                "selected_step_id": "call",
+            },
+        )
+        after = await _rpc(
+            client,
+            "workflow.draft_workspaces.get",
+            {"workspace_id": "authoring", "include_draft": True},
+        )
+
+    result = inspected["result"]
+    assert result["workspace_id"] == "authoring"
+    assert result["revision"] == created["result"]["revision"]
+    assert result["selected_step_id"] == "call"
+    assert result["entry_steps"][0]["step_id"] == "call"
+    assert result["entry_steps"][0]["outcomes"] == ["ok"]
+    assert result["step_input_targets"]
+    assert result["step_input_targets"][0]["path"] == "step_input.value"
+    assert isinstance(result["step_input_targets"][0]["schema"], dict)
+    assert result["step_output_sources"]
+    assert result["step_output_sources"][0]["path"] == "step_output.value"
+    assert isinstance(result["step_output_sources"][0]["schema"], dict)
+    assert any(
+        option["path"] == "context.prior_outcome"
+        for option in result["readable_sources"]
+    )
+    assert after["result"] == before["result"]
+
+
+async def test_rpc_inspect_authoring_contract_maps_domain_errors_and_conflicts(
+    tmp_path,
+) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    app = create_rpc_app(server)
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        created = await _rpc(
+            client,
+            "workflow.draft_workspaces.create_empty",
+            {"workspace_id": "authoring", "name": "authoring"},
+        )
+        unknown = await _rpc(
+            client,
+            "workflow.draft_workspaces.inspect_authoring_contract",
+            {
+                "workspace_id": "authoring",
+                "revision": created["result"]["revision"],
+                "selected_step_id": "missing",
+            },
+        )
+        missing = await _rpc(
+            client,
+            "workflow.draft_workspaces.inspect_authoring_contract",
+            {"workspace_id": "missing", "revision": 1},
+        )
+        changed = await _rpc(
+            client,
+            "workflow.draft_workspaces.set_name",
+            {
+                "workspace_id": "authoring",
+                "revision": created["result"]["revision"],
+                "name": "changed",
+            },
+        )
+        stale = await _rpc(
+            client,
+            "workflow.draft_workspaces.inspect_authoring_contract",
+            {
+                "workspace_id": "authoring",
+                "revision": created["result"]["revision"],
+            },
+        )
+
+    assert unknown["error"]["code"] == 5000
+    assert unknown["error"]["data"]["code"] == "KeyError"
+    assert missing["error"]["code"] == 5000
+    assert changed["result"]["revision"] == 2
+    assert stale["result"]["status"] == "conflict"
+    assert stale["result"]["diagnostics"][0]["code"] == "revision_conflict"
+
+
+async def test_rpc_inspect_authoring_contract_handles_invalid_persisted_draft(
+    tmp_path,
+) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    app = create_rpc_app(server)
+    transport = httpx.ASGITransport(app=app)
+    async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
+        created = await _rpc(
+            client,
+            "workflow.draft_workspaces.create_empty",
+            {
+                "workspace_id": "invalid_authoring",
+                "name": "invalid_authoring",
+                "input_schema": {
+                    "type": "object",
+                    "properties": {"request": {"type": "string"}},
+                },
+            },
+        )
+        before = await _rpc(
+            client,
+            "workflow.draft_workspaces.get",
+            {"workspace_id": "invalid_authoring", "include_draft": True},
+        )
+        malformed = await _rpc(
+            client,
+            "workflow.draft_workspaces.inspect_authoring_contract",
+            {
+                "workspace_id": "invalid_authoring",
+                "revision": created["result"]["revision"],
+                "selected_step_id": "",
+            },
+        )
+        patched = await _rpc(
+            client,
+            "workflow.draft_workspaces.patch",
+            {
+                "workspace_id": "invalid_authoring",
+                "revision": created["result"]["revision"],
+                "patch": [
+                    {
+                        "op": "replace",
+                        "path": "/steps",
+                        "value": {"broken": {"unknown_kind": {}}},
+                    },
+                    {"op": "replace", "path": "/start", "value": "broken"},
+                ],
+            },
+        )
+        inspected = await _rpc(
+            client,
+            "workflow.draft_workspaces.inspect_authoring_contract",
+            {
+                "workspace_id": "invalid_authoring",
+                "revision": patched["result"]["revision"],
+                "selected_step_id": "broken",
+            },
+        )
+        after = await _rpc(
+            client,
+            "workflow.draft_workspaces.get",
+            {"workspace_id": "invalid_authoring", "include_draft": True},
+        )
+
+    assert malformed["error"]["code"] == -32602
+    assert before["result"]["revision"] == 1
+    assert patched["result"]["status"] == "invalid"
+    assert inspected["result"]["selected_step_id"] == "broken"
+    assert inspected["result"]["entry_steps"] == []
+    assert inspected["result"]["readable_sources"][0]["path"] == "input.request"
+    assert any("broken" in warning for warning in inspected["result"]["warnings"])
+    assert after["result"]["revision"] == patched["result"]["revision"]
+
+
 @pytest.mark.parametrize(
     ("method", "params"),
     [

tests/wf_transport_rpc_http/test_client.py

@@ -400,6 +400,33 @@ async def test_rpc_client_sends_exact_draft_lifecycle_payloads() -> None:
     ]
 
 
+async def test_rpc_client_sends_exact_authoring_contract_payload() -> None:
+    calls: list[dict[str, Any]] = []
+
+    class Client(RpcDraftClientMixin):
+        async def _call(self, method: str, params: dict[str, object]):
+            calls.append({"method": method, "params": params})
+            return {"workspace_id": "ws", "revision": 4, "selected_step_id": None}
+
+    client = Client()
+    result = await client.inspect_draft_authoring_contract(
+        workspace_id="ws",
+        revision=4,
+    )
+
+    assert result["revision"] == 4
+    assert calls == [
+        {
+            "method": "workflow.draft_workspaces.inspect_authoring_contract",
+            "params": {
+                "workspace_id": "ws",
+                "revision": 4,
+                "selected_step_id": None,
+            },
+        }
+    ]
+
+
 async def test_rpc_client_sends_exact_stateless_draft_payloads() -> None:
     calls: list[dict[str, Any]] = []
 

tests/wf_transport_rpc_http/test_openrpc_contract.py

@@ -492,6 +492,41 @@ def test_openrpc_exposes_typed_draft_workspace_results(
     )
 
 
+def test_openrpc_exposes_typed_authoring_contract_inventory(
+    openrpc_document: dict[str, Any],
+) -> None:
+    method = _method_by_name(
+        openrpc_document,
+        "workflow.draft_workspaces.inspect_authoring_contract",
+    )
+    schemas = openrpc_document["components"]["schemas"]
+
+    assert method["result"]["schema"]["anyOf"] == [
+        {"$ref": "#/components/schemas/AuthoringContractInventoryPayload"},
+        {"$ref": "#/components/schemas/DraftWorkspaceResult"},
+    ]
+    params = method["params"]
+    assert [param["name"] for param in params] == [
+        "workspace_id",
+        "revision",
+        "selected_step_id",
+    ]
+    assert params[0]["required"] is True
+    assert params[1]["required"] is True
+    assert params[2]["required"] is False
+    assert params[2]["schema"]["anyOf"][0] == {
+        "type": "string",
+        "minLength": 1,
+    }
+    assert params[2]["schema"]["anyOf"][1] == {"type": "null"}
+    assert schemas["AuthoringContractInventoryPayload"]["properties"][
+        "selected_step_id"
+    ]["anyOf"] == [{"type": "string"}, {"type": "null"}]
+    option = schemas["AuthoringPathOptionPayload"]
+    assert option["properties"]["reason"]["type"] == "string"
+    assert "reason" not in option["required"]
+
+
 def test_openrpc_separates_step_input_and_workflow_output_binding_unions(
     openrpc_document: dict[str, Any],
 ) -> None: