list deployment compaction & inspect deployment

docs/current_roadmap.md

@@ -1,46 +1,62 @@
 # Current Roadmap
 
-This is the short active roadmap after the core type-shape cleanup. It is based
-on both the current docs and the implementation state.
+This is the short active roadmap after the core type-shape cleanup and MCP
+workflow authoring cleanup pass. It is based on both the current docs and the
+implementation state.
 
-## Next Work
+## Completed Cleanup Pass
 
 1. **Docs index and prune**
-   - Add/maintain a clear docs entry point.
-   - Keep current architecture docs separate from historical plans and scratch
+   - Current architecture docs are separated from historical plans and scratch
      notes.
+   - The active roadmap now lives here instead of being scattered through older
+     planning files.
 
 2. **MCP workflow authoring UX**
-   - Make the LLM/client path progressive: inspect sources, create a draft,
-     patch, validate, compile, save, run.
-   - Prefer smaller discovery/inspection responses over one huge payload.
-   - Done: the operator manual now categorizes workflow tools into discovery,
-     draft workspace, stateless draft, artifact/deployment, and raw escape
+   - The operator manual categorizes workflow tools into discovery, draft
+     workspace, stateless draft, artifact/deployment, run/debug, and raw escape
      hatch groups.
-   - Next: tighten the actual tool responses around that map so list calls stay
-     compact and inspect/run calls carry the detailed payloads.
+   - List-style tools are more compact, while inspect/run tools carry the
+     detailed payloads.
 
 3. **Wrapper creation ergonomics**
-   - Help create workflow-ready wrappers from raw capabilities.
-   - Suggest state schema, input bindings, output bindings, default `ok` /
-     `error` handling, and missing decisions.
+   - Wrapper draft helpers can suggest state schema, input bindings, output
+     bindings, default `ok` / `error` handling, and missing decisions.
+   - The end-to-end runbook documents the wrapper path from capability
+     discovery through deployment/run.
 
 4. **Run and deployment story**
-   - Tighten list/inspect/run/debug for artifacts and deployments.
-   - Keep dependency validation and trace/error output compact and actionable.
+   - Deployment listing is summary-first, with dedicated inspection for detail.
+   - `run_deployment` returns compact status by default and exposes trace slices
+     through an explicit `trace_range`.
+   - Dependency validation and error output remain part of the run path.
 
 5. **Source inventory polish**
-   - Make `list_sources` / `inspect_source` clearly show raw capabilities,
-     workflow-ready node specs, admin-only tools, docs/resources, enabled state,
-     and changes after reload.
+   - `list_sources` / `inspect_source` now present source-owned capabilities
+     progressively.
+   - Source inventory distinguishes external sources, local workflow-facing
+     sources, docs/resources, and admin-only control surfaces.
 
-## Runtime Work To Revisit Later
+## Runtime and Platform Roadmap
 
-- **Native subgraphs**: add child run state, child trace preservation, interrupt
-  bubbling, and resume back into the child workflow.
+- **Native subgraphs / graph-as-node**: add child run state, child trace
+  preservation, interrupt bubbling, and resume back into the child workflow.
+  Wrapper artifacts currently execute as deployments and return run status;
+  true graph-as-node outcome propagation belongs here.
 - **Async parallel foreach**: add explicit scheduling, reducer/merge semantics,
   and failure policy. Do not model this as plain parallel calls over sync
   handlers.
+- **Persistent run history**: add a run store before adding stable `run_id`,
+  `inspect_run`, or `read_run_trace(run_id, range)` APIs. Current traces are
+  returned directly from immediate run responses.
+- **Protocol-native long-running runs**: investigate MCP tasks/progress
+  notifications for long-running workflow execution. Avoid inventing a custom
+  "start" convention unless protocol-native behavior is insufficient.
+- **Dynamic saved workflows as tools**: defer until the stable run/inspect
+  surface is strong. Many MCP clients do not refresh tool lists reliably, so
+  `wf.workflow.run_deployment` remains the dependable front door.
+- **Dashboard/source controls**: future UI should consume the same source
+  inventory and deployment metadata instead of reverse-engineering MCP tools.
 
 Frame stress points to solve before either feature:
 
@@ -57,6 +73,8 @@ Frame stress points to solve before either feature:
 
 ## Why This Order
 
