feat: Kafka Schema Registry — Avro/Protobuf key+value decode/encode#52
Merged
Conversation
Approved design (brainstormed 2026-07-18) and its 16-task TDD implementation plan for opt-in Avro/Protobuf key+value decode/encode via Confluent Schema Registry across consume/assert/listen/produce/mock. Co-Authored-By: Claude <noreply@anthropic.com>
…rmat defaults
Foundation for the Kafka Schema Registry / Avro / Protobuf feature. Adds
Pydantic v2 models for the additive v3 kafka: block extensions specified in
DESIGN §6.1/§6.2:
- SchemaRegistryConfig (auth: plaintext|basic|mtls; basic_auth: BasicAuth;
ssl: KafkaSSL reuse) on KafkaCluster.schema_registry.
- KafkaTopicConfig (cluster, value_format, key_format, subject_strategy) under
KafkaConfig.topics.
- KafkaCluster cluster-level format defaults (value_format="json",
key_format="string").
- BasicAuth helper model for SR basic credentials.
No cross-field validation here (topic->cluster, format-requires-SR,
auth-shape checks land in config/validator.py as Task 2). ${ENV} interpolation
unchanged (still runs in the loader before model validation). Pydantic Literal
types are the only enforcement — invalid enum values raise ValidationError at
parse time, which the loader maps to ConfigError.
Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Claude <noreply@anthropic.com>
Co-Authored-By: Claude <noreply@anthropic.com>
The validator already distinguished two error paths when a resolved value_format/key_format is avro/protobuf but the cluster has no schema_registry_url: kafka.topics.<t> (topic override drove the need) vs kafka.clusters.<c> (cluster default drove it). Only the topic-override path was tested, leaving the cluster-default branch unguarded against a silent collapse. Add two tests covering the cluster-default path (negative: error at kafka.clusters.default; positive: with SR URL, no error). 16/16 passing. Co-Authored-By: Claude <noreply@anthropic.com>
…mance failures Co-Authored-By: Claude <noreply@anthropic.com>
Adds agctl/serialization/ subpackage with a pure-stdlib wire kernel that encodes/decodes the Confluent Schema Registry wire format (1-byte magic 0x00 + 4-byte big-endian schema id + payload): - MAGIC_BYTE = 0x00 - is_confluent_frame(raw) -> bool (len>=5 and raw[0]==MAGIC_BYTE) - parse_wire(raw) -> (schema_id, payload); ValueError on non-frames - build_wire(schema_id, payload) -> bytes Foundation for the Avro/Protobuf codecs (later tasks). Deliberately import-light (only stdlib struct at module top) so it is unit-testable with no extras; the heavy codecs lazy-import fastavro/protobuf on top. Co-Authored-By: Claude <noreply@anthropic.com>
…caching + conf builder Adds agctl/serialization/registry.py: - build_schema_registry_conf(url, sr): pure translation from SchemaRegistryConfig to the dotted-key confluent SR conf dict (url always; basic.auth.user.info when basic_auth populated; ssl.* keys only for non-empty KafkaSSL fields). - SchemaRegistryClient: thin cached wrapper. __init__ builds the conf and obtains the underlying client either via the client_factory test seam or by lazy-importing confluent_kafka.schema_registry (ImportError -> ConfigError pointing at 'agctl[kafka]'). get_schema/register_schema cache by id and (subject, schema_str); all three I/O methods map underlying errors to ConnectionFailure naming the URL. check_reachable issues a lightweight get_subjects() probe for the Task 9 startup check. Tests cover conf translation (plaintext/basic/mTLS), cache hits (fake not re-called), check_reachable success+raise, the missing-extra ConfigError path via sys.modules poisoning, and one importorskip real- construction test. confluent_kafka.schema_registry stays lazy so the module imports cleanly without the kafka extra. Co-Authored-By: Claude <noreply@anthropic.com>
…rror
SchemaRegistryClient.__init__ mapped any ImportError on the lazy
confluent_kafka.schema_registry import to a ConfigError whose message was
only the generic 'pip install agctl[kafka]' hint. The cause was chained
via (visible in a traceback) but not surfaced in the message.
This is misleading in the common confluent-kafka-installed-but-authlib-missing
case: authlib is a transitive import of the SR submodule and is not pinned
by the kafka extra, so the operator sees 'install agctl[kafka]' when the
kafka extra is already installed.
The ConfigError message now interpolates the underlying error text (so the
missing module name, e.g. 'No module named authlib', is visible without a
traceback) while keeping the spec-mandated install pointer. Added a
regression test that injects ModuleNotFoundError('No module named authlib')
via __import__ patching and asserts the missing module name appears in the
surfaced message.
Co-Authored-By: Claude <noreply@anthropic.com>
Adds agctl/serialization/avro_codec.py with four names per Task 6:
- _require_fastavro(): lazy import; missing 'avro' extra surfaces as
ConfigError pointing at pip install 'agctl[avro]', never bare ImportError.
- parse_schema(schema_str): module-level cached parse (keyed by schema
string) so repeated decode/encode does not re-parse per message.
- decode_avro(raw, schema_str): fastavro.schemaless_reader on the payload
AFTER wire.parse_wire strips framing; returns dict/list/scalar.
- encode_avro(value, schema_str): fastavro.schemaless_writer(strict=True)
to bytes BEFORE wire.build_wire adds framing; extra fields raise.
Module top imports only stdlib (io, json) — fastavro lives inside the
functions. Framing remains the Task 7 API layer's job.
Tests: tests/unit/test_serialization_avro.py gated on
pytest.importorskip("fastavro") (12 cases: round-trip, encode-decode,
extra-field raises, cached parse count, lazy-import ConfigError, module
purity). 12 passed.
Co-Authored-By: Claude <noreply@anthropic.com>
…lution (Avro)
Task 7 — the agent-facing decode/encode entry points that compose the wire
kernel (T4), SchemaRegistryClient (T5), and Avro codec (T6):
- agctl/serialization/api.py: Format enum (json/avro/protobuf/string),
decode_payload, encode_payload, resolve_subject, decode_message.
JSON path needs no SR; AVRO path resolves writer schema by id (decode)
or fetches latest-by-subject (encode); PROTOBUF branch raises
ConfigError pointing at the 'protobuf' extra (wired in T14).
Codec failures propagate as SerializationError carrying fmt/subject/
schema_id; missing SR for non-JSON is ConfigError.
- agctl/serialization/__init__.py: re-export the public api surface.
- registry.py: extend SchemaRegistryClient with get_latest_schema(subject)
-> (type, str, id), cached per subject; SUBJECT_NOT_FOUND (40404)
surfaces as ConfigError ("register it before producing"), all other
errors as ConnectionFailure — the encode hot path's v1 contract is
no auto-registration.
Tests: tests/unit/test_serialization_api.py covers brief cases (a)-(g)
plus protobuf placeholder, missing-SR, decode_message; Avro cases gated
on pytest.importorskip("fastavro"). tests/unit/test_serialization_registry.py
adds get_latest_schema success/cache/40404/network-error cases via an
extended _FakeSRClient.get_latest_version seam.
Co-Authored-By: Claude <noreply@anthropic.com>
…-less ConfigError tests
Address two Minor findings from the Task 7 review (test-only).
Minor 2: extend _FakeSR with call counters and a register_schema spy;
tighten test_encode_payload_avro_uses_get_latest_schema and add a focused
happy-path test asserting get_latest_schema_calls == 1 and
register_schema_calls == 0. A regression that silently auto-registers
(or resolves via the wrong SR method) now trips an assertion.
Minor 4: drop pytest.importorskip("fastavro") from the two
*_without_sr_raises_config_error tests — the sr is None -> ConfigError
path fires before any codec call, so they now always run on fastavro-less
CI. Empirically verified they pass with fastavro hidden while the
genuinely-codec tests still skip.
Co-Authored-By: Claude <noreply@anthropic.com>
…oduce
Adds a single dependency-injection `codec` kwarg to KafkaClient.__init__:
{"value": {"fmt": Format, "subject_strategy": str|None} | None,
"key": {"fmt": Format, "subject_strategy": str|None} | None,
"sr": SchemaRegistryClient | None}
codec=None preserves today's behavior byte-for-byte (raw JSON values,
string keys) — verified by the existing 232-test kafka unit suite passing
unchanged. When set:
- produce() encodes value/key via encode_payload + resolve_subject
(non-JSON/non-string formats only); SerializationError propagates.
- consume_window/find_in_window/consume_loop decode per side via
decode_payload. Single-side decode failures are NON-fatal: reported
through a new optional `on_decode_error` callback and the failed side
becomes None (the other side still decodes — a bad key does not lose
a healthy value).
- Tombstones (value=None) decode to value=None, never counted as errors.
Decode-error surfacing mechanism chosen: `on_decode_error` callback on
each consume method (vs a returned counter). Reasons: (1) keeps return
types unchanged for backward compat; (2) caller wires its own counter
(Task 9 will surface `decode_errors: int` in the result envelope); (3)
optional, so callers who don't care see no API change.
Co-Authored-By: Claude <noreply@anthropic.com>
…--key-format flags Task 9 — command-layer wiring of the T1-T8 surfaces: - resolve_topic_format / resolve_subject_strategy / resolve_schema_registry_client / probe_schema_registry in kafka_commands.py. Topic > cluster > default precedence; SR client memoized per-invocation (module-level cache keyed by cluster name); probe wraps check_reachable and on failure raises ConnectionFailure naming BOTH the cluster and the URL (T5 review point resolved here — the cluster name is only available at this layer). - new_kafka_client gains *, codec=None forwarded to KafkaClient unchanged (T8 seam shape preserved). - --value-format <json|avro|protobuf> and --key-format <string|avro|protobuf> CLI flags on kafka produce / consume / assert (level-1 precedence override over topic/cluster resolution). - Codec wiring in the three cores: codec=None for pure-JSON topics so the legacy byte-identical decode path runs (avoids a Format.JSON-codec divergence flagged in T8 review); probe runs exactly once when a non-JSON format resolves AND an SR client exists; consume/assert result envelopes gain decode_errors: int (counter wired to the T8 on_decode_error callback). - Defense-in-depth: a non-JSON format with no schema_registry_url raises ConfigError at the command layer (validator already flags this). Tests: tests/unit/test_kafka_commands_format.py — 20 cases covering brief (a-h): resolver precedence (topic beats cluster beats default), key-side mirror, unknown-topic fall-through, SR-client memoization, probe naming cluster+URL, Click-level --value-format threading Format.AVRO into the codec via monkeypatched new_kafka_client, and consume/assert carrying decode_errors: 0 on clean runs. Plus regression guards for codec=None pure-JSON path and avro-without-SR ConfigError. TDD: RED (20 errors — _sr_client_cache and resolver symbols absent) → GREEN (20 passed). Full suite: 1553 passed, 1 skipped, 0 regressions (pure-JSON paths unchanged). Co-Authored-By: Claude <noreply@anthropic.com>
…ern items Co-Authored-By: Claude <noreply@anthropic.com>
…de_errors in summary
Task 11. Wires the Task 8 KafkaClient codec seam into the kafka listen
capture daemon so Avro (and future Protobuf) values are decoded before
the envelope is written to <topic>.ndjson.
ListenEngine.start (cfg-aware path):
- Resolves each topic's value/key Format via the Task 9 resolvers (CLI
override -> topic-level -> cluster-level -> default).
- Resolves the cluster's SchemaRegistryClient once (memoized via
Task 9's _sr_client_cache).
- Probes the SR exactly once when any topic resolves to a non-JSON
format AND an SR client exists; a probe failure raises
ConnectionFailure BEFORE any started line is emitted.
- Builds a codec-aware KafkaClient per non-JSON topic via the new
kafka_client_factory test seam (or new_kafka_client by default).
Pure-JSON topics keep the legacy byte-identical decode path
(codec=None, same guidance as Task 9's _resolve_codec).
CaptureLoop:
- run() forwards on_decode_error=self._on_decode_error to consume_loop.
- _on_decode_error emits a non-fatal decode.error event
{event, topic, error, fatal:false}, increments the per-topic
decode_errors counter, and arms a per-message skip flag.
- _handle checks the flag first: if set, clear and COMMIT without
writing (no partial/null envelope for downstream listen assert
predicates to trip on).
Summary event gains decode_errors per topic (read from each
CaptureLoop's counter).
Legacy JSON-only listeners are unchanged: cfg=None keeps every topic
on self._client with codec=None.
Tests: tests/unit/test_listen_codec.py covers (a) decoded Avro written,
(b) decode.error emitted + bad message skipped, (c) summary
decode_errors, (d) probe_schema_registry called for AVRO+SR and a
probe failure surfacing as a startup error before any started line.
Existing _FakeKafkaClient.consume_loop signature gains on_decode_error.
Co-Authored-By: Claude <noreply@anthropic.com>
…ctors Task 12: per-direction codec independence for mocks.kafka reactors. The trigger topic's format drives DECODE (the trigger client's consume_loop decodes trigger values before _handle runs); the reaction topic's format drives ENCODE (the reactor encodes via encode_payload before client.produce(_raw=True)). A reactor may decode a JSON trigger and emit an Avro reaction, or any other combination. JSON/JSON path is byte-identical to pre-Task-12. - KafkaReactor gains reaction_codec kwarg + _encode_reaction helper + _on_decode_error callback (wired via run -> consume_loop). - KafkaClient.produce gains _raw=False kwarg (caller pre-encoded). - MockEngine threads a reaction_codecs dict to each reactor. - mock_run resolves trigger+reaction codecs per reactor via _resolve_codec (probe=False on the second call per cluster to deduplicate SR probing across trigger/reaction and across sibling reactors). - _resolve_codec gains probe=True kwarg (mock_run opts out for batched resolution; existing callers unchanged). Skip/error semantics: - Trigger decode failure (codec seam's on_decode_error fires before _handle) -> kafka.skipped with "decode failed: <label>" reason (non-fatal, COMMIT; consistent with today's non-object skip). - Reaction encode failure (SerializationError from encode_payload) reuses today's retry-then-error flow -> kafka.error on the final attempt, fatal: fail_fast. Tests: tests/unit/test_mock_reactor_codec.py covers all four brief scenarios (AVRO trig -> JSON reaction, JSON trig -> AVRO reaction, decode failure -> kafka.skipped, encode failure -> kafka.error). 4/4 PASS. Existing FakeKafkaClient signatures updated to accept the new on_decode_error kwarg. Full suite 1591 passed, 21 skipped (env). Co-Authored-By: Claude <noreply@anthropic.com>
KafkaReactor._encode_reaction duplicated ~30 lines of encode logic from KafkaClient._encode_payload and the copy diverged: the key-side resolve_subject(...) passed `value` as the 4th argument instead of `key`. Dormant under the default "topic" strategy but LIVE under "record"/"topic_record": a reactor whose reaction KEY uses a non-string format (Avro/Protobuf) would resolve the wrong subject (reading __record_name__ off the value instead of the key). Root-cause fix: extract a single module-level helper _encode_payload_with_codec(codec, topic, value, key) -> (value_bytes, key_bytes) in kafka_client.py. Both call sites delegate to it, so the two paths can no longer diverge. KafkaClient.produce(_raw=True) still passes pre-encoded bytes verbatim (reactor path unchanged). Added regression test (e): reactor with AVRO key format under subject_strategy="record" with __record_name__ on the key (not the value) — asserts the SR is queried at the key's record-name subject, not the topic-name fallback the bug produced. Verified the test catches the regression by temporarily reintroducing the bug. Co-Authored-By: Claude <noreply@anthropic.com>
T8-T12 wired Avro key+value decode/encode across consume/assert/listen/ produce/mock reactors, so the prior "Non-JSON Kafka values (Avro/Protobuf) ... effectively un-mockable" limitation rows in DESIGN and ARCHITECTURE are now false for Avro. Narrow to Protobuf (codec deferred until the protobuf codec task; raises ConfigError on resolve). Co-Authored-By: Claude <noreply@anthropic.com>
…(single-file v1) Co-Authored-By: Claude <noreply@anthropic.com>
… + surface coverage
Replace the Task-7 PROTOBUF placeholder (ConfigError pointing at the
'protobuf' extra) in decode_payload / encode_payload with a dispatch to
protobuf_codec.decode_protobuf / encode_protobuf, mirroring the AVRO
branch shape (SR resolves the schema; codec does no wire-framing). AVRO
paths are byte-identical / untouched.
Adds one Protobuf case to each surface test file (gated on
pytest.importorskip("google.protobuf") + grpc_tools):
- test_serialization_api.py: decode/encode round-trip; non-frame
SerializationError; sr=None ConfigError; no-auto-register contract.
- test_kafka_client_codec.py: produce emits framed protobuf; consume
decodes.
- test_kafka_commands_format.py: --value-format protobuf threads
Format.PROTOBUF into the codec and produces framed bytes.
- test_listen_codec.py: decoded protobuf message lands in capture file.
- test_mock_reactor_codec.py: JSON trigger -> protobuf-encoded reaction.
No surface production-code changes were required — the Task 8 codec
seam and the Task 12 shared _encode_payload_with_codec helper are
genuinely format-agnostic (they branch on Format.JSON / KEY_STRING
defaults, treating everything else as "needs the codec", so PROTOBUF
flows through identically to AVRO).
Co-Authored-By: Claude <noreply@anthropic.com>
…ive deps Adds three new optional extras for the Kafka Schema Registry / Avro / Protobuf feature (Tasks 1-14): avro = ["fastavro>=1.8"] protobuf = ["protobuf>=4.25"] # standalone, no grpcio schema-registry = ["agctl[avro,protobuf]"] # convenience meta-extra Extends the `integration` extra with `agctl[schema-registry]` so live SR integration tests can decode Avro/Protobuf payloads. Fixes a critical packaging gap: confluent_kafka.schema_registry imports authlib, cachetools, attrs, certifi, and httpx at module load time, but confluent-kafka declares these only under its own [schemaregistry] extra, not as core deps. A plain `confluent-kafka>=2.4` pin left SR broken for `pip install 'agctl[kafka]'` users (ModuleNotFoundError on authlib, then cachetools). Switching the kafka extra to `confluent-kafka[schemaregistry]>=2.4` pulls all five transitive deps the way upstream intends, making SR work out of the box and tracking future upstream additions automatically. Adds tests/unit/test_packaging_extras.py with 5 tests covering the new extras, the no-grpcio rule for protobuf, the meta-extra structure, and the kafka authlib fix. Updates the existing test_jq_extra_exists assertion for the new kafka extra value. Package version unchanged (release-time bump). Co-Authored-By: Claude <noreply@anthropic.com>
…e/assert/listen/mock Adds the require_schema_registry fixture (cp-schema-registry container paired with the existing Kafka container under AGCTL_TEST_LIVE=1) and a self-skipping test module covering: (a) Avro produce/consume/assert on value+key, (b) Avro capture via kafka listen lifecycle, (c) Avro mock reactor decode-trigger + encode-reaction, (d) Protobuf value round-trip, (e) --value-format CLI override, (f) plaintext SR auth path. Basic-auth/mTLS SR documented as deferred (TLS-provisioning harness is a follow-up). Self-skip is exact: all 6 tests SKIP without AGCTL_TEST_LIVE or when Docker is absent. Co-Authored-By: Claude <noreply@anthropic.com>
Code shipped on this branch but three doc families were stale or absent: ARCHITECTURE.md (HOW): - §3 module map: add the agctl/serialization/ subpackage (api/wire/registry/ avro_codec/protobuf_codec one-liners) - §6 streaming: kafka.skipped now doubles as the per-message decode-failure signal under non-JSON reactors (mock run); add decode.error event + decode_errors summary field to kafka listen run - §7 error table: add SerializationError (exit 2) row - §8 KafkaClient: document the codec seam (codec= kwarg, on_decode_error callback, tombstone guard, _encode_payload_with_codec shared with KafkaReactor._encode_reaction) and _resolve_codec startup probe - §11 packaging: new avro/protobuf/schema-registry extras; the kafka extra is now confluent-kafka[schemaregistry] (authlib/cachetools/attrs/certifi/ httpx needed by confluent_kafka.schema_registry at module load) - §15 limitations: replace stale "No SR / Avro/Protobuf decoding" and "Protobuf codec deferred" lines with the honest single-file-Protobuf caveat and the JSON-Schema/non-Confluent-registry deferral DESIGN.md (WHAT/WHY): - §2.1 schema reference: document kafka.clusters.<c>.schema_registry auth/TLS block, the cluster-level value_format/key_format defaults, and the new kafka.topics.<t> map (value_format/key_format/subject_strategy/cluster) - §3.2 commands: add --value-format/--key-format flags to kafka produce/consume/assert; document value/key format resolution chain - §4.2 output: add decode_errors to kafka.consume / kafka.assert result shapes; add decode.error event + decode_errors summary field to kafka listen run; note kafka.skipped doubles as decode-failure signal in mock run - §7 packaging skeleton: new extras; kafka extra pin updated - §10 limitations: replace stale "Schema Registry integration" deferred row (it shipped) with the JSON-Schema/non-Confluent deferral and the Protobuf multi-file deferral; replace stale "Protobuf Kafka values" limitation row with the honest multi-file-Protobuf caveat skills/agctl-config (consumer-facing config authoring): - reference/kafka-pattern.md: replace stale "agctl treats values as raw JSON today (no Avro/Protobuf decode)" with the opt-in reality (kafka.topics.<t>.value_format + schema_registry_url) - SKILL.md structural checklist: add the kafka.topics cluster + format- requires-SR + SR-auth-shape validator checks as a fallback item Co-Authored-By: Claude <noreply@anthropic.com>
An Avro-trigger + JSON-reaction reactor mis-routed the reaction through the trigger client's codec on the publish path. The pre-fix two-branch reaction publish called `client.produce(rendered_value, ...)` WITHOUT `_raw=True` when `reaction_codec=None`, delegating encoding to the trigger client's own (Avro) codec — mis-encoding the JSON reaction as Confluent-Avro-framed bytes against `<reaction_topic>-value` (DESIGN §7.4 violation: formats resolve INDEPENDENTLY per direction). The fix collapses the publish into ONE path that ALWAYS calls `_encode_reaction` (legacy `json.dumps` bytes when reaction_codec=None) and ALWAYS publishes via `produce(..., _raw=True)`. The reaction's wire format now depends ONLY on `reaction_codec`; the trigger codec is consulted on consume (decode) only. End-to-end published bytes are identical for the legacy JSON case. Also deletes an unreachable dead `return` in `_encode_reaction`. Test fallout + new coverage: - FakeKafkaClient.produce now accepts `_raw` (the pre-existing fake predated the kwarg, so the collapsed path raised TypeError → 11 failing tests). Updated the 11 produce-call assertions to the new pre-encoded-bytes wire shape (json.dumps(...).encode() value bytes, utf-8 key bytes, `_raw is True`) — same intent, new boundary. - New regression test (Avro trigger + JSON reaction via a REAL KafkaClient with producer/consumer factory seams) pins that the reaction bytes are JSON, NOT Avro-framed. - New per-side decode independence tests (value-fails/key-fine and symmetric) — the healthy side keeps its decoded value, on_decode_error fires once with the correct side label, message not dropped. - New `kafka consume` test surfaces `decode_errors == 1` and exit 0 when the resolved codec's decode raises on one message (non-fatal). Co-Authored-By: Claude <noreply@anthropic.com>
…out) The reactor _raw fix (012708f) made every reactor publish via produce(_raw=True). The unit-suite FakeKafkaClient was updated there, but tests/integration/test_mock_capture_e2e.py has its OWN local _FakeKafkaClient whose produce() did not accept _raw — so the chatx context-echo acceptance test TypeError'd on the reaction publish and reported 0 produce calls. Accept _raw (record it) and update the assertion to the new convention: value is now pre-encoded JSON bytes (json.loads round-trip), _raw is True. Co-Authored-By: Claude <noreply@anthropic.com>
F1 CRITICAL — get_latest_schema 4-tuple unpack broken in production. get_latest_version returns a non-iterable RegisteredSchema; the unpack TypeError'd, was swallowed by except-Exception, and surfaced as ConnectionFailure — breaking every Avro/Protobuf encode. Unpack via attributes (.schema / .schema_id); the unit fake now mirrors the real shape so a future regression is caught. Two new regression tests pin both the happy path and the failure direction. F2 — decode_payload raised uncaught TypeError on None (tombstone) raw on the AVRO/PROTOBUF path (parse_wire(None) inside except ValueError). Added an early None-guard at the public API. F3 — subject_strategy record/topic_record silently degraded to TopicNameStrategy when __record_name__ was absent (chicken-and-egg with the schema fetch). Now raises ConfigError (fail-loud); the explicit __record_name__ path is preserved. F4 — integration test hardcoded .venv/bin/python; replaced with sys.executable so it skips cleanly in non-venv envs. F5 — consume_window / find_in_window counted corrupt messages toward --expect-count and the assert match (false-green: a 5-corrupt-Avro window returned count:5 exit:0). Added a per-message decode-failure tracker; corrupt messages stay in messages[] (failed side None, debug visibility) but cannot satisfy the count or the match. Doc updates in DESIGN.md and ARCHITECTURE.md codec-seam bullet. F6 — Protobuf well-known-type imports (google/protobuf/timestamp.proto) failed protoc because grpc_tools' bundled _proto dir wasn't on --proto_path. Appended it; schemas importing WKTs now compile. Full suite: 1634 passed, 26 skipped. Co-Authored-By: Claude <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Opt-in Avro + Protobuf message key+value decode/encode via Confluent Schema Registry, across all four Kafka surfaces — closes the raw-JSON-only false-negative surface (
kafka assert/consumecouldn't read Avro/Protobuf topics; mocks emittedkafka.skipped).schema_registry_url, previously parsed-but-unused, is now wired end-to-end.Strict opt-in: no format declared = today's raw-JSON behavior, byte-for-byte. No config migration, no version bump.
What's included
agctl/serialization/kernel (pure-Python at module top, lazyfastavro/protobuf): Confluent wire-frame (wire.py), lazySchemaRegistryClientwith by-id/by-subject/by-latest caching (registry.py), Avro codec, Protobuf codec (reuses thegrpc_descriptorskernel +DynamicMessage), and theFormat/decode_payload/encode_payload/resolve_subjectAPI.codecseam onKafkaClient— decode on consume (value+key independent, tombstone-safe, per-message decode failure non-fatal), encode on produce. Zero surface production-code changes when Protobuf was added — the seam is format-agnostic by construction.resolve_topic_format/resolve_schema_registry_client/probe_schema_registry) — fail-loud on unreachable/auth-failing SR before any operation.kafka.clusters.<c>.schema_registryblock (plaintext / basic-auth / mTLS) + clustervalue_format/key_formatdefaults; newkafka.topics.<t>map (value_format/key_format/subject_strategy/cluster). 4-level precedence: CLI flag > topic > cluster > default.--value-format/--key-formatonkafka produce/consume/assert.decode_errorsonkafka.consume/kafka.assert;decode.errorevent +decode_errorsinkafka listensummary;kafka.skipped reason="decode failed"for mock reactors.SerializationError(exit 2) for encode schema-conformance failure;ConnectionFailurefor SR-unreachable;ConfigErrorfor bad config / missing extra.avro/protobuf(standalone, no grpcio) /schema-registryextras;kafkaextra →confluent-kafka[schemaregistry](pulls the SR submodule's transitive deps —authlib/cachetools/attrs/certifi/httpx— so SR works out of the box).discover --category kafka-patterns --name <x>now shows resolvedvalue_format/key_format.Design references
docs/superpowers/specs/active/2026-07-18-kafka-schema-registry-formats-design.mddocs/superpowers/plans/active/2026-07-18-kafka-schema-registry-formats.md(16 TDD tasks)skills/agctl-config.Testing
AGCTL_TEST_LIVE=1; the auth-matrix (basic-auth/mTLS) is unit-covered and documented as deferred for live runs.Known v1 limitations (documented honestly)
_normalize_messagestaticmethod.🤖 Generated with Claude Code