fix: format cli operation errors

docs/wf_cli.md

@@ -99,6 +99,14 @@ Detail and mutation commands are JSON-only unless documented otherwise.
 
 There is no `table` format in v1.
 
+By default, expected operation failures are shown as compact CLI errors without
+Python tracebacks. Use the root `--verbose` flag when debugging internal
+failures:
+
+```bash
+wf --verbose --url http://127.0.0.1:8765/rpc source inspect missing.source
+```
+
 ## Lifecycle
 
 The normal CLI workflow is:

src/wf_cli/app.py

@@ -48,13 +48,22 @@ def root(
         float | None,
         typer.Option("--timeout", min=0.1, help="Override RPC timeout seconds."),
     ] = None,
+    verbose: Annotated[
+        bool,
+        typer.Option(
+            "--verbose",
+            help="Show full tracebacks for unexpected internal errors.",
+        ),
+    ] = False,
 ) -> None:
     """Run workflow platform commands."""
+    app.pretty_exceptions_short = not verbose
     ctx.obj = CliTyperState(
         config_path=config,
         force_local=local,
         rpc_url=url,
         rpc_timeout_seconds=timeout,
+        verbose=verbose,
     )
 
 

src/wf_cli/commands/admin.py

@@ -1,12 +1,11 @@
 from __future__ import annotations
-
-import asyncio
 from typing import Annotated
 
 import typer
 
 from wf_cli.context import load_cli_context_from_typer
 from wf_cli.formats import ListOutputFormat, emit_list_payload
+from wf_cli.remote_errors import run_cli_operation
 
 from . import source_registry
 
@@ -28,7 +27,7 @@ def list_connections(
 ) -> None:
     """List configured upstream connections known to the target."""
     context = load_cli_context_from_typer(ctx)
