diff --git a/.superpowers/sdd/python-workflow-client/task-7-report.md b/.superpowers/sdd/python-workflow-client/task-7-report.md
new file mode 100644
index 00000000..14fcf376
--- /dev/null
+++ b/.superpowers/sdd/python-workflow-client/task-7-report.md
@@ -0,0 +1,37 @@
+# Task 7 Report
+
+## Delivered
+
+- Added a real `httpx.ASGITransport` proof for capability discovery, local
+  authoring, remote validation, immutable artifact save, deployment selection,
+  and durable run execution.
+- Added bounded, inert `repr()` and `_repr_html_()` implementations for the
+  Python client's loaded capability, artifact, deployment, validation, run,
+  diagnostic, and trace objects. The shared private renderer HTML-escapes,
+  bounds nested previews, and redacts credential-shaped keys without touching
+  the client port.
+- Added the `wf_client` package map, source/API boundary notes, and a concrete
+  Python walkthrough that explains the artifact -> deployment -> run model.
+- Added an opt-out seam for server draft composition. `drafts=False` skips
+  draft service construction, does not require a draft store, and omits draft
+  JSON-RPC methods; the existing implementation remains available to explicit
+  draft-enabled callers.
+
+## Verification
+
+- `uv run pytest tests/wf_client -q` — 46 passed.
+- Focused/cross-layer Task 7 selection — 288 passed.
+- `uv run pytest tests/wf_api/test_durable_context.py tests/wf_transport_rpc_http/test_app.py::test_rpc_app_can_omit_draft_methods -q` — passed.
+- Ruff check and basedpyright for changed client/API/transport surfaces — passed.
+- `uv run python -m wf_contract_manifest check` — passed.
+- `pnpm --dir web --filter @lda/workflow-rpc contract:check` — passed.
+- `pnpm --dir web --filter @lda/workflow-rpc test` — 151 passed, 3 skipped.
+
+The broader repository format check still reports pre-existing formatting
+differences in `src/wf_api/deployments.py`, `src/wf_authoring/builder/core.py`,
+and `tests/wf_client/test_authoring.py`; no formatting errors remain in the
+changed Task 7 files. The full `uv run pytest -q` run reached 2,613 passed,
+1 skipped, and 1 xfailed; three failures were external to this change: two
+legacy direct-service/draft tests were fixed by retaining the default-enabled
+constructor compatibility, while the remaining thesis asset test expects
+untracked PDF figures absent from the base worktree.
diff --git a/docs/current_roadmap.md b/docs/current_roadmap.md
index 2bc3d149..6871f9c3 100644
--- a/docs/current_roadmap.md
+++ b/docs/current_roadmap.md
@@ -22,6 +22,15 @@ The durable product path is now `wf-rpc-server` plus neutral `wf_config` /
 `wf_server` composition. The old `wf-mcp` script remains a legacy/special-purpose
 MCP entrypoint and compatibility surface.
 
+Completed: the async `wf_client` Python slice is verified against the real
+JSON-RPC ASGI application. It covers capability discovery, local graph
+authoring, remote validation, immutable artifact save, deployment selection,
+and durable run execution. Rich client objects have bounded, secret-safe
+`repr()` and `_repr_html_()` views that never perform remote I/O. Draft
+workspaces remain a separate server/admin surface and are not part of the
+client; server registration is explicit so the artifact -> deployment -> run
+path does not require draft storage.
+
 ## Active Initiative: Workflow Console And Defense Demo
 
 The next product-facing push is a local-first web console and defense demo that
diff --git a/docs/project_map.md b/docs/project_map.md
index f807179a..695e9574 100644
--- a/docs/project_map.md
+++ b/docs/project_map.md
@@ -25,6 +25,7 @@ For verified Python 3.14 dependency constraints and their removal criteria, see
 | `wf_api` | Workflow application surface over core/artifacts/platform: capabilities, drafts, artifacts, deployments, runs, and source/admin surfaces. | `wf_cli`, `wf_server`, JSON-RPC clients, future transports. |
 | `wf_server` | Durable server composition boundary around `WorkflowApi` plus optional admin/source-registry surfaces. Owns the `wf-rpc-server` startup CLI/policy. | Transport packages and server startup code. |
 | `wf_transport_rpc_http` | JSON-RPC-over-HTTP app/client and compatibility CLI shim. | Remote `wf` clients and local server smoke tests. |
+| `wf_client` | Async-native Python client for capability discovery, local authoring, immutable artifacts, deployments, and durable runs. | Python applications and notebooks using a workflow server. |
 | `wf_sources_mcp` | MCP-as-upstream-source implementation: ids, registry DTOs, auth/catalog stores, discovery, SDK client/facade, runtime pool, wrappers. | `wf_server`, broker glue, MCP source tests. |
 | `wf_mcp` | MCP frontend/compatibility package: legacy `wf-mcp` entrypoints, broker glue, proxy/admin tools, and shims while extraction continues. | Compatibility callers and MCP transport work. |
 | `wf_cli` | Command-line frontend over local or remote workflow APIs. | Humans, scripts, agent skills. |
@@ -87,6 +88,8 @@ permanent graph node or expose it as a final workflow-output source.
 - `wf_server.WorkflowServer`: durable workflow server composition object.
 - `wf_transport_rpc_http.RpcWorkflowApiClient`: JSON-RPC client implementing
   the workflow/admin surfaces over HTTP.
