From 438098068bb0e1bf49a97f37c485f99310339a36 Mon Sep 17 00:00:00 2001 From: krishung5 Date: Tue, 3 Nov 2026 15:49:45 -0800 Subject: [PATCH 1/3] feat(protocol): add encoder-output handoff contracts Signed-off-by: krishung5 --- proto/openengine/v1/generation.proto | 32 ++++++++++++++++++++++++++++ proto/openengine/v1/model.proto | 10 +++++++++ proto/openengine/v1/openengine.proto | 4 ++++ 3 files changed, 46 insertions(+) diff --git a/proto/openengine/v1/generation.proto b/proto/openengine/v1/generation.proto index c1f81dd..a2527ee 100644 --- a/proto/openengine/v1/generation.proto +++ b/proto/openengine/v1/generation.proto @@ -69,6 +69,27 @@ message GenerateRequest { // this field empty. Clients treat it as best-effort and never depend on it // for correctness. Carried as a Struct so JSON types survive the wire. google.protobuf.Struct extra = 12; + + // Encoder execution or an engine-owned encoder-cache handoff. Consumers must + // negotiate support before setting this field; ignoring it is not valid. + EncoderOptions encoder = 13; +} + +message EncoderCacheRef { + string transfer_backend = 1; + // Versioned, engine-specific reference. Tensor contents remain engine-owned. + bytes opaque_state = 2; +} + +message EncoderOptions { + // Produce EncoderReady without generating language tokens. + bool encode_only = 1; + // Consume a previous encoder result. Mutually exclusive with encode_only. + EncoderCacheRef cache = 2; +} + +message EncoderReady { + EncoderCacheRef cache = 1; } message SamplingParams { @@ -156,6 +177,8 @@ message GenerateResponse { PrefillReady prefill_ready = 4; GenerationFinished finished = 5; EngineError error = 6; + // Request-scoped terminal for a successful encoder-only request. + EncoderReady encoder_ready = 7; } Usage usage = 10; // Cumulative request usage; only set on the final response. @@ -221,3 +244,12 @@ message Usage { optional uint32 cached_prompt_tokens = 4; optional uint32 reasoning_tokens = 5; } + +// Engine-to-engine encoder-output transfer. The destination descriptor is +// backend-specific and must be validated by the negotiated transfer backend. +message TransferEncoderCacheRequest { + EncoderCacheRef cache = 1; + bytes destination = 2; +} + +message TransferEncoderCacheResponse {} diff --git a/proto/openengine/v1/model.proto b/proto/openengine/v1/model.proto index f8b8c9c..49234ff 100644 --- a/proto/openengine/v1/model.proto +++ b/proto/openengine/v1/model.proto @@ -38,6 +38,16 @@ message ModelInfo { // contract: clients read it opportunistically and never depend on it for // correctness. Carried as a Struct so JSON types survive the wire. google.protobuf.Struct extra = 28; + + // Negotiated encoder role and supported encoder-cache transports. + EncoderCapabilities encoder = 29; +} + +message EncoderCapabilities { + uint32 schema_version = 1; + bool encoder_only = 2; + bool accepts_cache = 3; + repeated string transfer_backends = 4; } message GenerationCapabilities { diff --git a/proto/openengine/v1/openengine.proto b/proto/openengine/v1/openengine.proto index 153d071..03be4ec 100644 --- a/proto/openengine/v1/openengine.proto +++ b/proto/openengine/v1/openengine.proto @@ -57,6 +57,10 @@ service Control { rpc UnloadLora(UnloadLoraRequest) returns (UnloadLoraResponse); rpc ListLoras(ListLorasRequest) returns (ListLorasResponse); + // Returns only after the encoder backend has completed the data transfer. + // Single-use handoffs must not be retried after an uncertain outcome. + rpc TransferEncoderCache(TransferEncoderCacheRequest) returns (TransferEncoderCacheResponse); + // Disaggregated serving / KV transfer. Connector info: ServerInfo.kv_connector. rpc GetKvEventSources(GetKvEventSourcesRequest) returns (GetKvEventSourcesResponse); rpc SubscribeKvEvents(SubscribeKvEventsRequest) returns (stream SubscribeKvEventsResponse); From 9b6f34f9efb17aa2c6b7e0ad737b78106a8ecca3 Mon Sep 17 00:00:00 2001 From: krishung5 Date: Wed, 4 Nov 2026 08:57:47 -0800 Subject: [PATCH 2/3] docs(protocol): define encoder handoff validation and lifetime Signed-off-by: krishung5 --- docs/api.md | 100 +++++++++++++++++++++++++-- proto/openengine/v1/generation.proto | 2 +- 2 files changed, 97 insertions(+), 5 deletions(-) diff --git a/docs/api.md b/docs/api.md index 646ffcb..0a0c9a5 100644 --- a/docs/api.md +++ b/docs/api.md @@ -41,6 +41,9 @@ service Control { rpc UnloadLora(UnloadLoraRequest) returns (UnloadLoraResponse); rpc ListLoras(ListLorasRequest) returns (ListLorasResponse); + // Engine-to-engine encoder-output transfer. + rpc TransferEncoderCache(TransferEncoderCacheRequest) returns (TransferEncoderCacheResponse); + // Disaggregated serving / KV transfer. Connector info: ServerInfo.kv_connector. rpc GetKvEventSources(GetKvEventSourcesRequest) returns (GetKvEventSourcesResponse); rpc SubscribeKvEvents(SubscribeKvEventsRequest) returns (stream SubscribeKvEventsResponse); @@ -205,6 +208,14 @@ message ModelInfo { string reasoning_parser = 25; string tool_call_parser = 26; google.protobuf.Struct extra = 28; // Engine-specific, non-portable; read opportunistically. + EncoderCapabilities encoder = 29; +} + +message EncoderCapabilities { + uint32 schema_version = 1; + bool encoder_only = 2; + bool accepts_cache = 3; + repeated string transfer_backends = 4; } message GenerationCapabilities { @@ -338,6 +349,7 @@ message GenerateRequest { repeated MediaItem media = 10; string lora_name = 11; google.protobuf.Struct extra = 12; // Engine-specific, non-portable; may be ignored. + EncoderOptions encoder = 13; } message TokenIds { @@ -479,6 +491,7 @@ message GenerateResponse { PrefillReady prefill_ready = 4; GenerationFinished finished = 5; EngineError error = 6; + EncoderReady encoder_ready = 7; } Usage usage = 10; @@ -544,7 +557,7 @@ message Usage { } ``` -`PromptOutput` and `PrefillReady` are request-scoped. `TokenOutput` and +`PromptOutput`, `PrefillReady`, and `EncoderReady` are request-scoped. `TokenOutput` and `GenerationFinished` are output-scoped and must carry `output_index`, including index zero. An index is stable for the request and ranges from `0` through `num_sequences - 1`. @@ -561,13 +574,15 @@ text; neither field is cumulative. After `GenerationFinished` for an index, the engine must not emit another token or terminal event for that index. A successful aggregated or decode request emits exactly one `GenerationFinished` for each requested output. `PrefillReady` is the terminal -success event for a prefill-only request. +success event for a prefill-only request. `EncoderReady` is the terminal success +event for an encoder-only request; it does not also emit `GenerationFinished` +or language tokens. `EngineError` is request-scoped and terminal. It terminates all outputs and may replace `GenerationFinished` for outputs that had not completed. No event may follow it. `Usage` is cumulative across every output in the request and is set only on the final response, which is the last `GenerationFinished`, -`PrefillReady`, or the terminal `EngineError`. +`PrefillReady`, `EncoderReady`, or the terminal `EngineError`. When `FinishReason` is `STOP`, `stop_match` identifies the matched caller stop token, stop string, or model EOS token. `cached_prompt_tokens` is a subset of @@ -575,6 +590,82 @@ token, stop string, or model EOS token. `cached_prompt_tokens` is a subset of --- +## Encoder execution and output handoff + +An encoder-only engine prepares multimodal model inputs without running language +generation. A consuming engine uses those inputs for generation. Encoder-output +handoff and prefill/decode KV handoff are separate operations; support for one +does not imply support for the other or for their composition. + +```protobuf +message EncoderCacheRef { + string transfer_backend = 1; + bytes opaque_state = 2; +} + +message EncoderOptions { + bool encode_only = 1; + EncoderCacheRef cache = 2; +} + +message EncoderReady { + EncoderCacheRef cache = 1; +} + +message TransferEncoderCacheRequest { + EncoderCacheRef cache = 1; + bytes destination = 2; +} + +message TransferEncoderCacheResponse {} +``` + +### Capability negotiation and validation + +`ModelInfo.encoder.schema_version = 1` identifies this encoder contract. +An absent capability, zero version, or version the client does not understand +does not authorize encoder requests. `encoder_only` advertises encoder-only +execution; `accepts_cache` advertises consumption. Both roles list their +supported backend names in `transfer_backends`. Clients select a common backend +before dispatch; an empty list offers no compatible transport. + +Omitting `GenerateRequest.encoder` preserves ordinary generation behavior on +generation engines. An encoder-only engine requires `encoder.encode_only = true`. +When `encoder` is present, exactly one of `encode_only = true` or `cache` being +present is required. Empty options and simultaneous encoding/consumption are +invalid. A cache reference requires nonempty `transfer_backend` and +`opaque_state`, and the consumer must advertise that backend. Servers reject +invalid or unsupported combinations before acceptance rather than silently +ignoring the encoder option. Ordinary model/input and modality checks still +apply; encoder capability alone does not enable every modality or topology. + +### Encoding, transfer, and ownership + +1. The client sends the prompt and media to an encoder-only engine using + `Generate` with `encoder.encode_only = true`. +2. Successful encoding emits one request-scoped terminal `EncoderReady` holding + an engine-owned cache reference. An accepted failure emits `EngineError`. +3. The client forwards that reference unchanged in `encoder.cache` to a + compatible generation engine. Tensor contents remain outside the inference + RPC payload; the backend owns transfer and validation of the reference. +4. A backend that needs receiver coordination prepares its destination buffers + and calls `Control.TransferEncoderCache` on the encoder engine. `destination` + is a nonempty backend-specific descriptor, not a portable tensor layout. + The backend validates the descriptor and request/model identity before + starting the transfer. A successful unary response means the data transfer + has completed, not merely that it was queued. + +Cache references are single-use and backend-versioned. The backend owns their +expiration, quotas, and cleanup on consumption, cancellation, timeout, and +shutdown. Callers cannot assume a reference survives indefinitely. Cancellation +or a lost transfer response does not prove that transfer stopped or that the +reference remains unused: never replay that handoff after an uncertain outcome. +Implementations retain buffers while transfer may still access them and reclaim +them only after the backend establishes completion or failure. These ownership +rules apply even when the waiting RPC is cancelled. + +--- + ## Disaggregated serving and KV API The core API makes prefill/decode handoff explicit through @@ -974,7 +1065,8 @@ accepted stream phase and report failures with non-OK gRPC status. `GenerationFinished` is terminal for its `output_index`; other output indexes may continue. The last `GenerationFinished` ends a successful aggregated or -decode stream, `PrefillReady` ends a successful prefill stream, and +decode stream, `PrefillReady` ends a successful prefill stream, +`EncoderReady` ends a successful encoder-only stream, and `EngineError` ends any failed generation stream. An `EngineError` also terminates a KV-event subscription. No response may follow a terminal `EngineError`. Application failure is not a `GenerationFinished` reason. diff --git a/proto/openengine/v1/generation.proto b/proto/openengine/v1/generation.proto index a2527ee..d45cb79 100644 --- a/proto/openengine/v1/generation.proto +++ b/proto/openengine/v1/generation.proto @@ -169,7 +169,7 @@ message JsonObjectConstraint {} message GenerateResponse { string request_id = 1; - // PromptOutput, PrefillReady, and EngineError are request-scoped. + // PromptOutput, PrefillReady, EncoderReady, and EngineError are request-scoped. // TokenOutput and GenerationFinished are output-scoped. oneof event { PromptOutput prompt = 2; From 7b4171659dad80d4820f1024345b18e397158641 Mon Sep 17 00:00:00 2001 From: Kris Hung Date: Thu, 5 Nov 2026 20:38:37 -0800 Subject: [PATCH 3/3] feat(protocol): release abandoned encoder cache references Signed-off-by: Kris Hung --- .github/workflows/buf.yml | 12 +++++ docs/api.md | 16 ++++++- proto/openengine/v1/lifecycle.proto | 4 ++ tests/test_encoder_abort_contract.py | 70 ++++++++++++++++++++++++++++ 4 files changed, 101 insertions(+), 1 deletion(-) create mode 100644 tests/test_encoder_abort_contract.py diff --git a/.github/workflows/buf.yml b/.github/workflows/buf.yml index 9c1bfff..a05f228 100644 --- a/.github/workflows/buf.yml +++ b/.github/workflows/buf.yml @@ -6,11 +6,13 @@ on: - main paths: - "proto/**" + - "tests/**" - "buf.yaml" - ".github/workflows/buf.yml" pull_request: paths: - "proto/**" + - "tests/**" - "buf.yaml" - ".github/workflows/buf.yml" @@ -34,3 +36,13 @@ jobs: push: false archive: false pr_comment: false + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Check wire contracts + run: | + python -m pip install grpcio-tools==1.67.1 protobuf==5.27.2 + python -m unittest discover -s tests -v diff --git a/docs/api.md b/docs/api.md index 0a0c9a5..bf3b770 100644 --- a/docs/api.md +++ b/docs/api.md @@ -655,6 +655,16 @@ apply; encoder capability alone does not enable every modality or topology. starting the transfer. A successful unary response means the data transfer has completed, not merely that it was queued. +If a completed encoder result will not be dispatched, the caller releases it +at the producer using `Control.Abort` with `encoder_cache`. `ABORTED` means the +unclaimed export was removed. `ALREADY_FINISHED` means it was absent, consumed, +or already claimed; a claimed transfer retains ownership until settlement. +Release is idempotent and must not interrupt a transfer or free its buffers. +If release races a claimed transfer that later fails preflight, the backend +must reclaim the export after settlement rather than restore it for reuse. +Malformed references return `INVALID_ARGUMENT`; unsupported release returns +`UNIMPLEMENTED`. A lost release response can be retried with the same reference. + Cache references are single-use and backend-versioned. The backend owns their expiration, quotas, and cleanup on consumption, cancellation, timeout, and shutdown. Callers cannot assume a reference survives indefinitely. Cancellation @@ -662,7 +672,10 @@ or a lost transfer response does not prove that transfer stopped or that the reference remains unused: never replay that handoff after an uncertain outcome. Implementations retain buffers while transfer may still access them and reclaim them only after the backend establishes completion or failure. These ownership -rules apply even when the waiting RPC is cancelled. +rules apply even when the waiting RPC is cancelled. If `EncoderReady` may +have been delivered but the caller never learned its reference, cancellation +cannot safely revoke that reference: this uncertain-delivery case remains +bounded by the backend's expiration, quotas, and producer shutdown cleanup. --- @@ -909,6 +922,7 @@ message AbortRequest { string request_id = 1; KvSessionRef kv_session = 2; AllRequests all_requests = 3; + EncoderCacheRef encoder_cache = 4; } } diff --git a/proto/openengine/v1/lifecycle.proto b/proto/openengine/v1/lifecycle.proto index 9c4d0fb..8ff4ce2 100644 --- a/proto/openengine/v1/lifecycle.proto +++ b/proto/openengine/v1/lifecycle.proto @@ -7,6 +7,7 @@ syntax = "proto3"; package openengine.v1; +import "openengine/v1/generation.proto"; import "openengine/v1/kv.proto"; import "openengine/v1/server.proto"; @@ -47,6 +48,9 @@ message AbortRequest { string request_id = 1; KvSessionRef kv_session = 2; AllRequests all_requests = 3; + // Release an unclaimed single-use encoder export at its producer. This never + // interrupts a claimed transfer or releases buffers still in use by it. + EncoderCacheRef encoder_cache = 4; } } diff --git a/tests/test_encoder_abort_contract.py b/tests/test_encoder_abort_contract.py new file mode 100644 index 0000000..75467e9 --- /dev/null +++ b/tests/test_encoder_abort_contract.py @@ -0,0 +1,70 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Wire compatibility for the additive encoder-cache abort target.""" + +import tempfile +import unittest +from pathlib import Path + +import grpc_tools +from google.protobuf import descriptor_pb2, descriptor_pool, message_factory +from grpc_tools import protoc + + +class EncoderAbortContract(unittest.TestCase): + @classmethod + def setUpClass(cls): + root = Path(__file__).resolve().parents[1] + with tempfile.TemporaryDirectory() as temporary: + output = Path(temporary) / "schema.pb" + result = protoc.main( + [ + "protoc", + f"-I{root / 'proto'}", + f"-I{Path(grpc_tools.__file__).parent / '_proto'}", + f"--descriptor_set_out={output}", + "--include_imports", + str(root / "proto/openengine/v1/lifecycle.proto"), + ] + ) + if result: + raise RuntimeError("Could not compile the lifecycle contract") + descriptors = descriptor_pb2.FileDescriptorSet.FromString( + output.read_bytes() + ) + pool = descriptor_pool.DescriptorPool() + for descriptor in descriptors.file: + pool.Add(descriptor) + cls.request = message_factory.GetMessageClass( + pool.FindMessageTypeByName("openengine.v1.AbortRequest") + ) + + def test_existing_targets_and_encoder_reference_roundtrip(self): + fields = self.request.DESCRIPTOR.fields_by_name + self.assertEqual( + {name: field.number for name, field in fields.items()}, + {"request_id": 1, "kv_session": 2, "all_requests": 3, "encoder_cache": 4}, + ) + for name, value in ( + ("request_id", "request"), + ("kv_session", {"session_id": "session"}), + ("all_requests", {}), + ("encoder_cache", {"transfer_backend": "test", "opaque_state": b"opaque"}), + ): + with self.subTest(target=name): + request = self.request(**{name: value}) + decoded = self.request.FromString(request.SerializeToString()) + self.assertEqual(decoded, request) + self.assertEqual(decoded.WhichOneof("target"), name) + self.assertEqual( + fields["encoder_cache"].message_type.full_name, + "openengine.v1.EncoderCacheRef", + ) + request.encoder_cache.SetInParent() + request.request_id = "replacement" + self.assertFalse(request.HasField("encoder_cache")) + + +if __name__ == "__main__": + unittest.main()