-    payload = asyncio.run(context.admin.list_connections())
+    payload = run_cli_operation(context, context.admin.list_connections())
     emit_list_payload(
         payload,
         collection_key="connections",
@@ -47,7 +46,7 @@ def get_connection_statuses(
 ) -> None:
     """List connection catalog/status summaries."""
     context = load_cli_context_from_typer(ctx)
-    payload = asyncio.run(context.admin.get_connection_statuses())
+    payload = run_cli_operation(context, context.admin.get_connection_statuses())
     emit_list_payload(
         payload,
         collection_key="statuses",
@@ -66,7 +65,7 @@ def list_events(
 ) -> None:
     """List recorded workflow platform events."""
     context = load_cli_context_from_typer(ctx)
-    payload = asyncio.run(context.admin.list_events())
+    payload = run_cli_operation(context, context.admin.list_events())
     emit_list_payload(
         payload,
         collection_key="events",

src/wf_cli/commands/artifacts.py

@@ -1,6 +1,4 @@
 from __future__ import annotations
-
-import asyncio
 from typing import Annotated, Literal
 
 import typer
@@ -8,6 +6,7 @@ import typer
 from wf_cli.context import load_cli_context_from_typer as load_cli_context
 from wf_cli.formats import ListOutputFormat, emit_list_payload
 from wf_cli.io import emit_json
+from wf_cli.remote_errors import run_cli_operation
 
 app = typer.Typer(
     name="artifact",
@@ -38,13 +37,14 @@ def list_artifacts(
 ) -> None:
     """List compact saved artifact summaries."""
     context = load_cli_context(ctx)
-    payload = asyncio.run(
+    payload = run_cli_operation(
+        context,
         context.handlers.list_artifacts(
             query=query,
             kind=kind,
             cursor=cursor,
             limit=limit,
-        )
+        ),
     )
     emit_list_payload(
         payload,
@@ -64,7 +64,8 @@ def inspect_artifact(
     """Inspect one saved artifact version."""
     context = load_cli_context(ctx)
     emit_json(
-        asyncio.run(
+        run_cli_operation(
-            context.handlers.inspect_artifact(artifact_id=artifact_id, version=version)
+            context,
+            context.handlers.inspect_artifact(artifact_id=artifact_id, version=version),
         )
     )

src/wf_cli/commands/caps.py

@@ -1,6 +1,4 @@
 from __future__ import annotations
-
-import asyncio
 from typing import Annotated
 
 import typer
@@ -8,6 +6,7 @@ import typer
 from wf_cli.context import load_cli_context_from_typer
 from wf_cli.formats import ListOutputFormat, emit_list_payload
 from wf_cli.io import emit_json
+from wf_cli.remote_errors import run_cli_operation
 
 app = typer.Typer(
     name="cap",
@@ -38,13 +37,14 @@ def list_capabilities(
 ) -> None:
     """List compact planner-visible workflow capabilities."""
     context = load_cli_context_from_typer(ctx)
-    payload = asyncio.run(
+    payload = run_cli_operation(
+        context,
         context.handlers.list_capabilities(
             query=query,
             source_id=source_id,
             cursor=cursor,
             limit=limit,
-        )
+        ),
     )
     emit_list_payload(
         payload,
@@ -62,7 +62,8 @@ def inspect_capability(
 ) -> None:
     """Inspect one workflow capability contract."""
     context = load_cli_context_from_typer(ctx)
-    payload = asyncio.run(
+    payload = run_cli_operation(
-        context.handlers.inspect_capability(qualified_name=qualified_name)
+        context,
+        context.handlers.inspect_capability(qualified_name=qualified_name),
     )
     emit_json(payload)

src/wf_cli/commands/deployments.py

@@ -1,6 +1,4 @@
 from __future__ import annotations
-
-import asyncio
 from pathlib import Path
 from typing import Annotated
 
@@ -9,6 +7,7 @@ import typer
 from wf_cli.context import load_cli_context_from_typer as load_cli_context
 from wf_cli.formats import ListOutputFormat, emit_list_payload
 from wf_cli.io import CliInputError, emit_json, parse_bindings, parse_json_input
+from wf_cli.remote_errors import run_cli_operation
 
 app = typer.Typer(
     name="deploy",
@@ -31,11 +30,12 @@ def validate_deployment(
 ) -> None:
     """Validate one saved workflow deployment."""
     context = load_cli_context(ctx)
-    payload = asyncio.run(
+    payload = run_cli_operation(
+        context,
         context.handlers.validate_deployment(
             deployment_id=deployment_id,
             live_check=live,
-        )
+        ),
     )
     emit_json(payload)
 
@@ -49,7 +49,7 @@ def list_deployments(
 ) -> None:
     """List saved workflow deployments."""
     context = load_cli_context(ctx)
-    payload = asyncio.run(context.handlers.list_deployments())
+    payload = run_cli_operation(context, context.handlers.list_deployments())
     emit_list_payload(
         payload,
         collection_key="deployments",
@@ -67,7 +67,10 @@ def inspect_deployment(
     """Inspect one saved deployment."""
     context = load_cli_context(ctx)
     emit_json(
-        asyncio.run(context.handlers.inspect_deployment(deployment_id=deployment_id))
+        run_cli_operation(
+            context,
+            context.handlers.inspect_deployment(deployment_id=deployment_id),
+        )
     )
 
 
@@ -107,7 +110,7 @@ def save_deployment(
     except CliInputError as exc:
         raise typer.BadParameter(str(exc)) from exc
     context = load_cli_context(ctx)
-    emit_json(asyncio.run(context.handlers.save_deployment(payload)))
+    emit_json(run_cli_operation(context, context.handlers.save_deployment(payload)))
 
 
 @app.command("delete")
@@ -118,7 +121,10 @@ def delete_deployment(
     """Delete one saved deployment."""
     context = load_cli_context(ctx)
     emit_json(
-        asyncio.run(context.handlers.delete_deployment(deployment_id=deployment_id))
+        run_cli_operation(
+            context,
+            context.handlers.delete_deployment(deployment_id=deployment_id),
+        )
     )
 
 

src/wf_cli/commands/drafts.py

@@ -1,6 +1,4 @@
 from __future__ import annotations
-
-import asyncio
 from pathlib import Path
 from typing import Annotated, Literal
 
@@ -9,6 +7,7 @@ import typer
 from wf_cli.context import load_cli_context_from_typer as load_cli_context
 from wf_cli.formats import ListOutputFormat, emit_list_payload
 from wf_cli.io import CliInputError, emit_json, parse_bindings, parse_json_value
+from wf_cli.remote_errors import run_cli_operation
 
 app = typer.Typer(
     name="draft",
@@ -26,7 +25,7 @@ def list_drafts(
 ) -> None:
     """List stored draft workspaces."""
     context = load_cli_context(ctx)
-    payload = asyncio.run(context.handlers.list_draft_workspaces())
+    payload = run_cli_operation(context, context.handlers.list_draft_workspaces())
     emit_list_payload(
         payload,
         collection_key="workspaces",
@@ -47,11 +46,12 @@ def inspect_draft(
     """Inspect one draft workspace."""
     context = load_cli_context(ctx)
     emit_json(
-        asyncio.run(
+        run_cli_operation(
+            context,
             context.handlers.get_draft_workspace(
                 workspace_id=workspace_id,
                 include_draft=include_draft,
-            )
+            ),
         )
     )
 
@@ -71,13 +71,14 @@ def create_from_capability(
     """Bootstrap a draft workspace from inspect_capability wrapper hints."""
     context = load_cli_context(ctx)
     emit_json(
-        asyncio.run(
+        run_cli_operation(
+            context,
             context.handlers.create_draft_workspace_from_capability(
                 workspace_id=workspace_id,
                 capability_name=capability_name,
                 name=name,
                 title=title,
-            )
+            ),
         )
     )
 
@@ -105,12 +106,13 @@ def patch_draft(
         raise typer.BadParameter("draft patch input must be a JSON array")
     context = load_cli_context(ctx)
     emit_json(
-        asyncio.run(
+        run_cli_operation(
+            context,
             context.handlers.patch_draft_workspace(
                 workspace_id=workspace_id,
                 revision=revision,
                 patch=patch,
-            )
+            ),
         )
     )
 
@@ -123,8 +125,9 @@ def validate_draft(
     """Validate one stored draft workspace."""
     context = load_cli_context(ctx)
     emit_json(
-        asyncio.run(
+        run_cli_operation(
-            context.handlers.validate_draft_workspace(workspace_id=workspace_id)
+            context,
+            context.handlers.validate_draft_workspace(workspace_id=workspace_id),
         )
     )
 
@@ -158,7 +161,8 @@ def save_draft(
         raise typer.BadParameter(str(exc)) from exc
     context = load_cli_context(ctx)
     if kind == "wrapper":
-        payload = asyncio.run(
+        payload = run_cli_operation(
+            context,
             context.handlers.create_wrapper_from_workspace(
                 workspace_id=workspace_id,
                 artifact_id=artifact_id,
@@ -167,10 +171,11 @@ def save_draft(
                 outcomes=tuple(outcome or ["ok"]),
                 description=description,
                 source_bindings=source_bindings or None,
-            )
+            ),
         )
     else:
-        payload = asyncio.run(
+        payload = run_cli_operation(
+            context,
             context.handlers.create_artifact_from_workspace(
                 workspace_id=workspace_id,
                 artifact_id=artifact_id,
@@ -180,6 +185,6 @@ def save_draft(
                 kind=kind,
                 description=description,
                 source_bindings=source_bindings or None,
-            )
+            ),
         )
     emit_json(payload)

src/wf_cli/commands/runs.py

@@ -1,6 +1,4 @@
 from __future__ import annotations
-
-import asyncio
 from pathlib import Path
 from typing import Annotated
 
@@ -8,6 +6,7 @@ import typer
 
 from wf_cli.context import load_cli_context_from_typer
 from wf_cli.io import CliInputError, emit_json, parse_json_input
+from wf_cli.remote_errors import run_cli_operation
 from wf_api import TraceRange
 
 app = typer.Typer(
@@ -47,12 +46,13 @@ def start_run(
         raise typer.BadParameter(str(exc)) from exc
     context = load_cli_context_from_typer(ctx)
     trace_range = _optional_trace_range(start=trace_from, limit=trace_limit)
-    payload = asyncio.run(
+    payload = run_cli_operation(
+        context,
         context.handlers.run_deployment(
             deployment_id=deployment_id,
             workflow_input=workflow_input,
             trace_range=trace_range,
-        )
+        ),
     )
     emit_json(payload)
 
@@ -64,7 +64,7 @@ def inspect_run(
 ) -> None:
     """Inspect a durable run without trace entries."""
     context = load_cli_context_from_typer(ctx)
-    emit_json(asyncio.run(context.handlers.inspect_run(run_id=run_id)))
+    emit_json(run_cli_operation(context, context.handlers.inspect_run(run_id=run_id)))
 
 
 @app.command("resume")
@@ -108,13 +108,14 @@ def resume_run(
         raise typer.BadParameter(str(exc)) from exc
     context = load_cli_context_from_typer(ctx)
     trace_range = _optional_trace_range(start=trace_from, limit=trace_limit)
-    payload = asyncio.run(
+    payload = run_cli_operation(
+        context,
         context.handlers.resume_run(
             run_id=run_id,
             resume_payload=resume_payload,
             resume_outcome=outcome,
             trace_range=trace_range,
-        )
+        ),
     )
     emit_json(payload)
 
@@ -134,11 +135,12 @@ def trace_run(
 ) -> None:
     """Read a bounded debug trace slice."""
     context = load_cli_context_from_typer(ctx)
-    payload = asyncio.run(
+    payload = run_cli_operation(
+        context,
         context.handlers.read_run_trace(
             run_id=run_id,
             trace_range=TraceRange(start=trace_from, limit=limit),
-        )
+        ),
     )
     emit_json(payload)
 

src/wf_cli/commands/source_registry.py

@@ -1,6 +1,4 @@
 from __future__ import annotations
-
-import asyncio
 import json
 from pathlib import Path
 from typing import Annotated, Any
@@ -10,6 +8,7 @@ import typer
 from wf_cli.context import CliContext, load_cli_context_from_typer
 from wf_cli.formats import ListOutputFormat, emit_list_payload
 from wf_cli.io import emit_json
+from wf_cli.remote_errors import run_cli_operation
 
 app = typer.Typer(
     name="registry",
@@ -43,7 +42,10 @@ def list_registry_entries(
     """List desired persisted source registry entries."""
     context = load_cli_context_from_typer(ctx)
     admin = _require_registry_admin(context)
-    payload = asyncio.run(admin.list_registry_entries(cursor=cursor, limit=limit))
+    payload = run_cli_operation(
+        context,
+        admin.list_registry_entries(cursor=cursor, limit=limit),
+    )
     emit_list_payload(
         payload,
         collection_key="entries",
@@ -61,7 +63,10 @@ def inspect_registry_entry(
     """Inspect one desired persisted source registry entry."""
     context = load_cli_context_from_typer(ctx)
     admin = _require_registry_admin(context)
-    payload = asyncio.run(admin.inspect_registry_entry(source_id=source_id))
+    payload = run_cli_operation(
+        context,
+        admin.inspect_registry_entry(source_id=source_id),
+    )
     emit_json(payload)
 
 
@@ -80,7 +85,7 @@ def add_registry_entry(
     context = load_cli_context_from_typer(ctx)
     admin = _require_registry_admin(context)
     entry = _read_json_arg(input_json, input_file, "--input/--input-file")
-    payload = asyncio.run(admin.add_registry_entry(entry=entry))
+    payload = run_cli_operation(context, admin.add_registry_entry(entry=entry))
     emit_json(payload)
 
 
@@ -99,7 +104,10 @@ def update_registry_entry(
     context = load_cli_context_from_typer(ctx)
     admin = _require_registry_admin(context)
     patch = _read_json_arg(patch_json, patch_file, "--patch/--patch-file")
-    payload = asyncio.run(admin.update_registry_entry(source_id=source_id, patch=patch))
+    payload = run_cli_operation(
+        context,
+        admin.update_registry_entry(source_id=source_id, patch=patch),
+    )
     emit_json(payload)
 
 
@@ -111,7 +119,10 @@ def enable_registry_entry(
     """Enable a desired source registry entry."""
     context = load_cli_context_from_typer(ctx)
     admin = _require_registry_admin(context)
-    payload = asyncio.run(admin.enable_registry_entry(source_id=source_id))
+    payload = run_cli_operation(
+        context,
+        admin.enable_registry_entry(source_id=source_id),
+    )
     emit_json(payload)
 
 
@@ -123,7 +134,10 @@ def disable_registry_entry(
     """Disable a desired source registry entry."""
     context = load_cli_context_from_typer(ctx)
     admin = _require_registry_admin(context)
-    payload = asyncio.run(admin.disable_registry_entry(source_id=source_id))
+    payload = run_cli_operation(
+        context,
+        admin.disable_registry_entry(source_id=source_id),
+    )
     emit_json(payload)
 
 
@@ -140,7 +154,10 @@ def remove_registry_entry(
     admin = _require_registry_admin(context)
     if not confirm:
         raise typer.BadParameter("removal requires --confirm flag")
-    payload = asyncio.run(admin.remove_registry_entry(source_id=source_id))
+    payload = run_cli_operation(
+        context,
+        admin.remove_registry_entry(source_id=source_id),
+    )
     emit_json(payload)
 
 
@@ -149,7 +166,7 @@ def apply_registry_changes(ctx: typer.Context) -> None:
     """Apply desired registry state to the running server."""
     context = load_cli_context_from_typer(ctx)
     admin = _require_registry_admin(context)
-    payload = asyncio.run(admin.apply_registry_changes())
+    payload = run_cli_operation(context, admin.apply_registry_changes())
     emit_json(payload)
 
 

src/wf_cli/commands/sources.py

@@ -1,6 +1,4 @@
 from __future__ import annotations
-
-import asyncio
 from typing import Annotated
 
 import typer
@@ -8,6 +6,7 @@ import typer
 from wf_cli.context import load_cli_context_from_typer
 from wf_cli.formats import ListOutputFormat, emit_list_payload
 from wf_cli.io import emit_json
+from wf_cli.remote_errors import run_cli_operation
 
 app = typer.Typer(
     name="source",
@@ -31,7 +30,10 @@ def list_sources(
 ) -> None:
     """List compact workflow source summaries."""
     context = load_cli_context_from_typer(ctx)
-    payload = asyncio.run(context.source_admin.list_sources(cursor=cursor, limit=limit))
+    payload = run_cli_operation(
+        context,
+        context.source_admin.list_sources(cursor=cursor, limit=limit),
+    )
     emit_list_payload(
         payload,
         collection_key="sources",
@@ -48,5 +50,8 @@ def inspect_source(
 ) -> None:
     """Inspect one workflow source inventory."""
     context = load_cli_context_from_typer(ctx)
-    payload = asyncio.run(context.source_admin.inspect_source(source_id=source_id))
+    payload = run_cli_operation(
+        context,
+        context.source_admin.inspect_source(source_id=source_id),
+    )
     emit_json(payload)

src/wf_cli/context.py

@@ -40,6 +40,7 @@ class CliContext:
     source_admin: WorkflowSourceAdminSurface
     admin: WorkflowAdminSurface
     source_registry_admin: WorkflowSourceRegistrySurface | None = None
+    verbose: bool = False
 
 
 @dataclass(frozen=True)
@@ -63,6 +64,7 @@ class CliTyperState:
     force_local: bool = False
     rpc_url: str | None = None
     rpc_timeout_seconds: float | None = None
+    verbose: bool = False
 
     @classmethod
     def from_context(cls, ctx: typer.Context) -> CliTyperState:
@@ -87,6 +89,7 @@ class CliTyperState:
             rpc_timeout_seconds=(
                 float(timeout) if isinstance(timeout, float | int) else None
             ),
+            verbose=bool(obj.get("verbose", cls.verbose)),
         )
 
 
@@ -101,6 +104,7 @@ def load_cli_context(
     force_local: bool = False,
     rpc_url: str | None = None,
     rpc_timeout_seconds: float | None = None,
+    verbose: bool = False,
 ) -> CliContext:
     """Load config and build workflow-surface handlers for CLI commands."""
     resolved_config_path = Path(config_path)
@@ -123,6 +127,7 @@ def load_cli_context(
             source_admin=client,
             admin=client,
             source_registry_admin=client,
+            verbose=verbose,
         )
 
     if _is_legacy_mcp_config(resolved_config_path):
@@ -137,6 +142,7 @@ def load_cli_context(
                 connections=service.connection_service,
                 events=service.events,
             ),
+            verbose=verbose,
         )
 
     config = load_workflow_config(resolved_config_path)
@@ -152,6 +158,7 @@ def load_cli_context(
             handlers=server.api,
             source_admin=server.source_admin,
             admin=server.admin,
+            verbose=verbose,
         )
     if isinstance(target, RpcHttpTargetConfig):
         client = RpcWorkflowApiClient(
@@ -169,6 +176,7 @@ def load_cli_context(
             source_admin=client,
             admin=client,
             source_registry_admin=client,
+            verbose=verbose,
         )
     raise ValueError(f"unsupported workflow target {target!r}")
 
@@ -218,6 +226,7 @@ def load_cli_context_from_typer(ctx: typer.Context) -> CliContext:
             force_local=force_local_from_context(ctx),
             rpc_url=rpc_url_from_context(ctx),
             rpc_timeout_seconds=rpc_timeout_from_context(ctx),
+            verbose=CliTyperState.from_context(ctx).verbose,
         )
     except ValueError as exc:
         raise typer.BadParameter(str(exc)) from exc

src/wf_cli/remote_errors.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Coroutine
+from typing import Any, TypeVar
+
+import click
+import httpx
+
+from wf_cli.context import CliContext
+
+T = TypeVar("T")
+
+
+def run_cli_operation(context: CliContext, operation: Coroutine[Any, Any, T]) -> T:
+    """Run a CLI async operation and format non-verbose operation errors.
+
+    Non-verbose CLI output should be useful to users, not a Python crash report.
+    `--verbose` preserves the raw exception path so developers can still debug
+    internal failures with a traceback.
+    """
+
+    try:
+        return asyncio.run(operation)
+    except (RuntimeError, httpx.HTTPError) as exc:
+        if context.verbose:
+            raise
+        raise click.ClickException(_operation_error_message(exc)) from exc
+
+
+def _operation_error_message(exc: RuntimeError | httpx.HTTPError) -> str:
+    """Return a stable non-empty message for compact CLI error output."""
+    message = str(exc)
+    if message:
+        return message
+    return exc.__class__.__name__

tests/wf_cli/test_remote_target.py

@@ -2,18 +2,33 @@ from __future__ import annotations
 
 import asyncio
 import json
+from pathlib import Path
+from typing import Any, cast
 
 import httpx
 from typer.testing import CliRunner
 
 from wf_api.models import RawWorkflowPlan
 from wf_cli.app import app
-from wf_cli.context import load_cli_context, load_local_cli_context
+from wf_cli.context import CliContext, load_cli_context, load_local_cli_context
 from wf_core import END
 from wf_server import build_local_static_workflow_server
 from wf_transport_rpc_http import RpcWorkflowApiClient, create_rpc_app
 
 
+class BrokenSourceAdmin:
+    async def list_sources(
+        self,
+        *,
+        cursor: str | None = None,
+        limit: int = 50,
+    ) -> dict[str, Any]:
+        return {"sources": [], "next_cursor": None, "total": 0}
+
+    async def inspect_source(self, *, source_id: str) -> dict[str, Any]:
+        raise RuntimeError(f"broken source admin for {source_id}")
+
+
 def test_load_cli_context_uses_rpc_client_for_rpc_http_target(tmp_path) -> None:
     config_path = tmp_path / "wf.json"
     config_path.write_text(
@@ -332,6 +347,115 @@ def test_wf_source_commands_use_rpc_url_override(monkeypatch, tmp_path) -> None:
     assert '"id": "wf.std"' in inspected.output
 
 
+def test_wf_remote_source_inspect_formats_expected_rpc_error(
+    monkeypatch,
+    tmp_path,
+) -> None:
+    server = build_local_static_workflow_server(tmp_path / "store")
+    original_client = httpx.AsyncClient
+    monkeypatch.setattr(
+        "wf_transport_rpc_http.client.httpx.AsyncClient",
+        lambda *args, **kwargs: original_client(
+            transport=httpx.ASGITransport(app=create_rpc_app(server)),
+            base_url="http://test",
+        ),
+    )
+    config_path = tmp_path / "wf.json"
+    config_path.write_text('{"version": 1}', encoding="utf-8")
+    runner = CliRunner()
+    base_args = ["--config", str(config_path), "--url", "http://test/rpc"]
+
+    result = runner.invoke(app, [*base_args, "source", "inspect", "missing.source"])
+
+    assert result.exit_code != 0
+    assert "Error" in result.output
+    assert "Workflow operation failed" in result.output
+    assert "missing.source" in result.output
+    assert "Traceback" not in result.output
+    assert "RuntimeError" not in result.output
+
+
+def test_wf_remote_source_list_formats_transport_error(monkeypatch, tmp_path) -> None:
+    async def connection_failed(*args: Any, **kwargs: Any) -> dict[str, Any]:
+        raise httpx.ConnectError(
+            "connection refused",
+            request=httpx.Request("POST", "http://test/rpc"),
+        )
+
+    monkeypatch.setattr(
+        "wf_transport_rpc_http.client_sources.RpcSourceAdminClientMixin.list_sources",
+        connection_failed,
+    )
+    config_path = tmp_path / "wf.json"
+    config_path.write_text('{"version": 1}', encoding="utf-8")
+    runner = CliRunner()
+
+    result = runner.invoke(
+        app,
+        [
+            "--config",
+            str(config_path),
+            "--url",
+            "http://test/rpc",
+            "source",
+            "list",
+        ],
+    )
+
+    assert result.exit_code != 0
+    assert "Error" in result.output
+    assert "connection refused" in result.output
+    assert "Traceback" not in result.output
+    assert "ConnectError" not in result.output
+
+
+def test_wf_unexpected_error_uses_short_traceback_by_default(
+    monkeypatch,
+    tmp_path,
+) -> None:
+    fake_context = CliContext(
+        config_path=Path("dummy"),
+        service=cast(Any, object()),
+        handlers=build_local_static_workflow_server(tmp_path / "store").api,
+        source_admin=BrokenSourceAdmin(),
+        admin=cast(Any, object()),
+    )
+    monkeypatch.setattr(
+        "wf_cli.commands.sources.load_cli_context_from_typer",
+        lambda _ctx: fake_context,
+    )
+
+    result = CliRunner().invoke(app, ["source", "inspect", "wf.std"])
+
+    assert result.exit_code != 0
+    assert "broken source admin for wf.std" in result.output
+    assert "tests/wf_cli/test_remote_target.py" not in result.output
+
+
+def test_wf_verbose_shows_full_traceback_for_unexpected_error(
+    monkeypatch,
+    tmp_path,
+) -> None:
+    fake_context = CliContext(
+        config_path=Path("dummy"),
+        service=cast(Any, object()),
+        handlers=build_local_static_workflow_server(tmp_path / "store").api,
+        source_admin=BrokenSourceAdmin(),
+        admin=cast(Any, object()),
+        verbose=True,
+    )
+    monkeypatch.setattr(
+        "wf_cli.commands.sources.load_cli_context_from_typer",
+        lambda _ctx: fake_context,
+    )
+
+    result = CliRunner().invoke(app, ["--verbose", "source", "inspect", "wf.std"])
+
+    assert result.exit_code != 0
+    assert isinstance(result.exception, RuntimeError)
+    assert str(result.exception) == "broken source admin for wf.std"
+
+
 def test_wf_admin_commands_use_rpc_url_override(monkeypatch, tmp_path) -> None:
     server = build_local_static_workflow_server(tmp_path / "store")
     server.events.record_workflow_event(

tests/wf_cli/test_source_registry.py

@@ -47,8 +47,8 @@ def _patch_load_cli_context(
 
 def _patch_asyncio_run(monkeypatch: pytest.MonkeyPatch) -> None:
     monkeypatch.setattr(
-        "wf_cli.commands.source_registry.asyncio.run",
+        "wf_cli.commands.source_registry.run_cli_operation",
-        lambda coro: coro,
+        lambda _context, operation: operation,
     )