-`wf_core` is now coherent enough for the next bottleneck to be platform and DX:
-how a human or LLM discovers capabilities, turns them into workflow-ready
-pieces, saves them, and runs them again.
+The MCP workflow authoring path is now usable enough for real testing. The next
+bottleneck is runtime/platform correctness: resumable child execution,
+parallel scheduling, persistent run history, and protocol-native progress
+reporting. Those pieces should come before adding more high-level authoring
+sugar.

docs/wf_mcp_operator_manual.md

@@ -118,6 +118,9 @@ Supporting:
 - `wf.workflow.list_artifacts`: compact list of saved workflow and wrapper
   artifacts.
 - `wf.workflow.inspect_artifact`: full saved artifact payload.
+- `wf.workflow.list_deployments`: compact list of saved deployment summaries.
+- `wf.workflow.inspect_deployment`: full deployment payload including source
+  bindings.
 
 ### Draft Workspaces
 
@@ -173,6 +176,8 @@ Primary:
 
 - `wf.workflow.save_deployment`: bind one saved artifact version to concrete
   sources.
+- `wf.workflow.inspect_deployment`: inspect source bindings for one saved
+  deployment.
 - `wf.workflow.validate_deployment`: check dependency availability and drift.
 - `wf.workflow.run_deployment`: execute a saved deployment with input. The
   default response is compact and returns `trace_count`; pass `trace_range`
@@ -372,6 +377,9 @@ data.
 | Save a workflow definition from a draft | `wf.workflow.create_artifact_from_draft` |
 | Save a compiled raw workflow definition | `wf.workflow.create_artifact_from_plan` |
 | List saved workflows/wrappers | `wf.workflow.list_artifacts` |
+| Inspect one saved workflow/wrapper | `wf.workflow.inspect_artifact` |
+| List saved deployments | `wf.workflow.list_deployments` |
+| Inspect one saved deployment | `wf.workflow.inspect_deployment` |
 | Bind a saved workflow to concrete sources | `wf.workflow.save_deployment` |
 | Check whether a deployment can run | `wf.workflow.validate_deployment` |
 | Execute a saved workflow | `wf.workflow.run_deployment` |

docs/workflow_artifacts.md

@@ -69,6 +69,21 @@ semantics.
 Dynamic projection of saved workflows as individual MCP tools can exist later,
 but it should be optional. The stable run tool is the reliable base layer.
 
+Current `run_deployment` calls are synchronous request/response executions. They
+return compact status, output, diagnostics, and `trace_count`; optional ranged
+trace detail is for debugging only.
+
+Future run history should introduce a stable `run_id` only when there is a real
+run store behind it. A `run_id` without persisted state, trace paging, and
+status lookup would be misleading. The likely shape is:
+
+- `run_deployment` starts or completes a run and returns `run_id`
+- `inspect_run(run_id)` returns status, output, diagnostics, and trace metadata
+- `read_run_trace(run_id, range)` returns bounded trace slices
+
+Until that exists, clients should treat the current response as the complete
+ephemeral run result for this request.
+
 For long-running workflow execution, prefer MCP-native execution mechanisms
 where available:
 

src/wf_mcp/admin_surface/tools.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
-from typing import Any
+from typing import Annotated, Any
+
+from pydantic import Field
 
 from wf_mcp.broker.service import WfMcpService
 
@@ -84,8 +86,26 @@ def register_service_admin_tools(
         description="List compact configured capability source summaries.",
     )
     async def list_sources(
-        cursor: str | None = None,
-        limit: int = 50,
+        cursor: Annotated[
+            str | None,
+            Field(
+                description=(
+                    "Opaque pagination cursor returned by a previous list_sources "
+                    "call. Omit for the first page."
+                )
+            ),
+        ] = None,
+        limit: Annotated[
+            int,
+            Field(
+                ge=1,
+                le=100,
+                description=(
+                    "Maximum source summaries to return. Use inspect_source for "
+                    "one full source inventory."
+                ),
+            ),
+        ] = 50,
     ) -> dict[str, Any]:
         return handlers.list_sources(cursor=cursor, limit=limit)
 
@@ -94,7 +114,17 @@ def register_service_admin_tools(
         title="Inspect Source",
         description="Return the full inventory for one configured capability source.",
     )
