APPROVE. Conservative, well-scoped fix for #6427 — diff only ignores the four flow-control knobs (max_queued_records, max_queued_bytes, max_batch_size, max_worker_batch_size), transport config still triggers a modified connector, and the checkpoint-replay path copies the new limits onto the resumed connector so the bump actually takes effect. Both unit tests (positive: flow-control changes don't modify; negative: transport change still modifies) plus the platform test_bootstrap_input_flow_control_changes with bootstrap_policy=REJECT cover the contract end-to-end. Nice work.

Non-blocking notes:

1. normalize_for_input_checkpoint_replay is now a strict superset of equal_modulo_paused (paused-zero plus the four flow-control fields). To keep the two helpers from drifting, consider expressing equal_modulo_paused in terms of a smaller normalize_paused and having the replay variant call both, or at minimum a comment noting that any new flow-control field added in the future must be added in two places (normalize_for_input_checkpoint_replay and apply_input_checkpoint_replay_config_from). Easy thing to forget six months from now.

2. apply_input_checkpoint_replay_config_from deliberately doesn't touch paused, which is consistent with the prior inputs: checkpoint_config.inputs behavior on the unmodified path (paused state from the checkpoint wins). Worth a one-line comment to make that explicit — otherwise the asymmetry with the four flow-control fields invites a "should we copy paused too?" question on the next pass.

3. The platform test uses datagen with limit=10 then wait_for_completion(timeout_s=300) before the restart. Tight, good. One thought: assertion is only on COUNT(*) round-tripping to 10 after restart; if you wanted to be paranoid you could also assert the new max_queued_records=200 is actually observed on the running connector (via stats), but that's well past "non-blocking" — the bootstrap-doesn't-fire assertion is the real test.

4. Pure cosmetics in controller_init_tests: the if matches!(...) block now has a misaligned ), on the DatafusionMemoryExceedsBudget arm (looks like an artifact of the import-block edit nearby). cargo fmt will fix it.

Not blocking on any of these. Storage-compat and output-connector diff are correctly untouched. Ship it.

Re-approving on 1de9d5a8d7.

Good catch on the HTTP input replay merge ordering bug. The previous flow mutated config.inputs with the HTTP-input transfer before feeding it into checkpoint_inputs_with_replay_safe_config, which meant HTTP input connectors silently picked up the checkpoint's flow-control limits instead of the new ones for the unmodified-pipeline case — directly contradicting this PR's stated invariant. Factoring out inputs_for_checkpoint_replay and snapshotting new_inputs before the HTTP transfer fixes it cleanly, and the new checkpoint_http_input_replay_uses_pre_transfer_config test pins down exactly that ordering. Nice tightening.

LGTM.

Re-APPROVE on 1de9d5a. The two new commits fix a real ordering bug and pin it down with a test:

• Prior code path (84cbd28 and earlier). ControllerInit::new first transferred HTTP input endpoints from the checkpoint into config.inputs (overwriting the pipeline-manager's fresh config for those names), then called checkpoint_inputs_with_replay_safe_config(checkpoint.inputs, &config.inputs) with that mutated config.inputs. For an HTTP input, "config.inputs" at the merge site already held the checkpoint's transport block, so the replay-safe limit merge picked up the checkpoint's max_queued_* settings instead of the new ones. End result: HTTP inputs would replay using the old flow-control limits even when the user had bumped them in the new config.
• New path (inputs_for_checkpoint_replay). Clones config.inputs into new_inputs before the HTTP transfer, transfers HTTP inputs into config.inputs, then — only when the pipeline is unmodified — calls checkpoint_inputs_with_replay_safe_config(checkpoint_inputs, &new_inputs) with the pre-transfer snapshot. Now the replay-safe limits come from the genuine new config, regardless of whether the input is HTTP or not. The pipeline_diff.is_empty() gate replaces the !modified flag from the call site and is the cleaner expression of the condition.
• checkpoint_http_input_replay_uses_pre_transfer_config pins the merge order: an HTTP input present in both checkpoint and config gets the HTTP transport block from the checkpoint and the 200/400/201/202 limits from the new config. That's exactly the case the bug was breaking. Good targeted test.

One small thought: the modified-vs-unmodified branch inside inputs_for_checkpoint_replay returns the fully-mutated config_inputs (which now contains transferred HTTP inputs) when the pipeline is modified. That preserves the prior behavior for modified pipelines, but the comment says "If the pipeline is unmodified, we may need to replay journaled inputs." It might be worth one more sentence stating what the modified branch does ("Otherwise, we keep the new pipeline-manager config — including transferred HTTP/adhoc inputs needed to replay journaled inputs after a program-diff bootstrap.") so a future reader doesn't have to reverse-engineer it.

APPROVE.

I read this fairly carefully and I think it's correct and helpful.

Re-review after 4242a1b0 ("Ignore replay-safe input connector changes").

The refactor is clean:

• The old inline logic in ControllerInit::new (HTTP-input transfer + "unmodified pipeline → checkpoint inputs" branch) is now two named helpers: inputs_for_checkpoint_replay and checkpoint_inputs_with_replay_safe_config. Both are testable, and both tests you added exercise the interesting corner cases (flow-control adoption on the unmodified path, and pre-transfer new-config for the HTTP-input replay path).
• equal_for_input_checkpoint_replay replacing equal_modulo_paused is the right generalization: max_queued_records, max_queued_bytes, max_batch_size, max_worker_batch_size, and paused are all runtime flow-control knobs that don't invalidate journaled connector state, so treating them as "same connector" avoids forcing a bootstrap on an SLO tweak. The pipeline_diff unit test locks that behavior in.
• The Python integration test (test_bootstrap_input_flow_control_changes) verifies end-to-end that changing all four limits + paused between checkpoint and restart resumes cleanly without bootstrapping and preserves committed state. Good.

One asymmetry worth calling out (not necessarily blocking):

normalize_for_input_checkpoint_replay zeros paused (so a paused-flag change doesn't count as a modification), but apply_input_checkpoint_replay_config_from does not copy paused over. Net effect: if the user pauses a connector in the new config and restarts from checkpoint on an unmodified pipeline, the checkpoint's paused = false wins and the user's pause intent is silently dropped. This behavior existed under equal_modulo_paused too, so this PR doesn't regress it, but it's a papercut users will hit eventually. Two reasonable options: (a) include paused in apply_input_checkpoint_replay_config_from so config wins on replay, or (b) file a follow-up and document the current semantics near normalize_for_input_checkpoint_replay.

Minor style: checkpoint_inputs_with_replay_safe_config takes mut checkpoint_inputs and mutates in place before the filter/collect. Reads fine, but combining the mutation and the ClockInput filter into a single pass would drop one traversal — cosmetic, ignore if you prefer the current shape.

Approving; the paused-adoption question is worth a moment but I don't want to hold up a clean refactor with a good test suite over pre-existing behavior.