feat: add wf source resource refs

src/wf_api/__init__.py

@@ -35,6 +35,7 @@ from .runs import WorkflowRunApi
 from .runtime_dependencies import RuntimeDependencies, resolve_runtime_dependencies
 from .service import WorkflowApi
 from .source_admin import WorkflowSourceAdminApi
+from .source_refs import SourceResourceRef
 from .source_registry_admin import (
     WorkflowSourceRegistryApi,
     WorkflowSourceRegistryApplyProvider,
@@ -110,6 +111,7 @@ __all__ = [
     "WorkflowRuntimeRunner",
     "WorkflowRunApi",
     "WorkflowRunSurface",
+    "SourceResourceRef",
     "WorkflowSourceAdminApi",
     "WorkflowSourceAdminSurface",
     "WorkflowSourceRegistryApi",

src/wf_api/local_sources.py

@@ -3,6 +3,8 @@ from __future__ import annotations
 from collections.abc import Mapping
 from typing import TYPE_CHECKING, Any
 
+from pydantic import BaseModel, Field
+
 from wf_authoring import (
     NodeSpec,
     coalesce,
@@ -28,6 +30,7 @@ from wf_authoring import (
     runtime_error,
     truthy,
 )
+from wf_core import RuntimeContext
 from wf_core.runtime.ops.merges import DEFAULT_REDUCER_DEFINITIONS
 from wf_platform import (
     CapabilityBuckets,
@@ -37,6 +40,9 @@ from wf_platform import (
     SourceVisibility,
 )
 
+from .source_helpers import ReadResourceOutput, read_resource
+from .source_refs import SourceResourceRef
+
 if TYPE_CHECKING:
     from wf_core import ReducerSpec
 
@@ -79,6 +85,24 @@ AUTHORING_STD_SPECS: tuple[NodeSpec[Any, Any], ...] = (
 RECIPE_SPECS: tuple[NodeSpec[Any, Any], ...] = (extract_text_content,)
 """Composed first-party recipes exposed as workflow-facing capabilities."""
 
+SOURCE_SOURCE_ID = "wf.source"
+"""Internal source id for explicit source-ref helper nodes."""
+
+
+class ReadResourceInput(BaseModel):
+    """Input model for wf.source.read_resource node."""
+
+    ref: SourceResourceRef
+    max_chars: int = Field(default=4000, ge=1, le=20000)
+
+
+@node(name="read_resource")
+async def read_resource_node(
+    payload: ReadResourceInput,
+    ctx: RuntimeContext,
+) -> ReadResourceOutput:
+    return await read_resource(payload.ref, ctx, max_chars=payload.max_chars)
+
 
 def qualify_node_name(source_id: str, local_name: str) -> str:
     """Return one source-qualified node name without assuming MCP connections."""
@@ -187,6 +211,21 @@ def builtin_sources() -> dict[str, CapabilitySource]:
             policy=SourcePolicy(platform=True, binding_required=False),
             description="First-party workflow recipes composed from standard nodes.",
         ),
+        SOURCE_SOURCE_ID: CapabilitySource(
+            id=SOURCE_SOURCE_ID,
+            kind="system",
+            capabilities=CapabilityBuckets(
+                node_specs=_qualified_specs(SOURCE_SOURCE_ID, (read_resource_node,)),
+            ),
+            visibility=SourceVisibility(
+                planner=True,
+                client=True,
+                admin_dashboard=True,
+            ),
+            permissions=SourcePermissions(safe_for_workflow=True, calls_upstream=True),
+            policy=SourcePolicy(platform=True, binding_required=False),
+            description="Platform helpers for explicit source refs.",
+        ),
     }
 
 

src/wf_api/platform_context.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from collections.abc import Awaitable, Callable, Mapping
+from dataclasses import dataclass, field
+from typing import Any, Protocol
+
+
+class WorkflowPlatformContext(Protocol):
+    """Runtime platform services available only to explicit platform helper nodes."""
+
+    def resolve_source(self, logical_source: str) -> str: ...
+
+    async def read_resource(
+        self,
+        *,
+        source_id: str,
+        uri: str,
+        max_chars: int,
+    ) -> dict[str, Any]: ...
+
+
+ReadResourceHandler = Callable[[str, str, int], Awaitable[dict[str, Any]]]
+
+
+@dataclass(frozen=True, slots=True)
+class SourceBindingPlatformContext:
+    """Resolve logical source refs for explicit source-aware helper nodes."""
+
+    source_bindings: Mapping[str, str]
+    read_resource_handler: ReadResourceHandler | None
+    platform_sources: set[str] = field(default_factory=set)
+
+    def resolve_source(self, logical_source: str) -> str:
+        if logical_source in self.platform_sources:
+            return logical_source
+        try:
+            return self.source_bindings[logical_source]
+        except KeyError as exc:
+            raise KeyError(f"unbound logical source {logical_source!r}") from exc
+
+    async def read_resource(
+        self,
+        *,
+        source_id: str,
+        uri: str,
+        max_chars: int,
+    ) -> dict[str, Any]:
+        if self.read_resource_handler is None:
+            raise RuntimeError("source resource reads are not configured")
+        return await self.read_resource_handler(source_id, uri, max_chars)

src/wf_api/runtime_dependencies.py

@@ -117,16 +117,16 @@ def _resolve_reducers(
     sources: dict[str, CapabilitySource],
 ) -> dict[str, ReducerDefinition]:
     reducers: dict[str, ReducerDefinition] = {}
-    if deployment is None:
-        return reducers
+    bindings = deployment.binding_map() if deployment is not None else {}
 
     for logical_ref, required in required_capabilities.items():
         if required.kind != "reducer":
             continue
-        bound_source_id = deployment.binding_map().get(required.logical_source)
-        if bound_source_id is None:
-            continue
-        source = sources.get(bound_source_id)
+        source = _find_source_for_reducer(
+            logical_source=required.logical_source,
+            bindings=bindings,
+            sources=sources,
+        )
         if source is None:
             continue
         definition = _find_reducer_definition(
@@ -138,6 +138,27 @@ def _resolve_reducers(
     return reducers
 
 
+def _find_source_for_reducer(
+    *,
+    logical_source: str,
+    bindings: dict[str, str],
+    sources: dict[str, CapabilitySource],
+) -> CapabilitySource | None:
+    """Find the source that owns a reducer for a logical source.
+
+    Platform sources resolve by exact ``logical_source`` without a deployment
+    binding.  External sources require a binding entry.
+    """
+    platform_source = sources.get(logical_source)
+    if platform_source is not None and platform_source.policy.platform:
+        return platform_source
+
+    bound_source_id = bindings.get(logical_source)
+    if bound_source_id is None:
+        return None
+    return sources.get(bound_source_id)
+
+
 def _find_reducer_definition(
     *,
     source: CapabilitySource,

src/wf_api/source_helpers.py

@@ -0,0 +1,61 @@
+from __future__ import annotations
+
+from typing import cast
+
+from pydantic import BaseModel
+
+from wf_core import RuntimeContext
+
+from .platform_context import WorkflowPlatformContext
+from .source_refs import SourceResourceRef
+
+
+class ReadResourceOutput(BaseModel):
+    """Bounded resource read result suitable for workflow state/output."""
+
+    source_id: str
+    uri: str
+    mime_type: str | None = None
+    text: str | None = None
+    content_count: int
+    truncated: bool = False
+
+
+async def read_resource(
+    ref: SourceResourceRef,
+    ctx: RuntimeContext,
+    *,
+    max_chars: int = 4000,
+) -> ReadResourceOutput:
+    """Explicitly dereference one source resource ref through platform context."""
+    platform = ctx.platform
+    if platform is None:
+        raise RuntimeError("wf.source.read_resource requires platform context")
+    typed_platform = cast(WorkflowPlatformContext, platform)
+    source_id = typed_platform.resolve_source(ref.logical_source)
+    payload = await typed_platform.read_resource(
+        source_id=source_id,
+        uri=ref.uri,
+        max_chars=max_chars,
+    )
+    contents = payload.get("contents", [])
+    first = contents[0] if isinstance(contents, list) and contents else {}
+    text = first.get("text") if isinstance(first, dict) else None
+    if isinstance(text, str) and len(text) > max_chars:
+        text = text[:max_chars]
+        truncated = True
+    else:
+        truncated = False
+    mime_type: str | None = None
+    if isinstance(first, dict) and isinstance(first.get("mimeType"), str):
+        mime_type = first["mimeType"]
+    elif ref.mime_type is not None:
+        mime_type = ref.mime_type
+    return ReadResourceOutput(
+        source_id=source_id,
+        uri=ref.uri,
+        mime_type=mime_type,
+        text=text if isinstance(text, str) else None,
+        content_count=len(contents) if isinstance(contents, list) else 0,
+        truncated=truncated,
+    )

src/wf_api/source_refs.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+from typing import Literal
+
+from pydantic import BaseModel, Field
+
+
+class SourceResourceRef(BaseModel):
+    """Workflow-safe resource handle.
+
+    The ref is inert pass-by-value data. Only explicit source-aware helper nodes
+    dereference it through deployment source bindings and platform context.
+    """
+
+    kind: Literal["source_resource_ref"] = "source_resource_ref"
+    logical_source: str = Field(min_length=1)
+    uri: str = Field(min_length=1)
+    mime_type: str | None = None
+    name: str | None = None