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

feat: type source discovery results

Browse Source
This commit is contained in:
lda
2026-07-31 16:55:00 +07:00 Verified
parent 82f764f7a6
commit 3891a62138
10 changed files with 401 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ISSUES.md
+4 -4
View File
@@ -72,10 +72,10 @@
complete OpenRPC document for all 70 registered methods. Request payloads
retain useful Pydantic schemas, so OpenRPC is a viable transport input.
- Typed-result slices now give `workflow.health`, all artifact, deployment,
and run operations, every persisted draft-workspace operation, and the
capability discovery/call surface named transport-neutral result schemas:
50 of 70 methods. The remaining 20 success results still collapse to generic
objects across stateless draft patch/validate, source discovery, and
and run operations, every persisted draft-workspace operation, and both the
capability and source-discovery surfaces named transport-neutral result
schemas: 53 of 70 methods. The remaining 17 success results still collapse
to generic objects across stateless draft patch/validate and the
source-registry/admin operations. Continue introducing operation result DTOs
before adopting generated TypeScript contracts.
- The stock `@open-rpc/generator` TypeScript client is not suitable here. It
src/wf_api/models/__init__.py
+34
View File
@@ -74,6 +74,24 @@ from .runs import (
TraceEntryPayload,
WorkflowRefPayload,
)
from .sources import (
InspectSourceResult,
ListSourcesResult,
NodeSpecInventoryPayload,
ReducerInventoryPayload,
SourceAuthDiagnosisPayload,
SourceCapabilityHasMorePayload,
SourceCapabilityInventoryPayload,
SourceCapabilityPreviewPayload,
SourceCatalogDiagnosisPayload,
SourceDiagnosisResult,
SourceDiagnosticsUnavailablePayload,
SourcePermissionsPayload,
SourcePolicyPayload,
SourceStatusPayload,
SourceTransportDiagnosisPayload,
SourceVisibilityPayload,
)
__all__ = [
"ArtifactVersionPayload",
@@ -102,6 +120,7 @@ __all__ = [
"InterruptRoutePayload",
"InvalidDraftResult",
"InspectCapabilityResult",
"InspectSourceResult",
"JsonObject",
"JsonProjector",
"JsonSchema",
@@ -110,12 +129,15 @@ __all__ = [
"ListArtifactsResult",
"ListCapabilitiesResult",
"ListRunsResult",
"ListSourcesResult",
"NextActionPatchExamplePayload",
"NextActionsPayload",
"NodeSpecCapabilityDetail",
"NodeSpecCapabilitySummary",
"NodeSpecInventoryPayload",
"PageMetadataPayload",
"RawWorkflowPlan",
"ReducerInventoryPayload",
"ResumeReadiness",
"RunResult",
"RunStatus",
@@ -126,6 +148,18 @@ __all__ = [
"SavedDraftArtifactResult",
"SaveDeploymentResult",
"SourceBindingPayload",
"SourceAuthDiagnosisPayload",
"SourceCapabilityHasMorePayload",
"SourceCapabilityInventoryPayload",
"SourceCapabilityPreviewPayload",
"SourceCatalogDiagnosisPayload",
"SourceDiagnosisResult",
"SourceDiagnosticsUnavailablePayload",
"SourcePermissionsPayload",
"SourcePolicyPayload",
"SourceStatusPayload",
"SourceTransportDiagnosisPayload",
"SourceVisibilityPayload",
"TraceRange",
"TraceEntryPayload",
"UnsavedDraftArtifactResult",
src/wf_api/models/sources.py
+171
View File
@@ -0,0 +1,171 @@
from __future__ import annotations
from typing import Literal, NotRequired, TypedDict
from pydantic import ConfigDict, with_config
from .artifacts import CapabilityRefPayload
from .common import (
DependencyDiagnosticPayload,
JsonObject,
PageMetadataPayload,
)
class SourceVisibilityPayload(TypedDict):
"""JSON projection of source visibility flags."""
planner: bool
client: bool
admin_dashboard: bool
class SourcePermissionsPayload(TypedDict):
"""JSON projection of source permission flags."""
safe_for_workflow: bool
calls_upstream: bool
mutates_config: bool
mutates_auth: bool
class SourcePolicyPayload(TypedDict):
"""JSON projection of deployment-binding policy for one source."""
platform: bool
binding_required: bool
class SourceCapabilityPreviewPayload(TypedDict):
"""Small sorted capability-name sample used by compact source rows."""
tools: list[str]
node_specs: list[str]
reducers: list[str]
prompts: list[str]
resources: list[str]
class SourceCapabilityHasMorePayload(TypedDict):
"""Whether each compact capability preview omitted owned names."""
tools: bool
node_specs: bool
reducers: bool
prompts: bool
resources: bool
class NodeSpecInventoryPayload(TypedDict):
"""Serializable executable contract owned by one capability source."""
name: str
description: str | None
outcomes: list[str]
input_schema: JsonObject
output_schema: JsonObject
is_async: bool
accepts_context: bool
class ReducerInventoryPayload(TypedDict):
"""Serializable pure-reducer contract owned by one capability source."""
name: str
ref: CapabilityRefPayload
description: str | None
config_schema: JsonObject
class SourceCapabilityInventoryPayload(TypedDict):
"""Capability names and detailed executable contracts owned by a source."""
tools: list[str]
node_specs: list[str]
node_spec_details: list[NodeSpecInventoryPayload]
reducers: list[str]
reducer_details: list[ReducerInventoryPayload]
prompts: list[str]
resources: list[str]
class SourceStatusPayload(TypedDict):
"""Compact source metadata returned by source discovery."""
id: str
kind: Literal["system", "connection", "python"]
enabled: bool
visibility: SourceVisibilityPayload
permissions: SourcePermissionsPayload
policy: SourcePolicyPayload
description: str | None
tool_count: int
node_spec_count: int
reducer_count: int
prompt_count: int
resource_count: int
preview: SourceCapabilityPreviewPayload
has_more: SourceCapabilityHasMorePayload
class ListSourcesResult(PageMetadataPayload):
"""Cursor-paged compact source discovery result."""
sources: list[SourceStatusPayload]
class SourceTransportDiagnosisPayload(TypedDict):
"""Known transport-health fields reported by a diagnostics provider."""
kind: NotRequired[str | None]
configured: NotRequired[bool]
class SourceAuthDiagnosisPayload(TypedDict):
"""Known auth-health fields reported by a diagnostics provider."""
auth_ref: NotRequired[str | None]
record_present: NotRequired[bool | None]
scheme: NotRequired[str | None]
transport_supported: NotRequired[bool]
class SourceCatalogDiagnosisPayload(TypedDict):
"""Known catalog-health fields reported by a diagnostics provider."""
has_snapshot: NotRequired[bool]
fetched_at_epoch_ms: NotRequired[int | None]
max_age_seconds: NotRequired[int | None]
node_count: NotRequired[int]
resource_count: NotRequired[int]
prompt_count: NotRequired[int]
@with_config(ConfigDict(extra="allow"))
class SourceDiagnosisResult(TypedDict):
"""Stable source-diagnostics envelope with provider-specific extensions."""
source_id: str
status: str
diagnostics: list[DependencyDiagnosticPayload]
enabled: NotRequired[bool]
transport: NotRequired[SourceTransportDiagnosisPayload]
auth: NotRequired[SourceAuthDiagnosisPayload]
catalog: NotRequired[SourceCatalogDiagnosisPayload]
message: NotRequired[str]
class SourceDiagnosticsUnavailablePayload(TypedDict):
"""Fallback embedded in inspect results when diagnostics collection fails."""
status: Literal["error"]
message: str
class InspectSourceResult(SourceStatusPayload):
"""Full source inventory with optional diagnostics supplied by the runtime."""
capabilities: SourceCapabilityInventoryPayload
diagnostics: NotRequired[
SourceDiagnosisResult | SourceDiagnosticsUnavailablePayload
]
src/wf_api/source_admin.py
+45 -22
View File
@@ -1,14 +1,24 @@
from __future__ import annotations
import logging
from typing import Any, Protocol
from typing import Any, Protocol, cast
from wf_platform import page_items
from .models import (
InspectSourceResult,
JsonProjector,
ListSourcesResult,
SourceDiagnosisResult,
SourceDiagnosticsUnavailablePayload,
)
from .operation_context import WorkflowOperationContext
logger = logging.getLogger(__name__)
_PROJECT_SOURCE_DIAGNOSIS = JsonProjector(SourceDiagnosisResult)
_PROJECT_DIAGNOSTICS_UNAVAILABLE = JsonProjector(SourceDiagnosticsUnavailablePayload)
class WorkflowSourceDiagnosticsProvider(Protocol):
"""Optional source-specific diagnostics provider.
@@ -42,7 +52,7 @@ class WorkflowSourceAdminApi:
*,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
) -> ListSourcesResult:
summaries = [
source.as_status().model_dump(mode="json")
for source in sorted(
@@ -51,13 +61,18 @@ class WorkflowSourceAdminApi:
)
]
page = page_items(summaries, cursor=cursor, limit=limit)
return {
"sources": list(page.items),
"next_cursor": page.next_cursor,
"total": page.total,
}
# SourceStatus validated every row before model_dump produced these
# transport dictionaries.
return cast(
ListSourcesResult,
{
"sources": list(page.items),
"next_cursor": page.next_cursor,
"total": page.total,
},
)
async def inspect_source(self, *, source_id: str) -> dict[str, Any]:
async def inspect_source(self, *, source_id: str) -> InspectSourceResult:
try:
source = self.context.specs.capability_sources[source_id]
except KeyError as exc:
@@ -65,29 +80,37 @@ class WorkflowSourceAdminApi:
payload = source.as_inventory().model_dump(mode="json")
if self.diagnostics is not None:
try:
payload["diagnostics"] = self.diagnostics.diagnose_source(source_id)
payload["diagnostics"] = _PROJECT_SOURCE_DIAGNOSIS(
self.diagnostics.diagnose_source(source_id)
)
except Exception as exc:
logger.exception(
"Source diagnostics failed for source_id=%s: %s",
source_id,
exc,
)
payload["diagnostics"] = {
"status": "error",
"message": "Diagnostics unavailable",
}
return payload
payload["diagnostics"] = _PROJECT_DIAGNOSTICS_UNAVAILABLE(
{
"status": "error",
"message": "Diagnostics unavailable",
}
)
# SourceInventory validated the stable inventory before model_dump;
# only the optional provider diagnostics are projected above.
return cast(InspectSourceResult, payload)
async def diagnose_source(self, *, source_id: str) -> dict[str, Any]:
async def diagnose_source(self, *, source_id: str) -> SourceDiagnosisResult:
try:
self.context.specs.capability_sources[source_id]
except KeyError as exc:
raise KeyError(f"unknown source {source_id!r}") from exc
if self.diagnostics is None:
return {
"source_id": source_id,
"status": "unknown",
"diagnostics": [],
"message": "No source diagnostics provider is configured.",
}
return self.diagnostics.diagnose_source(source_id)
return _PROJECT_SOURCE_DIAGNOSIS(
{
"source_id": source_id,
"status": "unknown",
"diagnostics": [],
"message": "No source diagnostics provider is configured.",
}
)
return _PROJECT_SOURCE_DIAGNOSIS(self.diagnostics.diagnose_source(source_id))
src/wf_api/surface.py
+6 -3
View File
@@ -19,15 +19,18 @@ from .models import (
DeleteDraftWorkspaceResult,
DraftWorkspaceResult,
InspectCapabilityResult,
InspectSourceResult,
ListArtifactsResult,
ListCapabilitiesResult,
ListDeploymentsResult,
ListDraftWorkspacesResult,
ListRunsResult,
ListSourcesResult,
RunResult,
RunTraceResult,
SaveArtifactResult,
SaveDeploymentResult,
SourceDiagnosisResult,
ValidateDeploymentResult,
WorkflowArtifactPayload,
WorkflowDeploymentPayload,
@@ -496,19 +499,19 @@ class WorkflowSourceAdminSurface(Protocol):
*,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]: ...
) -> ListSourcesResult: ...
async def inspect_source(
self,
*,
source_id: str,
) -> dict[str, Any]: ...
) -> InspectSourceResult: ...
async def diagnose_source(
self,
*,
source_id: str,
) -> dict[str, Any]: ...
) -> SourceDiagnosisResult: ...
class WorkflowAdminSurface(Protocol):
src/wf_transport_rpc_http/client/sources.py
+29 -16
View File
@@ -1,6 +1,8 @@
from __future__ import annotations
from typing import Any
from typing import cast
from wf_api.models import InspectSourceResult, ListSourcesResult, SourceDiagnosisResult
from .base import RpcCaller
@@ -13,23 +15,34 @@ class RpcSourceAdminClientMixin:
*,
cursor: str | None = None,
limit: int = 50,
) -> dict[str, Any]:
return await self._call(
"workflow.sources.list",
{
"cursor": cursor,
"limit": limit,
},
) -> ListSourcesResult:
return cast(
ListSourcesResult,
await self._call(
"workflow.sources.list",
{
"cursor": cursor,
"limit": limit,
},
),
)
async def inspect_source(self: RpcCaller, *, source_id: str) -> dict[str, Any]:
return await self._call(
"workflow.sources.inspect",
{"source_id": source_id},
async def inspect_source(self: RpcCaller, *, source_id: str) -> InspectSourceResult:
return cast(
InspectSourceResult,
await self._call(
"workflow.sources.inspect",
{"source_id": source_id},
),
)
async def diagnose_source(self: RpcCaller, *, source_id: str) -> dict[str, Any]:
return await self._call(
"workflow.sources.diagnose",
{"source_id": source_id},
async def diagnose_source(
self: RpcCaller, *, source_id: str
) -> SourceDiagnosisResult:
return cast(
SourceDiagnosisResult,
await self._call(
"workflow.sources.diagnose",
{"source_id": source_id},
),
)
src/wf_transport_rpc_http/methods/sources.py
+8 -5
View File
@@ -1,9 +1,12 @@
from __future__ import annotations
"""Source discovery JSON-RPC method registration.
from typing import Any
Return annotations stay eagerly evaluated because fastapi-jsonrpc captures them
while registering nested handlers for response validation and OpenRPC output.
"""
import fastapi_jsonrpc as jsonrpc
from wf_api.models import InspectSourceResult, ListSourcesResult, SourceDiagnosisResult
from wf_server import WorkflowServer
from ..errors import WorkflowRpcError, raise_workflow_rpc_error
@@ -20,7 +23,7 @@ def register_methods(
@entrypoint.method(name="workflow.sources.list", errors=[WorkflowRpcError])
async def workflow_sources_list(
params: ListSourcesParams = RpcParams(),
) -> dict[str, Any]:
) -> ListSourcesResult:
try:
return await server.source_admin.list_sources(
cursor=params.cursor,
@@ -32,7 +35,7 @@ def register_methods(
@entrypoint.method(name="workflow.sources.inspect", errors=[WorkflowRpcError])
async def workflow_sources_inspect(
params: InspectSourceParams = RpcParams(),
) -> dict[str, Any]:
) -> InspectSourceResult:
try:
return await server.source_admin.inspect_source(source_id=params.source_id)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
@@ -41,7 +44,7 @@ def register_methods(
@entrypoint.method(name="workflow.sources.diagnose", errors=[WorkflowRpcError])
async def workflow_sources_diagnose(
params: DiagnoseSourceParams = RpcParams(),
) -> dict[str, Any]:
) -> SourceDiagnosisResult:
try:
return await server.source_admin.diagnose_source(source_id=params.source_id)
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
tests/wf_api/test_source_admin_api.py
+22
View File
@@ -243,6 +243,28 @@ def test_diagnose_source_uses_provider() -> None:
assert payload["auth"]["record_present"] is True
class _ExtendedDiagnostics:
def diagnose_source(self, source_id: str) -> dict[str, object]:
return {
"source_id": source_id,
"status": "degraded",
"diagnostics": [],
"provider_latency_ms": 12,
}
def test_diagnose_source_preserves_provider_extensions() -> None:
payload = asyncio.run(
_api_with_diagnostics(
_source("demo.personal"),
diagnostics=_ExtendedDiagnostics(),
).diagnose_source(source_id="demo.personal")
)
assert payload["status"] == "degraded"
assert dict(payload)["provider_latency_ms"] == 12
def test_diagnose_source_without_provider_returns_basic_status() -> None:
payload = asyncio.run(
_api_with_diagnostics(_source("demo.personal")).diagnose_source(
tests/wf_transport_rpc_http/test_app.py
+30
View File
@@ -163,6 +163,36 @@ async def test_rpc_health_and_capability_methods(tmp_path) -> None:
assert called["result"]["output"] == {"value": "hello direct rpc"}
async def test_rpc_source_discovery_preserves_inventory_contracts(tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
app = create_rpc_app(server)
transport = httpx.ASGITransport(app=app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as client:
listed = await _rpc(client, "workflow.sources.list", {"limit": 10})
inspected = await _rpc(
client,
"workflow.sources.inspect",
{"source_id": "wf.std"},
)
diagnosed = await _rpc(
client,
"workflow.sources.diagnose",
{"source_id": "wf.std"},
)
source = next(row for row in listed["result"]["sources"] if row["id"] == "wf.std")
assert source["kind"] == "system"
assert source["preview"]["node_specs"]
inventory = inspected["result"]["capabilities"]
assert inventory["node_spec_details"][0]["input_schema"]["type"] == "object"
assert inventory["reducer_details"][0]["ref"] == {
"source": "wf.std",
"capability_key": "add",
}
assert diagnosed["result"]["source_id"] == "wf.std"
assert diagnosed["result"]["status"] == "unknown"
async def test_rpc_capability_methods_preserve_saved_wrapper_fields(tmp_path) -> None:
server = build_local_static_workflow_server(tmp_path / "store")
await server.api.create_artifact_from_plan(
tests/wf_transport_rpc_http/test_openrpc_contract.py
+52
View File
@@ -117,6 +117,58 @@ def test_openrpc_exposes_typed_capability_call_result(
)
def test_openrpc_exposes_typed_source_list_result(
openrpc_document: dict[str, Any],
) -> None:
_assert_result_component(
openrpc_document,
method_name="workflow.sources.list",
component_name="ListSourcesResult",
properties={"sources", "next_cursor", "total"},
)
sources = openrpc_document["components"]["schemas"]["ListSourcesResult"][
"properties"
]["sources"]
assert sources["items"] == {"$ref": "#/components/schemas/SourceStatusPayload"}
assert set(
openrpc_document["components"]["schemas"]["ListSourcesResult"]["required"]
) == {"sources", "next_cursor", "total"}
def test_openrpc_exposes_typed_source_inspect_result(
openrpc_document: dict[str, Any],
) -> None:
_assert_result_component(
openrpc_document,
method_name="workflow.sources.inspect",
component_name="InspectSourceResult",
properties={"id", "kind", "capabilities"},
)
inspected = openrpc_document["components"]["schemas"]["InspectSourceResult"]
assert inspected["properties"]["capabilities"] == {
"$ref": "#/components/schemas/SourceCapabilityInventoryPayload"
}
assert inspected["properties"]["diagnostics"]["anyOf"][:2] == [
{"$ref": "#/components/schemas/SourceDiagnosisResult"},
{"$ref": "#/components/schemas/SourceDiagnosticsUnavailablePayload"},
]
assert {"id", "kind", "capabilities"} <= set(inspected["required"])
def test_openrpc_exposes_typed_source_diagnosis_result(
openrpc_document: dict[str, Any],
) -> None:
_assert_result_component(
openrpc_document,
method_name="workflow.sources.diagnose",
component_name="SourceDiagnosisResult",
properties={"source_id", "status", "diagnostics"},
)
diagnosed = openrpc_document["components"]["schemas"]["SourceDiagnosisResult"]
assert set(diagnosed["required"]) == {"source_id", "status", "diagnostics"}
assert diagnosed["additionalProperties"] is True
@pytest.mark.parametrize(
("method_name", "component_name", "properties"),
[