feat: live in-place hot-reload engine (processor swap + apply decision) — PR1 of §4 (Tier 1)

pkg/conduit/runtime.go

@@ -120,6 +120,10 @@ type lifecycleService interface {
 	// lifecycle-poc(pkg/lifecycle-poc).Service.StopAndWait for why the
 	// Preview.PipelineArchV2 implementation always refuses.
 	StopAndWait(ctx context.Context, pipelineID string) error
+	// ReconfigureProcessor keeps this interface a superset of
+	// provisioning.LifecycleService (see StopAndWait above) — it is used by the
+	// live in-place apply path. See lifecycle.Service.ReconfigureProcessor.
+	ReconfigureProcessor(ctx context.Context, pipelineID, processorID string) error
 	Init(ctx context.Context) error
 }
 

pkg/lifecycle-poc/service.go

@@ -353,6 +353,18 @@ func (s *Service) StopAndWait(context.Context, string) error {
 	return ce
 }
 
+// ReconfigureProcessor refuses under the experimental Preview.PipelineArchV2
+// lifecycle service, mirroring StopAndWait: live in-place apply is not supported
+// under this arch. Returning the same coded error means a live apply is cleanly
+// refused rather than silently downgraded.
+func (s *Service) ReconfigureProcessor(context.Context, string, string) error {
+	ce := conduiterr.New(CodeStopAndWaitUnsupported,
+		"live in-place apply (processor reconfigure on a running pipeline) is not supported under "+
+			"the experimental Preview.PipelineArchV2 lifecycle service")
+	ce.Suggestion = "disable Preview.PipelineArchV2, or stop the pipeline manually before applying changes"
+	return ce
+}
+
 // buildRunnablePipeline will build and connect all tasks configured in the pipeline.
 func (s *Service) buildRunnablePipeline(
 	ctx context.Context,

pkg/lifecycle-poc/stop_and_wait_test.go

@@ -49,3 +49,23 @@ func TestServiceLifecycle_StopAndWait_Unsupported(t *testing.T) {
 	is.Equal(ce.Code.Reason(), CodeStopAndWaitUnsupported.Reason())
 	is.True(ce.Suggestion != "")
 }
+
+// TestServiceLifecycle_ReconfigureProcessor_Unsupported pins the same parity
+// refusal for the live in-place hot-reload path (PR1 of §4): this preview
+// lifecycle service must refuse ReconfigureProcessor with the same stable code
+// rather than silently no-op, so provisioning.applyInPlace never runs an
+// in-place swap against the unaudited arch.
+func TestServiceLifecycle_ReconfigureProcessor_Unsupported(t *testing.T) {
+	is := is.New(t)
+
+	logger := log.New(zerolog.Nop())
+	ls := NewService(logger, testConnectorService{}, testProcessorService{}, testConnectorPluginService{}, testPipelineService{}, true)
+
+	err := ls.ReconfigureProcessor(context.Background(), uuid.NewString(), uuid.NewString())
+	is.True(err != nil)
+
+	ce, ok := conduiterr.Get(err)
+	is.True(ok)
+	is.Equal(ce.Code.Reason(), CodeStopAndWaitUnsupported.Reason())
+	is.True(ce.Suggestion != "")
+}

pkg/lifecycle/reconfigure.go

@@ -0,0 +1,82 @@
+// Copyright © 2026 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package lifecycle
+
+import (
+	"context"
+
+	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/conduit/pkg/lifecycle/stream"
+)
+
+// ErrProcessorNotLiveReconfigurable signals that a processor cannot be swapped in
+// place on the running pipeline, so the caller must fall back to a restart-class
+// apply. It is returned when the processor is not running as a plain,
+// single-worker stream.ProcessorNode — e.g. it is parallelized (Workers > 1, and
+// so runs as a stream.ParallelNode), or no matching node is present. Restarting
+// is always a safe superset: the restart path re-imports the config and rebuilds
+// every node, so it applies the change regardless of the running node's shape.
+var ErrProcessorNotLiveReconfigurable = cerrors.New("processor is not live-reconfigurable, a restart is required")
+
+// ReconfigureProcessor swaps a processor's configuration in a running pipeline in
+// place — without stopping or restarting the pipeline — by rebuilding the
+// runnable processor from its (already-updated) stored instance and swapping it
+// into the live stream.ProcessorNode via ProcessorNode.Reconfigure. See the
+// design doc "Live in-place hot-reload".
+//
+// Ordering contract: the caller MUST persist the processor's new config to its
+// instance before calling this (ReconfigureProcessor reads the current instance
+// from the processor service). The swap happens at a record boundary; open of
+// the new processor happens before teardown of the old, so if the new config
+// fails to open, the old processor keeps running and the error is returned — the
+// pipeline never drops (invariant 3).
+//
+// Returns ErrProcessorNotLiveReconfigurable if the processor is not a plain
+// single-worker node in the running pipeline; the caller should then fall back to
+// a restart. Returns an error (old processor kept) if building or opening the new
+// processor fails.
+func (s *Service) ReconfigureProcessor(ctx context.Context, pipelineID, processorID string) error {
+	rp, ok := s.runningPipelines.Get(pipelineID)
+	if !ok {
+		return cerrors.Errorf("pipeline %q is not running, cannot reconfigure processor %q in place", pipelineID, processorID)
+	}
+
+	// A single-worker processor runs as a *stream.ProcessorNode named for the
+	// processor ID. A parallel processor (Workers > 1) runs as a
+	// *stream.ParallelNode instead and is not matched here, so it falls through
+	// to the restart fallback below — as does a processor with no live node.
+	var node *stream.ProcessorNode
+	for _, n := range rp.n {
+		pn, isProcNode := n.(*stream.ProcessorNode)
+		if isProcNode && pn.ID() == processorID {
+			node = pn
+			break
+		}
+	}
+	if node == nil {
+		return cerrors.Errorf("%w: processor %q in pipeline %q", ErrProcessorNotLiveReconfigurable, processorID, pipelineID)
+	}
+
+	instance, err := s.processors.Get(ctx, processorID)
+	if err != nil {
+		return cerrors.Errorf("could not fetch processor %q for live reconfigure: %w", processorID, err)
+	}
+	runnableProc, err := s.processors.MakeRunnableProcessor(ctx, instance)
+	if err != nil {
+		return cerrors.Errorf("could not build runnable processor %q for live reconfigure: %w", processorID, err)
+	}
+
+	return node.Reconfigure(ctx, runnableProc)
+}

pkg/lifecycle/reconfigure_test.go

@@ -0,0 +1,69 @@
+// Copyright © 2026 Meroxa, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package lifecycle
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/conduitio/conduit/pkg/foundation/cerrors"
+	"github.com/conduitio/conduit/pkg/foundation/log"
+	"github.com/conduitio/conduit/pkg/lifecycle/stream"
+	"github.com/conduitio/conduit/pkg/pipeline"
+	"github.com/matryer/is"
+)
+
+func newReconfigureTestService() *Service {
+	return NewService(
+		log.Nop(),
+		testErrRecoveryCfg(),
+		testConnectorService{},
+		testProcessorService{},
+		testConnectorPluginService{},
+		testPipelineService{},
+	)
+}
+
+// TestService_ReconfigureProcessor_NotRunning: reconfiguring a processor in a
+// pipeline that is not currently running is an error (nothing live to swap).
+func TestService_ReconfigureProcessor_NotRunning(t *testing.T) {
+	is := is.New(t)
+	s := newReconfigureTestService()
+
+	err := s.ReconfigureProcessor(context.Background(), "does-not-exist", "proc1")
+	is.True(err != nil)
+	is.True(strings.Contains(err.Error(), "not running"))
+}
+
+// TestService_ReconfigureProcessor_NoPlainProcessorNode_FallsBackToRestart: when
+// the running pipeline has no plain single-worker ProcessorNode for the target
+// (it is absent, or parallelized as a ParallelNode), ReconfigureProcessor returns
+// ErrProcessorNotLiveReconfigurable so the caller falls back to a restart.
+func TestService_ReconfigureProcessor_NoPlainProcessorNode_FallsBackToRestart(t *testing.T) {
+	is := is.New(t)
+	s := newReconfigureTestService()
+
+	// A running pipeline whose nodes include no *stream.ProcessorNode named
+	// "proc1" — stands in for both the "parallelized" and "absent" cases, which
+	// both must degrade to a restart.
+	s.runningPipelines.Set("pl1", &runnablePipeline{
+		pipeline: &pipeline.Instance{ID: "pl1"},
+		n:        []stream.Node{&stream.FaninNode{}, &stream.FanoutNode{}},
+	})
+
+	err := s.ReconfigureProcessor(context.Background(), "pl1", "proc1")
+	is.True(cerrors.Is(err, ErrProcessorNotLiveReconfigurable))
+}

pkg/lifecycle/stream/base.go

@@ -82,6 +82,15 @@ func (n *pubSubNodeBase) Sub(in <-chan *Message) {
 	n.subNodeBase.Sub(in)
 }
 
+// In returns the inbound message channel this node reads from. It is exposed so
+// a PubSubNode (e.g. ProcessorNode) can select over the inbound channel together
+// with its own control signals (such as a live-reconfigure wake) instead of
+// blocking solely in Trigger's Receive. The channel is set by Sub during wiring;
+// In returns nil until then.
+func (n *pubSubNodeBase) In() <-chan *Message {
+	return n.subNodeBase.in
+}
+
 func (n *pubSubNodeBase) Pub() <-chan *Message {
 	return n.pubNodeBase.Pub()
 }