+- `wf_client.App`: transport-independent async Python facade over capabilities,
+  authored workflows, saved artifacts, deployments, and durable runs.
 - `wf_transport_rpc_http.create_rpc_app`: JSON-RPC HTTP adapter over an existing
   `WorkflowServer`.
 - `wf_sources_mcp.McpRuntimePool`: persistent MCP source runtime for stateful
@@ -121,6 +124,43 @@ permanent graph node or expose it as a final workflow-output source.
   `--keep-temp` to preserve the generated config/store on failure.
 - `examples/browser_click_workflow/` is a serial browser-click workflow
   example with bounded before/after snapshots and full lifecycle tests.
+
+### Python client walkthrough
+
+The Python client is intended for an application that already has a running
+workflow server. This is the complete shape of a real client call; the schema
+arguments may be JSON Schema dictionaries or the application's schema model
+values:
+
+```python
+from wf_client import App
+
+app = App.from_http_jsonrpc("http://localhost:8765/rpc")
+capability = await app.capability("wf.std.constant")
+graph = app.new_workflow(
+    "example",
+    input_schema=InputModel,
+    state_schema=StateModel,
+    output_schema=OutputModel,
+)
+step = graph.use(capability)
+graph.set_entry_point(step)
+validation = await graph.validate()
+validation.raise_for_errors()
+artifact = await graph.save(version=1)
+run = await artifact.run({})
+```
+
+The graph is a local, mutable builder. `validate()` checks its structure locally
+and then asks the server to validate the serialized plan. `save()` persists an
+immutable artifact version; it does not deploy or execute the graph.
+`artifact.run()` selects or creates a deployment, validates its source bindings,
+and starts a durable run. The returned run is a loaded snapshot; call
+`refresh()`, `resume()`, or bounded `trace(start=..., limit=...)` when more
+server state is needed.
+
+Draft workspaces are intentionally not part of `wf_client`. They are a separate
+server/admin surface and must be explicitly enabled when composing a server.
 - `examples/agent_challenges/` contains reusable opencode challenge harnesses
   for evaluating whether agents can use the public workflow CLI/server path.
 
diff --git a/docs/source_architecture.md b/docs/source_architecture.md
index a91bdf46..ed964a71 100644
--- a/docs/source_architecture.md
+++ b/docs/source_architecture.md
@@ -51,6 +51,7 @@ OpenAPI, or a built-in source. It should see `CapabilitySource` and executable
 | `wf_sources_python` | Trusted Python module registry loading and projection to `CapabilitySource`. | Authoring primitives, registry mutation/apply, sandboxing. |
 | `wf_server` | `WorkflowServer` composition from config/store/source providers. | JSON-RPC method definitions, MCP protocol frontend. |
 | `wf_transport_rpc_http` | JSON-RPC HTTP app/client around an existing `WorkflowServer`. | Server startup policy, source-provider composition. |
+| `wf_client` | Async Python facade over a narrow client port: capability discovery, local authoring, artifact/deployment snapshots, and durable runs. | Draft workspace authoring, server composition, transport registration. |
 | `wf_mcp` | Legacy/special-purpose MCP frontend, broker glue, proxy, compatibility shims. | New durable product behavior unless explicitly retiring old callers. |
 
 ## Data Flow
@@ -65,6 +66,17 @@ wf_config.server.sources[]
   -> transport or CLI
 ```
 
+Python applications consume the same server through `wf_client.App`; the
+client's local builder and immutable snapshots sit above the transport:
+
+```text
+wf_client.App
+  -> WorkflowClientPort
+  -> RpcWorkflowApiClient
+  -> JSON-RPC HTTP
+  -> WorkflowServer / WorkflowApi
+```
+
 The first shared provider seam is intentionally static:
 
 ```python
@@ -212,3 +224,18 @@ Then add:
 
 Do not add source-family branches inside `wf_api` run execution. Source-specific
 logic belongs in the provider package or server composition layer.
+
+## Python client boundary
+
+`wf_client` is a consumer-facing layer, not another server API.
+`App.from_http_jsonrpc()` creates a lazy HTTP transport; the first
+capability/artifact/deployment/run operation performs I/O. `EditableWorkflow`
+keeps graph construction local and uses the port only for remote validation and
+artifact persistence. The server remains responsible for durable artifact,
+deployment, and run stores and for resolving concrete source bindings.
+
+The client intentionally omits draft workspace operations. Drafts are a
+server-side authoring/admin surface, and a server composition must opt in to
+registering their JSON-RPC methods. This keeps a normal Python application
+focused on the stable artifact -> deployment -> run lifecycle while retaining
+the underlying draft implementation for explicit server users.
diff --git a/docs/wf_api_architecture.md b/docs/wf_api_architecture.md
index 3131c5c8..315ae5ca 100644
--- a/docs/wf_api_architecture.md
+++ b/docs/wf_api_architecture.md
@@ -20,6 +20,7 @@ frontends can share.
 | `wf_sources_mcp` | MCP-as-upstream-source implementation: source ids, source registry DTOs, auth/catalog stores, discovery, SDK client/facade, persistent runtime pool, and tool-wrapper helpers. |
 | `wf_mcp` | MCP frontend/compatibility package: old `wf-mcp` server entry points, broker glue around MCP-hosted services, proxy/admin tools, and compatibility shims while callers migrate. |
 | `wf_transport_rpc_http` | JSON-RPC-over-HTTP transport adapter and remote client over `WorkflowApiSurface`, not a reimplementation of workflow business logic. |
