lda.chat/lda-wf
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

move more to wf_api: Next actions and Wrapper hints

Browse Source
This commit is contained in:
lda
2026-06-01 19:05:53 +07:00 Verified
parent db91824861
commit 232525b078
10 changed files with 1269 additions and 626 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/superpowers/plans/2026-06-01-wf-api-slice-3b-guidance-helpers.md
+541
View File
@@ -0,0 +1,541 @@
# wf_api Slice 3B: Guidance Helpers Move Implementation Plan
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Move protocol-neutral wrapper guidance helpers from `wf_mcp.workflow_surface` into `wf_api`, while preserving old imports as compatibility shims.
**Architecture:** `wrapper_hints.py` and `next_actions.py` are coupled guidance helpers: `next_actions` imports `WrapperAuthoringHints`, and both describe workflow authoring UX rather than MCP transport behavior. This slice moves them together to avoid a half-moved dependency. The old `wf_mcp.workflow_surface` modules remain thin re-export shims so MCP schemas/tests and older imports keep working.
**Tech Stack:** Python 3.14+, Pydantic v2 models, pytest, ruff, basedpyright.
---
## Scope
### In Scope
- Create `src/wf_api/wrapper_hints.py`.
- Create `src/wf_api/next_actions.py`.
- Re-export selected public guidance types/functions from `src/wf_api/__init__.py`.
- Replace `wf_mcp.workflow_surface.wrapper_hints` with a compatibility shim.
- Replace `wf_mcp.workflow_surface.next_actions` with a compatibility shim.
- Update `src/wf_mcp/workflow_surface/handlers.py` to import canonical helpers from `wf_api`.
- Update `src/wf_mcp/workflow_surface/models.py` to import canonical next-action models from `wf_api`.
- Update direct tests to use canonical imports while preserving shim compatibility tests.
- Keep `wf_api` free of `wf_mcp` imports.
### Out Of Scope
- Do not move `models.py`.
- Do not move `run_lifecycle.py`.
- Do not move `runtime_dependencies.py`.
- Do not move `saved_subgraphs.py`.
- Do not rename `WorkflowSurfaceHandlers`.
- Do not change wrapper hint behavior.
- Do not change next action behavior.
- Do not change public payloads, MCP tool names, CLI command names, or JSON schema field names.
---
## File Structure
### New Canonical Files
| File | Responsibility |
| --- | --- |
| `src/wf_api/wrapper_hints.py` | Wrapper scaffolding hints, confidence, missing-decision models, and conservative schema mapping helper. |
| `src/wf_api/next_actions.py` | Advisory next-action models and factory helpers for wrapper/deployment/run responses. |
### Compatibility Shims
| File | Responsibility |
| --- | --- |
| `src/wf_mcp/workflow_surface/wrapper_hints.py` | Re-export wrapper hint helpers from `wf_api.wrapper_hints`. |
| `src/wf_mcp/workflow_surface/next_actions.py` | Re-export next action helpers from `wf_api.next_actions`. |
### Modified Consumers
| File | Change |
| --- | --- |
| `src/wf_api/__init__.py` | Re-export public guidance helpers. |
| `src/wf_mcp/workflow_surface/handlers.py` | Import `NextActions` and wrapper hint helpers from `wf_api`. |
| `src/wf_mcp/workflow_surface/models.py` | Import `NextActionPatchExample` and `NextActions` from `wf_api.next_actions`. |
| `tests/wf_mcp/test_workflow_wrapper_hints.py` | Use canonical `wf_api.wrapper_hints`; add shim identity test. |
| `tests/wf_mcp/workflow_surface/test_next_actions.py` | Use canonical `wf_api.next_actions`; add shim identity test. |
---
## Task 1: Create Canonical `wf_api.wrapper_hints`
**Files:**
- Create: `src/wf_api/wrapper_hints.py`
- [ ] **Step 1: Copy existing implementation**
Create `src/wf_api/wrapper_hints.py` by copying the complete current contents of:
```text
src/wf_mcp/workflow_surface/wrapper_hints.py
```
Do not change behavior. The copied module must not import `wf_mcp`.
- [ ] **Step 2: Add `__all__` at the end**
Append this block to the copied file:
```python
__all__ = [
"MissingDecision",
"MissingDecisionKind",
"OutcomeCandidate",
"OutcomeCandidateKind",
"WrapperAuthoringHints",
"WrapperHintConfidence",
"WrapperOutcomePolicy",
"workflow_output_schema_for_authoring",
"wrapper_hints_for_capability",
]
```
- [ ] **Step 3: Run import smoke check**
Run:
```powershell
uv run python -c "from wf_api.wrapper_hints import WrapperAuthoringHints, wrapper_hints_for_capability; print(WrapperAuthoringHints.__name__, wrapper_hints_for_capability.__name__)"
```
Expected output:
```text
WrapperAuthoringHints wrapper_hints_for_capability
```
---
## Task 2: Create Canonical `wf_api.next_actions`
**Files:**
- Create: `src/wf_api/next_actions.py`
- [ ] **Step 1: Copy existing implementation**
Create `src/wf_api/next_actions.py` by copying the complete current contents of:
```text
src/wf_mcp/workflow_surface/next_actions.py
```
- [ ] **Step 2: Keep local wrapper hint import canonical**
Ensure the copied file imports wrapper hints from the new `wf_api` package via:
```python
from .wrapper_hints import WrapperAuthoringHints
```
The copied module must not import `wf_mcp`.
- [ ] **Step 3: Add `__all__` at the end**
Append this block to the copied file:
```python
__all__ = [
"NextActionPatchExample",
"NextActionTool",
"NextActions",
]
```
- [ ] **Step 4: Run import smoke check**
Run:
```powershell
uv run python -c "from wf_api.next_actions import NextActionTool, NextActions; print(NextActionTool.RUN_DEPLOYMENT, NextActions.__name__)"
```
Expected output:
```text
wf.workflow.run_deployment NextActions
```
---
## Task 3: Re-export Guidance Helpers From `wf_api`
**Files:**
- Modify: `src/wf_api/__init__.py`
- [ ] **Step 1: Add imports**
Add these imports:
```python
from .next_actions import NextActionPatchExample, NextActionTool, NextActions
from .wrapper_hints import (
MissingDecision,
MissingDecisionKind,
OutcomeCandidate,
OutcomeCandidateKind,
WrapperAuthoringHints,
WrapperHintConfidence,
WrapperOutcomePolicy,
workflow_output_schema_for_authoring,
wrapper_hints_for_capability,
)
```
- [ ] **Step 2: Add names to `__all__`**
Ensure `__all__` includes these names:
```python
"MissingDecision",
"MissingDecisionKind",
"NextActionPatchExample",
"NextActionTool",
"NextActions",
"OutcomeCandidate",
"OutcomeCandidateKind",
"WrapperAuthoringHints",
"WrapperHintConfidence",
"WrapperOutcomePolicy",
"workflow_output_schema_for_authoring",
"wrapper_hints_for_capability",
```
- [ ] **Step 3: Run top-level import smoke check**
Run:
```powershell
uv run python -c "from wf_api import NextActions, WrapperAuthoringHints; print(NextActions.__name__, WrapperAuthoringHints.__name__)"
```
Expected output:
```text
NextActions WrapperAuthoringHints
```
---
## Task 4: Convert Old Workflow-Surface Guidance Modules To Shims
**Files:**
- Modify: `src/wf_mcp/workflow_surface/wrapper_hints.py`
- Modify: `src/wf_mcp/workflow_surface/next_actions.py`
- [ ] **Step 1: Replace `src/wf_mcp/workflow_surface/wrapper_hints.py`**
Replace the file with this shim:
```python
"""Compatibility shim for workflow API wrapper authoring hints.
New code should import from `wf_api.wrapper_hints`. This module stays so older
MCP workflow-surface imports keep working during extraction.
"""
from wf_api.wrapper_hints import (
MissingDecision,
MissingDecisionKind,
OutcomeCandidate,
OutcomeCandidateKind,
WrapperAuthoringHints,
WrapperHintConfidence,
WrapperOutcomePolicy,
workflow_output_schema_for_authoring,
wrapper_hints_for_capability,
)
__all__ = [
"MissingDecision",
"MissingDecisionKind",
"OutcomeCandidate",
"OutcomeCandidateKind",
"WrapperAuthoringHints",
"WrapperHintConfidence",
"WrapperOutcomePolicy",
"workflow_output_schema_for_authoring",
"wrapper_hints_for_capability",
]
```
- [ ] **Step 2: Replace `src/wf_mcp/workflow_surface/next_actions.py`**
Replace the file with this shim:
```python
"""Compatibility shim for workflow API next-action guidance.
New code should import from `wf_api.next_actions`. This module stays so older
MCP workflow-surface imports keep working during extraction.
"""
from wf_api.next_actions import NextActionPatchExample, NextActionTool, NextActions
__all__ = [
"NextActionPatchExample",
"NextActionTool",
"NextActions",
]
```
- [ ] **Step 3: Run shim import smoke check**
Run:
```powershell
uv run python -c "from wf_mcp.workflow_surface.wrapper_hints import WrapperAuthoringHints; from wf_mcp.workflow_surface.next_actions import NextActions; print(WrapperAuthoringHints.__name__, NextActions.__name__)"
```
Expected output:
```text
WrapperAuthoringHints NextActions
```
---
## Task 5: Update Production Imports To Canonical Paths
**Files:**
- Modify: `src/wf_mcp/workflow_surface/handlers.py`
- Modify: `src/wf_mcp/workflow_surface/models.py`
- [ ] **Step 1: Update `handlers.py` next action import**
Replace:
```python
from .next_actions import NextActions
```
with:
```python
from wf_api.next_actions import NextActions
```
- [ ] **Step 2: Update `handlers.py` wrapper hint import**
Replace:
```python
from .wrapper_hints import (
workflow_output_schema_for_authoring,
wrapper_hints_for_capability,
)
```
with:
```python
from wf_api.wrapper_hints import (
workflow_output_schema_for_authoring,
wrapper_hints_for_capability,
)
```
- [ ] **Step 3: Update `models.py` next action import**
Replace:
```python
from .next_actions import NextActionPatchExample, NextActions
```
with:
```python
from wf_api.next_actions import NextActionPatchExample, NextActions
```
- [ ] **Step 4: Run production import smoke check**
Run:
```powershell
uv run python -c "from wf_mcp.workflow_surface.handlers import WorkflowSurfaceHandlers; from wf_mcp.workflow_surface.models import WrapperDraftNextActions; print(WorkflowSurfaceHandlers.__name__, WrapperDraftNextActions.__name__)"
```
Expected output:
```text
WorkflowSurfaceHandlers NextActions
```
---
## Task 6: Update Direct Tests And Add Shim Compatibility Tests
**Files:**
- Modify: `tests/wf_mcp/test_workflow_wrapper_hints.py`
- Modify: `tests/wf_mcp/workflow_surface/test_next_actions.py`
- [ ] **Step 1: Update wrapper hints test imports**
In `tests/wf_mcp/test_workflow_wrapper_hints.py`, replace imports from:
```python
from wf_mcp.workflow_surface.wrapper_hints import (
```
with:
```python
from wf_api.wrapper_hints import (
```
- [ ] **Step 2: Add wrapper hints shim identity test**
Append this test to `tests/wf_mcp/test_workflow_wrapper_hints.py`:
```python
def test_workflow_surface_wrapper_hints_shim_reexports_canonical_helper() -> None:
from wf_api.wrapper_hints import wrapper_hints_for_capability
from wf_mcp.workflow_surface.wrapper_hints import (
wrapper_hints_for_capability as wrapper_hints_for_capability_shim,
)
assert wrapper_hints_for_capability_shim is wrapper_hints_for_capability
```
- [ ] **Step 3: Update next actions test imports**
In `tests/wf_mcp/workflow_surface/test_next_actions.py`, replace:
```python
from wf_mcp.workflow_surface.next_actions import NextActionTool, NextActions
```
with:
```python
from wf_api.next_actions import NextActionTool, NextActions
```
- [ ] **Step 4: Add next actions shim identity test**
Append this test to `tests/wf_mcp/workflow_surface/test_next_actions.py`:
```python
def test_workflow_surface_next_actions_shim_reexports_canonical_model() -> None:
from wf_api.next_actions import NextActions
from wf_mcp.workflow_surface.next_actions import NextActions as NextActionsShim
assert NextActionsShim is NextActions
```
- [ ] **Step 5: Run focused tests**
Run:
```powershell
uv run pytest tests/wf_mcp/test_workflow_wrapper_hints.py tests/wf_mcp/workflow_surface/test_next_actions.py tests/wf_api/test_import_direction.py -q
```
Expected: all pass.
---
## Task 7: Search For Remaining Canonical Import Opportunities
**Files:**
- Inspect only unless the search finds new low-risk direct consumers.
- [ ] **Step 1: Search old imports**
Run:
```powershell
rg -n "from \\.next_actions|from \\.wrapper_hints|from wf_mcp\\.workflow_surface\\.(next_actions|wrapper_hints)" src tests
```
Expected remaining matches:
```text
src/wf_mcp/workflow_surface/next_actions.py
src/wf_mcp/workflow_surface/wrapper_hints.py
tests/wf_mcp/test_workflow_wrapper_hints.py
tests/wf_mcp/workflow_surface/test_next_actions.py
```
If any other production module imports the old paths, update it to import from
`wf_api.next_actions` or `wf_api.wrapper_hints`.
- [ ] **Step 2: Search for accidental `wf_api -> wf_mcp` imports**
Run:
```powershell
rg -n "from wf_mcp|import wf_mcp|wf_mcp\\." src/wf_api
```
Expected: no matches.
---
## Task 8: Verification
- [ ] **Step 1: Run focused tests**
```powershell
uv run pytest tests/wf_api tests/wf_mcp/test_workflow_wrapper_hints.py tests/wf_mcp/workflow_surface/test_next_actions.py tests/wf_mcp/workflow_surface tests/wf_mcp/server/test_config.py -q
```
Expected: all pass.
- [ ] **Step 2: Run CLI tests that assert next_actions/wrapper_hints payloads**
```powershell
uv run pytest tests/wf_cli/test_discovery_lifecycle.py tests/wf_cli/test_run_deploy.py -q
```
Expected: all pass.
- [ ] **Step 3: Run ruff on touched files**
```powershell
uv run ruff check src/wf_api src/wf_mcp/workflow_surface/next_actions.py src/wf_mcp/workflow_surface/wrapper_hints.py src/wf_mcp/workflow_surface/handlers.py src/wf_mcp/workflow_surface/models.py tests/wf_mcp/test_workflow_wrapper_hints.py tests/wf_mcp/workflow_surface/test_next_actions.py tests/wf_api
```
Expected: all checks pass.
- [ ] **Step 4: Run basedpyright on touched files**
```powershell
uv run basedpyright --level error src/wf_api src/wf_mcp/workflow_surface/next_actions.py src/wf_mcp/workflow_surface/wrapper_hints.py src/wf_mcp/workflow_surface/handlers.py src/wf_mcp/workflow_surface/models.py tests/wf_mcp/test_workflow_wrapper_hints.py tests/wf_mcp/workflow_surface/test_next_actions.py tests/wf_api
```
Expected: `0 errors`.
- [ ] **Step 5: Optional full suite**
Run this if time allows:
```powershell
uv run pytest -q
```
Expected: full suite passes with the project’s existing skipped/xfailed counts.
---
## Self-Review Checklist
- `wf_api.wrapper_hints` imports no `wf_mcp`.
- `wf_api.next_actions` imports no `wf_mcp`.
- Old `wf_mcp.workflow_surface.wrapper_hints` import path still works.
- Old `wf_mcp.workflow_surface.next_actions` import path still works.
- `WorkflowSurfaceHandlers` imports canonical guidance helpers from `wf_api`.
- `wf_mcp.workflow_surface.models` imports canonical next-action models from `wf_api`.
- Wrapper hint behavior is unchanged.
- Next action behavior is unchanged.
- No public payload shape changed.
- No other workflow-surface helper moved in this slice.
src/wf_api/__init__.py
+24
View File
@@ -8,18 +8,42 @@ from .constants import (
DEFAULT_OK_OUTCOME,
RUNTIME_ERROR_CAPABILITY,
)
from .next_actions import NextActionPatchExample, NextActionTool, NextActions
from .refs import WorkflowSurfaceCapabilityId, parse_workflow_surface_capability_id
from .service import WorkflowApi
from .wrapper_hints import (
MissingDecision,
MissingDecisionKind,
OutcomeCandidate,
OutcomeCandidateKind,
WrapperAuthoringHints,
WrapperHintConfidence,
WrapperOutcomePolicy,
workflow_output_schema_for_authoring,
wrapper_hints_for_capability,
)
__all__ = [
"DEFAULT_CALL_STEP_ID",
"DEFAULT_ERROR_OUTCOME",
"DEFAULT_ERROR_STEP_ID",
"DEFAULT_OK_OUTCOME",
"MissingDecision",
"MissingDecisionKind",
"NextActionPatchExample",
"NextActionTool",
"NextActions",
"OutcomeCandidate",
"OutcomeCandidateKind",
"RUNTIME_ERROR_CAPABILITY",
"TraceRange",
"WorkflowApi",
"WorkflowApiBackend",
"WorkflowSurfaceCapabilityId",
"WrapperAuthoringHints",
"WrapperHintConfidence",
"WrapperOutcomePolicy",
"parse_workflow_surface_capability_id",
"workflow_output_schema_for_authoring",
"wrapper_hints_for_capability",
]
src/wf_api/next_actions.py
+328
View File
@@ -0,0 +1,328 @@
from __future__ import annotations
from collections.abc import Sequence
from enum import StrEnum
from typing import Any, Self
from pydantic import BaseModel, Field
from .wrapper_hints import WrapperAuthoringHints
class NextActionTool(StrEnum):
"""Stable MCP workflow tools that guidance may recommend."""
PATCH_DRAFT_WORKSPACE = "wf.workflow.patch_draft_workspace"
VALIDATE_DRAFT_WORKSPACE = "wf.workflow.validate_draft_workspace"
VALIDATE_DEPLOYMENT = "wf.workflow.validate_deployment"
RUN_DEPLOYMENT = "wf.workflow.run_deployment"
RESUME_RUN = "wf.workflow.resume_run"
READ_RUN_TRACE = "wf.workflow.read_run_trace"
class NextActionPatchExample(BaseModel):
"""Concrete example request for a recommended MCP workflow tool."""
description: str = Field(description="Human-readable reason for this example.")
tool: NextActionTool = Field(description="MCP workflow tool to call.")
request: dict[str, Any] = Field(
description="JSON request payload to pass to the tool."
)
class NextActions(BaseModel):
"""Advisory continuation hints for MCP workflow clients.
This object is guidance, not authority. Validation diagnostics and runtime
status remain the source of truth; clients should treat this as a compact
answer to "what tool should I call next?"
"""
can_continue: bool = Field(
description=(
"Whether there is an obvious next workflow-surface tool call. "
"Advisory only."
)
)
can_save_now: bool | None = Field(
default=None,
description=(
"Advisory wrapper-authoring signal. False means review is "
"recommended before saving; the server does not enforce this."
),
)
recommended_next_tool: NextActionTool | None = Field(
default=None,
description="Suggested next MCP workflow tool, if one is obvious.",
)
reason: str = Field(description="Short explanation for the recommendation.")
patch_examples: list[NextActionPatchExample] = Field(
default_factory=list,
description="Concrete JSON Patch examples for common missing decisions.",
)
warnings: list[str] = Field(
default_factory=list,
description="Non-blocking warnings copied from low-confidence hints.",
)
@classmethod
def from_wrapper_hints(
cls,
*,
workspace_id: str,
revision: int,
hints: WrapperAuthoringHints | dict[str, Any],
) -> Self:
"""Create guidance after bootstrapping a wrapper draft workspace."""
payload = (
hints.model_dump(mode="json")
if isinstance(hints, WrapperAuthoringHints)
else hints
)
confidence = str(payload.get("confidence", "low"))
missing_decisions = payload.get("missing_decisions")
notes = [note for note in payload.get("notes", []) if isinstance(note, str)]
has_missing = isinstance(missing_decisions, list) and len(missing_decisions) > 0
can_save_now = confidence == "high" and not has_missing
if can_save_now:
return cls(
can_continue=True,
can_save_now=True,
recommended_next_tool=NextActionTool.VALIDATE_DRAFT_WORKSPACE,
reason="Wrapper hints are high confidence and have no missing decisions.",
patch_examples=[],
warnings=[],
)
return cls(
can_continue=True,
can_save_now=False,
recommended_next_tool=NextActionTool.PATCH_DRAFT_WORKSPACE,
reason="Review missing wrapper decisions before saving.",
patch_examples=_wrapper_draft_patch_examples(
workspace_id=workspace_id,
revision=revision,
hints=payload,
),
warnings=notes,
)
@classmethod
def from_deployment_validation(
cls,
*,
deployment_id: str,
diagnostics: Sequence[object],
) -> Self:
"""Create guidance after validate_deployment."""
if not diagnostics:
return cls(
can_continue=True,
can_save_now=None,
recommended_next_tool=NextActionTool.RUN_DEPLOYMENT,
reason=(
f"Deployment {deployment_id!r} is runnable; call "
"wf.workflow.run_deployment with workflow_input."
),
patch_examples=[],
warnings=[],
)
codes = {_diagnostic_field(diagnostic, "code") for diagnostic in diagnostics}
warnings = [_diagnostic_warning(diagnostic) for diagnostic in diagnostics]
if "source_unreachable" in codes:
reason = (
"One or more live sources are unreachable; fix or reconnect the "
"source, then rerun wf.workflow.validate_deployment with live_check=true."
)
elif "source_missing" in codes or "binding_missing" in codes:
reason = (
"Deployment bindings or sources are missing; inspect the deployment "
"and save corrected bindings before running."
)
elif "capability_missing" in codes or "schema_changed" in codes:
reason = (
"A required capability is missing or drifted; inspect capabilities "
"or refresh sources, then validate again."
)
else:
reason = (
"Deployment is not runnable; inspect diagnostics, repair the "
"deployment or sources, then validate again."
)
return cls(
can_continue=True,
can_save_now=None,
recommended_next_tool=NextActionTool.VALIDATE_DEPLOYMENT,
reason=reason,
patch_examples=[],
warnings=warnings,
)
@classmethod
def from_run_result(
cls,
*,
run_id: str | None,
status: str,
trace_count: int,
diagnostics: Sequence[object],
) -> Self:
"""Create guidance after run_deployment, inspect_run, resume_run, or read_run_trace."""
warnings = [_diagnostic_warning(diagnostic) for diagnostic in diagnostics]
if status == "interrupted" and run_id is not None:
return cls(
can_continue=True,
can_save_now=None,
recommended_next_tool=NextActionTool.RESUME_RUN,
reason=(
"Run is interrupted; call wf.workflow.resume_run with this "
"run_id and the interrupt response payload."
),
patch_examples=[],
warnings=warnings,
)
if status in {"failed", "unrunnable"}:
examples = (
[_bounded_trace_example(run_id=run_id, trace_count=trace_count)]
if run_id is not None and trace_count > 0
else []
)
return cls(
can_continue=bool(examples),
can_save_now=None,
recommended_next_tool=(
NextActionTool.READ_RUN_TRACE if examples else None
),
reason=(
"Run failed; read a bounded trace slice for debugging."
if examples
else "Run failed before producing trace entries; inspect diagnostics and error."
),
patch_examples=examples,
warnings=warnings,
)
if status == "completed":
return cls(
can_continue=False,
can_save_now=None,
recommended_next_tool=None,
reason=(
"Run completed. No required next workflow tool; use read_run_trace "
"with a bounded trace_range only if debugging."
),
patch_examples=[],
warnings=warnings,
)
return cls(
can_continue=False,
can_save_now=None,
recommended_next_tool=None,
reason=f"Run status {status!r} has no obvious next workflow tool.",
patch_examples=[],
warnings=warnings,
)
def _wrapper_draft_patch_examples(
*,
workspace_id: str,
revision: int,
hints: dict[str, Any],
) -> list[NextActionPatchExample]:
"""Return conservative JSON Patch examples without guessing semantics."""
examples: list[NextActionPatchExample] = []
missing_decisions = hints.get("missing_decisions")
if not isinstance(missing_decisions, list):
return examples
decision_kinds = {
str(decision.get("kind"))
for decision in missing_decisions
if isinstance(decision, dict)
}
if {"choose_output_fields", "review_nested_output"} & decision_kinds:
examples.append(
NextActionPatchExample(
description=(
"Replace output bindings after choosing which capability "
"outputs should be written to workflow state."
),
tool=NextActionTool.PATCH_DRAFT_WORKSPACE,
request={
"workspace_id": workspace_id,
"revision": revision,
"patch": [
{
"op": "replace",
"path": "/draft/steps/call/output",
"value": [],
}
],
},
)
)
if "confirm_boolean_outcomes" in decision_kinds:
examples.append(
NextActionPatchExample(
description=(
"Review boolean output candidates before adding routing; "
"do not route on boolean fields automatically."
),
tool=NextActionTool.PATCH_DRAFT_WORKSPACE,
request={
"workspace_id": workspace_id,
"revision": revision,
"patch": [],
},
)
)
return examples
def _diagnostic_field(diagnostic: object, field: str) -> str | None:
"""Read a diagnostic field from either a Pydantic model or a JSON dict."""
if isinstance(diagnostic, dict):
value = diagnostic.get(field)
else:
value = getattr(diagnostic, field, None)
return value if isinstance(value, str) else None
def _diagnostic_warning(diagnostic: object) -> str:
"""Format one compact diagnostic warning for next_actions."""
code = _diagnostic_field(diagnostic, "code") or "diagnostic"
bound_source = _diagnostic_field(diagnostic, "bound_source")
logical_ref = _diagnostic_field(diagnostic, "logical_ref")
if bound_source:
return f"{code}: {bound_source}"
if logical_ref:
return f"{code}: {logical_ref}"
return code
def _bounded_trace_example(
*,
run_id: str,
trace_count: int,
) -> NextActionPatchExample:
"""Return a safe read_run_trace request; never suggest full trace reads."""
return NextActionPatchExample(
description=(
"Read a bounded debug trace slice. Increase start/limit only when needed."
),
tool=NextActionTool.READ_RUN_TRACE,
request={
"run_id": run_id,
"trace_range": {
"start": 0,
"limit": 25,
},
},
)
__all__ = [
"NextActionPatchExample",
"NextActionTool",
"NextActions",
]
src/wf_api/wrapper_hints.py
+319
View File
@@ -0,0 +1,319 @@
from __future__ import annotations
from enum import StrEnum
from typing import Any
from pydantic import BaseModel, Field
JsonObject = dict[str, Any]
CONTROL_BOOLEAN_NAMES = {
"success",
"ok",
"failed",
"error",
"is_error",
"needs_input",
"requires_approval",
"approved",
"rejected",
"has_more",
"done",
"complete",
}
class WrapperHintConfidence(StrEnum):
"""Coarse confidence for generated wrapper scaffolding hints."""
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
class WrapperOutcomePolicy(StrEnum):
"""How wrapper outcomes were chosen."""
PRESERVE_DECLARED = "preserve_declared"
MANUAL_MAPPING_REQUIRED = "manual_mapping_required"
class OutcomeCandidateKind(StrEnum):
"""Reason a field was offered as a possible outcome source."""
BOOLEAN_CONTROL_FIELD = "boolean_control_field"
class MissingDecisionKind(StrEnum):
"""Typed action item a human or LLM must decide before saving a wrapper."""
CHOOSE_OUTPUT_FIELDS = "choose_output_fields"
REVIEW_NESTED_OUTPUT = "review_nested_output"
CONFIRM_BOOLEAN_OUTCOMES = "confirm_boolean_outcomes"
CHOOSE_ERROR_MAPPING = "choose_error_mapping"
class OutcomeCandidate(BaseModel):
"""One possible outcome mapping that must not be applied automatically."""
kind: OutcomeCandidateKind
source: str = Field(description="Output path such as output.success.")
candidate_outcomes: list[str]
confidence: WrapperHintConfidence
reason: str
automatic: bool = False
class MissingDecision(BaseModel):
"""One explicit decision required before a wrapper should be saved."""
kind: MissingDecisionKind
message: str
class WrapperAuthoringHints(BaseModel):
"""Scaffold for creating a workflow wrapper around one capability."""
capability_name: str
confidence: WrapperHintConfidence
declared_outcomes: list[str]
suggested_wrapper_outcomes: list[str]
outcome_policy: WrapperOutcomePolicy
input_schema: JsonObject
state_schema: JsonObject
output_schema: JsonObject
input_map: dict[str, str]
output_map: dict[str, str]
outcome_candidates: list[OutcomeCandidate] = Field(default_factory=list)
missing_decisions: list[MissingDecision] = Field(default_factory=list)
notes: list[str] = Field(default_factory=list)
def wrapper_hints_for_capability(
*,
capability_name: str,
input_schema: JsonObject,
output_schema: JsonObject,
outcomes: list[str] | tuple[str, ...],
) -> WrapperAuthoringHints:
"""Derive conservative wrapper scaffolding for one workflow capability.
The helper deliberately preserves declared outcomes and only proposes
boolean output fields as candidates. It must not infer business semantics or
create routes by itself.
"""
input_properties = _object_properties(input_schema)
hint_output_schema = workflow_output_schema_for_authoring(output_schema)
output_properties = _object_properties(hint_output_schema)
input_map = {f"input.{name}": name for name in sorted(input_properties)}
output_map_properties = _default_output_map_properties(
output_schema, output_properties
)
output_map = {name: f"state.{name}" for name in sorted(output_map_properties)}
state_schema = {
"type": "object",
"properties": {
name: schema for name, schema in sorted(output_map_properties.items())
},
}
wrapper_output_schema = {
"type": "object",
"properties": {
name: schema for name, schema in sorted(output_map_properties.items())
},
}
missing_decisions = _missing_decisions_for_output(hint_output_schema)
outcome_candidates = _boolean_outcome_candidates(output_properties)
if outcome_candidates:
missing_decisions.append(
MissingDecision(
kind=MissingDecisionKind.CONFIRM_BOOLEAN_OUTCOMES,
message=(
"Confirm whether boolean output fields should control "
"wrapper routing."
),
)
)
confidence = _confidence_for_hint(
input_schema=input_schema,
output_schema=hint_output_schema,
missing_decisions=missing_decisions,
outcome_candidates=outcome_candidates,
)
notes = [
"Hints are scaffolding, not semantic guarantees.",
(
"Declared outcomes are preserved; output-field outcome "
"inference is not automatic."
),
]
if _has_raw_mcp_content(output_schema):
notes.append(
"Raw MCP content blocks are not workflow-shaped. Use an explicit "
"wrapper or extraction node to handle TextContent, ResourceLink, "
"images, or mixed content before writing to typed state."
)
return WrapperAuthoringHints(
capability_name=capability_name,
confidence=confidence,
declared_outcomes=list(outcomes),
suggested_wrapper_outcomes=list(outcomes),
outcome_policy=WrapperOutcomePolicy.PRESERVE_DECLARED,
input_schema=input_schema,
state_schema=state_schema,
output_schema=wrapper_output_schema,
input_map=input_map,
output_map=output_map,
outcome_candidates=outcome_candidates,
missing_decisions=missing_decisions,
notes=notes,
)
def workflow_output_schema_for_authoring(output_schema: JsonObject) -> JsonObject:
"""Return the workflow-author-facing output schema for one capability.
Raw MCP ``content`` blocks stay raw. A wrapper may choose to extract
``content[0].text`` or handle resources/images, but the authoring surface
must not invent that decision as a top-level schema field.
"""
return {"type": "object", "properties": _object_properties(output_schema)}
def _object_properties(schema: JsonObject) -> dict[str, JsonObject]:
"""Return object properties that are themselves JSON Schema objects."""
properties = schema.get("properties")
if not isinstance(properties, dict):
return {}
return {
str(name): value
for name, value in properties.items()
if isinstance(value, dict)
}
def _has_raw_mcp_content(schema: JsonObject) -> bool:
"""Return true when schema exposes MCP's raw content-block envelope."""
properties = _object_properties(schema)
content_schema = properties.get("content")
return isinstance(content_schema, dict) and content_schema.get("type") == "array"
def _default_output_map_properties(
raw_output_schema: JsonObject,
output_properties: dict[str, JsonObject],
) -> dict[str, JsonObject]:
"""Return fields safe enough to wire by default in generated hints.
Raw MCP ``content`` is a protocol envelope, not a workflow value. Keeping it
out of the default map prevents the scaffold from writing text/image/resource
blocks into typed state without an explicit extraction wrapper.
"""
if _has_raw_mcp_content(raw_output_schema):
return {
name: schema
for name, schema in output_properties.items()
if name != "content"
}
return output_properties
def _missing_decisions_for_output(output_schema: JsonObject) -> list[MissingDecision]:
"""Return explicit decisions required by output schema shape."""
properties = _object_properties(output_schema)
if not properties:
return [
MissingDecision(
kind=MissingDecisionKind.CHOOSE_OUTPUT_FIELDS,
message=(
"Capability output schema has no top-level object "
"properties to map."
),
)
]
decisions: list[MissingDecision] = []
for name, schema in sorted(properties.items()):
schema_type = schema.get("type")
if schema_type == "object" or schema_type == "array":
decisions.append(
MissingDecision(
kind=MissingDecisionKind.REVIEW_NESTED_OUTPUT,
message=(
f"Review output.{name}; nested or collection outputs "
"may need explicit mapping."
),
)
)
return decisions
def _boolean_outcome_candidates(
output_properties: dict[str, JsonObject],
) -> list[OutcomeCandidate]:
"""Return conservative candidate outcome mappings for control-like booleans."""
candidates: list[OutcomeCandidate] = []
for name, schema in sorted(output_properties.items()):
if schema.get("type") != "boolean":
continue
if name.casefold() not in CONTROL_BOOLEAN_NAMES:
continue
candidates.append(
OutcomeCandidate(
kind=OutcomeCandidateKind.BOOLEAN_CONTROL_FIELD,
source=f"output.{name}",
candidate_outcomes=_candidate_outcomes_for_boolean_name(name),
confidence=WrapperHintConfidence.MEDIUM,
reason="top-level boolean field with control-like name",
automatic=False,
)
)
return candidates
def _candidate_outcomes_for_boolean_name(name: str) -> list[str]:
"""Map known control-like boolean names to possible outcome labels."""
normalized = name.casefold()
if normalized in {"success", "ok", "done", "complete"}:
return ["success", "failure"]
if normalized in {"failed", "error", "is_error"}:
return ["error", "ok"]
if normalized in {"approved", "rejected"}:
return ["approved", "rejected"]
if normalized in {"needs_input", "requires_approval"}:
return [normalized, "done"]
if normalized == "has_more":
return ["has_more", "done"]
return ["true", "false"]
def _confidence_for_hint(
*,
input_schema: JsonObject,
output_schema: JsonObject,
missing_decisions: list[MissingDecision],
outcome_candidates: list[OutcomeCandidate],
) -> WrapperHintConfidence:
"""Assign coarse confidence from schema shape and pending decisions."""
if not _object_properties(input_schema) or not _object_properties(output_schema):
return WrapperHintConfidence.LOW
if any(
decision.kind == MissingDecisionKind.REVIEW_NESTED_OUTPUT
for decision in missing_decisions
):
return WrapperHintConfidence.LOW
if missing_decisions or outcome_candidates:
return WrapperHintConfidence.MEDIUM
return WrapperHintConfidence.HIGH
__all__ = [
"MissingDecision",
"MissingDecisionKind",
"OutcomeCandidate",
"OutcomeCandidateKind",
"WrapperAuthoringHints",
"WrapperHintConfidence",
"WrapperOutcomePolicy",
"workflow_output_schema_for_authoring",
"wrapper_hints_for_capability",
]
src/wf_mcp/workflow_surface/handlers.py
+2 -2
View File
@@ -61,7 +61,7 @@ from ..events import make_event
from ..models import RawWorkflowPlan
from ..shared import matches_query, paged_list_payload
from .models import TraceRange
from .next_actions import NextActions
from wf_api.next_actions import NextActions
from .saved_subgraphs import (
SavedSubgraphTree,
direct_wrapper_interrupt_diagnostic,
@@ -78,7 +78,7 @@ from .run_lifecycle import (
restore_interrupted_run,
validate_pinned_resume_environment,
)
from .wrapper_hints import (
from wf_api.wrapper_hints import (
workflow_output_schema_for_authoring,
wrapper_hints_for_capability,
)
src/wf_mcp/workflow_surface/models.py
+1 -1
View File
@@ -4,7 +4,7 @@ from typing import Annotated, Any, Literal
from pydantic import BaseModel, Field
from .next_actions import NextActionPatchExample, NextActions
from wf_api.next_actions import NextActionPatchExample, NextActions
from wf_artifacts import ArtifactKind
from wf_artifacts.draft_workspaces.models import WORKSPACE_ID_PATTERN
from wf_core.models.steps import InputBinding, OutputBinding
src/wf_mcp/workflow_surface/next_actions.py
+10 -318
View File
@@ -1,321 +1,13 @@
from __future__ import annotations
"""Compatibility shim for workflow API next-action guidance.
from collections.abc import Sequence
from enum import StrEnum
from typing import Any, Self
New code should import from `wf_api.next_actions`. This module stays so older
MCP workflow-surface imports keep working during extraction.
"""
from pydantic import BaseModel, Field
from wf_api.next_actions import NextActionPatchExample, NextActionTool, NextActions
from .wrapper_hints import WrapperAuthoringHints
class NextActionTool(StrEnum):
"""Stable MCP workflow tools that guidance may recommend."""
PATCH_DRAFT_WORKSPACE = "wf.workflow.patch_draft_workspace"
VALIDATE_DRAFT_WORKSPACE = "wf.workflow.validate_draft_workspace"
VALIDATE_DEPLOYMENT = "wf.workflow.validate_deployment"
RUN_DEPLOYMENT = "wf.workflow.run_deployment"
RESUME_RUN = "wf.workflow.resume_run"
READ_RUN_TRACE = "wf.workflow.read_run_trace"
class NextActionPatchExample(BaseModel):
"""Concrete example request for a recommended MCP workflow tool."""
description: str = Field(description="Human-readable reason for this example.")
tool: NextActionTool = Field(description="MCP workflow tool to call.")
request: dict[str, Any] = Field(
description="JSON request payload to pass to the tool."
)
class NextActions(BaseModel):
"""Advisory continuation hints for MCP workflow clients.
This object is guidance, not authority. Validation diagnostics and runtime
status remain the source of truth; clients should treat this as a compact
answer to "what tool should I call next?"
"""
can_continue: bool = Field(
description=(
"Whether there is an obvious next workflow-surface tool call. "
"Advisory only."
)
)
can_save_now: bool | None = Field(
default=None,
description=(
"Advisory wrapper-authoring signal. False means review is "
"recommended before saving; the server does not enforce this."
),
)
recommended_next_tool: NextActionTool | None = Field(
default=None,
description="Suggested next MCP workflow tool, if one is obvious.",
)
reason: str = Field(description="Short explanation for the recommendation.")
patch_examples: list[NextActionPatchExample] = Field(
default_factory=list,
description="Concrete JSON Patch examples for common missing decisions.",
)
warnings: list[str] = Field(
default_factory=list,
description="Non-blocking warnings copied from low-confidence hints.",
)
@classmethod
def from_wrapper_hints(
cls,
*,
workspace_id: str,
revision: int,
hints: WrapperAuthoringHints | dict[str, Any],
) -> Self:
"""Create guidance after bootstrapping a wrapper draft workspace."""
payload = (
hints.model_dump(mode="json")
if isinstance(hints, WrapperAuthoringHints)
else hints
)
confidence = str(payload.get("confidence", "low"))
missing_decisions = payload.get("missing_decisions")
notes = [note for note in payload.get("notes", []) if isinstance(note, str)]
has_missing = isinstance(missing_decisions, list) and len(missing_decisions) > 0
can_save_now = confidence == "high" and not has_missing
if can_save_now:
return cls(
can_continue=True,
can_save_now=True,
recommended_next_tool=NextActionTool.VALIDATE_DRAFT_WORKSPACE,
reason="Wrapper hints are high confidence and have no missing decisions.",
patch_examples=[],
warnings=[],
)
return cls(
can_continue=True,
can_save_now=False,
recommended_next_tool=NextActionTool.PATCH_DRAFT_WORKSPACE,
reason="Review missing wrapper decisions before saving.",
patch_examples=_wrapper_draft_patch_examples(
workspace_id=workspace_id,
revision=revision,
hints=payload,
),
warnings=notes,
)
@classmethod
def from_deployment_validation(
cls,
*,
deployment_id: str,
diagnostics: Sequence[object],
) -> Self:
"""Create guidance after validate_deployment."""
if not diagnostics:
return cls(
can_continue=True,
can_save_now=None,
recommended_next_tool=NextActionTool.RUN_DEPLOYMENT,
reason=(
f"Deployment {deployment_id!r} is runnable; call "
"wf.workflow.run_deployment with workflow_input."
),
patch_examples=[],
warnings=[],
)
codes = {_diagnostic_field(diagnostic, "code") for diagnostic in diagnostics}
warnings = [_diagnostic_warning(diagnostic) for diagnostic in diagnostics]
if "source_unreachable" in codes:
reason = (
"One or more live sources are unreachable; fix or reconnect the "
"source, then rerun wf.workflow.validate_deployment with live_check=true."
)
elif "source_missing" in codes or "binding_missing" in codes:
reason = (
"Deployment bindings or sources are missing; inspect the deployment "
"and save corrected bindings before running."
)
elif "capability_missing" in codes or "schema_changed" in codes:
reason = (
"A required capability is missing or drifted; inspect capabilities "
"or refresh sources, then validate again."
)
else:
reason = (
"Deployment is not runnable; inspect diagnostics, repair the "
"deployment or sources, then validate again."
)
return cls(
can_continue=True,
can_save_now=None,
recommended_next_tool=NextActionTool.VALIDATE_DEPLOYMENT,
reason=reason,
patch_examples=[],
warnings=warnings,
)
@classmethod
def from_run_result(
cls,
*,
run_id: str | None,
status: str,
trace_count: int,
diagnostics: Sequence[object],
) -> Self:
"""Create guidance after run_deployment, inspect_run, resume_run, or read_run_trace."""
warnings = [_diagnostic_warning(diagnostic) for diagnostic in diagnostics]
if status == "interrupted" and run_id is not None:
return cls(
can_continue=True,
can_save_now=None,
recommended_next_tool=NextActionTool.RESUME_RUN,
reason=(
"Run is interrupted; call wf.workflow.resume_run with this "
"run_id and the interrupt response payload."
),
patch_examples=[],
warnings=warnings,
)
if status in {"failed", "unrunnable"}:
examples = (
[_bounded_trace_example(run_id=run_id, trace_count=trace_count)]
if run_id is not None and trace_count > 0
else []
)
return cls(
can_continue=bool(examples),
can_save_now=None,
recommended_next_tool=(
NextActionTool.READ_RUN_TRACE if examples else None
),
reason=(
"Run failed; read a bounded trace slice for debugging."
if examples
else "Run failed before producing trace entries; inspect diagnostics and error."
),
patch_examples=examples,
warnings=warnings,
)
if status == "completed":
return cls(
can_continue=False,
can_save_now=None,
recommended_next_tool=None,
reason=(
"Run completed. No required next workflow tool; use read_run_trace "
"with a bounded trace_range only if debugging."
),
patch_examples=[],
warnings=warnings,
)
return cls(
can_continue=False,
can_save_now=None,
recommended_next_tool=None,
reason=f"Run status {status!r} has no obvious next workflow tool.",
patch_examples=[],
warnings=warnings,
)
def _wrapper_draft_patch_examples(
*,
workspace_id: str,
revision: int,
hints: dict[str, Any],
) -> list[NextActionPatchExample]:
"""Return conservative JSON Patch examples without guessing semantics."""
examples: list[NextActionPatchExample] = []
missing_decisions = hints.get("missing_decisions")
if not isinstance(missing_decisions, list):
return examples
decision_kinds = {
str(decision.get("kind"))
for decision in missing_decisions
if isinstance(decision, dict)
}
if {"choose_output_fields", "review_nested_output"} & decision_kinds:
examples.append(
NextActionPatchExample(
description=(
"Replace output bindings after choosing which capability "
"outputs should be written to workflow state."
),
tool=NextActionTool.PATCH_DRAFT_WORKSPACE,
request={
"workspace_id": workspace_id,
"revision": revision,
"patch": [
{
"op": "replace",
"path": "/draft/steps/call/output",
"value": [],
}
],
},
)
)
if "confirm_boolean_outcomes" in decision_kinds:
examples.append(
NextActionPatchExample(
description=(
"Review boolean output candidates before adding routing; "
"do not route on boolean fields automatically."
),
tool=NextActionTool.PATCH_DRAFT_WORKSPACE,
request={
"workspace_id": workspace_id,
"revision": revision,
"patch": [],
},
)
)
return examples
def _diagnostic_field(diagnostic: object, field: str) -> str | None:
"""Read a diagnostic field from either a Pydantic model or a JSON dict."""
if isinstance(diagnostic, dict):
value = diagnostic.get(field)
else:
value = getattr(diagnostic, field, None)
return value if isinstance(value, str) else None
def _diagnostic_warning(diagnostic: object) -> str:
"""Format one compact diagnostic warning for next_actions."""
code = _diagnostic_field(diagnostic, "code") or "diagnostic"
bound_source = _diagnostic_field(diagnostic, "bound_source")
logical_ref = _diagnostic_field(diagnostic, "logical_ref")
if bound_source:
return f"{code}: {bound_source}"
if logical_ref:
return f"{code}: {logical_ref}"
return code
def _bounded_trace_example(
*,
run_id: str,
trace_count: int,
) -> NextActionPatchExample:
"""Return a safe read_run_trace request; never suggest full trace reads."""
return NextActionPatchExample(
description=(
"Read a bounded debug trace slice. Increase start/limit only when needed."
),
tool=NextActionTool.READ_RUN_TRACE,
request={
"run_id": run_id,
"trace_range": {
"start": 0,
"limit": 25,
},
},
)
__all__ = [
"NextActionPatchExample",
"NextActionTool",
"NextActions",
]
src/wf_mcp/workflow_surface/wrapper_hints.py
+26 -303
View File
@@ -1,306 +1,29 @@
from __future__ import annotations
"""Compatibility shim for workflow API wrapper authoring hints.
from enum import StrEnum
from typing import Any
New code should import from `wf_api.wrapper_hints`. This module stays so older
MCP workflow-surface imports keep working during extraction.
"""
from pydantic import BaseModel, Field
from wf_api.wrapper_hints import (
MissingDecision,
MissingDecisionKind,
OutcomeCandidate,
OutcomeCandidateKind,
WrapperAuthoringHints,
WrapperHintConfidence,
WrapperOutcomePolicy,
workflow_output_schema_for_authoring,
wrapper_hints_for_capability,
)
JsonObject = dict[str, Any]
CONTROL_BOOLEAN_NAMES = {
"success",
"ok",
"failed",
"error",
"is_error",
"needs_input",
"requires_approval",
"approved",
"rejected",
"has_more",
"done",
"complete",
}
class WrapperHintConfidence(StrEnum):
"""Coarse confidence for generated wrapper scaffolding hints."""
HIGH = "high"
MEDIUM = "medium"
LOW = "low"
class WrapperOutcomePolicy(StrEnum):
"""How wrapper outcomes were chosen."""
PRESERVE_DECLARED = "preserve_declared"
MANUAL_MAPPING_REQUIRED = "manual_mapping_required"
class OutcomeCandidateKind(StrEnum):
"""Reason a field was offered as a possible outcome source."""
BOOLEAN_CONTROL_FIELD = "boolean_control_field"
class MissingDecisionKind(StrEnum):
"""Typed action item a human or LLM must decide before saving a wrapper."""
CHOOSE_OUTPUT_FIELDS = "choose_output_fields"
REVIEW_NESTED_OUTPUT = "review_nested_output"
CONFIRM_BOOLEAN_OUTCOMES = "confirm_boolean_outcomes"
CHOOSE_ERROR_MAPPING = "choose_error_mapping"
class OutcomeCandidate(BaseModel):
"""One possible outcome mapping that must not be applied automatically."""
kind: OutcomeCandidateKind
source: str = Field(description="Output path such as output.success.")
candidate_outcomes: list[str]
confidence: WrapperHintConfidence
reason: str
automatic: bool = False
class MissingDecision(BaseModel):
"""One explicit decision required before a wrapper should be saved."""
kind: MissingDecisionKind
message: str
class WrapperAuthoringHints(BaseModel):
"""Scaffold for creating a workflow wrapper around one capability."""
capability_name: str
confidence: WrapperHintConfidence
declared_outcomes: list[str]
suggested_wrapper_outcomes: list[str]
outcome_policy: WrapperOutcomePolicy
input_schema: JsonObject
state_schema: JsonObject
output_schema: JsonObject
input_map: dict[str, str]
output_map: dict[str, str]
outcome_candidates: list[OutcomeCandidate] = Field(default_factory=list)
missing_decisions: list[MissingDecision] = Field(default_factory=list)
notes: list[str] = Field(default_factory=list)
def wrapper_hints_for_capability(
*,
capability_name: str,
input_schema: JsonObject,
output_schema: JsonObject,
outcomes: list[str] | tuple[str, ...],
) -> WrapperAuthoringHints:
"""Derive conservative wrapper scaffolding for one workflow capability.
The helper deliberately preserves declared outcomes and only proposes
boolean output fields as candidates. It must not infer business semantics or
create routes by itself.
"""
input_properties = _object_properties(input_schema)
hint_output_schema = workflow_output_schema_for_authoring(output_schema)
output_properties = _object_properties(hint_output_schema)
input_map = {f"input.{name}": name for name in sorted(input_properties)}
output_map_properties = _default_output_map_properties(
output_schema, output_properties
)
output_map = {name: f"state.{name}" for name in sorted(output_map_properties)}
state_schema = {
"type": "object",
"properties": {
name: schema for name, schema in sorted(output_map_properties.items())
},
}
wrapper_output_schema = {
"type": "object",
"properties": {
name: schema for name, schema in sorted(output_map_properties.items())
},
}
missing_decisions = _missing_decisions_for_output(hint_output_schema)
outcome_candidates = _boolean_outcome_candidates(output_properties)
if outcome_candidates:
missing_decisions.append(
MissingDecision(
kind=MissingDecisionKind.CONFIRM_BOOLEAN_OUTCOMES,
message=(
"Confirm whether boolean output fields should control "
"wrapper routing."
),
)
)
confidence = _confidence_for_hint(
input_schema=input_schema,
output_schema=hint_output_schema,
missing_decisions=missing_decisions,
outcome_candidates=outcome_candidates,
)
notes = [
"Hints are scaffolding, not semantic guarantees.",
(
"Declared outcomes are preserved; output-field outcome "
"inference is not automatic."
),
]
if _has_raw_mcp_content(output_schema):
notes.append(
"Raw MCP content blocks are not workflow-shaped. Use an explicit "
"wrapper or extraction node to handle TextContent, ResourceLink, "
"images, or mixed content before writing to typed state."
)
return WrapperAuthoringHints(
capability_name=capability_name,
confidence=confidence,
declared_outcomes=list(outcomes),
suggested_wrapper_outcomes=list(outcomes),
outcome_policy=WrapperOutcomePolicy.PRESERVE_DECLARED,
input_schema=input_schema,
state_schema=state_schema,
output_schema=wrapper_output_schema,
input_map=input_map,
output_map=output_map,
outcome_candidates=outcome_candidates,
missing_decisions=missing_decisions,
notes=notes,
)
def workflow_output_schema_for_authoring(output_schema: JsonObject) -> JsonObject:
"""Return the workflow-author-facing output schema for one capability.
Raw MCP ``content`` blocks stay raw. A wrapper may choose to extract
``content[0].text`` or handle resources/images, but the authoring surface
must not invent that decision as a top-level schema field.
"""
return {"type": "object", "properties": _object_properties(output_schema)}
def _object_properties(schema: JsonObject) -> dict[str, JsonObject]:
"""Return object properties that are themselves JSON Schema objects."""
properties = schema.get("properties")
if not isinstance(properties, dict):
return {}
return {
str(name): value
for name, value in properties.items()
if isinstance(value, dict)
}
def _has_raw_mcp_content(schema: JsonObject) -> bool:
"""Return true when schema exposes MCP's raw content-block envelope."""
properties = _object_properties(schema)
content_schema = properties.get("content")
return isinstance(content_schema, dict) and content_schema.get("type") == "array"
def _default_output_map_properties(
raw_output_schema: JsonObject,
output_properties: dict[str, JsonObject],
) -> dict[str, JsonObject]:
"""Return fields safe enough to wire by default in generated hints.
Raw MCP ``content`` is a protocol envelope, not a workflow value. Keeping it
out of the default map prevents the scaffold from writing text/image/resource
blocks into typed state without an explicit extraction wrapper.
"""
if _has_raw_mcp_content(raw_output_schema):
return {
name: schema
for name, schema in output_properties.items()
if name != "content"
}
return output_properties
def _missing_decisions_for_output(output_schema: JsonObject) -> list[MissingDecision]:
"""Return explicit decisions required by output schema shape."""
properties = _object_properties(output_schema)
if not properties:
return [
MissingDecision(
kind=MissingDecisionKind.CHOOSE_OUTPUT_FIELDS,
message=(
"Capability output schema has no top-level object "
"properties to map."
),
)
]
decisions: list[MissingDecision] = []
for name, schema in sorted(properties.items()):
schema_type = schema.get("type")
if schema_type == "object" or schema_type == "array":
decisions.append(
MissingDecision(
kind=MissingDecisionKind.REVIEW_NESTED_OUTPUT,
message=(
f"Review output.{name}; nested or collection outputs "
"may need explicit mapping."
),
)
)
return decisions
def _boolean_outcome_candidates(
output_properties: dict[str, JsonObject],
) -> list[OutcomeCandidate]:
"""Return conservative candidate outcome mappings for control-like booleans."""
candidates: list[OutcomeCandidate] = []
for name, schema in sorted(output_properties.items()):
if schema.get("type") != "boolean":
continue
if name.casefold() not in CONTROL_BOOLEAN_NAMES:
continue
candidates.append(
OutcomeCandidate(
kind=OutcomeCandidateKind.BOOLEAN_CONTROL_FIELD,
source=f"output.{name}",
candidate_outcomes=_candidate_outcomes_for_boolean_name(name),
confidence=WrapperHintConfidence.MEDIUM,
reason="top-level boolean field with control-like name",
automatic=False,
)
)
return candidates
def _candidate_outcomes_for_boolean_name(name: str) -> list[str]:
"""Map known control-like boolean names to possible outcome labels."""
normalized = name.casefold()
if normalized in {"success", "ok", "done", "complete"}:
return ["success", "failure"]
if normalized in {"failed", "error", "is_error"}:
return ["error", "ok"]
if normalized in {"approved", "rejected"}:
return ["approved", "rejected"]
if normalized in {"needs_input", "requires_approval"}:
return [normalized, "done"]
if normalized == "has_more":
return ["has_more", "done"]
return ["true", "false"]
def _confidence_for_hint(
*,
input_schema: JsonObject,
output_schema: JsonObject,
missing_decisions: list[MissingDecision],
outcome_candidates: list[OutcomeCandidate],
) -> WrapperHintConfidence:
"""Assign coarse confidence from schema shape and pending decisions."""
if not _object_properties(input_schema) or not _object_properties(output_schema):
return WrapperHintConfidence.LOW
if any(
decision.kind == MissingDecisionKind.REVIEW_NESTED_OUTPUT
for decision in missing_decisions
):
return WrapperHintConfidence.LOW
if missing_decisions or outcome_candidates:
return WrapperHintConfidence.MEDIUM
return WrapperHintConfidence.HIGH
__all__ = [
"MissingDecision",
"MissingDecisionKind",
"OutcomeCandidate",
"OutcomeCandidateKind",
"WrapperAuthoringHints",
"WrapperHintConfidence",
"WrapperOutcomePolicy",
"workflow_output_schema_for_authoring",
"wrapper_hints_for_capability",
]
tests/wf_mcp/test_workflow_wrapper_hints.py
+10 -1
View File
@@ -1,6 +1,6 @@
from __future__ import annotations
from wf_mcp.workflow_surface.wrapper_hints import (
from wf_api.wrapper_hints import (
MissingDecision,
MissingDecisionKind,
OutcomeCandidate,
@@ -220,3 +220,12 @@ def test_wrapper_hints_mark_empty_output_schema_as_low_confidence() -> None:
assert dumped["input_map"] == {"input.text": "text"}
assert dumped["output_map"] == {}
assert dumped["missing_decisions"][0]["kind"] == "choose_output_fields"
def test_workflow_surface_wrapper_hints_shim_reexports_canonical_helper() -> None:
from wf_api.wrapper_hints import wrapper_hints_for_capability
from wf_mcp.workflow_surface.wrapper_hints import (
wrapper_hints_for_capability as wrapper_hints_for_capability_shim,
)
assert wrapper_hints_for_capability_shim is wrapper_hints_for_capability
tests/wf_mcp/workflow_surface/test_next_actions.py
+8 -1
View File
@@ -2,7 +2,7 @@ from __future__ import annotations
from wf_artifacts import DependencyDiagnostic, DiagnosticSeverity
from wf_mcp.workflow_surface.next_actions import NextActionTool, NextActions
from wf_api.next_actions import NextActionTool, NextActions
def test_next_actions_from_high_confidence_wrapper_hints_can_validate() -> None:
@@ -146,3 +146,10 @@ def test_next_actions_from_interrupted_run_recommends_resume() -> None:
assert dumped["recommended_next_tool"] == NextActionTool.RESUME_RUN.value
assert "resume_run" in dumped["reason"]
assert dumped["patch_examples"] == []
def test_workflow_surface_next_actions_shim_reexports_canonical_model() -> None:
from wf_api.next_actions import NextActions
from wf_mcp.workflow_surface.next_actions import NextActions as NextActionsShim
assert NextActionsShim is NextActions