-    async def inspect_source(source_id: str) -> dict[str, Any]:
+    async def inspect_source(
+        source_id: Annotated[
+            str,
+            Field(
+                description=(
+                    "Exact source id from list_sources, such as wf.std, wf.docs, "
+                    "or an enabled connection id like demo.personal."
+                )
+            ),
+        ],
+    ) -> dict[str, Any]:
         return handlers.inspect_source(source_id)
 
     @server.tool(

src/wf_mcp/proxy/runtime.py

@@ -55,6 +55,7 @@ _SEARCH_ALWAYS_VISIBLE_TOOL_NAMES = [
     "wf.workflow.create_wrapper_from_workspace",
     "wf.workflow.inspect_artifact",
     "wf.workflow.list_deployments",
+    "wf.workflow.inspect_deployment",
     "wf.workflow.save_deployment",
     "wf.workflow.validate_deployment",
     "wf.workflow.run_deployment",

src/wf_mcp/workflow_surface/handlers.py

@@ -819,11 +819,18 @@ class WorkflowSurfaceHandlers:
             return {"deployments": []}
         return {
             "deployments": [
-                deployment.model_dump(mode="json")
+                _deployment_summary(deployment)
                 for deployment in self.service.artifact_store.list_deployments()
             ]
         }
 
+    async def inspect_deployment(self, *, deployment_id: str) -> dict[str, Any]:
+        if self.service.artifact_store is None:
+            raise KeyError("workflow artifact store is not configured")
+        return self.service.artifact_store.get_deployment(deployment_id).model_dump(
+            mode="json"
+        )
+
     async def save_deployment(self, deployment: dict[str, Any]) -> dict[str, Any]:
         if self.service.artifact_store is None:
             raise KeyError("workflow artifact store is not configured")
@@ -1262,3 +1269,14 @@ def _run_payload(
         payload["trace"] = trace
         payload["trace_truncated"] = trace_truncated
     return payload
+
+
+def _deployment_summary(deployment: WorkflowDeployment) -> dict[str, Any]:
+    """Return compact deployment metadata for progressive list responses."""
+    return {
+        "id": deployment.id,
+        "artifact_id": deployment.artifact_id,
+        "artifact_version": deployment.artifact_version,
+        "binding_count": len(deployment.binding_map()),
+        "drift_policy": deployment.drift_policy.value,
+    }

src/wf_mcp/workflow_surface/tools.py

@@ -514,11 +514,22 @@ def register_workflow_tools(server: FastMCP[Any], service: WfMcpService) -> None
     @server.tool(
         name="wf.workflow.list_deployments",
         title="List Workflow Deployments",
-        description="List saved workflow deployments and their source bindings.",
+        description=(
+            "List compact saved workflow deployment summaries. Use "
+            "inspect_deployment for full source bindings."
+        ),
     )
     async def list_deployments() -> dict[str, Any]:
         return await handlers.list_deployments()
 
+    @server.tool(
+        name="wf.workflow.inspect_deployment",
+        title="Inspect Workflow Deployment",
+        description="Return one full workflow deployment including source bindings.",
+    )
+    async def inspect_deployment(deployment_id: str) -> dict[str, Any]:
+        return await handlers.inspect_deployment(deployment_id=deployment_id)
+
     @server.tool(
         name="wf.workflow.save_deployment",
         title="Save Workflow Deployment",

src/wf_platform/sources.py

@@ -10,6 +10,7 @@ from wf_platform.refs import CapabilityRef
 
 SourceKind = Literal["system", "connection"]
 JsonObject = dict[str, Any]
+SOURCE_PREVIEW_LIMIT = 3
 
 
 @dataclass(frozen=True, slots=True)
@@ -143,7 +144,6 @@ class CapabilitySource:
 
     def as_status(self) -> SourceStatus:
         """Return serializable source metadata without owned capability names."""
-        preview_limit = 3
         return SourceStatus(
             id=self.id,
             kind=self.kind,
@@ -166,21 +166,29 @@ class CapabilitySource:
             prompt_count=len(self.capabilities.prompts),
             resource_count=len(self.capabilities.resources),
             preview=SourceCapabilityPreview(
-                tools=_preview_names(self.capabilities.tools, preview_limit),
+                tools=_preview_names(self.capabilities.tools, SOURCE_PREVIEW_LIMIT),
                 node_specs=_preview_names(
                     self.capabilities.node_specs,
-                    preview_limit,
+                    SOURCE_PREVIEW_LIMIT,
+                ),
+                reducers=_preview_names(
+                    self.capabilities.reducers, SOURCE_PREVIEW_LIMIT
+                ),
+                prompts=_preview_names(
+                    self.capabilities.prompts, SOURCE_PREVIEW_LIMIT
+                ),
+                resources=_preview_names(
+                    self.capabilities.resources, SOURCE_PREVIEW_LIMIT
                 ),
-                reducers=_preview_names(self.capabilities.reducers, preview_limit),
-                prompts=_preview_names(self.capabilities.prompts, preview_limit),
-                resources=_preview_names(self.capabilities.resources, preview_limit),
             ),
             has_more=SourceCapabilityHasMore(
-                tools=_has_more(self.capabilities.tools, preview_limit),
-                node_specs=_has_more(self.capabilities.node_specs, preview_limit),
-                reducers=_has_more(self.capabilities.reducers, preview_limit),
-                prompts=_has_more(self.capabilities.prompts, preview_limit),
-                resources=_has_more(self.capabilities.resources, preview_limit),
+                tools=_has_more(self.capabilities.tools, SOURCE_PREVIEW_LIMIT),
+                node_specs=_has_more(
+                    self.capabilities.node_specs, SOURCE_PREVIEW_LIMIT
+                ),
+                reducers=_has_more(self.capabilities.reducers, SOURCE_PREVIEW_LIMIT),
+                prompts=_has_more(self.capabilities.prompts, SOURCE_PREVIEW_LIMIT),
+                resources=_has_more(self.capabilities.resources, SOURCE_PREVIEW_LIMIT),
             ),
         )
 

tests/wf_mcp/test_server.py

@@ -132,6 +132,14 @@ def test_server_exposes_upstream_admin_and_workflow_tools() -> None:
             assert create_workspace_schema is not None
             assert "workspace_id" in create_workspace_schema["properties"]
             assert "revision" in create_workspace_schema["properties"]
+            list_sources_schema = tools_by_name["wf.admin.list_sources"].inputSchema
+            assert "inspect_source" in list_sources_schema["properties"]["limit"][
+                "description"
+            ]
+            inspect_source_schema = tools_by_name["wf.admin.inspect_source"].inputSchema
+            assert "Exact source id" in inspect_source_schema["properties"][
+                "source_id"
+            ]["description"]
             minimal_workspace_input = tools_by_name[
                 "wf.workflow.create_minimal_draft_workspace"
             ].inputSchema
@@ -274,6 +282,7 @@ def test_server_search_mode_pins_stable_control_and_workflow_tools() -> None:
             assert "wf.workflow.create_wrapper_from_workspace" in names
             assert "wf.workflow.inspect_artifact" in names
             assert "wf.workflow.list_deployments" in names
+            assert "wf.workflow.inspect_deployment" in names
             assert "wf.workflow.save_deployment" in names
             assert "wf.workflow.validate_deployment" in names
             assert "wf.workflow.run_deployment" in names

tests/wf_mcp/test_workflow_surface.py

@@ -268,6 +268,33 @@ def test_workflow_surface_records_artifact_and_deployment_save_events() -> None:
     assert events[1].capability_id == "deployment.echo.personal"
 
 
+def test_workflow_surface_lists_compact_deployment_summaries_and_inspects_detail() -> (
+    None
+):
+    artifact_store = FileWorkflowArtifactStore(
+        local_temp_root() / "surface_deployment_list"
+    )
+    handlers = _handlers(artifact_store)
+    deployment = WorkflowDeployment(
+        id="echo.personal",
+        artifact_id="echo",
+        artifact_version=1,
+        bindings=[{"logical_source": "demo", "concrete_source": "demo.personal"}],
+    )
+    artifact_store.save_deployment(deployment)
+
+    listed = asyncio.run(handlers.list_deployments())
+    inspected = asyncio.run(
+        handlers.inspect_deployment(deployment_id="echo.personal")
+    )
+
+    assert listed["deployments"][0]["id"] == "echo.personal"
+    assert listed["deployments"][0]["binding_count"] == 1
+    assert "bindings" not in listed["deployments"][0]
+    assert inspected["bindings"][0]["logical_source"] == "demo"
+    assert inspected["bindings"][0]["concrete_source"] == "demo.personal"
+
+
 def test_workflow_surface_creates_wrapper_artifact_from_plan() -> None:
     artifact_store = FileWorkflowArtifactStore(
         local_temp_root() / "surface_wrapper_plan"