+| `wf_client` | Async Python consumer facade over a narrow capability/artifact/deployment/run port. It reconstructs immutable snapshots and keeps representations bounded and inert. |
 | future `wf_http` / WebSocket / MCP server transports | Additional transports over `WorkflowApiSurface`, not new workflow application APIs. |
 | `wf_cli` | CLI frontend over `WorkflowApiSurface`; it may run locally against process-local stores or target a remote JSON-RPC backend. |
 
@@ -107,6 +108,40 @@ Important rules:
 Do not add a catch-all `service` field to the context. If a domain API needs a
 new dependency, add a narrow protocol or explicit field.
 
+## Python client lifecycle
+
+The Python client makes the intended application flow explicit:
+
+```python
+app = App.from_http_jsonrpc("http://localhost:8765/rpc")
+capability = await app.capability("wf.std.constant")
+graph = app.new_workflow(
+    "example",
+    input_schema=InputModel,
+    state_schema=StateModel,
+    output_schema=OutputModel,
+)
+step = graph.use(capability)
+graph.set_entry_point(step)
+validation = await graph.validate()
+validation.raise_for_errors()
+artifact = await graph.save(version=1)
+run = await artifact.run({})
+```
+
+The graph is an in-process builder. Validation is local structural checking
+plus a server plan check. Saving creates an immutable, versioned artifact; it
+does not execute anything. A deployment is the server's runnable configuration
+for one exact artifact version, including logical-to-concrete source bindings
+and drift policy. A run is a durable execution record for that deployment;
+inspection and bounded trace reads return snapshots, while resume is an
+explicit operation for interrupted runs.
+
+`wf_client` does not expose draft workspaces. Draft API classes remain useful
+to server/admin and console callers, but normal server composition keeps draft
+JSON-RPC registration opt-in so artifact, deployment, and run durability do not
+depend on a draft store.
+
 ## WorkflowApiSurface And Domain Services
 
 `WorkflowApiSurface` is the public application contract shared by local and
diff --git a/src/wf_api/durable_context.py b/src/wf_api/durable_context.py
index 026e390f..102969f1 100644
--- a/src/wf_api/durable_context.py
+++ b/src/wf_api/durable_context.py
@@ -5,25 +5,27 @@ from .service import WorkflowApi
 from .stores import WorkflowStores
 
 
-def require_workflow_stores(context: WorkflowOperationContext) -> WorkflowStores:
+def require_workflow_stores(
+    context: WorkflowOperationContext,
+    *,
+    drafts: bool = True,
+) -> WorkflowStores:
     """Return required stores or fail before constructing durable frontends.
 
     `WorkflowOperationContext` keeps stores optional for compatibility tests and
-    lightweight MCP surfaces. Durable API surfaces need all stores up front so a
-    run cannot start without somewhere to persist artifacts, drafts, and stopped
-    execution state.
+    lightweight MCP surfaces. Durable API surfaces need artifact/run stores up
+    front; a draft store is only required when draft APIs are enabled.
     """
     missing = []
     if context.artifact_store is None:
         missing.append("artifact_store")
-    if context.draft_workspace_store is None:
+    if drafts and context.draft_workspace_store is None:
         missing.append("draft_workspace_store")
     if context.run_store is None:
         missing.append("run_store")
     if missing:
         raise ValueError("durable workflow API requires stores: " + ", ".join(missing))
     assert context.artifact_store is not None
-    assert context.draft_workspace_store is not None
     assert context.run_store is not None
     return WorkflowStores(
         artifact_store=context.artifact_store,
@@ -32,10 +34,14 @@ def require_workflow_stores(context: WorkflowOperationContext) -> WorkflowStores
     )
 
 
-def durable_workflow_api(context: WorkflowOperationContext) -> WorkflowApi:
-    """Construct a WorkflowApi only after durable store dependencies exist."""
-    require_workflow_stores(context)
-    return WorkflowApi(context)
+def durable_workflow_api(
+    context: WorkflowOperationContext,
+    *,
+    drafts: bool = True,
+) -> WorkflowApi:
+    """Construct a durable API, optionally omitting the draft product surface."""
+    require_workflow_stores(context, drafts=drafts)
+    return WorkflowApi(context, drafts=drafts)
 
 
 __all__ = ["durable_workflow_api", "require_workflow_stores"]
diff --git a/src/wf_api/service.py b/src/wf_api/service.py
index 641e9912..04ea7543 100644
--- a/src/wf_api/service.py
+++ b/src/wf_api/service.py
@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from collections.abc import Mapping, Sequence
-from typing import Any, Literal, overload
+from typing import Any, Literal, cast, overload
 
 from wf_artifacts import ArtifactKind, compile_workflow_draft
 from wf_artifacts.drafts.models import DraftStep
@@ -56,6 +56,19 @@ from .operation_context import WorkflowOperationContext
 from .runs import TraceRangeLike, WorkflowRunApi
 
 
