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

feat: type admin operation results

Browse Source
This commit is contained in:
lda
2026-08-01 09:31:27 +07:00 Verified
parent 18b9002d17
commit 26eb00aa10
9 changed files with 391 additions and 67 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ISSUES.md
+5 -7
View File
@@ -71,13 +71,11 @@
- A 2026-07-30 spike confirmed that `fastapi-jsonrpc` already exports a
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 both the
capability and source-discovery surfaces named transport-neutral result
schemas: 63 of 70 methods. The remaining seven success results still
collapse to generic objects across connections, events, and auth admin
operations. Continue introducing operation result DTOs before adopting
generated TypeScript contracts.
- All 70 JSON-RPC methods now expose named transport-neutral success-result
schemas, including connections, events, and secret-safe auth admin results.
No success result collapses to a generic object. The next contract-parity
step can consume OpenRPC through a small transport-neutral manifest rather
than adding more Python result DTOs.
- The stock `@open-rpc/generator` TypeScript client is not suitable here. It
exhausted a 4 GB Node heap on the full contract and emitted invalid dotted
class members plus `any` results for a minimal `workflow.health` contract.
src/wf_api/admin.py
+57 -18
View File
@@ -5,6 +5,28 @@ from dataclasses import asdict, is_dataclass
from typing import Any, Protocol
from wf_api.auth import AuthRecord
from wf_api.models import (
AdminEventPayload,
AuthRecordSummaryPayload,
ConnectionPayload,
ConnectionStatusPayload,
DeleteAuthRecordResult,
JsonProjector,
ListAdminEventsResult,
ListAuthRecordsResult,
ListConnectionsResult,
ListConnectionStatusesResult,
)
_project_connection = JsonProjector(ConnectionPayload)
_project_connection_status = JsonProjector(ConnectionStatusPayload)
_project_event = JsonProjector(AdminEventPayload)
_project_auth_record = JsonProjector(AuthRecordSummaryPayload)
_project_delete_auth = JsonProjector(DeleteAuthRecordResult)
_project_connections_result = JsonProjector(ListConnectionsResult)
_project_connection_statuses_result = JsonProjector(ListConnectionStatusesResult)
_project_events_result = JsonProjector(ListAdminEventsResult)
_project_auth_records_result = JsonProjector(ListAuthRecordsResult)
class WorkflowAdminConnectionProvider(Protocol):
@@ -51,39 +73,56 @@ class WorkflowAdminApi:
self.events = events
self.auth = auth
async def list_connections(self) -> dict[str, Any]:
async def list_connections(self) -> ListConnectionsResult:
connections = sorted(
(_payload(item) for item in self.connections.list_connections()),
(
_project_connection(_payload(item))
for item in self.connections.list_connections()
),
key=lambda item: str(item.get("id", item.get("connection_id", ""))),
)
return {"connections": connections, "total": len(connections)}
return _project_connections_result(
{"connections": connections, "total": len(connections)}
)
async def get_connection_statuses(self) -> dict[str, Any]:
async def get_connection_statuses(self) -> ListConnectionStatusesResult:
statuses = sorted(
(_payload(item) for item in self.connections.get_connection_statuses()),
(
_project_connection_status(_payload(item))
for item in self.connections.get_connection_statuses()
),
key=lambda item: str(item.get("connection_id", item.get("id", ""))),
)
return {"statuses": statuses, "total": len(statuses)}
return _project_connection_statuses_result(
{"statuses": statuses, "total": len(statuses)}
)
# Preserve provider order for events; event providers are expected to return
# chronological order and callers may rely on that ordering for diagnostics.
async def list_events(self) -> dict[str, Any]:
events = [_payload(event) for event in self.events.list_events()]
return {"events": events, "total": len(events)}
async def list_events(self) -> ListAdminEventsResult:
events = [
_project_event(_payload(event)) for event in self.events.list_events()
]
return _project_events_result({"events": events, "total": len(events)})
async def list_auth_records(self) -> dict[str, Any]:
async def list_auth_records(self) -> ListAuthRecordsResult:
if self.auth is None:
raise RuntimeError("auth admin is not available for this target")
records = sorted(
(_payload(item) for item in self.auth.list_auth_records()),
(
_project_auth_record(_payload(item))
for item in self.auth.list_auth_records()
),
key=lambda item: str(item.get("id", "")),
)
return {"auth_records": records, "total": len(records)}
return _project_auth_records_result(
{"auth_records": records, "total": len(records)}
)
async def inspect_auth_record(self, auth_ref: str) -> dict[str, Any]:
async def inspect_auth_record(self, auth_ref: str) -> AuthRecordSummaryPayload:
if self.auth is None:
raise RuntimeError("auth admin is not available for this target")
return _payload(self.auth.inspect_auth_record(auth_ref))
return _project_auth_record(_payload(self.auth.inspect_auth_record(auth_ref)))
async def save_auth_record(
self,
@@ -92,7 +131,7 @@ class WorkflowAdminApi:
scheme: str,
payload: Mapping[str, object],
metadata: Mapping[str, object] | None = None,
) -> dict[str, Any]:
) -> AuthRecordSummaryPayload:
if self.auth is None:
raise RuntimeError("auth admin is not available for this target")
record = AuthRecord(
@@ -101,12 +140,12 @@ class WorkflowAdminApi:
payload=dict(payload),
metadata=dict(metadata or {}),
)
return _payload(self.auth.save_auth_record(record))
return _project_auth_record(_payload(self.auth.save_auth_record(record)))
async def delete_auth_record(self, auth_ref: str) -> dict[str, Any]:
async def delete_auth_record(self, auth_ref: str) -> DeleteAuthRecordResult:
if self.auth is None:
raise RuntimeError("auth admin is not available for this target")
return _payload(self.auth.delete_auth_record(auth_ref))
return _project_delete_auth(_payload(self.auth.delete_auth_record(auth_ref)))
def _payload(value: Mapping[str, Any] | object) -> dict[str, Any]:
src/wf_api/models/__init__.py
+20
View File
@@ -1,5 +1,16 @@
"""Transport-neutral workflow API models."""
from .admin import (
AdminEventPayload,
AuthRecordSummaryPayload,
ConnectionPayload,
ConnectionStatusPayload,
DeleteAuthRecordResult,
ListAdminEventsResult,
ListAuthRecordsResult,
ListConnectionsResult,
ListConnectionStatusesResult,
)
from .artifacts import (
ArtifactCatalogEntryPayload,
ArtifactKindPayload,
@@ -108,8 +119,10 @@ from .sources import (
)
__all__ = [
"AdminEventPayload",
"ApplyRegistryChangesResult",
"ArtifactVersionPayload",
"AuthRecordSummaryPayload",
"ArtifactCatalogEntryPayload",
"ArtifactKindPayload",
"CapabilityKindPayload",
@@ -119,9 +132,12 @@ __all__ = [
"CapabilitySummaryPayload",
"CompileDraftWorkspaceResult",
"CompileDraftWorkspaceSuccess",
"ConnectionPayload",
"ConnectionStatusPayload",
"CreateArtifactFromWorkspaceResult",
"CreateDraftWorkspaceFromCapabilityResult",
"DeleteArtifactResult",
"DeleteAuthRecordResult",
"DeleteDeploymentResult",
"DeleteDraftWorkspaceResult",
"DependencyDiagnosticPayload",
@@ -141,6 +157,10 @@ __all__ = [
"JsonProjector",
"JsonSchema",
"ListDeploymentsResult",
"ListAdminEventsResult",
"ListAuthRecordsResult",
"ListConnectionsResult",
"ListConnectionStatusesResult",
"ListDraftWorkspacesResult",
"ListArtifactsResult",
"ListCapabilitiesResult",
src/wf_api/models/admin.py
+97
View File
@@ -0,0 +1,97 @@
from __future__ import annotations
from typing import NotRequired, TypedDict
from pydantic import ConfigDict, with_config
from .common import JsonObject
@with_config(ConfigDict(extra="allow"))
class ConnectionPayload(TypedDict):
"""Configured connection inventory row with provider extensions preserved."""
id: str
server: str
account: str
enabled: bool
metadata: JsonObject
source_config_ownership: NotRequired[str]
class ListConnectionsResult(TypedDict):
"""Sorted configured connection inventory."""
connections: list[ConnectionPayload]
total: int
@with_config(ConfigDict(extra="allow"))
class ConnectionStatusPayload(TypedDict):
"""Connection readiness row with optional catalog snapshot facts."""
connection_id: str
enabled: bool
server: NotRequired[str]
account: NotRequired[str]
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]
class ListConnectionStatusesResult(TypedDict):
"""Sorted connection readiness inventory."""
statuses: list[ConnectionStatusPayload]
total: int
@with_config(ConfigDict(extra="allow"))
class AdminEventPayload(TypedDict):
"""Chronological platform event with provider-specific payload preserved."""
kind: str
timestamp_epoch_ms: int
connection_id: NotRequired[str | None]
capability_id: NotRequired[str | None]
workflow_name: NotRequired[str | None]
payload: JsonObject
class ListAdminEventsResult(TypedDict):
"""Chronological platform event history."""
events: list[AdminEventPayload]
total: int
@with_config(ConfigDict(extra="forbid"))
class AuthRecordSummaryPayload(TypedDict):
"""Auth record summary without credential payload values.
``metadata`` is explicitly non-secret display data. Credential material
belongs in the omitted auth payload and is represented only by key names.
"""
id: str
scheme: str
metadata: JsonObject
payload_keys: list[str]
class ListAuthRecordsResult(TypedDict):
"""Sorted secret-safe auth record inventory."""
auth_records: list[AuthRecordSummaryPayload]
total: int
@with_config(ConfigDict(extra="forbid"))
class DeleteAuthRecordResult(TypedDict):
"""Outcome of deleting one auth record."""
deleted: bool
id: str
src/wf_api/surface.py
+13 -7
View File
@@ -11,19 +11,25 @@ from .draft_authoring import RouteSource
from .draft_updates import CapabilityStepUpdate
from .models import (
ApplyRegistryChangesResult,
AuthRecordSummaryPayload,
CapabilityCallResult,
CompileDraftWorkspaceResult,
CreateArtifactFromWorkspaceResult,
CreateDraftWorkspaceFromCapabilityResult,
DeleteArtifactResult,
DeleteAuthRecordResult,
DeleteDeploymentResult,
DeleteDraftWorkspaceResult,
DraftWorkspaceResult,
InspectCapabilityResult,
InspectRegistryEntryResult,
InspectSourceResult,
ListAdminEventsResult,
ListArtifactsResult,
ListAuthRecordsResult,
ListCapabilitiesResult,
ListConnectionsResult,
ListConnectionStatusesResult,
ListDeploymentsResult,
ListDraftWorkspacesResult,
ListRegistryEntriesResult,
@@ -537,15 +543,15 @@ class WorkflowSourceAdminSurface(Protocol):
class WorkflowAdminSurface(Protocol):
"""Read-only connection/config admin methods exposed by platform frontends."""
async def list_connections(self) -> dict[str, Any]: ...
async def list_connections(self) -> ListConnectionsResult: ...
async def get_connection_statuses(self) -> dict[str, Any]: ...
async def get_connection_statuses(self) -> ListConnectionStatusesResult: ...
async def list_events(self) -> dict[str, Any]: ...
async def list_events(self) -> ListAdminEventsResult: ...
async def list_auth_records(self) -> dict[str, Any]: ...
async def list_auth_records(self) -> ListAuthRecordsResult: ...
async def inspect_auth_record(self, auth_ref: str) -> dict[str, Any]: ...
async def inspect_auth_record(self, auth_ref: str) -> AuthRecordSummaryPayload: ...
async def save_auth_record(
self,
@@ -554,9 +560,9 @@ class WorkflowAdminSurface(Protocol):
scheme: str,
payload: Mapping[str, object],
metadata: Mapping[str, object] | None = None,
) -> dict[str, Any]: ...
) -> AuthRecordSummaryPayload: ...
async def delete_auth_record(self, auth_ref: str) -> dict[str, Any]: ...
async def delete_auth_record(self, auth_ref: str) -> DeleteAuthRecordResult: ...
class WorkflowSourceRegistrySurface(Protocol):
src/wf_transport_rpc_http/client/admin.py
+49 -15
View File
@@ -1,7 +1,16 @@
from __future__ import annotations
from collections.abc import Mapping
from typing import Any
from typing import cast
from wf_api.models import (
AuthRecordSummaryPayload,
DeleteAuthRecordResult,
ListAdminEventsResult,
ListAuthRecordsResult,
ListConnectionsResult,
ListConnectionStatusesResult,
)
from .base import RpcCaller
@@ -9,22 +18,39 @@ from .base import RpcCaller
class RpcAdminClientMixin:
"""JSON-RPC implementation of read-only admin/config surface methods."""
async def list_connections(self: RpcCaller) -> dict[str, Any]:
return await self._call("workflow.admin.connections.list", {})
async def list_connections(self: RpcCaller) -> ListConnectionsResult:
return cast(
ListConnectionsResult,
await self._call("workflow.admin.connections.list", {}),
)
async def get_connection_statuses(self: RpcCaller) -> dict[str, Any]:
return await self._call("workflow.admin.connection_statuses.list", {})
async def get_connection_statuses(self: RpcCaller) -> ListConnectionStatusesResult:
return cast(
ListConnectionStatusesResult,
await self._call("workflow.admin.connection_statuses.list", {}),
)
async def list_events(self: RpcCaller) -> dict[str, Any]:
return await self._call("workflow.admin.events.list", {})
async def list_events(self: RpcCaller) -> ListAdminEventsResult:
return cast(
ListAdminEventsResult,
await self._call("workflow.admin.events.list", {}),
)
async def list_auth_records(self: RpcCaller) -> dict[str, Any]:
return await self._call("workflow.admin.auth.list", {})
async def list_auth_records(self: RpcCaller) -> ListAuthRecordsResult:
return cast(
ListAuthRecordsResult,
await self._call("workflow.admin.auth.list", {}),
)
async def inspect_auth_record(self: RpcCaller, auth_ref: str) -> dict[str, Any]:
return await self._call(
async def inspect_auth_record(
self: RpcCaller, auth_ref: str
) -> AuthRecordSummaryPayload:
return cast(
AuthRecordSummaryPayload,
await self._call(
"workflow.admin.auth.inspect",
{"auth_ref": auth_ref},
),
)
async def save_auth_record(
@@ -34,8 +60,10 @@ class RpcAdminClientMixin:
scheme: str,
payload: Mapping[str, object],
metadata: Mapping[str, object] | None = None,
) -> dict[str, Any]:
return await self._call(
) -> AuthRecordSummaryPayload:
return cast(
AuthRecordSummaryPayload,
await self._call(
"workflow.admin.auth.save",
{
"auth_ref": auth_ref,
@@ -43,10 +71,16 @@ class RpcAdminClientMixin:
"payload": dict(payload),
"metadata": dict(metadata or {}),
},
),
)
async def delete_auth_record(self: RpcCaller, auth_ref: str) -> dict[str, Any]:
return await self._call(
async def delete_auth_record(
self: RpcCaller, auth_ref: str
) -> DeleteAuthRecordResult:
return cast(
DeleteAuthRecordResult,
await self._call(
"workflow.admin.auth.delete",
{"auth_ref": auth_ref},
),
)
src/wf_transport_rpc_http/methods/admin.py
+19 -9
View File
@@ -1,9 +1,19 @@
from __future__ import annotations
"""Admin 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 (
AuthRecordSummaryPayload,
DeleteAuthRecordResult,
ListAdminEventsResult,
ListAuthRecordsResult,
ListConnectionsResult,
ListConnectionStatusesResult,
)
from wf_server import WorkflowServer
from ..errors import WorkflowRpcError, raise_workflow_rpc_error
@@ -28,7 +38,7 @@ def register_methods(
)
async def workflow_admin_connections_list(
params: AdminEmptyParams = RpcParams(),
) -> dict[str, Any]:
) -> ListConnectionsResult:
try:
return await server.admin.list_connections()
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
@@ -40,7 +50,7 @@ def register_methods(
)
async def workflow_admin_connection_statuses_list(
params: AdminEmptyParams = RpcParams(),
) -> dict[str, Any]:
) -> ListConnectionStatusesResult:
try:
return await server.admin.get_connection_statuses()
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
@@ -52,7 +62,7 @@ def register_methods(
)
async def workflow_admin_events_list(
params: AdminEmptyParams = RpcParams(),
) -> dict[str, Any]:
) -> ListAdminEventsResult:
try:
return await server.admin.list_events()
except (ValueError, KeyError, LookupError, FileNotFoundError) as exc:
@@ -64,7 +74,7 @@ def register_methods(
)
async def workflow_admin_auth_list(
params: AdminEmptyParams = RpcParams(),
) -> dict[str, Any]:
) -> ListAuthRecordsResult:
try:
return await server.admin.list_auth_records()
except (
@@ -82,7 +92,7 @@ def register_methods(
)
async def workflow_admin_auth_inspect(
params: InspectAuthParams = RpcParams(),
) -> dict[str, Any]:
) -> AuthRecordSummaryPayload:
try:
return await server.admin.inspect_auth_record(params.auth_ref)
except (
@@ -100,7 +110,7 @@ def register_methods(
)
async def workflow_admin_auth_save(
params: SaveAuthParams = RpcParams(),
) -> dict[str, Any]:
) -> AuthRecordSummaryPayload:
try:
return await server.admin.save_auth_record(
auth_ref=params.auth_ref,
@@ -123,7 +133,7 @@ def register_methods(
)
async def workflow_admin_auth_delete(
params: DeleteAuthParams = RpcParams(),
) -> dict[str, Any]:
) -> DeleteAuthRecordResult:
try:
return await server.admin.delete_auth_record(params.auth_ref)
except (
tests/wf_api/test_admin_api.py
+13
View File
@@ -4,6 +4,7 @@ from dataclasses import dataclass, field
from typing import Any
import pytest
from pydantic import ValidationError
from wf_api import WorkflowAdminApi, WorkflowAdminSurface
from wf_api.auth import AuthRecord
@@ -156,6 +157,18 @@ async def test_admin_inspects_auth_record_without_payload_values() -> None:
}
async def test_admin_rejects_auth_provider_payload_values() -> None:
class UnsafeAuthProvider(AuthProvider):
def inspect_auth_record(self, auth_ref: str) -> dict[str, Any]:
return {
**super().inspect_auth_record(auth_ref),
"payload": {"token": "secret"},
}
with pytest.raises(ValidationError, match="payload"):
await _api(UnsafeAuthProvider()).inspect_auth_record("github.work")
async def test_admin_auth_methods_report_unavailable_without_provider() -> None:
with pytest.raises(RuntimeError, match="auth admin is not available"):
await _api().list_auth_records()
tests/wf_transport_rpc_http/test_openrpc_contract.py
+107
View File
@@ -284,6 +284,113 @@ def test_openrpc_exposes_typed_source_registry_remove_and_apply_results(
] == {"$ref": "#/components/schemas/DependencyDiagnosticPayload"}
def test_openrpc_exposes_typed_admin_inventory_results(
openrpc_document: dict[str, Any],
) -> None:
for method_name, component_name, collection_name, item_name in [
(
"workflow.admin.connections.list",
"ListConnectionsResult",
"connections",
"ConnectionPayload",
),
(
"workflow.admin.connection_statuses.list",
"ListConnectionStatusesResult",
"statuses",
"ConnectionStatusPayload",
),
(
"workflow.admin.events.list",
"ListAdminEventsResult",
"events",
"AdminEventPayload",
),
(
"workflow.admin.auth.list",
"ListAuthRecordsResult",
"auth_records",
"AuthRecordSummaryPayload",
),
]:
_assert_result_component(
openrpc_document,
method_name=method_name,
component_name=component_name,
properties={collection_name, "total"},
)
schema = openrpc_document["components"]["schemas"][component_name]
assert schema["properties"][collection_name]["items"] == {
"$ref": f"#/components/schemas/{item_name}"
}
assert set(schema["required"]) == {collection_name, "total"}
schemas = openrpc_document["components"]["schemas"]
assert set(schemas["ConnectionPayload"]["required"]) == {
"id",
"server",
"account",
"enabled",
"metadata",
}
assert set(schemas["ConnectionStatusPayload"]["required"]) == {
"connection_id",
"enabled",
}
assert set(schemas["AdminEventPayload"]["required"]) == {
"kind",
"timestamp_epoch_ms",
"payload",
}
for component_name in [
"ConnectionPayload",
"ConnectionStatusPayload",
"AdminEventPayload",
]:
assert schemas[component_name]["additionalProperties"] is True
@pytest.mark.parametrize(
"method_name",
[
"workflow.admin.auth.inspect",
"workflow.admin.auth.save",
],
)
def test_openrpc_exposes_secret_safe_auth_record_results(
openrpc_document: dict[str, Any],
method_name: str,
) -> None:
_assert_result_component(
openrpc_document,
method_name=method_name,
component_name="AuthRecordSummaryPayload",
properties={"id", "scheme", "metadata", "payload_keys"},
)
schema = openrpc_document["components"]["schemas"]["AuthRecordSummaryPayload"]
assert set(schema["required"]) == {
"id",
"scheme",
"metadata",
"payload_keys",
}
assert "payload" not in schema["properties"]
assert schema["additionalProperties"] is False
def test_openrpc_exposes_typed_auth_delete_result(
openrpc_document: dict[str, Any],
) -> None:
_assert_result_component(
openrpc_document,
method_name="workflow.admin.auth.delete",
component_name="DeleteAuthRecordResult",
properties={"deleted", "id"},
)
schema = openrpc_document["components"]["schemas"]["DeleteAuthRecordResult"]
assert set(schema["required"]) == {"deleted", "id"}
@pytest.mark.parametrize(
("method_name", "component_name", "properties"),
[