fix(binding-kafka): honor per-message authorization on the cache fetch path

[jfallows]

Description

KafkaCacheServerFetchFactory decoded a fetched Kafka record's value once, while populating the local cache from the broker, with authorization hardcoded to NO_AUTHORIZATION — there is no requesting consumer yet at that point. The decoded result was stored in the partition's convertedFile and served to every future reader verbatim, including the per-consumer fetch stream (KafkaCacheClientFetchFactory), which constructed no model pipeline of its own at all and never threaded its own live authorization into anything — meaning the per-message authorization parameter ModelPipeline.transform(...) already carries (#2369) never reached a real, per-consumer value on this path.

Correction (superseding the original design)

The first version of this fix (now superseded by later commits on this branch) added a separate, abstract ModelHandler.supplyCacheable(...) method alongside supplyDecoder/supplyEncoder. CI caught a real regression in that design: the http.kafka.avro.json example hung for its full 30-minute timeout. Root cause — supplyCacheable and supplyDecoder used the same frontend parser and output format, differing only in which ext-handler fold ran. That's fine for a byte-for-byte identity decode, but any model whose decode does real, format-converting work (e.g. view: json) produces output that differs from its wire-format input. Caching that converted output and then re-decoding it with a wire-format frontend can never work: the avro model's view: json config made the populate-time decode cache JSON text, and the per-consumer fetch-time decode then fed that JSON into a parser expecting avro wire bytes.

The corrected design replaces the two-method split with a single supplyDecoder(envelope, transform, ModelCache), parameterized by a three-way cache context:

• NONE — today's plain decode, no cache involved.
• WRITE — decode ahead of any specific reader, producing the value a cache persists (the old supplyCacheable's role).
• READ — decode a value already in whatever form WRITE produced, for the reader requesting it now. Its frontend matches whatever WRITE actually emitted — for avro/protobuf's view: json, that's AvroJson.stream/ProtobufJson.parser, the same json-view frontend already used on the encode-shaped pipeline, reused rather than duplicated.

Each model's own *ModelExtHandler SPI mirrors this as a single decode(T, Cache) method — a local enum per model (AvroCache/ProtobufCache/JsonCache/CoreCache), not the shared engine ModelCache, so each model's extension surface stays independent — replacing the previous decode(T)/cacheable(T) pair.

Building this also surfaced and closed a related gap: protobuf's own message-index framing (separate from catalog framing) has no magic-byte-style validity guard, so a cached view:json value with no index framing at all would misparse its leading JSON bytes as an index varint on READ. Fixed by resolving the message via the same static catalog.record path the encode side already uses whenever cache == READ.

Finally, WRITE now persists the schema id it resolves as catalog framing ahead of its cached output (reusing the same CatalogHandler.encode(...) machinery the encode-direction pipeline already uses), so READ recovers it directly from the cached bytes via the existing resolveSchemaId/prefix logic, with no changes needed on the read side. This closes the one case that otherwise stays broken even with the frontend fix: a strategy: encoded schema id is per-message authoritative (schema evolution means two cached messages on the same topic can carry different ids), and a view-converting WRITE output previously discarded it entirely with no static fallback available — unlike strategy: topic/subject-pinned configs, whose schema was never message-embedded to begin with.

Verification

• AvroModelDecoderPipelineTest/ProtobufModelDecoderPipelineTest gained WRITE→READ round-trip tests for both the view: json and no-view cases, plus a case proving READ recovers the schema id from WRITE-injected framing even when the model's own static catalog reference is deliberately wrong — the exact strategy: encoded shape. These are the tests that would have caught the original regression; TestModel's identity decode structurally cannot exercise this failure mode. Exercising the framing case required extending the engine's TestCatalogHandler test double (resolve(DirectBufferEx,...)/decodePadding(...)) to honor its own prefix option symmetrically with its existing encode(...), per this repo's convention of extending type: test implementations additively rather than reaching for a production catalog implementation in test scope.
• CacheFetchIT gained a new scenario (shouldReceiveMessageValueAvroViewJson) driving a real avro + view: json config through a live engine populate-then-fetch round trip — the exact shape of the original regression — verified passing against the live engine, not just unit-level.
• Full runtime/binding-kafka suite (401 tests, all k3po ITs included), specs/binding-kafka.spec suite (peer-to-peer + config), runtime/model-avro/model-protobuf/model-json/model-core suites, and runtime/binding-http/binding-sse/binding-mqtt/binding-mcp suites all pass with no regressions; checkstyle and license checks clean.
• CI on this branch's merge with latest develop (commit d63e78ab9) is green across all 42 checks, including testing (http.kafka.avro.json) (2m27s, vs. the original 30-minute hang) and testing (mcp.proxy).

Fixes #2423

---

Generated by Claude Code