+class _DisabledDraftSurface:
+    """Placeholder that fails clearly if disabled draft methods are called.
+
+    Keeping this tiny seam avoids constructing draft services while preserving
+    the existing method layout on ``WorkflowApi`` for explicit draft callers.
+    """
+
+    def __getattr__(self, name: str) -> Any:
+        raise RuntimeError(
+            "workflow draft APIs are disabled; compose WorkflowApi with drafts=True"
+        )
+
+
 def _authoring_schema(
     value: object,
     *,
@@ -100,11 +113,26 @@ class WorkflowApi:
     callers share one operation surface without importing wf_mcp.
     """
 
-    def __init__(self, context: WorkflowOperationContext) -> None:
+    def __init__(
+        self,
+        context: WorkflowOperationContext,
+        *,
+        drafts: bool = True,
+    ) -> None:
         self.context = context
         self.capabilities = WorkflowCapabilityApi(context)
-        self.drafts = WorkflowDraftApi(context)
-        self.draft_authoring = WorkflowDraftAuthoringApi(context, self.drafts)
+        # ``drafts`` keeps server composition explicit: callers that omit draft
+        # storage must pass False, while the default preserves legacy direct
+        # WorkflowApi callers that use the draft service for validation only.
+        self.drafts_enabled = drafts
+        if self.drafts_enabled:
+            self.drafts = WorkflowDraftApi(context)
+            self.draft_authoring = WorkflowDraftAuthoringApi(context, self.drafts)
+        else:
+            self.drafts = cast(WorkflowDraftApi, _DisabledDraftSurface())
+            self.draft_authoring = cast(
+                WorkflowDraftAuthoringApi, _DisabledDraftSurface()
+            )
         self.artifacts = WorkflowArtifactApi(context)
         self.deployments = WorkflowDeploymentApi(context)
         self.runs = WorkflowRunApi(context)
diff --git a/src/wf_api/stores.py b/src/wf_api/stores.py
index 805a2ea7..68507a58 100644
--- a/src/wf_api/stores.py
+++ b/src/wf_api/stores.py
@@ -18,7 +18,7 @@ class WorkflowStores:
     """Protocol-neutral persistence dependencies for workflow APIs."""
 
     artifact_store: WorkflowArtifactStore
-    draft_workspace_store: DraftWorkspaceStore
+    draft_workspace_store: DraftWorkspaceStore | None
     run_store: RunStore
 
 
diff --git a/src/wf_client/_repr.py b/src/wf_client/_repr.py
new file mode 100644
index 00000000..46e27241
--- /dev/null
+++ b/src/wf_client/_repr.py
@@ -0,0 +1,103 @@
+"""Small, inert renderers shared by the public workflow-client snapshots.
+
+Representations are a debugging aid, not another client operation.  This
+module deliberately accepts already-loaded values and never knows about the
+workflow transport port.  The same bounded projector is used for plain and
+HTML representations so notebooks cannot accidentally expose an unbounded
+trace, output, or credential-shaped value.
+"""
+
+from __future__ import annotations
+
+import html
+import json
+from collections.abc import Mapping, Sequence
+
+_SECRET_KEY_PARTS = (
+    "authorization",
+    "cookie",
+    "set-cookie",
+    "token",
+    "secret",
+    "password",
+    "api_key",
+    "api-key",
+)
+_MAX_DEPTH = 2
+_MAX_ITEMS = 8
+_MAX_STRING = 160
+_MAX_RENDERED = 1_200
+
+
+def _secret_key(key: object) -> bool:
+    lowered = str(key).lower()
+    return any(part in lowered for part in _SECRET_KEY_PARTS)
+
+
+def bounded_value(value: object, *, depth: int = 0) -> object:
+    """Project loaded JSON-like data into a small, secret-safe preview."""
+    if depth >= _MAX_DEPTH:
+        return "[truncated]"
+    if isinstance(value, str):
+        return value if len(value) <= _MAX_STRING else value[:_MAX_STRING] + "…"
+    if value is None or isinstance(value, bool | int | float):
+        return value
+    if isinstance(value, Mapping):
+        items = list(value.items())
+        preview = {
+            str(key): "[redacted]"
+            if _secret_key(key)
+            else bounded_value(item, depth=depth + 1)
+            for key, item in items[:_MAX_ITEMS]
+        }
+        if len(items) > _MAX_ITEMS:
+            preview["…"] = f"{len(items) - _MAX_ITEMS} more entries"
+        return preview
+    if isinstance(value, Sequence) and not isinstance(value, str | bytes | bytearray):
+        items = list(value)
+        preview = [bounded_value(item, depth=depth + 1) for item in items[:_MAX_ITEMS]]
+        if len(items) > _MAX_ITEMS:
+            preview.append(f"… {len(items) - _MAX_ITEMS} more items")
+        return preview
+    rendered = repr(value)
+    return rendered if len(rendered) <= _MAX_STRING else rendered[:_MAX_STRING] + "…"
+
+
+def preview(value: object) -> str:
+    """Render a bounded value without allowing an object's repr to grow freely."""
+    try:
+        rendered = json.dumps(bounded_value(value), sort_keys=True, default=str)
+    except TypeError, ValueError:
+        rendered = str(bounded_value(value))
+    return rendered[:_MAX_RENDERED] + ("…" if len(rendered) > _MAX_RENDERED else "")
+
+
+def short_repr(type_name: str, **fields: object) -> str:
+    """Build a compact Python repr from already-loaded field values."""
+    body = ", ".join(f"{name}={preview(value)}" for name, value in fields.items())
+    rendered = f"{type_name}({body})"
+    return rendered[:_MAX_RENDERED] + ("…" if len(rendered) > _MAX_RENDERED else "")
+
+
+def html_repr(type_name: str, **fields: object) -> str:
+    """Build a bounded HTML table suitable for IPython rich display."""
+
+    def html_preview(value: object) -> str:
+        rendered = preview(value)
+        return rendered[:400] + ("…" if len(rendered) > 400 else "")
+
+    rows = "".join(
+        '<tr><th scope="row">'
+        + html.escape(name)
+        + "</th><td><code>"
+        + html.escape(html_preview(value))
+        + "</code></td></tr>"
+        for name, value in fields.items()
+    )
+    return (
+        '<div class="wf-client-repr"><strong>'
+        + html.escape(type_name)
+        + "</strong><table><tbody>"
+        + rows
+        + "</tbody></table></div>"
+    )
diff --git a/src/wf_client/capabilities.py b/src/wf_client/capabilities.py
index 08f7cef8..96aabcca 100644
--- a/src/wf_client/capabilities.py
+++ b/src/wf_client/capabilities.py
@@ -13,6 +13,7 @@ from wf_artifacts.models import DependencyDiagnostic
 from wf_core.models.schemas import NodeDef, SchemaRef
 from wf_platform import CapabilityRef
 
+from ._repr import html_repr, short_repr
 from .codec import decode_capability_call, decode_capability_diagnostics
 from .errors import InvalidResponse
 from .protocols import WorkflowClientPort
@@ -39,6 +40,24 @@ class CapabilitySummary:
         """Compatibility alias for the wire row's ``name`` field."""
         return self.qualified_name
 
+    def __repr__(self) -> str:
+        return short_repr(
+            type(self).__name__,
+            name=self.qualified_name,
+            source=self.source_id,
+            outcomes=self.outcomes,
+        )
+
+    def _repr_html_(self) -> str:
+        return html_repr(
+            type(self).__name__,
+            name=self.qualified_name,
+            source=self.source_id,
+            outcomes=self.outcomes,
+            inputs=f"{len(self.input_fields)} fields",
+            outputs=f"{len(self.output_fields)} fields",
+        )
+
 
 @dataclass(frozen=True, slots=True)
 class CapabilityResult:
@@ -48,6 +67,22 @@ class CapabilityResult:
     output: dict[str, Any] | None
     diagnostics: tuple[DependencyDiagnostic, ...]
 
+    def __repr__(self) -> str:
+        return short_repr(
+            type(self).__name__,
+            outcome=self.outcome,
+            output=self.output,
+            diagnostics=f"{len(self.diagnostics)} diagnostics",
+        )
+
+    def _repr_html_(self) -> str:
+        return html_repr(
+            type(self).__name__,
+            outcome=self.outcome,
+            output=self.output,
+            diagnostics=f"{len(self.diagnostics)} diagnostics",
+        )
+
 
 def _check_schema(schema: object, *, operation: str) -> dict[str, Any]:
     if not isinstance(schema, Mapping):
@@ -106,6 +141,27 @@ class RemoteCapability:
         )
         object.__setattr__(self, "outcomes", tuple(self.outcomes))
 
