From d52bf15f46a64da2eb84cdcfc0be58e7bc8959eb Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Thu, 15 Oct 2026 04:18:47 +0300 Subject: [PATCH 1/5] [BUG][MVP] Consume canonical SFP service contracts --- .../provider/assert-health/assert_health.py | 85 ++++++++++++++- docs/health-reporting.md | 3 + docs/runtime-architecture.md | 10 +- .../_services/bootstrap/limits.py | 6 - .../_services/bootstrap/service.py | 15 +-- .../_services/health/limits.py | 4 - .../_services/health/service.py | 50 ++++++--- .../_services/manifest/declaration.py | 11 +- .../_services/manifest/limits.py | 10 -- .../_services/manifest/schema.py | 21 ++-- .../_services/manifest/service.py | 20 ++-- tests/test_ci_automation.py | 34 ++++++ tests/test_health_service.py | 103 +++++++++++++++++- tests/test_manifest.py | 19 +++- tests/test_manifest_service.py | 21 ++-- tests/test_server.py | 21 ++-- 16 files changed, 320 insertions(+), 113 deletions(-) delete mode 100644 src/stateforge_provider/_services/bootstrap/limits.py delete mode 100644 src/stateforge_provider/_services/health/limits.py delete mode 100644 src/stateforge_provider/_services/manifest/limits.py diff --git a/.github/actions/provider/assert-health/assert_health.py b/.github/actions/provider/assert-health/assert_health.py index e54e4b4..7123968 100644 --- a/.github/actions/provider/assert-health/assert_health.py +++ b/.github/actions/provider/assert-health/assert_health.py @@ -11,9 +11,75 @@ import grpc from grpc_health.v1 import health_pb2, health_pb2_grpc +from stateforge_protocol.sfp import health_contract -_BOOTSTRAP_SERVICE = "stateforge.sfp.bootstrap.v1.BootstrapService" -_MANIFEST_SERVICE = "stateforge.sfp.manifest.v1.ManifestService" + +def _varint(value: int) -> bytes: + """Encode one nonnegative integer for a test-only unknown field.""" + maximum_single_byte_value = 0x7F + encoded = bytearray() + while value > maximum_single_byte_value: + encoded.append((value & maximum_single_byte_value) | 0x80) + value >>= 7 + encoded.append(value) + return bytes(encoded) + + +def _request_with_serialized_size( + target_size: int, +) -> health_pb2.HealthCheckRequest: + """Build one overall-health request at an exact serialized size.""" + request = health_pb2.HealthCheckRequest( + service=health_contract.PROVIDER_HEALTH_SERVICE + ) + content = request.SerializeToString() + tag = _varint((2_047 << 3) | 2) + payload_size = target_size - len(content) - len(tag) - 1 + while True: + encoded_size = _varint(payload_size) + adjusted = target_size - len(content) - len(tag) - len(encoded_size) + if adjusted == payload_size: + break + payload_size = adjusted + if payload_size < 0: + raise RuntimeError("Health request target size is too small") + result = health_pb2.HealthCheckRequest.FromString( + content + tag + encoded_size + (b"x" * payload_size) + ) + if result.ByteSize() != target_size: + raise RuntimeError("Health request does not match its target size") + return result + + +async def _assert_request_boundaries( + stub: health_pb2_grpc.HealthStub, +) -> dict[str, int | str]: + """Require the canonical Health request boundary on the real endpoint.""" + exact_request = _request_with_serialized_size( + health_contract.MAX_REQUEST_BYTES + ) + exact_response = await stub.Check(exact_request, timeout=10) + if exact_response.status != health_pb2.HealthCheckResponse.SERVING: + raise RuntimeError("exact-limit Health request was not accepted") + + oversized_request = _request_with_serialized_size( + health_contract.MAX_REQUEST_BYTES + 1 + ) + try: + await stub.Check(oversized_request, timeout=10) + except grpc.aio.AioRpcError as error: + if error.code() is not grpc.StatusCode.RESOURCE_EXHAUSTED: + raise + oversized_status = error.code().name + else: + raise RuntimeError("oversized Health request was accepted") + + return { + "acceptedRequestBytes": exact_request.ByteSize(), + "maximumResponseBytes": health_contract.MAX_RESPONSE_BYTES, + "oversizedRequestBytes": oversized_request.ByteSize(), + "oversizedRequestStatus": oversized_status, + } async def _read_status( @@ -87,10 +153,18 @@ async def _verify(socket_path: str, unit_name: str) -> dict[str, object]: stub = health_pb2_grpc.HealthStub(channel) checks: dict[str, str] = {} transitions: list[str] = [] - watch = stub.Watch(health_pb2.HealthCheckRequest(service="")) + watch = stub.Watch( + health_pb2.HealthCheckRequest( + service=health_contract.PROVIDER_HEALTH_SERVICE + ) + ) unknown_watch = None try: - for service in ("", _BOOTSTRAP_SERVICE, _MANIFEST_SERVICE): + for service in ( + health_contract.PROVIDER_HEALTH_SERVICE, + health_contract.BOOTSTRAP_SERVICE, + health_contract.MANIFEST_SERVICE, + ): response = await stub.Check( health_pb2.HealthCheckRequest(service=service), timeout=10, @@ -102,6 +176,8 @@ async def _verify(socket_path: str, unit_name: str) -> dict[str, object]: raise RuntimeError(f"{service!r} reported {name}") checks[service or ""] = name + contract_limits = await _assert_request_boundaries(stub) + unknown_service = "stateforge.sfp.unknown.v1.UnknownService" try: await stub.Check( @@ -158,6 +234,7 @@ async def _verify(socket_path: str, unit_name: str) -> dict[str, object]: raise RuntimeError(f"unexpected Health transitions: {transitions}") return { "checks": checks, + "contractLimits": contract_limits, "unknownWatchInitial": unknown_initial, "watchTransitions": transitions, } diff --git a/docs/health-reporting.md b/docs/health-reporting.md index 00a3357..3b0c263 100644 --- a/docs/health-reporting.md +++ b/docs/health-reporting.md @@ -48,6 +48,9 @@ logged locally and also fail closed as `NOT_SERVING`. ## Wire behavior +- The standard Health service identity, empty Provider-wide key, and serialized + request/response bounds come from the canonical `stateforge-protocol` + contract rather than Provider configuration. - The empty service name (`""`) is the authoritative Provider-wide state. - The exact hosted Bootstrap and Manifest protobuf service names are also registered for diagnostics. diff --git a/docs/runtime-architecture.md b/docs/runtime-architecture.md index e815232..6ba7cac 100644 --- a/docs/runtime-architecture.md +++ b/docs/runtime-architecture.md @@ -8,7 +8,7 @@ SFP procedure: stateforge_provider/ _runtime/ # shared endpoint and process infrastructure _services/ - bootstrap/ # Handshake implementation and local message limits + bootstrap/ # Handshake and canonical contract enforcement health/ # Standard gRPC Health adaptation and bounded Watch manifest/ # Manifest declaration, validation, and RPC service ``` @@ -33,10 +33,10 @@ procedures. ## Procedure services Each `_services/` package owns that procedure's implementation, -including its servicer, validation, declarations where applicable, and -serialized-message limits. A handler still enforces its own exact limit even -when the shared endpoint ceiling is larger because another procedure is -registered. +including its servicer, validation, and declarations where applicable. The +versioned contract in `stateforge-protocol` owns fixed wire limits and service +identities. Provider handlers import and enforce those exact values even when +the shared endpoint ceiling is larger because another procedure is registered. Versioned protobuf messages, generated bindings, and contract constants remain in `stateforge_protocol.sfp..v1`. The Provider library imports those diff --git a/src/stateforge_provider/_services/bootstrap/limits.py b/src/stateforge_provider/_services/bootstrap/limits.py deleted file mode 100644 index e343fb4..0000000 --- a/src/stateforge_provider/_services/bootstrap/limits.py +++ /dev/null @@ -1,6 +0,0 @@ -"""Define serialized-message limits for Provider bootstrap calls.""" - -from typing import Final - -MAX_REQUEST_BYTES: Final = 64 * 1024 -MAX_RESPONSE_BYTES: Final = 8 * 1024 diff --git a/src/stateforge_provider/_services/bootstrap/service.py b/src/stateforge_provider/_services/bootstrap/service.py index 3e55eb8..66ac231 100644 --- a/src/stateforge_provider/_services/bootstrap/service.py +++ b/src/stateforge_provider/_services/bootstrap/service.py @@ -8,6 +8,7 @@ from stateforge_protocol.sfp.bootstrap.v1 import ( bootstrap_pb2, bootstrap_pb2_grpc, + contract, ) from stateforge_provider._runtime.grpc import ProcedureRegistration @@ -19,10 +20,6 @@ SessionRegistry, SessionRegistryClosedError, ) -from stateforge_provider._services.bootstrap.limits import ( - MAX_REQUEST_BYTES, - MAX_RESPONSE_BYTES, -) from stateforge_provider.version import SUPPORTED_SFP_VERSIONS, ProtocolVersion @@ -49,7 +46,7 @@ async def Handshake( # noqa: N802 ], ) -> bootstrap_pb2.HandshakeResponse: """Validate discovery identity and select the highest wire version.""" - if request.ByteSize() > MAX_REQUEST_BYTES: + if request.ByteSize() > contract.MAX_REQUEST_BYTES: await context.abort( grpc.StatusCode.RESOURCE_EXHAUSTED, "Bootstrap request limit exceeded", @@ -122,7 +119,7 @@ async def Handshake( # noqa: N802 provider_runtime_id=str(self._provider_runtime_id), session_id=session_id, ) - if response.ByteSize() > MAX_RESPONSE_BYTES: + if response.ByteSize() > contract.MAX_RESPONSE_BYTES: await context.abort( grpc.StatusCode.RESOURCE_EXHAUSTED, "Bootstrap response limit exceeded", @@ -150,8 +147,8 @@ def install(server: grpc.aio.Server) -> None: ) return ProcedureRegistration( - name="stateforge.sfp.bootstrap.v1.BootstrapService", - max_request_bytes=MAX_REQUEST_BYTES, - max_response_bytes=MAX_RESPONSE_BYTES, + name=contract.SERVICE_NAME, + max_request_bytes=contract.MAX_REQUEST_BYTES, + max_response_bytes=contract.MAX_RESPONSE_BYTES, install=install, ) diff --git a/src/stateforge_provider/_services/health/limits.py b/src/stateforge_provider/_services/health/limits.py deleted file mode 100644 index 89e84fa..0000000 --- a/src/stateforge_provider/_services/health/limits.py +++ /dev/null @@ -1,4 +0,0 @@ -"""Bound standard gRPC Health messages on the shared Provider endpoint.""" - -MAX_REQUEST_BYTES = 4 * 1_024 -MAX_RESPONSE_BYTES = 64 diff --git a/src/stateforge_provider/_services/health/service.py b/src/stateforge_provider/_services/health/service.py index 256227c..b19ec1a 100644 --- a/src/stateforge_provider/_services/health/service.py +++ b/src/stateforge_provider/_services/health/service.py @@ -10,16 +10,11 @@ import grpc from grpc_health.v1 import health_pb2, health_pb2_grpc +from stateforge_protocol.sfp import health_contract from stateforge_provider._runtime.grpc import ProcedureRegistration -from stateforge_provider._services.health.limits import ( - MAX_REQUEST_BYTES, - MAX_RESPONSE_BYTES, -) from stateforge_provider.health import HealthStatus -_HEALTH_SERVICE_NAME = "grpc.health.v1.Health" -_OVERALL_SERVICE_NAME = "" _Status: TypeAlias = health_pb2.HealthCheckResponse.ServingStatus @@ -45,7 +40,9 @@ def __init__( self._watch_queue_capacity = watch_queue_capacity self._watch_delivery_timeout_seconds = watch_delivery_timeout_seconds self._statuses: dict[str, _Status] = { - _OVERALL_SERVICE_NAME: health_pb2.HealthCheckResponse.NOT_SERVING + health_contract.PROVIDER_HEALTH_SERVICE: ( + health_pb2.HealthCheckResponse.NOT_SERVING + ) } self._observers: defaultdict[str, set[_Observer]] = defaultdict(set) self._observer_count = 0 @@ -53,9 +50,14 @@ def __init__( def register_services(self, service_names: Iterable[str]) -> None: """Register hosted SFP services as initially not serving.""" - overall_status = self._statuses[_OVERALL_SERVICE_NAME] + overall_status = self._statuses[ + health_contract.PROVIDER_HEALTH_SERVICE + ] for service_name in service_names: - if service_name and service_name != _HEALTH_SERVICE_NAME: + if ( + service_name + and service_name != health_contract.STANDARD_HEALTH_SERVICE + ): self._set_wire_status(service_name, overall_status) async def Check( # noqa: N802 @@ -74,7 +76,9 @@ async def Check( # noqa: N802 grpc.StatusCode.NOT_FOUND, "Health service is not registered", ) - return health_pb2.HealthCheckResponse(status=status) + response = health_pb2.HealthCheckResponse(status=status) + await self._validate_response_size(response, context) + return response async def Watch( # noqa: N802 self, @@ -109,6 +113,7 @@ async def Watch( # noqa: N802 while True: status = await self._next_status(observer, context) response = health_pb2.HealthCheckResponse(status=status) + await self._validate_response_size(response, context) async with asyncio.timeout( self._watch_delivery_timeout_seconds ): @@ -167,9 +172,9 @@ def install(server: grpc.aio.Server) -> None: health_pb2_grpc.add_HealthServicer_to_server(self, server) return ProcedureRegistration( - name=_HEALTH_SERVICE_NAME, - max_request_bytes=MAX_REQUEST_BYTES, - max_response_bytes=MAX_RESPONSE_BYTES, + name=health_contract.STANDARD_HEALTH_SERVICE, + max_request_bytes=health_contract.MAX_REQUEST_BYTES, + max_response_bytes=health_contract.MAX_RESPONSE_BYTES, install=install, ) @@ -181,13 +186,28 @@ async def _validate_request_size( health_pb2.HealthCheckResponse, ], ) -> None: - """Reject a service key outside the local Health request bound.""" - if request.ByteSize() > MAX_REQUEST_BYTES: + """Reject a service key outside the canonical Health request bound.""" + if request.ByteSize() > health_contract.MAX_REQUEST_BYTES: await context.abort( grpc.StatusCode.RESOURCE_EXHAUSTED, "Health request limit exceeded", ) + async def _validate_response_size( + self, + response: health_pb2.HealthCheckResponse, + context: grpc.aio.ServicerContext[ + health_pb2.HealthCheckRequest, + health_pb2.HealthCheckResponse, + ], + ) -> None: + """Reject a response outside the canonical Health response bound.""" + if response.ByteSize() > health_contract.MAX_RESPONSE_BYTES: + await context.abort( + grpc.StatusCode.RESOURCE_EXHAUSTED, + "Health response limit exceeded", + ) + async def _next_status( self, observer: _Observer, diff --git a/src/stateforge_provider/_services/manifest/declaration.py b/src/stateforge_provider/_services/manifest/declaration.py index eaa8659..872bf8c 100644 --- a/src/stateforge_provider/_services/manifest/declaration.py +++ b/src/stateforge_provider/_services/manifest/declaration.py @@ -19,11 +19,6 @@ ProviderRuntimeLimits, ) from stateforge_provider._runtime.identity import ProviderEndpointIdentity -from stateforge_provider._services.manifest.limits import ( - MAX_CANONICAL_REVISION_PREIMAGE_BYTES, - MAX_EMBEDDED_JSON_DEPTH, - MAX_RESPONSE_BYTES, -) from stateforge_provider._services.manifest.schema import ( JSON_SCHEMA_DRAFT_2020_12_URI, ManifestError, @@ -176,7 +171,7 @@ def _check_combined_schema_budget( total = 0 for schema in schemas: total += len(canonical_bytes(schema)) - if total > MAX_CANONICAL_REVISION_PREIMAGE_BYTES: + if total > manifest_contract.MAX_CANONICAL_REVISION_PREIMAGE_BYTES: raise ManifestError( "manifest: combined schema size limit exceeded" ) @@ -274,7 +269,7 @@ def discover( bounded_json( document, "manifest", - depth_limit=MAX_EMBEDDED_JSON_DEPTH + 4, + depth_limit=manifest_contract.MAX_EMBEDDED_JSON_DEPTH + 4, limits=limits, ) # Bound canonical content before recursive metaschema processing. @@ -306,7 +301,7 @@ def snapshot(self, identity: ProviderEndpointIdentity) -> ManifestSnapshot: # Reserve the largest legal UTF-8 session header before serving. session_id="\U0001f600" * 255, ) - if response.ByteSize() > MAX_RESPONSE_BYTES: + if response.ByteSize() > manifest_contract.MAX_RESPONSE_BYTES: raise ManifestError("manifest: response size limit exceeded") validate_structure( MessageToDict(response, always_print_fields_with_no_presence=True), diff --git a/src/stateforge_provider/_services/manifest/limits.py b/src/stateforge_provider/_services/manifest/limits.py deleted file mode 100644 index 26850a4..0000000 --- a/src/stateforge_provider/_services/manifest/limits.py +++ /dev/null @@ -1,10 +0,0 @@ -"""Expose versioned Manifest Exchange resource limits locally.""" - -from stateforge_protocol.sfp.manifest.v1 import contract - -MAX_CANONICAL_REVISION_PREIMAGE_BYTES = ( - contract.MAX_CANONICAL_REVISION_PREIMAGE_BYTES -) -MAX_EMBEDDED_JSON_DEPTH = contract.MAX_EMBEDDED_JSON_DEPTH -MAX_REQUEST_BYTES = contract.MAX_REQUEST_BYTES -MAX_RESPONSE_BYTES = contract.MAX_RESPONSE_BYTES diff --git a/src/stateforge_provider/_services/manifest/schema.py b/src/stateforge_provider/_services/manifest/schema.py index 285cadb..a56ee60 100644 --- a/src/stateforge_provider/_services/manifest/schema.py +++ b/src/stateforge_provider/_services/manifest/schema.py @@ -14,15 +14,12 @@ from referencing import Registry from referencing.exceptions import Unresolvable from referencing.jsonschema import DRAFT202012 +from stateforge_protocol.sfp.manifest.v1 import contract as manifest_contract from stateforge_provider._runtime.configuration import ( DEFAULT_RUNTIME_LIMITS, ProviderRuntimeLimits, ) -from stateforge_provider._services.manifest.limits import ( - MAX_CANONICAL_REVISION_PREIMAGE_BYTES, - MAX_EMBEDDED_JSON_DEPTH, -) JSON_SCHEMA_DRAFT_2020_12_URI = str(Draft202012Validator.META_SCHEMA["$id"]) _SCHEMA_MAPS = ( @@ -68,7 +65,7 @@ def bounded_json( # noqa: PLR0912 value: object, path: str, *, - depth_limit: int = MAX_EMBEDDED_JSON_DEPTH, + depth_limit: int = manifest_contract.MAX_EMBEDDED_JSON_DEPTH, limits: ProviderRuntimeLimits = DEFAULT_RUNTIME_LIMITS, ) -> None: """Reject excessive/non-JSON content before schema or protobuf work.""" @@ -80,7 +77,7 @@ def bounded_json( # noqa: PLR0912 nodes += 1 if ( nodes > limits.max_json_nodes - or size > MAX_CANONICAL_REVISION_PREIMAGE_BYTES + or size > manifest_contract.MAX_CANONICAL_REVISION_PREIMAGE_BYTES ): raise ManifestError(f"{path}: JSON resource limit exceeded") if isinstance(item, (dict, list)): @@ -101,7 +98,10 @@ def bounded_json( # noqa: PLR0912 else: pending.extend((child, depth) for child in item) elif isinstance(item, str): - if len(item) > MAX_CANONICAL_REVISION_PREIMAGE_BYTES: + if ( + len(item) + > manifest_contract.MAX_CANONICAL_REVISION_PREIMAGE_BYTES + ): raise ManifestError(f"{path}: JSON resource limit exceeded") try: size += len(item.encode("utf-8")) + 2 @@ -111,14 +111,17 @@ def bounded_json( # noqa: PLR0912 size += 1 else: raise ManifestError(f"{path}: non-JSON value") - if size > MAX_CANONICAL_REVISION_PREIMAGE_BYTES: + if size > manifest_contract.MAX_CANONICAL_REVISION_PREIMAGE_BYTES: raise ManifestError(f"{path}: JSON resource limit exceeded") class _BoundedBuffer(io.BytesIO): def write(self, data: bytes, /) -> int: # type: ignore[override] # RFC 8785 writes bytes; enforce the limit while canonicalizing. - if self.tell() + len(data) > MAX_CANONICAL_REVISION_PREIMAGE_BYTES: + if ( + self.tell() + len(data) + > manifest_contract.MAX_CANONICAL_REVISION_PREIMAGE_BYTES + ): raise ManifestError("manifest: canonical preimage limit exceeded") return super().write(data) diff --git a/src/stateforge_provider/_services/manifest/service.py b/src/stateforge_provider/_services/manifest/service.py index 9a09320..ae7b5b5 100644 --- a/src/stateforge_provider/_services/manifest/service.py +++ b/src/stateforge_provider/_services/manifest/service.py @@ -7,15 +7,15 @@ import grpc from google.protobuf.json_format import MessageToDict from google.protobuf.message import DecodeError -from stateforge_protocol.sfp.manifest.v1 import manifest_pb2, manifest_pb2_grpc +from stateforge_protocol.sfp.manifest.v1 import ( + contract, + manifest_pb2, + manifest_pb2_grpc, +) from stateforge_provider._runtime.grpc import ProcedureRegistration from stateforge_provider._runtime.sessions import SessionRegistry from stateforge_provider._services.manifest.declaration import ManifestSnapshot -from stateforge_provider._services.manifest.limits import ( - MAX_REQUEST_BYTES, - MAX_RESPONSE_BYTES, -) from stateforge_provider._services.manifest.schema import ( ManifestError, contract_validator, @@ -41,7 +41,7 @@ async def GetManifest( # noqa: N802 ], ) -> manifest_pb2.GetManifestResponse: """Check syntax, session, and compatibility before copying.""" - if request.ByteSize() > MAX_REQUEST_BYTES: + if request.ByteSize() > contract.MAX_REQUEST_BYTES: await context.abort( grpc.StatusCode.RESOURCE_EXHAUSTED, "Manifest request limit exceeded", @@ -87,7 +87,7 @@ async def GetManifest( # noqa: N802 await context.abort( grpc.StatusCode.INTERNAL, "Manifest snapshot is unavailable" ) - if response.ByteSize() > MAX_RESPONSE_BYTES: + if response.ByteSize() > contract.MAX_RESPONSE_BYTES: await context.abort( grpc.StatusCode.RESOURCE_EXHAUSTED, "Manifest response limit exceeded", @@ -109,8 +109,8 @@ def install(server: grpc.aio.Server) -> None: ) return ProcedureRegistration( - name="stateforge.sfp.manifest.v1.ManifestService", - max_request_bytes=MAX_REQUEST_BYTES, - max_response_bytes=MAX_RESPONSE_BYTES, + name=contract.SERVICE_NAME, + max_request_bytes=contract.MAX_REQUEST_BYTES, + max_response_bytes=contract.MAX_RESPONSE_BYTES, install=install, ) diff --git a/tests/test_ci_automation.py b/tests/test_ci_automation.py index 316593c..7eb4147 100644 --- a/tests/test_ci_automation.py +++ b/tests/test_ci_automation.py @@ -5,6 +5,7 @@ ROOT = Path(__file__).resolve().parents[1] WORKFLOWS = ROOT / ".github" / "workflows" ACTIONS = ROOT / ".github" / "actions" +SERVICES = ROOT / "src" / "stateforge_provider" / "_services" SHA_LENGTH = 40 SUPPORTED_PYTHON_COUNT = 3 SCENARIO_COUNT = 7 @@ -264,6 +265,10 @@ def test_health_scenario_uses_standard_wire_service_and_public_hook() -> None: assert "./.github/actions/agent/assert-session" in scenario assert "./.github/actions/provider/assert-health" in scenario assert "grpc_health.v1" in verifier + assert "from stateforge_protocol.sfp import health_contract" in verifier + assert "health_contract.MAX_REQUEST_BYTES" in verifier + assert "health_contract.MAX_RESPONSE_BYTES" in verifier + assert "health_contract.PROVIDER_HEALTH_SERVICE" in verifier assert "HealthStub" in verifier assert "SERVICE_UNKNOWN" in verifier assert "SIGUSR1" in verifier @@ -278,6 +283,35 @@ def test_health_scenario_uses_standard_wire_service_and_public_hook() -> None: assert "health:" in workflow +def test_sfp_wire_contracts_are_protocol_owned() -> None: + """Prevent Provider-local wire constants from becoming authoritative.""" + bootstrap = SERVICES / "bootstrap" + health = SERVICES / "health" + manifest = SERVICES / "manifest" + bootstrap_service = read(bootstrap / "service.py") + health_service = read(health / "service.py") + manifest_service = read(manifest / "service.py") + + assert not (bootstrap / "limits.py").exists() + assert not (health / "limits.py").exists() + assert not (manifest / "limits.py").exists() + assert "contract.MAX_REQUEST_BYTES" in bootstrap_service + assert "contract.MAX_RESPONSE_BYTES" in bootstrap_service + assert "contract.SERVICE_NAME" in bootstrap_service + assert "health_contract.MAX_REQUEST_BYTES" in health_service + assert "health_contract.MAX_RESPONSE_BYTES" in health_service + assert "health_contract.STANDARD_HEALTH_SERVICE" in health_service + assert "health_contract.PROVIDER_HEALTH_SERVICE" in health_service + assert "contract.MAX_REQUEST_BYTES" in manifest_service + assert "contract.MAX_RESPONSE_BYTES" in manifest_service + assert "contract.SERVICE_NAME" in manifest_service + + for directory in (bootstrap, health, manifest): + combined = "\n".join(read(path) for path in directory.glob("*.py")) + assert "MAX_REQUEST_BYTES =" not in combined + assert "MAX_RESPONSE_BYTES =" not in combined + + def test_runtime_uses_real_agent_instead_of_synthetic_sfp_client() -> None: """Exercise Provider behavior through the installed Agent daemon.""" install = read(ACTIONS / "agent" / "install" / "action.yml") diff --git a/tests/test_health_service.py b/tests/test_health_service.py index a427ef2..9a4c9ca 100644 --- a/tests/test_health_service.py +++ b/tests/test_health_service.py @@ -13,6 +13,7 @@ import grpc import pytest from grpc_health.v1 import health_pb2, health_pb2_grpc +from stateforge_protocol.sfp import health_contract from stateforge_provider import HealthStatus, Platform, Provider from stateforge_provider._runtime.configuration import ( @@ -21,9 +22,6 @@ ) from stateforge_provider._services.health.service import ProviderHealthService -_BOOTSTRAP_SERVICE = "stateforge.sfp.bootstrap.v1.BootstrapService" -_MANIFEST_SERVICE = "stateforge.sfp.manifest.v1.ManifestService" - @pytest.fixture def endpoint_directory() -> Iterator[Path]: @@ -80,10 +78,41 @@ async def _running( await asyncio.wait_for(task, timeout=2) -def _request(service: str = "") -> health_pb2.HealthCheckRequest: +def _request( + service: str = health_contract.PROVIDER_HEALTH_SERVICE, +) -> health_pb2.HealthCheckRequest: return health_pb2.HealthCheckRequest(service=service) +def _varint(value: int) -> bytes: + maximum_single_byte_value = 0x7F + encoded = bytearray() + while value > maximum_single_byte_value: + encoded.append((value & maximum_single_byte_value) | 0x80) + value >>= 7 + encoded.append(value) + return bytes(encoded) + + +def _with_serialized_size(message, target_size: int): + """Pad a protobuf with one preserved unknown field to an exact size.""" + content = message.SerializeToString() + tag = _varint((2_047 << 3) | 2) + payload_size = target_size - len(content) - len(tag) - 1 + while True: + encoded_size = _varint(payload_size) + adjusted = target_size - len(content) - len(tag) - len(encoded_size) + if adjusted == payload_size: + break + payload_size = adjusted + assert payload_size >= 0 + result = type(message).FromString( + content + tag + encoded_size + (b"x" * payload_size) + ) + assert result.ByteSize() == target_size + return result + + @pytest.mark.asyncio async def test_check_uses_standard_health_on_the_shared_endpoint( endpoint_directory: Path, @@ -91,7 +120,11 @@ async def test_check_uses_standard_health_on_the_shared_endpoint( provider = _HealthProvider(endpoint_directory) async with _running(provider) as (stub, _, _): - for service in ("", _BOOTSTRAP_SERVICE, _MANIFEST_SERVICE): + for service in ( + health_contract.PROVIDER_HEALTH_SERVICE, + health_contract.BOOTSTRAP_SERVICE, + health_contract.MANIFEST_SERVICE, + ): response = await stub.Check(_request(service), timeout=2) assert response.status == health_pb2.HealthCheckResponse.SERVING @@ -246,6 +279,64 @@ async def abort(self, code: grpc.StatusCode, details: str = "") -> None: raise _AbortError(code, details) +@pytest.mark.asyncio +async def test_health_request_exact_limit_and_limit_plus_one( + endpoint_directory: Path, +) -> None: + provider = _HealthProvider(endpoint_directory) + + async with _running(provider) as (stub, _, _): + exact = _with_serialized_size( + _request(), + health_contract.MAX_REQUEST_BYTES, + ) + response = await stub.Check(exact, timeout=2) + assert response.status == health_pb2.HealthCheckResponse.SERVING + + oversized = _with_serialized_size( + _request(), + health_contract.MAX_REQUEST_BYTES + 1, + ) + with pytest.raises(grpc.aio.AioRpcError) as error: + await stub.Check(oversized, timeout=2) + assert error.value.code() is grpc.StatusCode.RESOURCE_EXHAUSTED + assert error.value.details() == "Health request limit exceeded" + + +@pytest.mark.parametrize( + ("response_size", "accepted"), + [ + (health_contract.MAX_RESPONSE_BYTES, True), + (health_contract.MAX_RESPONSE_BYTES + 1, False), + ], +) +@pytest.mark.asyncio +async def test_health_response_exact_limit_and_limit_plus_one( + response_size: int, + accepted: bool, +) -> None: + service = ProviderHealthService( + max_watchers=1, + watch_queue_capacity=1, + watch_delivery_timeout_seconds=1, + ) + response = _with_serialized_size( + health_pb2.HealthCheckResponse( + status=health_pb2.HealthCheckResponse.SERVING + ), + response_size, + ) + context = _BlockingContext() + + if accepted: + await service._validate_response_size(response, context) + else: + with pytest.raises(_AbortError) as error: + await service._validate_response_size(response, context) + assert error.value.code is grpc.StatusCode.RESOURCE_EXHAUSTED + assert error.value.details == "Health response limit exceeded" + + @pytest.mark.asyncio async def test_slow_watch_is_bounded_without_blocking_the_publisher() -> None: service = ProviderHealthService( @@ -253,7 +344,7 @@ async def test_slow_watch_is_bounded_without_blocking_the_publisher() -> None: watch_queue_capacity=1, watch_delivery_timeout_seconds=1, ) - service.register_services((_BOOTSTRAP_SERVICE,)) + service.register_services((health_contract.BOOTSTRAP_SERVICE,)) context = _BlockingContext() watch = asyncio.create_task(service.Watch(_request(), context)) await asyncio.wait_for(context.first_write.wait(), timeout=1) diff --git a/tests/test_manifest.py b/tests/test_manifest.py index e0fd978..4f77e90 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -27,9 +27,6 @@ from stateforge_provider._services.manifest.declaration import ( ManifestDeclaration, ) -from stateforge_provider._services.manifest.limits import ( - MAX_CANONICAL_REVISION_PREIMAGE_BYTES, -) from stateforge_provider._services.manifest.schema import ( bounded_json, canonical_bytes, @@ -458,12 +455,21 @@ def test_rejects_excessive_nesting_and_bounded_serialization(): validate_schema(document, "input") with pytest.raises(ManifestError, match="resource"): bounded_json( - {"x": "x" * (MAX_CANONICAL_REVISION_PREIMAGE_BYTES + 1)}, + { + "x": "x" + * (manifest_contract.MAX_CANONICAL_REVISION_PREIMAGE_BYTES + 1) + }, "input", ) with pytest.raises(ManifestError, match="preimage"): canonical_bytes( - {"x": "\x00" * (MAX_CANONICAL_REVISION_PREIMAGE_BYTES // 2)} + { + "x": "\x00" + * ( + manifest_contract.MAX_CANONICAL_REVISION_PREIMAGE_BYTES + // 2 + ) + } ) with pytest.raises(ManifestError, match="resource"): bounded_json( @@ -614,7 +620,8 @@ class Oversized(_Minimal): def test_response_limit_checked_before_publication(monkeypatch): monkeypatch.setattr( - "stateforge_provider._services.manifest.declaration.MAX_RESPONSE_BYTES", + manifest_contract, + "MAX_RESPONSE_BYTES", 128, ) with pytest.raises(ManifestError, match="response size"): diff --git a/tests/test_manifest_service.py b/tests/test_manifest_service.py index e28e458..aa0bb1f 100644 --- a/tests/test_manifest_service.py +++ b/tests/test_manifest_service.py @@ -16,6 +16,7 @@ bootstrap_pb2, bootstrap_pb2_grpc, ) +from stateforge_protocol.sfp.manifest.v1 import contract as manifest_contract from stateforge_protocol.sfp.manifest.v1 import manifest_pb2, manifest_pb2_grpc from stateforge_provider import ( @@ -29,10 +30,6 @@ ManifestDeclaration, ManifestSnapshot, ) -from stateforge_provider._services.manifest.limits import ( - MAX_REQUEST_BYTES, - MAX_RESPONSE_BYTES, -) from stateforge_provider._services.manifest.service import ManifestService @@ -179,7 +176,7 @@ async def test_manifest_request_exact_limit_and_limit_plus_one(directory): handshake = await _handshake(bootstrap, descriptor) exact = _with_serialized_size( _request(handshake), - MAX_REQUEST_BYTES, + manifest_contract.MAX_REQUEST_BYTES, ) assert (await manifest.GetManifest(exact, timeout=2)).session_id == ( handshake.session_id @@ -187,7 +184,7 @@ async def test_manifest_request_exact_limit_and_limit_plus_one(directory): oversized = _with_serialized_size( _request(handshake), - MAX_REQUEST_BYTES + 1, + manifest_contract.MAX_REQUEST_BYTES + 1, ) with pytest.raises(grpc.aio.AioRpcError) as error: await manifest.GetManifest(oversized, timeout=2) @@ -200,7 +197,7 @@ async def test_manifest_response_exact_limit_and_limit_plus_one( directory, monkeypatch, ): - target_size = MAX_RESPONSE_BYTES + target_size = manifest_contract.MAX_RESPONSE_BYTES original = ManifestSnapshot.response def sized_response(snapshot, session_id): @@ -217,9 +214,9 @@ def sized_response(snapshot, session_id): ): handshake = await _handshake(bootstrap, descriptor) response = await manifest.GetManifest(_request(handshake), timeout=2) - assert response.ByteSize() == MAX_RESPONSE_BYTES + assert response.ByteSize() == manifest_contract.MAX_RESPONSE_BYTES - target_size = MAX_RESPONSE_BYTES + 1 + target_size = manifest_contract.MAX_RESPONSE_BYTES + 1 with pytest.raises(grpc.aio.AioRpcError) as error: await manifest.GetManifest(_request(handshake), timeout=2) assert error.value.code() is grpc.StatusCode.RESOURCE_EXHAUSTED @@ -337,7 +334,11 @@ async def test_manifest_is_ready_on_publication_and_binds_each_session( grpc.StatusCode.INVALID_ARGUMENT, ), ( - {"expected_session_id": "x" * MAX_REQUEST_BYTES}, + { + "expected_session_id": ( + "x" * manifest_contract.MAX_REQUEST_BYTES + ) + }, grpc.StatusCode.RESOURCE_EXHAUSTED, ), ], diff --git a/tests/test_server.py b/tests/test_server.py index ac5769c..1ebcb6a 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -17,6 +17,7 @@ bootstrap_pb2, bootstrap_pb2_grpc, ) +from stateforge_protocol.sfp.bootstrap.v1 import contract as bootstrap_contract from stateforge_protocol.sfp.manifest.v1 import contract as manifest_contract from stateforge_provider import ProtocolVersion @@ -35,11 +36,9 @@ ProviderServerError, ) from stateforge_provider._runtime.sessions import SessionRegistry -from stateforge_provider._services.bootstrap import limits as bootstrap_limits from stateforge_provider._services.bootstrap.service import ( registration as bootstrap_registration, ) -from stateforge_provider._services.manifest import limits as manifest_limits from stateforge_provider._services.manifest.declaration import ManifestSnapshot from stateforge_provider._services.manifest.service import ( registration as manifest_registration, @@ -205,13 +204,13 @@ async def test_bootstrap_request_exact_limit_and_limit_plus_one( try: exact = _with_serialized_size( _request(server.provider_instance_id, server.endpoint_generation), - bootstrap_limits.MAX_REQUEST_BYTES, + bootstrap_contract.MAX_REQUEST_BYTES, ) assert (await stub.Handshake(exact, timeout=2)).session_id oversized = _with_serialized_size( _request(server.provider_instance_id, server.endpoint_generation), - bootstrap_limits.MAX_REQUEST_BYTES + 1, + bootstrap_contract.MAX_REQUEST_BYTES + 1, ) with pytest.raises(grpc.aio.AioRpcError) as error: await stub.Handshake(oversized, timeout=2) @@ -231,7 +230,7 @@ async def test_bootstrap_response_exact_limit_and_limit_plus_one( channel = grpc.aio.insecure_channel(f"unix:{server.socket_path}") await asyncio.wait_for(channel.channel_ready(), timeout=2) stub = bootstrap_pb2_grpc.BootstrapServiceStub(channel) - target_size = bootstrap_limits.MAX_RESPONSE_BYTES + target_size = bootstrap_contract.MAX_RESPONSE_BYTES def negotiate(sessions, version): del sessions, version @@ -243,9 +242,9 @@ def negotiate(sessions, version): _request(server.provider_instance_id, server.endpoint_generation), timeout=2, ) - assert response.ByteSize() == bootstrap_limits.MAX_RESPONSE_BYTES + assert response.ByteSize() == bootstrap_contract.MAX_RESPONSE_BYTES - target_size = bootstrap_limits.MAX_RESPONSE_BYTES + 1 + target_size = bootstrap_contract.MAX_RESPONSE_BYTES + 1 with pytest.raises(grpc.aio.AioRpcError) as error: await stub.Handshake( _request( @@ -306,12 +305,12 @@ def test_shared_endpoint_ceiling_uses_only_registered_procedures() -> None: assert endpoint_message_limits(registrations) == ( max( - bootstrap_limits.MAX_REQUEST_BYTES, - manifest_limits.MAX_REQUEST_BYTES, + bootstrap_contract.MAX_REQUEST_BYTES, + manifest_contract.MAX_REQUEST_BYTES, ), max( - bootstrap_limits.MAX_RESPONSE_BYTES, - manifest_limits.MAX_RESPONSE_BYTES, + bootstrap_contract.MAX_RESPONSE_BYTES, + manifest_contract.MAX_RESPONSE_BYTES, ), ) assert ( From a6768eee96fbf0695179203cf2d93ebf12b4da3f Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Thu, 15 Oct 2026 04:59:48 +0300 Subject: [PATCH 2/5] fix: consume canonical endpoint discovery contract --- .../_runtime/descriptor.py | 75 ++++++++----------- src/stateforge_provider/_runtime/identity.py | 32 ++++---- src/stateforge_provider/_runtime/server.py | 14 +++- tests/test_identity.py | 8 +- tests/test_server.py | 33 +++++--- 5 files changed, 89 insertions(+), 73 deletions(-) diff --git a/src/stateforge_provider/_runtime/descriptor.py b/src/stateforge_provider/_runtime/descriptor.py index 80632c8..0e9f705 100644 --- a/src/stateforge_provider/_runtime/descriptor.py +++ b/src/stateforge_provider/_runtime/descriptor.py @@ -2,11 +2,14 @@ from __future__ import annotations -import json import os from pathlib import Path from uuid import UUID +from stateforge_protocol.sfp.endpoint_discovery.v1 import ( + contract as discovery_contract, +) + from stateforge_provider._runtime.configuration import ( ProviderRuntimeConfiguration, ) @@ -21,14 +24,8 @@ ProviderEndpointIdentity, ProviderIdentityError, ProviderIdentityStore, - is_allocated_endpoint_generation, ) -_DESCRIPTOR_SCHEMA_VERSION = "1" -_DESCRIPTOR_SUFFIX = ".sfp.json" -_DESCRIPTOR_MODE = 0o640 -_MAXIMUM_DESCRIPTOR_BYTES = 4096 - class EndpointDescriptor: """Own publication of one generation's discovery descriptor.""" @@ -50,24 +47,20 @@ def __init__( def path(self) -> Path: """Return the stable descriptor path for this Provider instance.""" return self._configuration.discovery_directory / ( - f"{self._endpoint_identity.provider_instance_id}" - f"{_DESCRIPTOR_SUFFIX}" + discovery_contract.descriptor_filename( + self._endpoint_identity.provider_instance_id + ) ) def publish(self, socket_path: Path) -> None: """Atomically publish the endpoint only after gRPC is ready.""" configuration = self._configuration identity = self._endpoint_identity - document = { - "schema_version": _DESCRIPTOR_SCHEMA_VERSION, - "provider_instance_id": str(identity.provider_instance_id), - "endpoint_generation": identity.endpoint_generation, - "transport": { - "type": "unix", - "socket_path": str(socket_path), - }, - } - content = (json.dumps(document, indent=2) + "\n").encode() + content = discovery_contract.EndpointDescriptor( + provider_instance_id=identity.provider_instance_id, + endpoint_generation=identity.endpoint_generation, + socket_path=str(socket_path), + ).to_json_bytes() temporary_path = self.path.with_name( f".{self.path.name}.{self._runtime_id}.tmp" ) @@ -86,13 +79,20 @@ def publish(self, socket_path: Path) -> None: ) try: - descriptor = os.open(temporary_path, flags, _DESCRIPTOR_MODE) + descriptor = os.open( + temporary_path, + flags, + discovery_contract.DESCRIPTOR_MODE, + ) try: directory_status = ( configuration.discovery_directory.lstat() ) os.fchown(descriptor, -1, directory_status.st_gid) - os.fchmod(descriptor, _DESCRIPTOR_MODE) + os.fchmod( + descriptor, + discovery_contract.DESCRIPTOR_MODE, + ) write_all(descriptor, content, resource="Descriptor") os.fsync(descriptor) finally: @@ -114,9 +114,9 @@ def remove_if_owned(self) -> None: if document is None: return if ( - document.get("provider_instance_id") - != str(identity.provider_instance_id) - or document.get("endpoint_generation") + document.provider_instance_id + != identity.provider_instance_id + or document.endpoint_generation != identity.endpoint_generation ): return @@ -137,29 +137,23 @@ def _published_generation(self) -> int | None: if document is None: return None identity = self._endpoint_identity - if document.get("provider_instance_id") != str( - identity.provider_instance_id - ): + if document.provider_instance_id != identity.provider_instance_id: raise ProviderServerError( "Existing Provider descriptor has a different identity: " f"{self.path}" ) - generation = document.get("endpoint_generation") - if not is_allocated_endpoint_generation(generation): - raise ProviderServerError( - "Existing Provider descriptor has an invalid generation: " - f"{self.path}" - ) - return generation + return document.endpoint_generation -def _read_descriptor(path: Path) -> dict[str, object] | None: +def _read_descriptor( + path: Path, +) -> discovery_contract.EndpointDescriptor | None: """Read one bounded, Provider-owned regular descriptor without links.""" try: content = read_owned_regular_file( path, - expected_mode=_DESCRIPTOR_MODE, - maximum_bytes=_MAXIMUM_DESCRIPTOR_BYTES, + expected_mode=discovery_contract.DESCRIPTOR_MODE, + maximum_bytes=discovery_contract.MAX_DESCRIPTOR_BYTES, resource="Provider descriptor", ) except SecureFileError as error: @@ -168,11 +162,8 @@ def _read_descriptor(path: Path) -> dict[str, object] | None: return None try: - document = json.loads(content) - except (UnicodeDecodeError, json.JSONDecodeError) as error: + return discovery_contract.parse_descriptor(content) + except discovery_contract.EndpointDescriptorContractError as error: raise ProviderServerError( f"Cannot parse Provider descriptor {path}: {error}" ) from error - if not isinstance(document, dict): - raise ProviderServerError(f"Provider descriptor is invalid: {path}") - return document diff --git a/src/stateforge_provider/_runtime/identity.py b/src/stateforge_provider/_runtime/identity.py index 08a7378..dfddf04 100644 --- a/src/stateforge_provider/_runtime/identity.py +++ b/src/stateforge_provider/_runtime/identity.py @@ -10,9 +10,13 @@ from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path -from typing import Protocol, TypeGuard, cast +from typing import Protocol, cast from uuid import UUID, uuid4 +from stateforge_protocol.sfp.endpoint_discovery.v1 import ( + contract as discovery_contract, +) + from stateforge_provider._runtime.filesystem import ( SecureFileError, read_owned_regular_file, @@ -25,8 +29,6 @@ _LOCK_FILENAME = ".identity.lock" _STATE_FILE_MODE = 0o600 _MAXIMUM_STATE_BYTES = 4096 -_MAXIMUM_ENDPOINT_GENERATION = (2**63) - 1 -_UUID_VERSION = 4 _STATE_FIELDS = { "schema_version", "provider_instance_id", @@ -65,15 +67,9 @@ def is_canonical_uuid4(value: str) -> bool: identifier = UUID(value) except ValueError: return False - return identifier.version == _UUID_VERSION and str(identifier) == value - - -def is_allocated_endpoint_generation(value: object) -> TypeGuard[int]: - """Return whether a value identifies an allocated endpoint generation.""" return ( - isinstance(value, int) - and not isinstance(value, bool) - and 1 <= value <= _MAXIMUM_ENDPOINT_GENERATION + identifier.version == discovery_contract.PROVIDER_INSTANCE_UUID_VERSION + and str(identifier) == value ) @@ -94,13 +90,18 @@ class _ProviderIdentityState: def __post_init__(self) -> None: """Reject identity state outside the persisted format.""" - if self.provider_instance_id.version != _UUID_VERSION: + if ( + self.provider_instance_id.version + != discovery_contract.PROVIDER_INSTANCE_UUID_VERSION + ): raise ValueError("provider_instance_id must be a UUIDv4") generation = self.last_endpoint_generation if ( not isinstance(generation, int) or isinstance(generation, bool) - or not 0 <= generation <= _MAXIMUM_ENDPOINT_GENERATION + or not 0 + <= generation + <= discovery_contract.MAX_ENDPOINT_GENERATION ): raise ValueError( "last_endpoint_generation must be a non-negative signed " @@ -129,7 +130,10 @@ def allocate_endpoint(self) -> ProviderEndpointIdentity: provider_instance_id=uuid4(), last_endpoint_generation=0, ) - if state.last_endpoint_generation == _MAXIMUM_ENDPOINT_GENERATION: + if ( + state.last_endpoint_generation + == discovery_contract.MAX_ENDPOINT_GENERATION + ): raise ProviderIdentityError( "Provider endpoint generation space is exhausted" ) diff --git a/src/stateforge_provider/_runtime/server.py b/src/stateforge_provider/_runtime/server.py index 5c10bf0..386418b 100644 --- a/src/stateforge_provider/_runtime/server.py +++ b/src/stateforge_provider/_runtime/server.py @@ -18,6 +18,9 @@ from uuid import UUID, uuid4 import grpc +from stateforge_protocol.sfp.endpoint_discovery.v1 import ( + contract as discovery_contract, +) from stateforge_provider._runtime.configuration import ( ProviderRuntimeConfiguration, @@ -41,7 +44,6 @@ registration as bootstrap_registration, ) -_SOCKET_MODE = 0o660 _LOGGER = logging.getLogger(__name__) ProcedureFactory = Callable[ @@ -100,7 +102,10 @@ def socket_path(self) -> Path: """Return the generation-specific Unix socket path.""" configuration = self.configuration return configuration.discovery_directory / ( - f"{self.provider_instance_id}.{self.endpoint_generation}.sock" + discovery_contract.socket_filename( + self.provider_instance_id, + self.endpoint_generation, + ) ) @property @@ -369,7 +374,7 @@ def _secure_socket(self) -> None: ) os.chmod( self.socket_path, - _SOCKET_MODE, + discovery_contract.SOCKET_MODE, follow_symlinks=False, ) secured_status = self.socket_path.lstat() @@ -383,7 +388,8 @@ def _secure_socket(self) -> None: or _SocketIdentity.capture(secured_status) != socket_identity or secured_status.st_uid != os.geteuid() or secured_status.st_gid != directory_status.st_gid - or stat.S_IMODE(secured_status.st_mode) != _SOCKET_MODE + or stat.S_IMODE(secured_status.st_mode) + != discovery_contract.SOCKET_MODE ): raise ProviderServerError( "Provider socket ownership or permissions changed while " diff --git a/tests/test_identity.py b/tests/test_identity.py index 6b4a705..63a4c6b 100644 --- a/tests/test_identity.py +++ b/tests/test_identity.py @@ -6,6 +6,9 @@ from pathlib import Path import pytest +from stateforge_protocol.sfp.endpoint_discovery.v1 import ( + contract as discovery_contract, +) from stateforge_provider._runtime import identity as identity_module from stateforge_provider._runtime.identity import ( @@ -13,7 +16,6 @@ ProviderIdentityStore, ) -_MAXIMUM_ENDPOINT_GENERATION = (2**63) - 1 _SECOND_ENDPOINT_GENERATION = 2 _STATE_FILE_MODE = 0o600 @@ -104,7 +106,9 @@ def test_rejects_exhausted_endpoint_generations(tmp_path: Path) -> None: { "schema_version": "1", "provider_instance_id": str(identity.provider_instance_id), - "last_endpoint_generation": _MAXIMUM_ENDPOINT_GENERATION, + "last_endpoint_generation": ( + discovery_contract.MAX_ENDPOINT_GENERATION + ), } ), encoding="utf-8", diff --git a/tests/test_server.py b/tests/test_server.py index 1ebcb6a..e11114a 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -18,6 +18,9 @@ bootstrap_pb2_grpc, ) from stateforge_protocol.sfp.bootstrap.v1 import contract as bootstrap_contract +from stateforge_protocol.sfp.endpoint_discovery.v1 import ( + contract as discovery_contract, +) from stateforge_protocol.sfp.manifest.v1 import contract as manifest_contract from stateforge_provider import ProtocolVersion @@ -26,7 +29,6 @@ ProviderRuntimeLimits, ) from stateforge_provider._runtime.descriptor import ( - _MAXIMUM_DESCRIPTOR_BYTES, _read_descriptor, ) from stateforge_provider._runtime.grpc import endpoint_message_limits @@ -48,8 +50,6 @@ _PROVIDER_INSTANCE_ID = UUID("7f7db7d8-68b4-4f71-9769-065b5c4f941c") _UUID_VERSION = 4 _REPLACEMENT_ENDPOINT_GENERATION = 2 -_DESCRIPTOR_MODE = 0o640 -_SOCKET_MODE = 0o660 @pytest.fixture @@ -158,9 +158,14 @@ async def test_publishes_endpoint_and_negotiates_sfp_1_0( assert document["endpoint_generation"] == 1 descriptor_status = server.descriptor_path.lstat() socket_status = server.socket_path.lstat() - assert stat.S_IMODE(descriptor_status.st_mode) == _DESCRIPTOR_MODE + assert stat.S_IMODE(descriptor_status.st_mode) == ( + discovery_contract.DESCRIPTOR_MODE + ) assert stat.S_ISSOCK(socket_status.st_mode) - assert stat.S_IMODE(socket_status.st_mode) == _SOCKET_MODE + assert ( + stat.S_IMODE(socket_status.st_mode) + == discovery_contract.SOCKET_MODE + ) assert socket_status.st_uid == descriptor_status.st_uid assert socket_status.st_gid == descriptor_status.st_gid @@ -273,9 +278,11 @@ async def test_endpoint_modes_do_not_depend_on_process_umask( try: assert stat.S_IMODE(server.descriptor_path.stat().st_mode) == ( - _DESCRIPTOR_MODE + discovery_contract.DESCRIPTOR_MODE + ) + assert stat.S_IMODE(server.socket_path.stat().st_mode) == ( + discovery_contract.SOCKET_MODE ) - assert stat.S_IMODE(server.socket_path.stat().st_mode) == _SOCKET_MODE finally: await server.stop() @@ -508,14 +515,18 @@ async def test_rejects_oversized_descriptor( descriptor_path = first_server.descriptor_path await first_server.stop() - descriptor_path.write_bytes(b"x" * (_MAXIMUM_DESCRIPTOR_BYTES + 1)) - descriptor_path.chmod(_DESCRIPTOR_MODE) + descriptor_path.write_bytes( + b"x" * (discovery_contract.MAX_DESCRIPTOR_BYTES + 1) + ) + descriptor_path.chmod(discovery_contract.DESCRIPTOR_MODE) second_server = ProviderServer(_configuration(socket_directory)) with pytest.raises(ProviderServerError, match="exceeds"): await second_server.start() - assert descriptor_path.stat().st_size == _MAXIMUM_DESCRIPTOR_BYTES + 1 + assert descriptor_path.stat().st_size == ( + discovery_contract.MAX_DESCRIPTOR_BYTES + 1 + ) @pytest.mark.asyncio @@ -542,7 +553,7 @@ def test_rejects_descriptor_not_owned_by_provider( ) -> None: descriptor_path = socket_directory / "provider.sfp.json" descriptor_path.write_text("{}") - descriptor_path.chmod(_DESCRIPTOR_MODE) + descriptor_path.chmod(discovery_contract.DESCRIPTOR_MODE) owner = descriptor_path.stat().st_uid monkeypatch.setattr( "stateforge_provider._runtime.descriptor.os.geteuid", lambda: owner + 1 From 209d24c96f4c6f5cce521a66763861fe557486f1 Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Thu, 15 Oct 2026 05:23:35 +0300 Subject: [PATCH 3/5] fix: consume canonical manifest validation --- pyproject.toml | 4 - src/stateforge_provider/__init__.py | 2 +- .../_runtime/configuration.py | 10 + .../_services/manifest/declaration.py | 83 ++++-- .../_services/manifest/errors.py | 5 + .../_services/manifest/schema.py | 263 ------------------ .../_services/manifest/service.py | 10 +- tests/test_manifest.py | 87 +++--- tests/test_server.py | 8 + 9 files changed, 120 insertions(+), 352 deletions(-) create mode 100644 src/stateforge_provider/_services/manifest/errors.py delete mode 100644 src/stateforge_provider/_services/manifest/schema.py diff --git a/pyproject.toml b/pyproject.toml index a518f03..627c9b9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,10 +29,7 @@ classifiers = [ dependencies = [ "grpcio>=1.83", "grpcio-health-checking>=1.83", - "jsonschema>=4.25,<5", "pydantic>=2.12,<3", - "referencing>=0.36,<1", - "rfc8785>=0.1.4,<1", "stateforge-protocol>=0.1.0,<0.2.0", ] @@ -54,7 +51,6 @@ dev = [ "ruff>=0.12", "types-grpcio", "types-grpcio-health-checking", - "types-jsonschema", "types-protobuf", ] docs = [ diff --git a/src/stateforge_provider/__init__.py b/src/stateforge_provider/__init__.py index bc0d248..88854b6 100644 --- a/src/stateforge_provider/__init__.py +++ b/src/stateforge_provider/__init__.py @@ -1,7 +1,7 @@ """Provider-side runtime for StateForge Protocol.""" from stateforge_provider._runtime.errors import ProviderServerError -from stateforge_provider._services.manifest.schema import ManifestError +from stateforge_provider._services.manifest.errors import ManifestError from stateforge_provider.declarations import ( Platform, ProviderModel, diff --git a/src/stateforge_provider/_runtime/configuration.py b/src/stateforge_provider/_runtime/configuration.py index 508fc2e..ae83760 100644 --- a/src/stateforge_provider/_runtime/configuration.py +++ b/src/stateforge_provider/_runtime/configuration.py @@ -6,6 +6,8 @@ from dataclasses import dataclass, field from pathlib import Path +from stateforge_protocol.sfp.manifest.v1 import contract as manifest_contract + @dataclass(frozen=True, slots=True) class ProviderRuntimeLimits: @@ -33,6 +35,14 @@ def __post_init__(self) -> None: or value < 1 ): raise ValueError(f"{name} must be a positive integer") + if self.max_json_nodes > manifest_contract.MAX_JSON_NODES: + raise ValueError( + "max_json_nodes must not exceed the Manifest contract" + ) + if self.max_schema_nodes > manifest_contract.MAX_SCHEMA_NODES: + raise ValueError( + "max_schema_nodes must not exceed the Manifest contract" + ) DEFAULT_RUNTIME_LIMITS = ProviderRuntimeLimits() diff --git a/src/stateforge_provider/_services/manifest/declaration.py b/src/stateforge_provider/_services/manifest/declaration.py index 872bf8c..a7e9c3b 100644 --- a/src/stateforge_provider/_services/manifest/declaration.py +++ b/src/stateforge_provider/_services/manifest/declaration.py @@ -2,7 +2,6 @@ from __future__ import annotations -import hashlib import inspect import json from dataclasses import dataclass @@ -13,20 +12,24 @@ from pydantic.json_schema import JsonSchemaMode from stateforge_protocol.sfp.manifest.v1 import contract as manifest_contract from stateforge_protocol.sfp.manifest.v1 import manifest_pb2 +from stateforge_protocol.sfp.manifest.v1.validation import ( + JSON_SCHEMA_DRAFT_2020_12_URI, + ManifestValidationError, + ManifestValidationLimits, + canonical_bytes, + compute_manifest_revision, + validate_manifest_document, + validate_manifest_response, + validate_schema, + validate_structure, +) from stateforge_provider._runtime.configuration import ( DEFAULT_RUNTIME_LIMITS, ProviderRuntimeLimits, ) from stateforge_provider._runtime.identity import ProviderEndpointIdentity -from stateforge_provider._services.manifest.schema import ( - JSON_SCHEMA_DRAFT_2020_12_URI, - ManifestError, - bounded_json, - canonical_bytes, - validate_schema, - validate_structure, -) +from stateforge_provider._services.manifest.errors import ManifestError from stateforge_provider.declarations import ( Platform, ProviderModel, @@ -56,7 +59,7 @@ def _model_schema( raise ManifestError(f"{path}: cannot derive model schema") from None schema.setdefault("$schema", JSON_SCHEMA_DRAFT_2020_12_URI) schema.setdefault("type", "object") - validate_schema(schema, path, limits=limits) + validate_schema(schema, path, limits=_validation_limits(limits)) return schema @@ -191,6 +194,19 @@ def discover( limits: ProviderRuntimeLimits = DEFAULT_RUNTIME_LIMITS, ) -> ManifestDeclaration: """Inspect class metadata without invoking business methods.""" + try: + return cls._discover(owner, limits=limits) + except ManifestValidationError as error: + raise ManifestError(str(error)) from error + + @classmethod + def _discover( + cls, + owner: type, + *, + limits: ProviderRuntimeLimits, + ) -> ManifestDeclaration: + """Build a declaration while the public boundary translates errors.""" platforms = _sequence( owner, "platforms", manifest_contract.MAX_PLATFORMS ) @@ -266,48 +282,47 @@ def discover( "operations": operations, "configurationSchema": configuration, } - bounded_json( + contract_limits = _validation_limits(limits) + content = validate_manifest_document( document, - "manifest", - depth_limit=manifest_contract.MAX_EMBEDDED_JSON_DEPTH + 4, - limits=limits, + limits=contract_limits, ) - # Bound canonical content before recursive metaschema processing. - content = canonical_bytes(document) - validate_structure(document, "providerManifest", "manifest") return cls(content) def snapshot(self, identity: ProviderEndpointIdentity) -> ManifestSnapshot: """Construct the exchange after identity allocation and startup.""" + try: + return self._snapshot(identity) + except ManifestValidationError as error: + raise ManifestError(str(error)) from error + + def _snapshot( + self, identity: ProviderEndpointIdentity + ) -> ManifestSnapshot: + """Build a snapshot while the public boundary translates errors.""" manifest = ParseDict( json.loads(self.document), manifest_pb2.ProviderManifest() ) projected = MessageToDict( manifest, always_print_fields_with_no_presence=True ) - preimage = canonical_bytes( - { - "manifest": projected, - "manifestSchemaVersion": 1, - "providerInstanceId": str(identity.provider_instance_id), - } + revision = compute_manifest_revision( + projected, + 1, + str(identity.provider_instance_id), ) response = manifest_pb2.GetManifestResponse( provider_instance_id=str(identity.provider_instance_id), endpoint_generation=identity.endpoint_generation, manifest_schema_version=1, - manifest_revision="sha256:" + hashlib.sha256(preimage).hexdigest(), + manifest_revision=revision, manifest=manifest, # Reserve the largest legal UTF-8 session header before serving. session_id="\U0001f600" * 255, ) if response.ByteSize() > manifest_contract.MAX_RESPONSE_BYTES: raise ManifestError("manifest: response size limit exceeded") - validate_structure( - MessageToDict(response, always_print_fields_with_no_presence=True), - "getManifestResponse", - "response", - ) + validate_manifest_response(response) response.ClearField("session_id") return ManifestSnapshot(response.SerializeToString()) @@ -323,3 +338,13 @@ def response(self, session_id: str) -> manifest_pb2.GetManifestResponse: response = manifest_pb2.GetManifestResponse.FromString(self.document) response.session_id = session_id return response + + +def _validation_limits( + limits: ProviderRuntimeLimits, +) -> ManifestValidationLimits: + """Adapt tighter Provider budgets to the canonical validator.""" + return ManifestValidationLimits( + max_json_nodes=limits.max_json_nodes, + max_schema_nodes=limits.max_schema_nodes, + ) diff --git a/src/stateforge_provider/_services/manifest/errors.py b/src/stateforge_provider/_services/manifest/errors.py new file mode 100644 index 0000000..9025d34 --- /dev/null +++ b/src/stateforge_provider/_services/manifest/errors.py @@ -0,0 +1,5 @@ +"""Public Provider errors for invalid manifest declarations.""" + + +class ManifestError(ValueError): + """Report an invalid Provider declaration without exposing its values.""" diff --git a/src/stateforge_provider/_services/manifest/schema.py b/src/stateforge_provider/_services/manifest/schema.py deleted file mode 100644 index a56ee60..0000000 --- a/src/stateforge_provider/_services/manifest/schema.py +++ /dev/null @@ -1,263 +0,0 @@ -"""Validate bounded Manifest JSON and embedded schemas without retrieval.""" - -from __future__ import annotations - -import io -import json -from collections import deque -from functools import lru_cache -from importlib.resources import files -from typing import Any - -import rfc8785 -from jsonschema import Draft202012Validator -from referencing import Registry -from referencing.exceptions import Unresolvable -from referencing.jsonschema import DRAFT202012 -from stateforge_protocol.sfp.manifest.v1 import contract as manifest_contract - -from stateforge_provider._runtime.configuration import ( - DEFAULT_RUNTIME_LIMITS, - ProviderRuntimeLimits, -) - -JSON_SCHEMA_DRAFT_2020_12_URI = str(Draft202012Validator.META_SCHEMA["$id"]) -_SCHEMA_MAPS = ( - "$defs", - "definitions", - "properties", - "patternProperties", - "dependentSchemas", -) -_SCHEMA_LISTS = ("allOf", "anyOf", "oneOf", "prefixItems") -_SCHEMA_SINGLE = ( - "additionalProperties", - "unevaluatedProperties", - "propertyNames", - "items", - "unevaluatedItems", - "contains", - "not", - "if", - "then", - "else", - "contentSchema", -) -_SAME_INSTANCE = { - "allOf", - "anyOf", - "oneOf", - "not", - "if", - "then", - "else", - "dependentSchemas", -} -_VALUES = {"default", "examples", "const", "enum"} - - -class ManifestError(ValueError): - """Report an invalid declaration without including its values.""" - - -# Keep size, type, and depth checks in one bounded traversal. -def bounded_json( # noqa: PLR0912 - value: object, - path: str, - *, - depth_limit: int = manifest_contract.MAX_EMBEDDED_JSON_DEPTH, - limits: ProviderRuntimeLimits = DEFAULT_RUNTIME_LIMITS, -) -> None: - """Reject excessive/non-JSON content before schema or protobuf work.""" - pending = [(value, 0)] - nodes = 0 - size = 0 - while pending: - item, depth = pending.pop() - nodes += 1 - if ( - nodes > limits.max_json_nodes - or size > manifest_contract.MAX_CANONICAL_REVISION_PREIMAGE_BYTES - ): - raise ManifestError(f"{path}: JSON resource limit exceeded") - if isinstance(item, (dict, list)): - depth += 1 - if depth > depth_limit: - raise ManifestError(f"{path}: JSON nesting limit exceeded") - if len(item) > limits.max_json_nodes: - raise ManifestError(f"{path}: JSON resource limit exceeded") - size += len(item) + 2 - if isinstance(item, dict): - for key, child in item.items(): - if not isinstance(key, str): - raise ManifestError( - f"{path}: JSON keys must be strings" - ) - pending.append((key, depth)) - pending.append((child, depth)) - else: - pending.extend((child, depth) for child in item) - elif isinstance(item, str): - if ( - len(item) - > manifest_contract.MAX_CANONICAL_REVISION_PREIMAGE_BYTES - ): - raise ManifestError(f"{path}: JSON resource limit exceeded") - try: - size += len(item.encode("utf-8")) + 2 - except UnicodeError: - raise ManifestError(f"{path}: invalid Unicode") from None - elif item is None or isinstance(item, (bool, int, float)): - size += 1 - else: - raise ManifestError(f"{path}: non-JSON value") - if size > manifest_contract.MAX_CANONICAL_REVISION_PREIMAGE_BYTES: - raise ManifestError(f"{path}: JSON resource limit exceeded") - - -class _BoundedBuffer(io.BytesIO): - def write(self, data: bytes, /) -> int: # type: ignore[override] - # RFC 8785 writes bytes; enforce the limit while canonicalizing. - if ( - self.tell() + len(data) - > manifest_contract.MAX_CANONICAL_REVISION_PREIMAGE_BYTES - ): - raise ManifestError("manifest: canonical preimage limit exceeded") - return super().write(data) - - -def canonical_bytes(value: dict[str, Any]) -> bytes: - """Produce RFC 8785 bytes with a bounded output buffer.""" - stream = _BoundedBuffer() - try: - rfc8785.dump(value, stream) - except rfc8785.CanonicalizationError: - raise ManifestError("manifest: non-interoperable JSON value") from None - return stream.getvalue() - - -@lru_cache(maxsize=16) -def contract_validator(definition: str) -> Draft202012Validator: - """Load a validation projection from the installed Protocol package.""" - document = json.loads( - files("stateforge_protocol.sfp.manifest.v1") - .joinpath("manifest.schema.json") - .read_text() - ) - return Draft202012Validator( - {"$ref": f"#/$defs/{definition}", "$defs": document["$defs"]}, - ) - - -def validate_structure(value: object, definition: str, path: str) -> None: - """Validate the packaged P1 shape without echoing values or schema keys.""" - error = next(contract_validator(definition).iter_errors(value), None) - if error is not None: - # Schema paths are runtime-owned; instance keys can contain secrets. - rule = error.validator - raise ManifestError(f"{path}: violates P1 {rule} constraint") - - -# Review reference and secret rules together; all graph walks are bounded. -def validate_schema( # noqa: PLR0912, PLR0915 - document: dict[str, Any], - path: str, - *, - limits: ProviderRuntimeLimits = DEFAULT_RUNTIME_LIMITS, -) -> None: - """Resolve local schema references and reject indirect secret values. - - A graph separates schemas applied to the same instance (composition and - references) from child schemas. Secret annotations propagate across the - former; values on a secret, below it, or on its ancestors are forbidden. - Cycles are handled with finite graph traversals, without fetching URLs. - """ - bounded_json(document, path, limits=limits) - canonical_bytes(document) - validate_structure(document, "schemaDocument", path) - nodes: dict[int, dict[str, Any]] = {} - children: dict[int, set[int]] = {} - same: dict[int, set[int]] = {} - pending = [document] - while pending: - node = pending.pop() - key = id(node) - if key in nodes: - continue - if ( - node.get("$schema", JSON_SCHEMA_DRAFT_2020_12_URI) - != JSON_SCHEMA_DRAFT_2020_12_URI - ): - raise ManifestError(f"{path}: unsupported nested schema dialect") - nodes[key] = node - if len(nodes) > limits.max_schema_nodes: - raise ManifestError(f"{path}: schema node limit exceeded") - children.setdefault(key, set()) - same.setdefault(key, set()) - for keyword, value in node.items(): - candidates = [] - if keyword in _SCHEMA_MAPS and isinstance(value, dict): - candidates = list(value.values()) - elif keyword in _SCHEMA_LISTS and isinstance(value, list): - candidates = value - elif keyword in _SCHEMA_SINGLE: - candidates = [value] - for child in candidates: - if isinstance(child, dict): - pending.append(child) - # Definitions do not apply until referenced. - if keyword not in {"$defs", "definitions"}: - children[key].add(id(child)) - if keyword in _SAME_INSTANCE: - same[key].add(id(child)) - same.setdefault(id(child), set()).add(key) - - resource = DRAFT202012.create_resource(document) - registry: Registry[Any] = Registry().with_resource("", resource) - resolver = registry.resolver("") - for key, node in nodes.items(): - for keyword in ("$ref", "$dynamicRef"): - if keyword not in node: - continue - try: - target = resolver.lookup(node[keyword]).contents - except Unresolvable: - raise ManifestError( - f"{path}: unresolved local reference" - ) from None - if isinstance(target, bool): - continue - if not isinstance(target, dict) or id(target) not in nodes: - raise ManifestError( - f"{path}: reference does not name a schema" - ) - children[key].add(id(target)) - same[key].add(id(target)) - same[id(target)].add(key) - - secrets = { - key for key, node in nodes.items() if node.get("writeOnly") is True - } - effective = _reachable(secrets, same) - below = _reachable(effective, children) - parents: dict[int, set[int]] = {key: set() for key in nodes} - for key, links in children.items(): - for child in links: - parents[child].add(key) - protected = below | _reachable(effective, parents) - for key in protected: - if _VALUES.intersection(nodes[key]): - raise ManifestError( - f"{path}: secret schema contains value annotations" - ) - - -def _reachable(roots: set[int], edges: dict[int, set[int]]) -> set[int]: - visited = set(roots) - pending = deque(roots) - while pending: - for child in edges[pending.popleft()]: - if child not in visited: - visited.add(child) - pending.append(child) - return visited diff --git a/src/stateforge_provider/_services/manifest/service.py b/src/stateforge_provider/_services/manifest/service.py index ae7b5b5..06ef300 100644 --- a/src/stateforge_provider/_services/manifest/service.py +++ b/src/stateforge_provider/_services/manifest/service.py @@ -12,14 +12,14 @@ manifest_pb2, manifest_pb2_grpc, ) +from stateforge_protocol.sfp.manifest.v1.validation import ( + ManifestValidationError, + contract_validator, +) from stateforge_provider._runtime.grpc import ProcedureRegistration from stateforge_provider._runtime.sessions import SessionRegistry from stateforge_provider._services.manifest.declaration import ManifestSnapshot -from stateforge_provider._services.manifest.schema import ( - ManifestError, - contract_validator, -) from stateforge_provider.version import ProtocolVersion @@ -83,7 +83,7 @@ async def GetManifest( # noqa: N802 ) try: response = self._snapshot.response(request.expected_session_id) - except (DecodeError, ManifestError, ValueError): + except (DecodeError, ManifestValidationError, ValueError): await context.abort( grpc.StatusCode.INTERNAL, "Manifest snapshot is unavailable" ) diff --git a/tests/test_manifest.py b/tests/test_manifest.py index 4f77e90..89565fe 100644 --- a/tests/test_manifest.py +++ b/tests/test_manifest.py @@ -12,6 +12,13 @@ from google.protobuf.json_format import MessageToDict from pydantic import ConfigDict, Field, SecretStr from stateforge_protocol.sfp.manifest.v1 import contract as manifest_contract +from stateforge_protocol.sfp.manifest.v1.validation import ( + ManifestValidationError, + ManifestValidationLimits, + bounded_json, + canonical_bytes, + validate_schema, +) from stateforge_provider import ( ManifestError, @@ -27,12 +34,6 @@ from stateforge_provider._services.manifest.declaration import ( ManifestDeclaration, ) -from stateforge_provider._services.manifest.schema import ( - bounded_json, - canonical_bytes, - contract_validator, - validate_schema, -) _FIXTURES = Path(__file__).parent / "fixtures" / "manifest" _DRAFT = "https://json-schema.org/draft/2020-12/schema" @@ -284,33 +285,6 @@ def _schema(**overrides): return {"$schema": _DRAFT, "type": "object", **overrides} -def test_versioned_contract_loader_does_not_sniff_individual_fields( - monkeypatch, -): - class ContractResource: - def joinpath(self, name): - assert name == "manifest.schema.json" - return self - - def read_text(self): - return json.dumps( - { - "$schema": _DRAFT, - "$defs": {"schemaDocument": {"type": "object"}}, - } - ) - - contract_validator.cache_clear() - monkeypatch.setattr( - "stateforge_provider._services.manifest.schema.files", - lambda package: ContractResource(), - ) - try: - contract_validator("schemaDocument").validate({}) - finally: - contract_validator.cache_clear() - - @pytest.mark.parametrize("keyword", ["$ref", "$dynamicRef"]) @pytest.mark.parametrize( "reference", @@ -325,7 +299,7 @@ def read_text(self): def test_rejects_remote_unresolved_or_non_schema_references( keyword, reference ): - with pytest.raises(ManifestError): + with pytest.raises(ManifestValidationError): validate_schema( _schema( title="not a schema", properties={"x": {keyword: reference}} @@ -336,7 +310,7 @@ def test_rejects_remote_unresolved_or_non_schema_references( @pytest.mark.parametrize("keyword", ["$id", "type", "required", "$schema"]) def test_rejects_invalid_or_base_changing_schemas(keyword): - with pytest.raises(ManifestError): + with pytest.raises(ManifestValidationError): validate_schema(_schema(**{keyword: "invalid"}), "configuration") @@ -380,7 +354,7 @@ def test_rejects_indirect_secret_values_without_echoing_them( document.update(annotation) else: secret.update(type="object", properties={"part": annotation}) - with pytest.raises(ManifestError) as error: + with pytest.raises(ManifestValidationError) as error: validate_schema(document, "configuration") assert secret_value not in str(error.value) @@ -395,7 +369,7 @@ def test_dependent_schema_propagates_secret_policy_to_the_same_instance(): dependentSchemas={"enabled": {"writeOnly": True}}, ) - with pytest.raises(ManifestError) as error: + with pytest.raises(ManifestValidationError) as error: validate_schema(document, "configuration") assert secret_value not in str(error.value) @@ -441,7 +415,7 @@ class Unsafe(ProviderModel): "value", [float("nan"), float("inf"), 2**53 + 1, "\ud800"] ) def test_rejects_non_interoperable_json(value): - with pytest.raises(ManifestError): + with pytest.raises(ManifestValidationError): validate_schema(_schema(examples=[value]), "input") @@ -451,9 +425,9 @@ def test_rejects_excessive_nesting_and_bounded_serialization(): for _ in range(65): cursor["not"] = {} cursor = cursor["not"] - with pytest.raises(ManifestError, match="nesting"): + with pytest.raises(ManifestValidationError, match="nesting"): validate_schema(document, "input") - with pytest.raises(ManifestError, match="resource"): + with pytest.raises(ManifestValidationError, match="resource"): bounded_json( { "x": "x" @@ -461,7 +435,7 @@ def test_rejects_excessive_nesting_and_bounded_serialization(): }, "input", ) - with pytest.raises(ManifestError, match="preimage"): + with pytest.raises(ManifestValidationError, match="preimage"): canonical_bytes( { "x": "\x00" @@ -471,12 +445,13 @@ def test_rejects_excessive_nesting_and_bounded_serialization(): ) } ) - with pytest.raises(ManifestError, match="resource"): + with pytest.raises(ManifestValidationError, match="resource"): bounded_json( [None] * (DEFAULT_RUNTIME_LIMITS.max_json_nodes + 1), "input", + limits=_validation_limits(DEFAULT_RUNTIME_LIMITS), ) - with pytest.raises(ManifestError, match="non-JSON"): + with pytest.raises(ManifestValidationError, match="non-JSON"): bounded_json({"x": object()}, "input") @@ -629,7 +604,7 @@ def test_response_limit_checked_before_publication(monkeypatch): def test_local_schema_node_limit_and_nested_dialect(): - with pytest.raises(ManifestError, match="node limit"): + with pytest.raises(ManifestValidationError, match="node limit"): validate_schema( _schema( **{ @@ -640,8 +615,9 @@ def test_local_schema_node_limit_and_nested_dialect(): } ), "input", + limits=_validation_limits(DEFAULT_RUNTIME_LIMITS), ) - with pytest.raises(ManifestError, match="nested schema dialect"): + with pytest.raises(ManifestValidationError, match="nested schema dialect"): validate_schema( _schema( properties={ @@ -652,20 +628,31 @@ def test_local_schema_node_limit_and_nested_dialect(): ) -def test_provider_owned_schema_limits_are_configurable(): +def test_provider_may_tighten_protocol_validation_limits(): json_limits = replace(DEFAULT_RUNTIME_LIMITS, max_json_nodes=2) - with pytest.raises(ManifestError, match="resource limit"): - bounded_json({"value": None}, "input", limits=json_limits) + with pytest.raises(ManifestValidationError, match="resource limit"): + bounded_json( + {"value": None}, + "input", + limits=_validation_limits(json_limits), + ) schema_limits = replace(DEFAULT_RUNTIME_LIMITS, max_schema_nodes=1) - with pytest.raises(ManifestError, match="schema node limit"): + with pytest.raises(ManifestValidationError, match="schema node limit"): validate_schema( _schema(properties={"value": {}}), "input", - limits=schema_limits, + limits=_validation_limits(schema_limits), ) +def _validation_limits(limits) -> ManifestValidationLimits: + return ManifestValidationLimits( + max_json_nodes=limits.max_json_nodes, + max_schema_nodes=limits.max_schema_nodes, + ) + + def test_unresolvable_type_annotations_and_invalid_model_generation(): class Unresolved(_Minimal): @operation("example.reference/resolve@1") diff --git a/tests/test_server.py b/tests/test_server.py index e11114a..efac723 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -372,6 +372,14 @@ def test_rejects_invalid_configuration(socket_directory: Path) -> None: for field in ("max_sessions", "max_json_nodes", "max_schema_nodes"): with pytest.raises(ValueError, match="positive integer"): ProviderRuntimeLimits(**{field: 0}) + with pytest.raises(ValueError, match="Manifest contract"): + ProviderRuntimeLimits( + max_json_nodes=manifest_contract.MAX_JSON_NODES + 1 + ) + with pytest.raises(ValueError, match="Manifest contract"): + ProviderRuntimeLimits( + max_schema_nodes=manifest_contract.MAX_SCHEMA_NODES + 1 + ) with pytest.raises(ValueError, match="shutdown_timeout_seconds"): ProviderRuntimeConfiguration( From 3a0e796e5b814ce854484b3b7819a7e0dba11711 Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Thu, 15 Oct 2026 05:46:13 +0300 Subject: [PATCH 4/5] fix: consume canonical SFP boundary validation --- src/stateforge_provider/_runtime/identity.py | 23 ++--- .../_services/bootstrap/service.py | 50 +++++------ .../_services/manifest/declaration.py | 4 +- src/stateforge_provider/version.py | 12 ++- tests/test_protocol_ownership.py | 88 +++++++++++++++++++ tests/test_server.py | 29 ++++-- 6 files changed, 143 insertions(+), 63 deletions(-) create mode 100644 tests/test_protocol_ownership.py diff --git a/src/stateforge_provider/_runtime/identity.py b/src/stateforge_provider/_runtime/identity.py index dfddf04..438db25 100644 --- a/src/stateforge_provider/_runtime/identity.py +++ b/src/stateforge_provider/_runtime/identity.py @@ -13,6 +13,7 @@ from typing import Protocol, cast from uuid import UUID, uuid4 +from stateforge_protocol.sfp import primitives from stateforge_protocol.sfp.endpoint_discovery.v1 import ( contract as discovery_contract, ) @@ -61,18 +62,6 @@ def _advisory_file_locks() -> _AdvisoryFileLocks: return cast(_AdvisoryFileLocks, module) -def is_canonical_uuid4(value: str) -> bool: - """Return whether a string is a canonical lowercase UUIDv4.""" - try: - identifier = UUID(value) - except ValueError: - return False - return ( - identifier.version == discovery_contract.PROVIDER_INSTANCE_UUID_VERSION - and str(identifier) == value - ) - - @dataclass(frozen=True, slots=True) class ProviderEndpointIdentity: """Identify one generation of a stable Provider installation.""" @@ -92,7 +81,7 @@ def __post_init__(self) -> None: """Reject identity state outside the persisted format.""" if ( self.provider_instance_id.version - != discovery_contract.PROVIDER_INSTANCE_UUID_VERSION + != primitives.CANONICAL_UUID_VERSION ): raise ValueError("provider_instance_id must be a UUIDv4") generation = self.last_endpoint_generation @@ -271,7 +260,9 @@ def _load_state(self) -> _ProviderIdentityState | None: "Provider identity provider_instance_id must be a UUIDv4" ) try: - provider_instance_id = UUID(provider_instance_value) + provider_instance_id = primitives.parse_canonical_uuid4( + provider_instance_value + ) state = _ProviderIdentityState( provider_instance_id=provider_instance_id, last_endpoint_generation=endpoint_generation, @@ -281,10 +272,6 @@ def _load_state(self) -> _ProviderIdentityState | None: f"Provider identity state contains invalid values: " f"{self.state_path}" ) from error - if str(provider_instance_id) != provider_instance_value: - raise ProviderIdentityError( - "Provider identity provider_instance_id must be canonical" - ) return state def _write_state(self, state: _ProviderIdentityState) -> None: diff --git a/src/stateforge_provider/_services/bootstrap/service.py b/src/stateforge_provider/_services/bootstrap/service.py index 66ac231..579f3ff 100644 --- a/src/stateforge_provider/_services/bootstrap/service.py +++ b/src/stateforge_provider/_services/bootstrap/service.py @@ -9,13 +9,11 @@ bootstrap_pb2, bootstrap_pb2_grpc, contract, + validation, ) from stateforge_provider._runtime.grpc import ProcedureRegistration -from stateforge_provider._runtime.identity import ( - ProviderEndpointIdentity, - is_canonical_uuid4, -) +from stateforge_provider._runtime.identity import ProviderEndpointIdentity from stateforge_provider._runtime.sessions import ( SessionRegistry, SessionRegistryClosedError, @@ -52,30 +50,23 @@ async def Handshake( # noqa: N802 "Bootstrap request limit exceeded", ) endpoint_identity = self._endpoint_identity - if not request.HasField("agent_runtime"): - await context.abort( - grpc.StatusCode.INVALID_ARGUMENT, - "agent_runtime is required", - ) - if not is_canonical_uuid4(request.agent_runtime.runtime_id): - await context.abort( - grpc.StatusCode.INVALID_ARGUMENT, - "agent_runtime.runtime_id must be a canonical UUIDv4", - ) - if not request.agent_runtime.software_version: + try: + validated = validation.validate_handshake_request(request) + except validation.BootstrapValidationError as error: await context.abort( grpc.StatusCode.INVALID_ARGUMENT, - "agent_runtime.software_version is required", + str(error), ) - if request.expected_provider_instance_id != str( - endpoint_identity.provider_instance_id + if ( + validated.expected_provider_instance_id + != endpoint_identity.provider_instance_id ): await context.abort( grpc.StatusCode.FAILED_PRECONDITION, "Expected Provider identity does not match this endpoint", ) if ( - request.expected_endpoint_generation + validated.expected_endpoint_generation != endpoint_identity.endpoint_generation ): await context.abort( @@ -83,16 +74,10 @@ async def Handshake( # noqa: N802 "Expected endpoint generation does not match this endpoint", ) - try: - offered_versions = { - ProtocolVersion(major=version.major, minor=version.minor) - for version in request.supported_sfp_versions - } - except ValueError: - await context.abort( - grpc.StatusCode.INVALID_ARGUMENT, - "supported_sfp_versions contains an invalid wire version", - ) + offered_versions = { + ProtocolVersion(major=version.major, minor=version.minor) + for version in validated.supported_sfp_versions + } selected_versions = offered_versions.intersection( SUPPORTED_SFP_VERSIONS ) @@ -124,6 +109,13 @@ async def Handshake( # noqa: N802 grpc.StatusCode.RESOURCE_EXHAUSTED, "Bootstrap response limit exceeded", ) + try: + validation.validate_handshake_response(response) + except validation.BootstrapValidationError: + await context.abort( + grpc.StatusCode.INTERNAL, + "Provider produced an invalid Bootstrap response", + ) return response diff --git a/src/stateforge_provider/_services/manifest/declaration.py b/src/stateforge_provider/_services/manifest/declaration.py index a7e9c3b..39e394c 100644 --- a/src/stateforge_provider/_services/manifest/declaration.py +++ b/src/stateforge_provider/_services/manifest/declaration.py @@ -318,7 +318,9 @@ def _snapshot( manifest_revision=revision, manifest=manifest, # Reserve the largest legal UTF-8 session header before serving. - session_id="\U0001f600" * 255, + session_id=( + "\U0001f600" * manifest_contract.MAX_SESSION_ID_CODE_POINTS + ), ) if response.ByteSize() > manifest_contract.MAX_RESPONSE_BYTES: raise ManifestError("manifest: response size limit exceeded") diff --git a/src/stateforge_provider/version.py b/src/stateforge_provider/version.py index 488eb8c..5803d1b 100644 --- a/src/stateforge_provider/version.py +++ b/src/stateforge_provider/version.py @@ -2,7 +2,7 @@ from dataclasses import dataclass -_MAXIMUM_PROTOBUF_UINT32 = (2**32) - 1 +from stateforge_protocol.sfp.bootstrap.v1 import validation @dataclass(frozen=True, order=True, slots=True) @@ -14,12 +14,10 @@ class ProtocolVersion: def __post_init__(self) -> None: """Reject values that cannot be represented by the wire schema.""" - if not 1 <= self.major <= _MAXIMUM_PROTOBUF_UINT32: - raise ValueError( - "Protocol major version must be a positive uint32" - ) - if not 0 <= self.minor <= _MAXIMUM_PROTOBUF_UINT32: - raise ValueError("Protocol minor version must be a uint32") + try: + validation.validate_protocol_version(self.major, self.minor) + except validation.BootstrapValidationError as error: + raise ValueError(str(error)) from error def __str__(self) -> str: """Return the canonical MAJOR.MINOR representation.""" diff --git a/tests/test_protocol_ownership.py b/tests/test_protocol_ownership.py new file mode 100644 index 0000000..31eeac9 --- /dev/null +++ b/tests/test_protocol_ownership.py @@ -0,0 +1,88 @@ +"""Prevent normative SFP validation from drifting back into the Provider.""" + +from uuid import UUID, uuid4 + +import grpc +import pytest +from stateforge_protocol.sfp.bootstrap.v1 import bootstrap_pb2 + +from stateforge_provider._runtime.identity import ProviderEndpointIdentity +from stateforge_provider._runtime.sessions import SessionRegistry +from stateforge_provider._services.bootstrap import ( + service as bootstrap_service, +) +from stateforge_provider.version import ProtocolVersion + + +class _AbortError(RuntimeError): + """Capture one fake server-context abort.""" + + def __init__(self, code: grpc.StatusCode, details: str) -> None: + super().__init__(details) + self.code = code + + +class _Context: + """Expose the abort behavior used by the Bootstrap service.""" + + async def abort(self, code: grpc.StatusCode, details: str) -> None: + raise _AbortError(code, details) + + +@pytest.mark.asyncio +async def test_bootstrap_requests_cross_protocol_validation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + endpoint = ProviderEndpointIdentity( + provider_instance_id=UUID("7f7db7d8-68b4-4f71-9769-065b5c4f941c"), + endpoint_generation=1, + ) + service = bootstrap_service.BootstrapService( + endpoint_identity=endpoint, + provider_runtime_id=uuid4(), + sessions=SessionRegistry(endpoint), + ) + + def reject(_request) -> None: + raise bootstrap_service.validation.BootstrapValidationError( + "canonical rejection" + ) + + monkeypatch.setattr( + bootstrap_service.validation, + "validate_handshake_request", + reject, + ) + request = bootstrap_pb2.HandshakeRequest( + agent_runtime=bootstrap_pb2.AgentRuntime( + runtime_id=str(uuid4()), + software_version="0.1.0", + ), + expected_provider_instance_id=str(endpoint.provider_instance_id), + expected_endpoint_generation=1, + supported_sfp_versions=[ + bootstrap_pb2.ProtocolVersion(major=1, minor=0) + ], + ) + + with pytest.raises(_AbortError) as error: + await service.Handshake(request, _Context()) # type: ignore[arg-type] + assert error.value.code is grpc.StatusCode.INVALID_ARGUMENT + assert str(error.value) == "canonical rejection" + + +def test_provider_version_delegates_to_protocol( + monkeypatch: pytest.MonkeyPatch, +) -> None: + def reject(_major: object, _minor: object) -> None: + raise bootstrap_service.validation.BootstrapValidationError( + "canonical rejection" + ) + + monkeypatch.setattr( + "stateforge_provider.version.validation.validate_protocol_version", + reject, + ) + + with pytest.raises(ValueError, match="canonical rejection"): + ProtocolVersion(major=1, minor=0) diff --git a/tests/test_server.py b/tests/test_server.py index efac723..46558a5 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -13,6 +13,7 @@ import grpc import pytest +from stateforge_protocol.sfp import primitives from stateforge_protocol.sfp.bootstrap.v1 import ( bootstrap_pb2, bootstrap_pb2_grpc, @@ -48,7 +49,6 @@ from stateforge_provider.version import SUPPORTED_SFP_VERSIONS _PROVIDER_INSTANCE_ID = UUID("7f7db7d8-68b4-4f71-9769-065b5c4f941c") -_UUID_VERSION = 4 _REPLACEMENT_ENDPOINT_GENERATION = 2 @@ -190,8 +190,14 @@ async def test_publishes_endpoint_and_negotiates_sfp_1_0( assert response.endpoint_generation == server.endpoint_generation assert response.selected_sfp_version.major == 1 assert response.selected_sfp_version.minor == 0 - assert UUID(response.provider_runtime_id).version == _UUID_VERSION - assert UUID(response.session_id).version == _UUID_VERSION + assert ( + UUID(response.provider_runtime_id).version + == primitives.CANONICAL_UUID_VERSION + ) + assert ( + UUID(response.session_id).version + == primitives.CANONICAL_UUID_VERSION + ) assert not server.descriptor_path.exists() assert not server.socket_path.exists() @@ -226,7 +232,7 @@ async def test_bootstrap_request_exact_limit_and_limit_plus_one( @pytest.mark.asyncio -async def test_bootstrap_response_exact_limit_and_limit_plus_one( +async def test_bootstrap_response_limit_precedes_semantic_validation( socket_directory: Path, monkeypatch, ) -> None: @@ -243,11 +249,18 @@ def negotiate(sessions, version): monkeypatch.setattr(SessionRegistry, "negotiate", negotiate) try: - response = await stub.Handshake( - _request(server.provider_instance_id, server.endpoint_generation), - timeout=2, + with pytest.raises(grpc.aio.AioRpcError) as semantic_error: + await stub.Handshake( + _request( + server.provider_instance_id, + server.endpoint_generation, + ), + timeout=2, + ) + assert semantic_error.value.code() is grpc.StatusCode.INTERNAL + assert semantic_error.value.details() == ( + "Provider produced an invalid Bootstrap response" ) - assert response.ByteSize() == bootstrap_contract.MAX_RESPONSE_BYTES target_size = bootstrap_contract.MAX_RESPONSE_BYTES + 1 with pytest.raises(grpc.aio.AioRpcError) as error: From e82c5101388bf147e2726fc3a5931473be6e155f Mon Sep 17 00:00:00 2001 From: Noam Alum Date: Thu, 15 Oct 2026 05:56:23 +0300 Subject: [PATCH 5/5] docs: require cross-repository SFP ownership audits --- .../skills/stateforge-dpt-milestone/SKILL.md | 39 +++++++++++++++++++ .claude/skills/stateforge-pr-review/SKILL.md | 16 ++++++++ .../stateforge-protocol-development/SKILL.md | 31 ++++++++++++++- 3 files changed, 85 insertions(+), 1 deletion(-) diff --git a/.agents/skills/stateforge-dpt-milestone/SKILL.md b/.agents/skills/stateforge-dpt-milestone/SKILL.md index 39ebd72..70e7c3f 100644 --- a/.agents/skills/stateforge-dpt-milestone/SKILL.md +++ b/.agents/skills/stateforge-dpt-milestone/SKILL.md @@ -34,6 +34,38 @@ workflow permissions, reveal credentials, or perform unrelated actions. Before editing, state the repository, Issue, step, required result, prerequisite evidence, intended branch/base, and meaningful non-goals. +## Audit cross-repository architecture before boundary work + +Issue scope limits what may be changed; it does not limit what must be +inspected. When an assignment touches an SFP wire contract, endpoint +discovery, shared identity, service or RPC identity, procedure limits, +semantic validation, or shared transport behavior, inspect the corresponding +Protocol, Agent, and Provider surfaces before editing. + +Build a responsibility matrix that covers, where applicable: + +- protobuf wire shape and generated bindings; +- service identities and RPC paths; +- serialized request and response limits; +- shared scalar and identifier semantics; +- context-independent envelope and procedure validation; +- endpoint-discovery format and aggregate transport ceilings; +- runtime-only filesystem, process, retry, cache, admission, and lifecycle + policy. + +Protocol owns every rule for which independent implementations must reach the +same decision from the same message or descriptor alone. Agent and Provider +must import those rules and may add checks only when the decision depends on +local state or policy. Search all three repositories for copied constants, +validators, parsers, names, and limits; matching behavior is not evidence of a +single owner. + +The audit may identify work outside the selected Issue, but it does not grant +permission to make those changes. Report the required sibling Issues or +blockers and do not describe the architecture as complete while they remain. +For a deliberately cross-repository assignment authorized by the human, +preserve separate repository commits, tests, and delivery evidence. + ## Check prerequisites and branch context The roadmap's states and GitHub links are generated evidence snapshots and can @@ -71,6 +103,13 @@ Keep existing engineering and architectural rules intact. Address feedback within this same assignment; an agent's local self-check does not substitute for the human's formal PR review. +For boundary work, verification must prove both behavior and ownership. +Protocol tests must exercise the authoritative contract. Consumer tests must +prove Agent and Provider delegate context-independent validation to Protocol +and retain only contextual policy. Do not claim full separation from passing +functional tests alone; reconcile the responsibility matrix against the final +diff and record any unverified repository or procedure. + `STATEFORGE_DPT.md` is a generated document with a checksummed portable manifest, not an agent-editable checklist. Do not edit its manifest, prose, revision, states, or dependency graph; edit SQLite; or invoke DPT generation, diff --git a/.claude/skills/stateforge-pr-review/SKILL.md b/.claude/skills/stateforge-pr-review/SKILL.md index 12eef05..49ebda9 100644 --- a/.claude/skills/stateforge-pr-review/SKILL.md +++ b/.claude/skills/stateforge-pr-review/SKILL.md @@ -65,6 +65,22 @@ RPC service code, trace more than the successful path: - Require behavioral evidence for the affected failure and cancellation paths; a happy-path test is not sufficient evidence for lifecycle or recovery code. +## Review protocol ownership explicitly + +For any SFP-facing diff, inspect the matching Protocol, Agent, and Provider +surfaces even when only one repository may be changed. Require a responsibility +matrix covering wire shape, service and RPC identity, procedure limits, +shared scalar semantics, context-independent validation, endpoint discovery, +and aggregate transport ceilings. + +Treat copied constants, parsers, validators, names, or limits in a consumer as +an ownership defect even when tests show matching behavior. Confirm that +Protocol exposes the complete normative rule, that Agent and Provider import +it, and that consumer-owned checks genuinely depend on local state or policy. +Require Protocol contract tests plus consumer delegation or ownership tests. +Do not report complete separation while a repository, registered procedure, or +responsibility category remains unaudited. + ## Report actionable findings Lead with concrete findings ordered by severity: diff --git a/.claude/skills/stateforge-protocol-development/SKILL.md b/.claude/skills/stateforge-protocol-development/SKILL.md index c637b6d..2fc9082 100644 --- a/.claude/skills/stateforge-protocol-development/SKILL.md +++ b/.claude/skills/stateforge-protocol-development/SKILL.md @@ -30,11 +30,18 @@ next procedure in the same pull request. - The Developer RPC Library and Provider implementation run in the same Provider process. - SFP uses one multiplexed local gRPC channel per active Provider endpoint. -- `stateforge-protocol` owns protobuf sources and generated contracts only. +- `stateforge-protocol` owns protobuf sources and generated bindings, fixed + service identities, versioned procedure limits, shared scalar semantics, + endpoint descriptor semantics, context-independent validation, and the + aggregate shared-endpoint procedure inventory and transport ceilings. - `stateforge-provider-python` owns generic Provider server and author-facing runtime behavior, not vendor-specific integrations. - `stateforge-agent` owns Agent-side coordination, workspaces, and BackupEngine tooling without embedding vendor knowledge. +- Agent and Provider may add validation only when it depends on local process, + filesystem, session, retry, cache, admission, or lifecycle state. They map + Protocol validation failures into their local error surfaces rather than + redefining the rule. - Backup engines are in-process Agent tools and never SFP peers. - SF-AOP and the Control Plane are mandatory later MVP phases, but must not be invented during the active SFP phase. @@ -67,6 +74,28 @@ their semantics fit. Do not duplicate the standard gRPC health protocol. Do not add a protobuf solely to make every documented procedure appear structurally identical. +## Prove one authoritative owner + +Before implementing or reviewing a boundary change, inventory the matching +surface in Protocol, Agent, and Provider. Include protobuf shape, service and +RPC identity, message-size bounds, shared identifiers and scalar rules, +context-independent request and response validation, endpoint discovery, and +shared transport ceilings. Classify every rule as Protocol-owned or dependent +on runtime state; do not leave an unclassified literal, parser, or validator. + +Search consumers for copies even when their values currently match. Agent and +Provider must import the authoritative versioned contract instead of mirroring +numbers, names, regular expressions, UUID parsing, or semantic checks. A local +wrapper is justified only when it translates errors or combines canonical +validation with local state. + +Verification must include Protocol tests for the normative contract and +consumer evidence that Agent and Provider delegate to it. Passing equivalent +behavior in three independent implementations is a failure of ownership, not +proof of interoperability. Reconcile the inventory against the final diff and +name any repository or procedure that was not inspected before claiming the +boundary is complete. + ## Preserve the shared Provider endpoint - Keep the endpoint-wide gRPC receive/send ceilings separate from procedure