feat: Lanes B/C/D — Source/Destination lifecycle, config/record codec, acceptance harness + example

[devarismeroxa]

Summary

Implements Lanes B, C, D of the Python connector SDK per docs/design/20260707-python-connector-sdk.md, completing the Phase-1 (v0.19 core) author-facing surface on top of Lane A (handshake + generated gRPC stubs, already merged).

Lane C — config/record:

• conduit/record.py: Record/Change/Operation/Data (bytes | Mapping[str, Any]), Metadata well-known-key constants + typed accessors.
• conduit/config.py: BaseConfig (pydantic v2) + Field + to_parameters() — introspects model_fields into config.Parameter/Validation with no codegen. gt=/lt= map exactly; ge=/le= are approximated as exclusive gt/lt (exact for int, epsilon-nudged for float, documented). TYPE_DURATION/TYPE_EXCLUSION raise NotImplementedError (documented open A-gaps, never guessed).
• conduit/errors.py: BackoffRetry, ConnectorError, and BatchWriteError — the B1 fix. Construction requires an exhaustive, disjoint success/failures accounting (or a written= prefix); incomplete accounting raises ValueError at construction time.

Lane B — lifecycle over gRPC v2:

• conduit/source.py / conduit/destination.py: Source/Destination as abc.ABC generic over config, dual sync/async dispatch (conduit/_dispatch.py), plus the internal _SourceServicer/_DestinationServicer adapters.
• conduit/serve.py: serve() entry point — handshake validation (reusing Lane A's _handshake.py), grpc.aio server, health/specifier/connector servicer registration, and _ShutdownCoordinator — the hung-event-loop watchdog (▶ MUST-FIX 3).
• conduit/_grpc/_controller.py: hand-written GRPCController.Shutdown (go-plugin's own internal proto, not part of conduit-connector-protocol).
• conduit/_grpc/adapters.py: Record/Data/Change ↔ proto conversion, including the B3 Struct int→float boundary at its one documented call site.
• Invariant 1 enforced structurally: Source.ack() is called only from the ack-consumer loop, driven only by Conduit's ack_positions; Destination acks/nacks are driven only by write() returning cleanly or BatchWriteError's pre-validated accounting.
• Read-loop backoff reuses the Go SDK's exact constants (Factor=2, Min=100ms, Max=5s).

Lane D — acceptance harness + example:

• conduit/testing/acceptance.py: AcceptanceTestDriver Protocol, ConfigurableAcceptanceTestDriver, AcceptanceTestSuite (contract version 2026-07.v1) covering all six design-doc categories.
• conduit/testing/fixtures.py: golden record-shape factories.
• examples/http-poll-source/: the design doc §2.7 worked example, made fully runnable, plus its own pyproject.toml.

Failure-mode analysis (Tier 1)

• Partial-batch destination write (B1): an incomplete/absent per-index accounting is a ValueError at BatchWriteError.__init__ time — the adapter has no code path that computes "ack everything not explicitly failed." Any non-BatchWriteError exception nacks the whole batch. Regression tests: test_destination_partial_write_nacks_all's three cases in tests/test_destination.py, plus a duck-typing-resemblance case (_RaisesNaiveIncompleteMapping) proving only isinstance(exc, BatchWriteError) grants partial credit.
• Early source ack: Source.ack() has exactly one call site (_consume_acks), driven only by ack_positions arriving on the Run request stream. The read loop (_read_loop) never references ack(). Verified by tests/test_source.py::TestAckOnlyAfterConduitConfirms.
• GRPCController.Shutdown not implemented / silently falls through: implemented and tested end-to-end (real grpc.aio server, real client, real RPC) in tests/test_serve.py::TestDeterministicShutdownRpc — asserts the RPC succeeds and teardown() ran exactly once beforehand via a spy, not a timing race (▶ MUST-FIX 2's tightened bar).
• Hung event loop (▶ MUST-FIX 3, no Go analog): _ShutdownCoordinator installs SIGTERM via low-level signal.signal (not loop.add_signal_handler) and starts an independent threading.Timer watchdog. Tested directly (test_watchdog_fires_when_shutdown_never_confirmed, test_watchdog_does_not_fire_if_confirmed_before_deadline, test_start_watchdog_is_idempotent) and against a genuinely wedged real event loop in a background thread (test_watchdog_forces_exit_even_with_a_genuinely_wedged_event_loop), proving the watchdog fires independent of the loop's own thread. Scope note: this does not exercise real OS SIGTERM delivery timing to a wedged main thread — that interaction is flagged as its own, separately hard-to-construct-deterministically concern (see Self-review below), appropriate for compat-nightly/Conduit-repo-side verification, not this unit suite.
• server.stop(grace=None) racing the Shutdown RPC's own response: investigated because grpc.aio.Server.stop's docstring says a None grace period "aborts all existing RPCs immediately" — a plausible race where the Shutdown RPC's own response gets cut off by the subsequent stop() call from the concurrent drive_shutdown task. Empirically stress-tested (30 repeated runs of the shutdown test class, no flakiness) and the full suite (5 repeated full runs, no flakiness). Read as: grpc.aio's internal RPC-completion bookkeeping for an already-returned handler resolves before stop()'s "wait for last handler to terminate" clause proceeds. Flagged in Self-review as verified-by-repetition, not verified by reading grpc-core's C-level implementation.
• B3 Struct int→float precision loss: documented and pinned exactly (not glossed over with == laxity) in tests/test_record_codec.py::test_large_int_in_structured_data_loses_precision_silently. Contrasted with the loud-failure bytes-inside-structured-data case.
• Rollback: this is a new-file-only PR (no existing behavior changed); rollback is reverting the merge commit. No serialized-format/protocol changes.

Self-review (adversarial pass)

1. Any code path where an unaccounted destination-write index gets acked? Checked _acks_from_batch_write_error: the only ack branch is if i in exc.success; the else branch always nacks (using exc.failures.get(i, <generic RuntimeError>) — never falls through to an ack). Verified this holds even under a hypothetically buggy connector that mismatches BatchWriteError's declared batch_size against the real records list length (traced both under- and over-declaration cases) — no path over-acks. No issue found.
2. Any path where a source record could be acked before ack_positions confirms it? self._source.ack has exactly one call site, _consume_acks, itself driven only by iterating the Run request stream. _read_loop (the producer side) never references ack. No issue found.
3. Watchdog timer thread leak on the normal (non-wedged) path? start_watchdog() is only ever invoked from _on_sigterm — meaning if SIGTERM never arrives (the ordinary GRPCController.Shutdown-only path), no threading.Timer is created at all, nothing to leak. When SIGTERM does arrive and shutdown completes normally, confirm_clean_exit() sets the confirmed flag before calling timer.cancel(), closing the race window against _force_exit (which checks the flag first). No issue found.
4. Struct int/float boundary mishandled anywhere outside the one documented site? Grepped for Struct/MessageToDict/structured_data across hand-written code — exactly one encode site (_data_to_proto) and one decode site (_data_from_proto), both in conduit/_grpc/adapters.py, both documented. No issue found.
5. New finding during self-review, documented above under failure modes: the server.stop(grace=None)-vs-Shutdown-RPC-response race. Resolved via empirical stress-testing rather than static reasoning alone — flagged so DeVaris can decide whether that's sufficient confidence for this property or whether it warrants a grpc-core-level investigation before sign-off.

This is Tier 1 (data path) and needs DeVaris's explicit sign-off — do not merge.

Test plan

• ruff format --check . — pass (26 files)
• ruff check . — pass
• mypy — pass, 16 source files, no errors
• pytest — 98 passed (5 repeated full runs, no flakiness; 30 repeated runs of the shutdown-RPC test class specifically, no flakiness)
• git status --porcelain -- src/conduit/_grpc/connector src/conduit/_grpc/opencdc src/conduit/_grpc/config — empty (generated dirs untouched)
• Acceptance suite passes against a synthetic driver (tests/test_acceptance_harness.py) and against the real, unmodified example connector (tests/test_example_http_poll_source.py) — 8/8 categories each, not skipped or stubbed
• Real Conduit-binary launch + conduit pipelines stop graceful-shutdown verification — out of scope for this repo's CI per the design doc (compat-nightly.yml/Conduit-repo-side, not built here)

🤖 Generated with Claude Code
https://claude.ai/code/session_015GQFzakPShAYj8CcwajYDD

[devarismeroxa]

DX audit follow-up: 5 must-fixes applied (5 new commits on this branch)

Per the coordinator's request, applied all five DX-audit items to this same branch/PR. Summary per item:

1. Duration config → real timedelta mapping (deae18d)

• config.Parameter.Type.TYPE_DURATION is now a real, round-tripping mapping instead of NotImplementedError. New public conduit.config.format_go_duration()/parse_go_duration() implement Go's time.Duration string syntax ("5s", "1h30m", "500ms", "1.5h", ns/us/µs/ms/s/m/h units) using exact fractions.Fraction arithmetic, not floats.
• to_parameters() serializes a timedelta default via format_go_duration(); gt=/lt=/ge=/le= on duration fields use exact ±1-microsecond boundary adjustment (timedelta is discrete, like int — no epsilon needed).
• BaseConfig gained a model_validator(mode="before") that parses a Go-duration string into a real timedelta before pydantic validation, so Configure's map<string,string> round-trips correctly. Direct construction with a real timedelta still works.
• TYPE_EXCLUSION remains untouched (still an open, documented A-gap, still NotImplementedError per your instruction not to expand scope there).
• Tests: tests/test_go_duration.py (known-Go-output pins + Hypothesis round-trip property + malformed-syntax rejection), tests/test_config.py.

2. Source.ack() genuinely inert by default (ac1e90e)

• Verified: already a plain return None, no log line, no raise. Docstring now states explicitly: "you don't need to override ack() unless you're acknowledging against the source system itself (e.g. committing a Kafka offset, deleting a queue message)." Same note added to examples/http-poll-source/README.md.

3. Lifecycle hook rename (ac1e90e)

• lifecycle_on_created/lifecycle_on_updated/lifecycle_on_deleted → on_created/on_updated/on_deleted on both Source and Destination. Grepped the whole tree (including _dispatch.py/_introspect.py, which had no references) — no stragglers. tests/test_destination.py updated.

4. BatchWriteError.partial() + Configure ValidationError forwarding (ac1e90e)

• BatchWriteError.partial(batch_size, written=N, cause=exc): the new recommended constructor — every failed index gets your real cause exception instead of a generic placeholder, reusing the existing exhaustive-accounting path under the hood (still no code path that computes "ack everything not explicitly failed").
• Configure (both Source and Destination) now catches pydantic.ValidationError explicitly and context.abort()s with INVALID_ARGUMENT + a per-field detail message (errors.format_validation_error()), instead of grpc.aio's generic UNKNOWN-status "Unexpected : ..." wrapping.
• Test: tests/test_configure_errors.py — a real grpc.aio server + real client, asserting the actual gRPC status code and that the field name/message are present in the status detail (not a mock of pydantic or the transport).

5. conduit-connector-sdk build CLI (392e3e4, 036cf02)

• New console-script entry point, build subcommand: packages a connector project into one self-contained, directly-executable artifact with an absolute-interpreter-path shebang (closes design doc §1.1.6's packaging gap — no PATH lookup at exec time).
• Vendors from the current environment's installed distributions (importlib.metadata: file manifests + transitive Requires-Dist, extras-filtered) rather than a fresh pip install — this SDK isn't on PyPI yet, so a fresh resolve would fail; this also means build needs no network and runs in ~0.2–0.4s. conduit-connector-sdk itself is vendored by copying its actual installed location directly (works for this repo's own editable dev install too).
• Not a plain zipapp: grpcio and pydantic-core both ship compiled extensions, which zipimport cannot load from inside a zip. Found this the hard way mid-implementation — an earlier version hard-errored on any compiled extension, which would have made this SDK's own core deps impossible to vendor. Fixed with a small, dependency-free bootstrap __main__.py that extracts the real payload to a per-build cache directory on first run (the same fundamental approach shiv/pex use), then runs the connector's real entry point from those extracted files.
• Real bug found and fixed via this same end-to-end testing: sending a real SIGTERM to a running serve() process whose connector's teardown() raised (e.g. HTTPPollSource.teardown() accessing self._client before open() ever ran) used to silently hang for the full watchdog deadline instead of shutting down promptly. Root cause: _sigterm_shutdown()'s coroutine runs via run_coroutine_threadsafe with its Future never awaited, so an exception inside it was silently swallowed and shutdown_requested was never set. Fixed: shutdown_requested.set() now runs unconditionally in a finally block, with a diagnostic distinguishing "teardown raised" from "loop genuinely wedged." Also hardened the example connector's own teardown() against running before open() did.
• Windows CI failure found and fixed (036cf02): Windows has no POSIX executable bits and no shebang-based direct execution (OSError: [WinError 193]) — confirmed via the actual CI failure, not guessed. Fixed the test suite to invoke via sys.executable on Windows (documenting the platform gap explicitly) and skip the SIGTERM-graceful-shutdown test there (Popen.send_signal(SIGTERM) maps to an unconditional TerminateProcess() on Windows, not something our signal handler ever sees — a real, separate feature gap, not solved here, matching the design doc's already-flagged "Windows subprocess launch specifics" risk).
• Test: tests/test_build.py builds the real example connector and execs the resulting artifact directly as a subprocess (mirroring exactly how Conduit's dispenser launches a plugin) — asserts a valid handshake line, absolute-path shebang, compiled extensions present as real extracted files, cache reuse on a second launch, and the SIGTERM regression test above (POSIX only).

Gate results (final, this branch's HEAD, all green in CI)

• ruff format --check . — pass (37 files)
• ruff check . — pass
• mypy — pass, 18 source files, 0 errors
• pytest — 163 passed, 0 skipped on macOS/Linux (2 skipped on Windows for the documented signal-handling reasons above), not faked
• git status --porcelain -- src/conduit/_grpc/{connector,opencdc,config} — empty
• CI matrix (ubuntu/macos/windows × Python 3.12/3.13) + ruff + mypy: all 8 checks green — https://github.com/ConduitIO/conduit-connector-sdk-python/actions/runs/30041753713 / .../30041753696

Still Tier 1 — needs DeVaris's explicit sign-off, still not merged

This PR remains Tier 1 (data path / wire adapter / config codec). The self-review from the original PR description still applies; this comment adds no new unresolved risk beyond what's documented above (the Windows graceful-shutdown gap and the extraction-cache concurrency note in _build.py's docstring are both explicitly flagged as known, out-of-scope-for-this-fix limitations, not silently glossed over).

🤖 Generated with Claude Code
https://claude.ai/code/session_015GQFzakPShAYj8CcwajYDD

[devarismeroxa]

Invariant-7 gap fix + design-doc overclaim fix (Tier-1 review)

The gap (confirmed): serve.py's _sigterm_shutdown ran teardown() immediately on SIGTERM with no regard for whether Run() was actively streaming — it never signaled the read/write loop to stop and never awaited an in-flight Destination.write()/Source.read(). Confirmed not a data-loss bug (a raced write nacks the whole batch via the generic exception branch → Conduit redelivers on restart), but a real invariant-7 (graceful shutdown by default) gap on a Tier-1 SDK.

Fix:

• _SourceServicer.drain() / _DestinationServicer.drain() (new) factor out the same stop-then-wait ordering Stop() already performs for the deterministic path (Conduit Stop RPC → Run ends → Teardown RPC). _DestinationServicer gained the same _stop_event/_stopped_event/try…finally shape _SourceServicer already had, so drain() waits for the entire Run() generator to finish — including its already-computed ack response reaching the grpc.aio framework — not just for write() itself to return. (Waiting only on write() was tried first and raced server.stop() against an already-earned-but-unflushed ack in testing; waiting on the generator's finally closes that too.)
• serve.py's _sigterm_shutdown now awaits drain() before run_teardown_once(). Bounded by the existing hung-loop watchdog deadline, unchanged — the watchdog thread starts before _sigterm_shutdown is even scheduled, so a stuck connector still force-exits on schedule.
• Both drain() and teardown() remain individually exception-safe (shutdown_requested.set() unconditional in finally) — preserves the existing "buggy teardown can't hang shutdown forever" fix.

Doc overclaim fixed: Risks & open questions §3 asserted the ordinary SIGTERM-mid-write case was "covered by the ordinary SIGTERM-mid-write test" — no such test existed; the only SIGTERM test (test_build.py::test_sigterm_triggers_prompt_graceful_shutdown) fires before Open(). Added the actual tests and pointed the doc at them by name.

New tests (tests/test_serve.py::TestSigtermDrainsInFlightOperation):

• test_sigterm_mid_read_drains_before_teardown
• test_sigterm_mid_write_drains_before_teardown

Both build the real grpc.aio server via _build_plugin_server, drive a real bidi-streaming Run() call against it, block the connector mid-read()/mid-write(), invoke the coordinator's real _on_sigterm directly (deterministic, not OS-signal-timing-dependent), and assert the in-flight call's completion strictly precedes teardown() in an ordered event log — plus that the ack/record response was actually delivered and the watchdog never fired.

Gates: ruff format --check clean, ruff check clean, mypy --strict clean (18 source files), full pytest 165 passed (was 163) including the two new tests. Generated _grpc/ dirs unchanged (diff touches only docs/design/...md, src/conduit/{source,destination,serve}.py, tests/test_serve.py).

Not merging — Tier-1, needs DeVaris sign-off per CLAUDE.md.