+    def __repr__(self) -> str:
+        return short_repr(
+            type(self).__name__,
+            name=self.qualified_name,
+            outcomes=self.outcomes,
+            input_schema=f"{len(self.input_schema)} keys",
+            output_schema=f"{len(self.output_schema)} keys",
+        )
+
+    def _repr_html_(self) -> str:
+        return html_repr(
+            type(self).__name__,
+            name=self.qualified_name,
+            description=self.description,
+            outcomes=self.outcomes,
+            **{
+                "input schema": f"{len(self.input_schema)} keys",
+                "output schema": f"{len(self.output_schema)} keys",
+            },
+        )
+
     async def __call__(
         self,
         payload: Mapping[str, Any] | None = None,
diff --git a/src/wf_client/deployments.py b/src/wf_client/deployments.py
index 7388b262..3535ac59 100644
--- a/src/wf_client/deployments.py
+++ b/src/wf_client/deployments.py
@@ -8,6 +8,7 @@ from typing import Any
 
 from wf_artifacts import DependencyDiagnostic, DriftPolicy, WorkflowDeployment
 
+from ._repr import html_repr, short_repr
 from .codec import (
     decode_dependency_diagnostics,
     decode_deployment,
@@ -33,6 +34,22 @@ class DeploymentValidation:
     def runnable(self) -> bool:
         return self.status == "runnable"
 
+    def __repr__(self) -> str:
+        return short_repr(
+            type(self).__name__,
+            deployment_id=self.deployment_id,
+            status=self.status,
+            diagnostics=f"{len(self.diagnostics)} diagnostics",
+        )
+
+    def _repr_html_(self) -> str:
+        return html_repr(
+            type(self).__name__,
+            deployment_id=self.deployment_id,
+            status=self.status,
+            diagnostics=f"{len(self.diagnostics)} diagnostics",
+        )
+
 
 @dataclass(frozen=True, slots=True)
 class Deployment:
@@ -51,6 +68,25 @@ class Deployment:
     def deployment_id(self) -> str:
         return self.model.id
 
+    def __repr__(self) -> str:
+        return short_repr(
+            type(self).__name__,
+            deployment_id=self.deployment_id,
+            artifact=f"{self.artifact_id}.v{self.artifact_version}",
+            runnable=self.runnable,
+            diagnostics=f"{len(self.diagnostics)} diagnostics",
+        )
+
+    def _repr_html_(self) -> str:
+        return html_repr(
+            type(self).__name__,
+            deployment_id=self.deployment_id,
+            artifact=f"{self.artifact_id}.v{self.artifact_version}",
+            bindings=f"{len(self.bindings)} bindings",
+            runnable=self.runnable,
+            diagnostics=f"{len(self.diagnostics)} diagnostics",
+        )
+
     @property
     def artifact_id(self) -> str:
         return self.model.artifact_id
diff --git a/src/wf_client/runs.py b/src/wf_client/runs.py
index 4b6f87b6..c2dadb2b 100644
--- a/src/wf_client/runs.py
+++ b/src/wf_client/runs.py
@@ -12,6 +12,7 @@ from wf_api import TraceRange
 from wf_artifacts import DependencyDiagnostic
 from wf_core import InterruptRequest, InterruptRoute, TraceEntry, WorkflowRef
 
+from ._repr import html_repr, short_repr
 from .codec import DecodedRunResult, decode_run_result, decode_trace_result
 from .errors import DeploymentNotRunnable, InvalidResponse
 from .protocols import WorkflowClientPort
@@ -27,6 +28,24 @@ class TracePage:
     truncated: bool
     trace_count: int
 
+    def __repr__(self) -> str:
+        return short_repr(
+            type(self).__name__,
+            start=self.start,
+            limit=self.limit,
+            frames=f"{len(self.frames)} loaded/{self.trace_count} total",
+            truncated=self.truncated,
+        )
+
+    def _repr_html_(self) -> str:
+        return html_repr(
+            type(self).__name__,
+            start=self.start,
+            limit=self.limit,
+            frames=f"{len(self.frames)} loaded/{self.trace_count} total",
+            truncated=self.truncated,
+        )
+
 
 def _interrupt(
     payload: Mapping[str, Any] | None,
@@ -110,6 +129,30 @@ class Run:
     diagnostics: tuple[DependencyDiagnostic, ...]
     trace_count: int
 
+    def __repr__(self) -> str:
+        return short_repr(
+            type(self).__name__,
+            run_id=self.run_id,
+            deployment_id=self.deployment_id,
+            status=self.status,
+            outcome=self.outcome,
+            output=self.output,
+            diagnostics=f"{len(self.diagnostics)} diagnostics",
+            trace=f"{self.trace_count} frames",
+        )
+
+    def _repr_html_(self) -> str:
+        return html_repr(
+            type(self).__name__,
+            run_id=self.run_id,
+            deployment_id=self.deployment_id,
+            status=self.status,
+            outcome=self.outcome,
+            output=self.output,
+            diagnostics=f"{len(self.diagnostics)} diagnostics",
+            trace=f"{self.trace_count} frames (use trace() for a bounded page)",
+        )
+
     @classmethod
     def from_payload(
         cls,
diff --git a/src/wf_client/workflows.py b/src/wf_client/workflows.py
index bed92300..74a6755e 100644
--- a/src/wf_client/workflows.py
+++ b/src/wf_client/workflows.py
@@ -14,6 +14,7 @@ from wf_artifacts.models import (
 )
 from wf_core import ValidationReport, Workflow
 
+from ._repr import html_repr, short_repr
 from .errors import InvalidResponse, ValidationFailed
 
 if TYPE_CHECKING:
@@ -30,6 +31,16 @@ class ArtifactRef:
     artifact_id: str
     version: int
 
+    def __repr__(self) -> str:
+        return short_repr(
+            type(self).__name__, artifact_id=self.artifact_id, version=self.version
+        )
+
+    def _repr_html_(self) -> str:
+        return html_repr(
+            type(self).__name__, artifact_id=self.artifact_id, version=self.version
+        )
+
 
 @dataclass(frozen=True, slots=True)
 class WorkflowDiagnostic:
@@ -41,6 +52,24 @@ class WorkflowDiagnostic:
     message: str
     repair_hint: str | None = None
 
+    def __repr__(self) -> str:
+        return short_repr(
+            type(self).__name__,
+            severity=self.severity,
+            code=self.code,
+            path=self.path,
+            message=self.message,
+        )
+
+    def _repr_html_(self) -> str:
+        return html_repr(
+            type(self).__name__,
+            severity=self.severity,
+            code=self.code,
+            path=self.path,
+            message=self.message,
+        )
+
 
 # Keep the short name used by the public design available without requiring a
 # second diagnostic implementation.
@@ -59,6 +88,22 @@ class WorkflowValidation:
     def ok(self) -> bool:
         return self.local.ok and self.remote_status == "valid"
 
+    def __repr__(self) -> str:
+        return short_repr(
+            type(self).__name__,
+            ok=self.ok,
+            remote_status=self.remote_status,
+            diagnostics=f"{len(self.remote_diagnostics)} diagnostics",
+        )
+
+    def _repr_html_(self) -> str:
+        return html_repr(
+            type(self).__name__,
+            ok=self.ok,
+            remote_status=self.remote_status,
+            diagnostics=f"{len(self.remote_diagnostics)} diagnostics",
+        )
+
     def raise_for_errors(self) -> None:
         """Raise a useful error for either local or remote validation failures."""
         self.local.raise_for_errors()
@@ -86,6 +131,23 @@ class WorkflowArtifact:
     def ref(self) -> ArtifactRef:
         return ArtifactRef(self.artifact.id, self.artifact.version)
 
+    def __repr__(self) -> str:
+        return short_repr(
+            type(self).__name__,
+            ref=self.ref,
+            title=self.title,
+            required_capabilities=f"{len(self.required_capabilities)} capabilities",
+        )
+
+    def _repr_html_(self) -> str:
+        return html_repr(
+            type(self).__name__,
+            ref=self.ref,
+            title=self.title,
+            description=self.description,
+            required_capabilities=f"{len(self.required_capabilities)} capabilities",
+        )
+
     @property
     def title(self) -> str:
         return self.artifact.title
@@ -137,8 +199,13 @@ class WorkflowArtifact:
                 "drift_policy": drift_policy,
             }
         )
-        if not isinstance(saved, Mapping) or saved.get("deployment_id") != deployment_id:
-            saved_id = saved.get("deployment_id") if isinstance(saved, Mapping) else None
+        if (
+            not isinstance(saved, Mapping)
+            or saved.get("deployment_id") != deployment_id
+        ):
+            saved_id = (
+                saved.get("deployment_id") if isinstance(saved, Mapping) else None
+            )
             raise InvalidResponse(
                 operation="workflow.deployments.save",
                 details=(
diff --git a/src/wf_transport_rpc_http/app.py b/src/wf_transport_rpc_http/app.py
index 47014dd6..b578b673 100644
--- a/src/wf_transport_rpc_http/app.py
+++ b/src/wf_transport_rpc_http/app.py
@@ -24,7 +24,12 @@ from .methods.source_registry import (
 from .methods.sources import register_methods as register_source_methods
 
 
-def create_rpc_app(server: WorkflowServer, *, rpc_path: str = "/rpc") -> jsonrpc.API:
+def create_rpc_app(
+    server: WorkflowServer,
+    *,
+    rpc_path: str = "/rpc",
+    drafts: bool | None = None,
+) -> jsonrpc.API:
     """Build a JSON-RPC HTTP app over an existing WorkflowServer.
 
     Transport code owns only JSON-RPC envelope handling. Workflow semantics stay
@@ -49,7 +54,9 @@ def create_rpc_app(server: WorkflowServer, *, rpc_path: str = "/rpc") -> jsonrpc
         }
 
     register_capability_methods(entrypoint, server)
-    register_draft_methods(entrypoint, server)
+    drafts_enabled = server.api.drafts_enabled if drafts is None else drafts
+    if drafts_enabled:
+        register_draft_methods(entrypoint, server)
     register_artifact_methods(entrypoint, server)
     register_deployment_methods(entrypoint, server)
     register_run_methods(entrypoint, server)
diff --git a/tests/wf_api/test_durable_context.py b/tests/wf_api/test_durable_context.py
index 82f76b4a..1a2a1218 100644
--- a/tests/wf_api/test_durable_context.py
+++ b/tests/wf_api/test_durable_context.py
@@ -54,3 +54,18 @@ def test_durable_workflow_api_returns_workflow_api_with_same_context(tmp_path) -
 
     assert isinstance(api, WorkflowApi)
     assert api.context is context
+
+
+def test_durable_workflow_api_can_opt_out_of_draft_store(tmp_path) -> None:
+    stores = file_workflow_stores(tmp_path / "workflow_stores")
+    service = WfMcpService(
+        store=FileStore(tmp_path / "mcp"),
+        artifact_store=stores.artifact_store,
+        draft_workspace_store=None,
+        run_store=stores.run_store,
+    )
+    context = context_from_service(service)
+
+    api = durable_workflow_api(context, drafts=False)
+
+    assert api.drafts_enabled is False
diff --git a/tests/wf_client/test_http_integration.py b/tests/wf_client/test_http_integration.py
new file mode 100644
index 00000000..207f6626
--- /dev/null
+++ b/tests/wf_client/test_http_integration.py
@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+import httpx
+import pytest
+
+from wf_authoring import input_from, input_value, output_to, state_path
+from wf_client import App, ArtifactRef
+from wf_server import build_local_static_workflow_server
+from wf_transport_rpc_http import RpcWorkflowApiClient, create_rpc_app
+
+
+@pytest.mark.asyncio
+async def test_http_app_calls_authors_saves_deploys_and_runs(tmp_path) -> None:
+    """Prove the public client lifecycle against the real JSON-RPC ASGI app."""
+    server = build_local_static_workflow_server(tmp_path / "store")
+    rpc_app = create_rpc_app(server)
+    transport = httpx.ASGITransport(app=rpc_app)
+
+    async with httpx.AsyncClient(
+        transport=transport,
+        base_url="http://test",
+    ) as http_client:
+        app = App._from_port(
+            RpcWorkflowApiClient(
+                url="http://test/rpc",
+                http_client=http_client,
+            )
+        )
+        constant = await app.capability("wf.std.constant")
+        graph = app.new_workflow(
+            "http_client_proof",
+            input_schema={"type": "object", "properties": {}},
+            state_schema={
+                "type": "object",
+                "properties": {"value": {"type": "string"}},
+            },
+            output_schema={
+                "type": "object",
+                "properties": {"value": {"type": "string"}},
+                "required": ["value"],
+            },
+        )
+        step = graph.use(
+            constant,
+            id="constant",
+            input=[input_value("value", "hello")],
+            output=[output_to("value", state_path("value"))],
+        )
+        end = graph.end("ok", id="end_ok")
+        graph.set_entry_point(step)
+        graph.connect(step, "ok", end)
+        graph.set_output([input_from(state_path("value"), "value")])
+
+        validation = await graph.validate()
+        artifact = await graph.save(version=1, title="HTTP client proof")
+        run = await artifact.run({})
+
+    assert validation.ok is True
+    assert artifact.ref == ArtifactRef("http_client_proof", 1)
+    assert run.status == "completed"
+    assert run.output == {"value": "hello"}
diff --git a/tests/wf_client/test_repr.py b/tests/wf_client/test_repr.py
new file mode 100644
index 00000000..34847157
--- /dev/null
+++ b/tests/wf_client/test_repr.py
@@ -0,0 +1,117 @@
+from __future__ import annotations
+
+from typing import Any, cast
+
+from wf_client.capabilities import CapabilityResult, RemoteCapability
+from wf_client.deployments import Deployment, DeploymentValidation
+from wf_client.protocols import WorkflowClientPort
+from wf_client.runs import Run, TracePage
+from wf_client.workflows import (
+    ArtifactRef,
+    WorkflowDiagnostic,
+    WorkflowValidation,
+)
+from wf_core import ValidationReport
+from wf_platform import CapabilityRef
+
+
+class _Port:
+    """A port that records accidental representation-time remote operations."""
+
+    def __init__(self) -> None:
+        self.calls: list[tuple[str, dict[str, Any]]] = []
+
+    def __getattr__(self, name: str) -> Any:
+        async def operation(**params: Any) -> object:
+            self.calls.append((name, params))
+            return {}
+
+        return operation
+
+
+def _port() -> _Port:
+    return _Port()
+
+
+def test_capability_html_repr_is_bounded_and_does_not_call_port() -> None:
+    port = _port()
+    remote = RemoteCapability(
+        _port=cast(WorkflowClientPort, port),
+        ref=CapabilityRef.parse("app.default.search"),
+        qualified_name="app.default.search",
+        description="Search things <carefully>",
+        input_schema={"type": "object", "properties": {"query": {"type": "string"}}},
+        output_schema={"type": "object", "properties": {"items": {"type": "array"}}},
+        outcomes=("ok",),
+        is_async=False,
+    )
+
+    rendered = remote._repr_html_()
+
+    assert "app.default.search" in rendered
+    assert "input schema" in rendered.lower()
+    assert "&lt;carefully&gt;" in rendered
+    assert port.calls == []
+
+
+def test_rich_representations_bound_large_values_and_redact_secret_like_fields() -> (
+    None
+):
+    port = _port()
+    result = CapabilityResult(
+        outcome="ok",
+        output={"token": "do-not-show", "items": ["x" * 400] * 20},
+        diagnostics=(),
+    )
+    run = Run(
+        _port=cast(WorkflowClientPort, port),
+        run_id="run-1",
+        deployment_id="deployment-1",
+        status="completed",
+        outcome="ok",
+        output=result.output,
+        interrupt=None,
+        diagnostics=(),
+        trace_count=1000,
+    )
+
+    rendered = repr(run)
+    html = run._repr_html_()
+
+    assert len(rendered) <= 1_201
+    assert len(html) <= 2_500
+    assert "do-not-show" not in rendered
+    assert "do-not-show" not in html
+    assert "1000 frames" in rendered
+    assert port.calls == []
+
+
+def test_all_rich_objects_render_without_port_access() -> None:
+    port = cast(WorkflowClientPort, _port())
+    diagnostic = WorkflowDiagnostic("error", "bad", "state.x", "broken")
+    local = ValidationReport()
+    objects = [
+        ArtifactRef("artifact", 1),
+        diagnostic,
+        WorkflowValidation(local, "valid", (diagnostic,)),
+        CapabilityResult("ok", {"value": 1}, ()),
+        DeploymentValidation("deployment", "artifact", 1, "runnable", ()),
+        TracePage(0, 25, (), False, 0),
+    ]
+
+    for value in objects:
+        assert repr(value)
+        assert value._repr_html_()
+    assert repr(
+        Deployment.from_payload(
+            port,
+            {
+                "id": "deployment",
+                "artifact_id": "artifact",
+                "artifact_version": 1,
+                "bindings": [],
+                "drift_policy": "block",
+            },
+        )
+    )
+    assert port.calls == []
diff --git a/tests/wf_transport_rpc_http/test_app.py b/tests/wf_transport_rpc_http/test_app.py
index 0a650a47..9beea833 100644
--- a/tests/wf_transport_rpc_http/test_app.py
+++ b/tests/wf_transport_rpc_http/test_app.py
@@ -32,6 +32,17 @@ async def _rpc(
     return response.json()
 
 
+def test_rpc_app_can_omit_draft_methods(tmp_path) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+
+    assert server.api.drafts_enabled is True
+    app = create_rpc_app(server, drafts=False)
+    methods = {method["name"] for method in app.get_openrpc()["methods"]}
+
+    assert "workflow.capabilities.list" in methods
+    assert "workflow.draft_workspaces.list" not in methods
+
+
 def _rpc_constant_draft() -> dict[str, Any]:
     """Return the canonical keyed draft shared by stateless RPC tests."""
     return {