Right shape for the first slice. The new InputTransportEndpointFactory / OutputTransportEndpointFactory traits + per-transport *InputFactory structs decompose the previous monolithic match in controller.rs::connect_input / connect_output cleanly. builtin_*_transport_registry() keeps every built-in registration in one place, and the Option<Box<dyn ...>> return convention preserves the "fall through to integrated connector" path correctly. Behavior is unchanged: the registry lookup hits the same factories the old match arms used to instantiate, error paths still go through ControllerError::input_transport_error / output_transport_error, secret resolution still happens up front. Net -89 lines of dispatch despite the new trait surface is a good sign.

A few thoughts:

• registered: BTreeMap<&'static str, ...> in InputTransportRegistry. This is fine for built-ins (where keys are TransportConfig::FILE_INPUT etc. &'static str constants), but it boxes any future third-party / user-provided factory into needing a 'static name. Dnreikronos's follow-up comment on #6529 ("the registry key change... belongs in this slice") is exactly right — switching to String here costs nothing at the call sites in builtin_*_transport_registry() (.to_string()) and unblocks any dynamic registration later. It's a one-line change in the type definition; doing it here saves a churn-the-API-again later. Suggest folding the #6529 key-type widening into this PR.
• get(&self, name: &str): BTreeMap<&'static str, _>::get(&str) needs name.to_owned() or a Borrow<str> workaround at the call site — the current code does .get(&config.name()) which is &String against &&'static str. This compiles only because name() returns &'static str, but a future caller passing an owned String will get a confusing type error. Same root cause as the previous bullet.
• Factories implement create() by matching one specific TransportConfig variant. That's the right shape, but it means the factory registry and the TransportConfig enum stay coupled: adding a new transport requires both editing the enum and adding a factory. The next slice (#6529) addresses this by moving config parsing into the factories, which is the real prize. Worth a one-line code comment at InputTransportEndpointFactory noting that the current shape is transitional and that #6529 generalizes it — readers picking up either PR in isolation will be confused otherwise.
• KafkaInputFactory always creates KafkaFtInputEndpoint. I read this twice; the previous code did the same thing (the non-FT path was already dead). Carries forward, not a regression — but worth a note in the commit body that KafkaInput and KafkaFtInput configs both go through KafkaFtInputEndpoint so future readers don't expect a non-FT branch here.
• Test coverage. The PR is a pure refactor (no behavior change), and every transport is covered by existing integration tests, so the refactor is implicitly tested. No new tests needed. (No "tests for refactor" demand from me on this one — that's the right call for a no-op extraction.)
• Cosmetic. Several factory struct *Factory; definitions are followed by a blank line then impl. Consistent. The #[cfg(feature = "...")] attributes on each factory + its impl are paired correctly. A possible compression: a small factory! macro to generate the per-transport boilerplate, but only worth it once a few more transports land; not worth doing now.

Direction is right; happy to APPROVE after the &'static str → String widening (or your call to defer it explicitly).

The &'static str → String widening landed in Allow dynamic transport registry keys (adapterlib/src/transport.rs): BTreeMap<String, ...> on both registries, register(name: impl Into<String>, ...), and get(name: &str) reads cleanly (config.name().as_str()). The builtin &'static str constants flow in for free via Into, and a future dynamic factory can register an owned String without ceremony. The two ergonomic loose ends I flagged are gone.

LGTM.

Fourth commit tightens the registry contract nicely. Two changes worth calling out:

• InputTransportEndpointFactory::create / OutputTransportEndpointFactory::create now return Result<Box<...>> (not Result<Option<Box<...>>>). Right call — a factory that has been dispatched to by transport name should never Ok(None). The unexpected_input_transport_config / unexpected_output_transport_config helpers convert the "wrong variant" case into a loud error with the factory name and transport name, so any dispatch bug is visible at runtime instead of silently short-circuiting. The selected_factory_rejects_mismatched_transport_config test pins this behaviour.
• New input_transport_uses_registry / output_transport_uses_registry predicates enumerate every TransportConfig variant that goes through the registry. When the built-in registry doesn't return a factory (feature compiled out), the top-level helpers now return UnknownInputTransport / UnknownOutputTransport rather than the previous Ok(None) fall-through. Three #[cfg(not(feature = ...))] tests confirm the compiled-out kafka/redis paths surface the missing-factory error. This is the right shape — compiling out kafka should fail loudly on a kafka pipeline config, not produce a mysterious null endpoint.

Diff is +245/-111 on transport.rs, +30/-12 on controller.rs, +6/-7 on adapterlib/transport.rs; the churn is almost entirely mechanical Option<Box<...>> → Box<...>. No functional regressions on the happy path.

Non-blocking nits:

• The two _uses_registry helpers are hand-maintained enumerations of TransportConfig variants. If someone adds a new variant and forgets these, the compiled-out case will silently regress to Ok(None). Consider a #[non_exhaustive]-style compile-time check, or at minimum a comment on TransportConfig reminding contributors to update the predicates.
• unexpected_input_transport_config / unexpected_output_transport_config return anyhow!(...) strings that get wrapped in ControllerError::input_transport_error. That path is fine, but this class of failure is a programming bug (registry dispatched to the wrong factory), not a config error — a debug_assert! alongside the error would help surface it in tests without changing production behaviour.

Verdict stays APPROVE. Nothing here that should block merge.

This looks fine to me, but I am not the expert on connectors to make the final decision.