api/dev/knowledge_fs_product_contract.py
Rama main · réplica de lectura del repositorio original.
"""Cross-service KnowledgeFS product-operation contract validation.
The checked product manifest is the reviewable boundary, while the Dify product registry, Dify
Capability issuer registry, exported KnowledgeFS OpenAPI, and exported TypeScript request-guard
policy remain executable sources of truth. A product operation must appear exactly once as ready
or as an explicit gap; internal KFS-only operations require a named exclusion.
"""
from __future__ import annotations
from collections.abc import Iterable, Mapping
from typing import Any, Literal, NamedTuple, TypedDict, cast
class ProductOperationLimits(TypedDict):
kfsMaxResponseBytes: int
productMaxRequestBytes: int
productMaxResponseBytes: int
class ProductOperationStream(TypedDict):
kfsResponseKind: str
productKind: str
class ProductOperationManifestEntry(TypedDict):
action: str
kfsOperationId: str
limits: ProductOperationLimits
method: str
path: str
productOperationId: str
resource: str
stream: ProductOperationStream
transport: str
type ProductOperationManifestField = Literal[
"action",
"kfsOperationId",
"limits",
"method",
"path",
"productOperationId",
"resource",
"stream",
"transport",
]
PRODUCT_OPERATION_MANIFEST_FIELDS: tuple[ProductOperationManifestField, ...] = (
"action",
"kfsOperationId",
"limits",
"method",
"path",
"productOperationId",
"resource",
"stream",
"transport",
)
class ProductOperationManifest(TypedDict):
operations: list[ProductOperationManifestEntry]
schemaVersion: int
class ProductOperationGapEntry(ProductOperationManifestEntry):
reason: str
reasonCode: str
replacementProductOperationIds: list[str]
class InternalKfsOperationExclusion(TypedDict):
kfsOperationId: str
reason: str
reasonCode: str
class ProductOperationGapManifest(TypedDict):
gaps: list[ProductOperationGapEntry]
internalKfsOperationExclusions: list[InternalKfsOperationExclusion]
schemaVersion: int
class ProductOperationRuntimeContract(NamedTuple):
action: str
kfs_operation_id: str
max_request_bytes: int
max_response_bytes: int
method: str
path: str
product_operation_id: str
ready: bool
resource: str
stream_kind: str
transport: str
class CapabilityOperationRuntimeContract(NamedTuple):
action: str
allowed_caller_kinds: tuple[str, ...]
method: str
operation_id: str
path: str
resource: str
class CapabilityOperationPolicy(NamedTuple):
action: str
allowed_caller_kinds: tuple[str, ...]
method: str
operation_id: str
parent_resource_binding: dict[str, object] | None
path: str
resource: str
resource_binding: dict[str, object]
def product_operation_runtime_contracts() -> tuple[ProductOperationRuntimeContract, ...]:
"""Read the complete typed Dify product registry without normalizing away gaps."""
from services.knowledge_fs.product_operations import (
KNOWLEDGE_FS_PRODUCT_OPERATIONS,
is_product_operation_ready,
)
contracts: list[ProductOperationRuntimeContract] = []
for product_operation_id, operation in KNOWLEDGE_FS_PRODUCT_OPERATIONS.items():
if operation.capability_operation_id is None or operation.kfs_path is None or operation.action is None:
raise ValueError(f"Dify product operation {product_operation_id} has no Capability contract")
contracts.append(
ProductOperationRuntimeContract(
action=operation.action,
kfs_operation_id=operation.capability_operation_id,
max_request_bytes=operation.max_request_bytes,
max_response_bytes=operation.max_response_bytes,
method=operation.method,
path=operation.kfs_path,
product_operation_id=product_operation_id,
ready=is_product_operation_ready(product_operation_id),
resource=operation.resource_resolver,
stream_kind=operation.stream_kind,
transport=operation.transport,
)
)
return tuple(contracts)
def capability_operation_runtime_contracts() -> tuple[CapabilityOperationRuntimeContract, ...]:
"""Read the Python issuer policy used to mint each operation Capability."""
from services.knowledge_fs_capability import KNOWLEDGE_FS_CAPABILITY_OPERATIONS
return tuple(
CapabilityOperationRuntimeContract(
action=operation.action,
allowed_caller_kinds=operation.allowed_caller_kinds,
method=operation.method,
operation_id=operation_id,
path=operation.path,
resource=operation.resource_type,
)
for operation_id, operation in KNOWLEDGE_FS_CAPABILITY_OPERATIONS.items()
)
def parse_product_operation_manifest(value: dict[str, Any]) -> ProductOperationManifest:
"""Parse the ready product-operation inventory with a closed schema."""
if set(value) != {"operations", "schemaVersion"} or value.get("schemaVersion") != 2:
raise ValueError("KnowledgeFS product operation manifest must use schemaVersion 2")
operations = value.get("operations")
if not isinstance(operations, list) or not operations:
raise ValueError("KnowledgeFS product operation manifest operations must be a non-empty list")
return {
"operations": [_parse_product_entry(entry, gap=False) for entry in operations],
"schemaVersion": 2,
}
def parse_product_operation_gap_manifest(value: dict[str, Any]) -> ProductOperationGapManifest:
"""Parse explicit product gaps and KFS-only operation exclusions with a closed schema."""
expected_fields = {"gaps", "internalKfsOperationExclusions", "schemaVersion"}
if set(value) != expected_fields or value.get("schemaVersion") != 2:
raise ValueError("KnowledgeFS product operation gap manifest must use schemaVersion 2")
gaps = value.get("gaps")
exclusions = value.get("internalKfsOperationExclusions")
if not isinstance(gaps, list) or not isinstance(exclusions, list):
raise ValueError("KnowledgeFS product operation gap manifest lists are invalid")
parsed_exclusions: list[InternalKfsOperationExclusion] = []
for raw in exclusions:
if not isinstance(raw, dict) or set(raw) != {"kfsOperationId", "reason", "reasonCode"}:
raise ValueError("KnowledgeFS internal KFS operation exclusion fields are invalid")
for field in ("kfsOperationId", "reason", "reasonCode"):
_required_string(raw.get(field), f"internal exclusion {field}")
parsed_exclusions.append(cast(InternalKfsOperationExclusion, raw))
return {
"gaps": [cast(ProductOperationGapEntry, _parse_product_entry(entry, gap=True)) for entry in gaps],
"internalKfsOperationExclusions": parsed_exclusions,
"schemaVersion": 2,
}
def parse_capability_operation_policy(value: dict[str, Any]) -> tuple[CapabilityOperationPolicy, ...]:
"""Parse the runtime-exported TypeScript request-guard registry."""
if set(value) != {"operations", "schemaVersion"} or value.get("schemaVersion") != 1:
raise ValueError("KnowledgeFS Capability operation policy must use schemaVersion 1")
operations = value.get("operations")
if not isinstance(operations, list) or not operations:
raise ValueError("KnowledgeFS Capability operation policy must contain operations")
parsed: list[CapabilityOperationPolicy] = []
for raw in operations:
expected_fields = {
"action",
"allowedCallerKinds",
"method",
"operationId",
"parentResourceBinding",
"path",
"resourceBinding",
"resourceType",
}
if not isinstance(raw, dict) or set(raw) != expected_fields:
raise ValueError("KnowledgeFS Capability operation policy fields are invalid")
callers = raw.get("allowedCallerKinds")
resource_binding = raw.get("resourceBinding")
parent_binding = raw.get("parentResourceBinding")
if (
not isinstance(callers, list)
or not callers
or not all(isinstance(caller, str) and caller for caller in callers)
or not isinstance(resource_binding, dict)
or not resource_binding
or (parent_binding is not None and not isinstance(parent_binding, dict))
):
raise ValueError("KnowledgeFS Capability operation request binding is invalid")
for field in ("action", "method", "operationId", "path", "resourceType"):
_required_string(raw.get(field), f"Capability operation {field}")
parsed.append(
CapabilityOperationPolicy(
action=cast(str, raw["action"]),
allowed_caller_kinds=tuple(callers),
method=cast(str, raw["method"]),
operation_id=cast(str, raw["operationId"]),
parent_resource_binding=cast(dict[str, object] | None, parent_binding),
path=cast(str, raw["path"]),
resource=cast(str, raw["resourceType"]),
resource_binding=cast(dict[str, object], resource_binding),
)
)
_unique((operation.operation_id for operation in parsed), "Capability operation policy")
return tuple(parsed)
def validate_product_operation_contracts(
*,
capability_operations: tuple[CapabilityOperationRuntimeContract, ...],
capability_policy: tuple[CapabilityOperationPolicy, ...],
document: dict[str, Any],
gap_manifest: ProductOperationGapManifest,
manifest: ProductOperationManifest,
product_operations: tuple[ProductOperationRuntimeContract, ...],
) -> None:
"""Require exact agreement across manifests, Dify registries, KFS guard policy, and OpenAPI."""
ready_entries = {entry["productOperationId"]: entry for entry in manifest["operations"]}
gap_entries = {entry["productOperationId"]: entry for entry in gap_manifest["gaps"]}
if len(ready_entries) != len(manifest["operations"]):
raise ValueError("KnowledgeFS product operation manifest contains duplicate product ids")
if len(gap_entries) != len(gap_manifest["gaps"]):
raise ValueError("KnowledgeFS product operation gap manifest contains duplicate product ids")
if set(ready_entries) & set(gap_entries):
raise ValueError("KnowledgeFS product operation cannot be both ready and a gap")
runtime_by_product = {operation.product_operation_id: operation for operation in product_operations}
if set(ready_entries) | set(gap_entries) != set(runtime_by_product):
missing = sorted(set(runtime_by_product) - set(ready_entries) - set(gap_entries))
extra = sorted((set(ready_entries) | set(gap_entries)) - set(runtime_by_product))
raise ValueError(f"KnowledgeFS product operation completeness drifted: missing={missing}, extra={extra}")
python_capabilities = {operation.operation_id: operation for operation in capability_operations}
guard_capabilities = {operation.operation_id: operation for operation in capability_policy}
openapi_operations = _openapi_operations(document)
for product_operation_id, runtime in runtime_by_product.items():
entry = ready_entries.get(product_operation_id) or gap_entries[product_operation_id]
if runtime.ready != (product_operation_id in ready_entries):
state = "ready" if runtime.ready else "gap"
raise ValueError(f"Dify product operation {product_operation_id} must be declared as {state}")
_validate_product_entry(entry, runtime, openapi_operations, python_capabilities, guard_capabilities)
if product_operation_id in gap_entries:
_required_string(gap_entries[product_operation_id]["reason"], "product gap reason")
_required_string(gap_entries[product_operation_id]["reasonCode"], "product gap reasonCode")
replacements = gap_entries[product_operation_id]["replacementProductOperationIds"]
if not replacements or any(replacement not in ready_entries for replacement in replacements):
raise ValueError(f"KnowledgeFS product gap {product_operation_id} has invalid replacements")
exclusions = gap_manifest["internalKfsOperationExclusions"]
exclusion_by_id = {entry["kfsOperationId"]: entry for entry in exclusions}
if len(exclusion_by_id) != len(exclusions):
raise ValueError("KnowledgeFS internal KFS operation exclusions contain duplicate ids")
lifecycle_operation_ids = {
"activateDifyWorkspaceIntegration",
"freezeDifyWorkspaceIntegration",
}
if set(exclusion_by_id) != lifecycle_operation_ids:
raise ValueError("KnowledgeFS Dify integration lifecycle operations must be explicit internal exclusions")
product_capability_ids = {operation.kfs_operation_id for operation in product_operations}
for operation_id in exclusion_by_id:
if operation_id in product_capability_ids:
raise ValueError(f"Internal KFS operation {operation_id} cannot also be a product operation")
_validate_capability_alignment(
operation_id,
openapi_operations,
python_capabilities,
guard_capabilities,
)
python_operation = python_capabilities[operation_id]
if python_operation.allowed_caller_kinds != ("internal_worker",) or python_operation.resource != "namespace":
raise ValueError("Dify integration lifecycle operations must remain internal-worker namespace-only")
def _parse_product_entry(raw: object, *, gap: bool) -> ProductOperationManifestEntry:
base_fields = {
"action",
"kfsOperationId",
"limits",
"method",
"path",
"productOperationId",
"resource",
"stream",
"transport",
}
expected_fields = base_fields | ({"reason", "reasonCode", "replacementProductOperationIds"} if gap else set())
if not isinstance(raw, dict) or set(raw) != expected_fields:
raise ValueError("KnowledgeFS product operation entry fields are invalid")
for field in ("action", "kfsOperationId", "method", "path", "productOperationId", "resource", "transport"):
_required_string(raw.get(field), f"product operation {field}")
limits = raw.get("limits")
stream = raw.get("stream")
if not isinstance(limits, dict) or set(limits) != {
"kfsMaxResponseBytes",
"productMaxRequestBytes",
"productMaxResponseBytes",
}:
raise ValueError("KnowledgeFS product operation limits are invalid")
if not all(isinstance(limit, int) and not isinstance(limit, bool) and limit >= 0 for limit in limits.values()):
raise ValueError("KnowledgeFS product operation limits must be nonnegative integers")
if not isinstance(stream, dict) or set(stream) != {"kfsResponseKind", "productKind"}:
raise ValueError("KnowledgeFS product operation stream contract is invalid")
_required_string(stream.get("kfsResponseKind"), "KFS response kind")
_required_string(stream.get("productKind"), "product stream kind")
if gap:
_required_string(raw.get("reason"), "product gap reason")
_required_string(raw.get("reasonCode"), "product gap reasonCode")
replacements = raw.get("replacementProductOperationIds")
if not isinstance(replacements, list) or not all(isinstance(item, str) and item for item in replacements):
raise ValueError("KnowledgeFS product operation gap replacements are invalid")
return cast(ProductOperationManifestEntry, raw)
def _validate_product_entry(
entry: ProductOperationManifestEntry,
runtime: ProductOperationRuntimeContract,
openapi_operations: Mapping[str, tuple[str, str, dict[str, Any]]],
python_capabilities: Mapping[str, CapabilityOperationRuntimeContract],
guard_capabilities: Mapping[str, CapabilityOperationPolicy],
) -> None:
_validate_capability_alignment(
runtime.kfs_operation_id,
openapi_operations,
python_capabilities,
guard_capabilities,
)
_, _, openapi_operation = openapi_operations[runtime.kfs_operation_id]
expected: ProductOperationManifestEntry = {
"action": runtime.action,
"kfsOperationId": runtime.kfs_operation_id,
"limits": {
"kfsMaxResponseBytes": _openapi_max_response_bytes(openapi_operation),
"productMaxRequestBytes": runtime.max_request_bytes,
"productMaxResponseBytes": runtime.max_response_bytes,
},
"method": runtime.method,
"path": runtime.path,
"productOperationId": runtime.product_operation_id,
"resource": runtime.resource,
"stream": {
"kfsResponseKind": _openapi_response_kind(openapi_operation),
"productKind": runtime.stream_kind,
},
"transport": runtime.transport,
}
for field in PRODUCT_OPERATION_MANIFEST_FIELDS:
expected_value = expected[field]
if entry[field] != expected_value:
raise ValueError(
f"KnowledgeFS product operation {runtime.product_operation_id} field {field} drifted: "
f"expected {expected_value!r}, received {entry[field]!r}"
)
def _validate_capability_alignment(
operation_id: str,
openapi_operations: Mapping[str, tuple[str, str, dict[str, Any]]],
python_capabilities: Mapping[str, CapabilityOperationRuntimeContract],
guard_capabilities: Mapping[str, CapabilityOperationPolicy],
) -> None:
python_operation = python_capabilities.get(operation_id)
guard_operation = guard_capabilities.get(operation_id)
openapi_operation = openapi_operations.get(operation_id)
if python_operation is None or guard_operation is None or openapi_operation is None:
raise ValueError(
f"KnowledgeFS operation {operation_id} must exist in Python issuer, TypeScript guard, and OpenAPI"
)
method, path, _ = openapi_operation
expected = (
python_operation.action,
python_operation.method,
python_operation.path,
python_operation.resource,
)
received_guard = (guard_operation.action, guard_operation.method, guard_operation.path, guard_operation.resource)
if received_guard != expected:
raise ValueError(f"KnowledgeFS operation {operation_id} TypeScript guard drifted")
if (method, path) != (python_operation.method, python_operation.path):
raise ValueError(f"KnowledgeFS operation {operation_id} OpenAPI method/path drifted")
if guard_operation.allowed_caller_kinds != python_operation.allowed_caller_kinds:
raise ValueError(f"KnowledgeFS operation {operation_id} caller policy drifted")
def _openapi_operations(document: dict[str, Any]) -> dict[str, tuple[str, str, dict[str, Any]]]:
operations: dict[str, tuple[str, str, dict[str, Any]]] = {}
duplicates: set[str] = set()
for path, path_item in document.get("paths", {}).items():
if not isinstance(path, str) or not isinstance(path_item, dict):
continue
for method in ("delete", "get", "patch", "post", "put"):
operation = path_item.get(method)
if not isinstance(operation, dict):
continue
operation_id = operation.get("operationId")
if not isinstance(operation_id, str) or not operation_id:
continue
if operation_id in operations:
duplicates.add(operation_id)
operations[operation_id] = (method.upper(), path, operation)
if duplicates:
raise ValueError(f"KnowledgeFS OpenAPI contains duplicate operation ids: {sorted(duplicates)}")
return operations
def _openapi_response_kind(operation: dict[str, Any]) -> Literal["binary", "buffered", "stream"]:
media_types = {
media_type
for status, response in operation.get("responses", {}).items()
if status == "2XX" or (len(status) == 3 and status.startswith("2") and status.isdigit())
for media_type in response.get("content", {})
}
if "text/event-stream" in media_types:
return "stream"
if "application/octet-stream" in media_types:
return "binary"
return "buffered"
def _openapi_max_response_bytes(operation: dict[str, Any]) -> int:
value = operation.get("x-knowledge-fs-max-response-bytes")
if not isinstance(value, int) or isinstance(value, bool) or value <= 0:
raise ValueError("KnowledgeFS product OpenAPI operation has no positive response byte limit")
return value
def _required_string(value: object, field: str) -> str:
if not isinstance(value, str) or not value.strip() or value != value.strip():
raise ValueError(f"KnowledgeFS {field} must be a non-empty trimmed string")
return value
def _unique(values: Iterable[str], label: str) -> None:
sequence = tuple(values)
if len(sequence) != len(set(sequence)):
raise ValueError(f"KnowledgeFS {label} contains duplicate ids")
__all__ = [
"CapabilityOperationPolicy",
"CapabilityOperationRuntimeContract",
"ProductOperationGapManifest",
"ProductOperationManifest",
"ProductOperationRuntimeContract",
"capability_operation_runtime_contracts",
"parse_capability_operation_policy",
"parse_product_operation_gap_manifest",
"parse_product_operation_manifest",
"product_operation_runtime_contracts",
"validate_product_operation_contracts",
]