fix: durable signal delivery preserves source port for edge config eval

v0.4.0 had a real bug exposed by in-cluster validation: durable-port
persist captures msg.Data BEFORE MsgHandler runs edge config evaluation.
The signal stored the source's raw output (e.g. an http response), but
the target port (e.g. kv:store) expected an already-mapped target shape
(e.g. StoreRequest with primary key populated by edge expressions).

Reconciler delivery would then unmarshal raw source bytes as target type
and the component would reject with structural errors ("primary key
not found in document"), the reconciler would treat as transient,
retry forever, and signals would accumulate.

Fix:
- TinySignalReconciler now sends msg.From = signal.Spec.From (the
  original source port) when the spec carries it, instead of always
  FromSignal. This makes MsgHandler treat the message as a regular
  edge delivery and run edge config evaluation, mapping source's raw
  payload into the target port's expected shape.
- To prevent infinite re-persist (msg.From != FromSignal would normally
  retrigger the durable-port persist branch), reconciler marks the
  delivery context with utils.WithDurableDelivery, and scheduler.Handle
  checks this flag in addition to From.

Externally-injected signals (send_signal MCP, ticker fire) keep their
existing behavior: Spec.From is empty, msg.From defaults to FromSignal,
MsgHandler treats data as already-typed.

New test in scheduler_durable_test.go covers the
durable-delivery-context-skips-persist case so future regressions
can't reintroduce the loop.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: tpoxa

internal/controller/tinysignal_controller.go

@@ -128,6 +128,10 @@ func (r *TinySignalReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	deliveryCtx, cancel := context.WithTimeout(context.Background(), deliveryTimeout)
 	defer cancel()
 	deliveryCtx = utils.WithLeader(deliveryCtx, true)
+	// Mark this delivery so scheduler.Handle skips the durable-port persist
+	// step (we ARE the reconciler delivering a persisted signal — re-persisting
+	// would be an infinite loop).
+	deliveryCtx = utils.WithDurableDelivery(deliveryCtx)
 
 	// Inject a remote span context if the signal carries a TraceID so
 	// the resulting trace stitches under the caller-supplied root.
@@ -146,9 +150,19 @@ func (r *TinySignalReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		}
 	}
 
-	l.Info("delivering signal", "targetPort", targetPort)
+	// For durable-port-persisted signals, signal.Spec.From carries the
+	// original source port so MsgHandler runs edge config evaluation
+	// (target's "from this source" port-config maps source output → target
+	// input shape). For externally-injected signals (send_signal MCP),
+	// Spec.From is empty and we use FromSignal — MsgHandler treats the
+	// payload as already-typed in that case (existing legacy behavior).
+	deliveryFrom := runner.FromSignal
+	if signal.Spec.From != "" {
+		deliveryFrom = signal.Spec.From
+	}
+	l.Info("delivering signal", "targetPort", targetPort, "from", deliveryFrom)
 	_, deliverErr := r.Scheduler.Handle(deliveryCtx, &runner.Msg{
-		From:   runner.FromSignal,
+		From:   deliveryFrom,
 		To:     targetPort,
 		Data:   signal.Spec.Data,
 		EdgeID: signal.Spec.EdgeID,

internal/scheduler/scheduler.go

@@ -218,10 +218,18 @@ func (s *Schedule) Handle(ctx context.Context, msg *runner.Msg) (any, error) {
 	}
 
 	// Durable-port intake. If the target port is marked Durable AND this
-	// message did not arrive from the TinySignal reconciler (which would
-	// already be a delivery of a persisted signal), persist a TinySignal
-	// CRD and ack immediately. The reconciler will deliver it later.
-	if msg.From != runner.FromSignal {
+	// is NOT a reconciler-driven delivery of an already-persisted signal,
+	// persist a TinySignal CRD with the source's raw payload and ack
+	// immediately. The reconciler later replays the signal back through
+	// scheduler.Handle with utils.WithDurableDelivery(ctx), which skips
+	// this branch and lets MsgHandler evaluate the edge config (mapping
+	// source's output shape into target's expected port shape).
+	//
+	// Externally-injected signals (send_signal MCP, ticker fire) arrive
+	// with msg.From == runner.FromSignal; we let those through without
+	// persisting (they're already targeting a port and the existing
+	// signal-path treats data as already-typed).
+	if !utils.IsDurableDelivery(ctx) && msg.From != runner.FromSignal {
 		if p, ok := instance.GetPort(port); ok && p.Durable && s.manager != nil {
 			name := signalname.For(nodeName, port, msg.Data, msg.EdgeID, "")
 			if err := s.manager.PersistDurableSignal(ctx, name, nodeName, instance.Node().Namespace, port, msg.Data, msg.EdgeID, msg.From); err != nil {

internal/scheduler/scheduler_durable_test.go

@@ -19,6 +19,7 @@ import (
 	"github.com/tiny-systems/module/internal/scheduler/runner"
 	"github.com/tiny-systems/module/module"
 	"github.com/tiny-systems/module/pkg/state"
+	"github.com/tiny-systems/module/pkg/utils"
 	metricnoop "go.opentelemetry.io/otel/metric/noop"
 	tracenoop "go.opentelemetry.io/otel/trace/noop"
 	"k8s.io/apimachinery/pkg/runtime"
@@ -281,6 +282,41 @@ func TestDurablePort_SignalDeliveryDoesNotRePersist(t *testing.T) {
 	}
 }
 
+// TestDurablePort_DeliveryContextSkipsPersist verifies the "don't loop forever"
+// guard: when the reconciler replays a persisted signal back through
+// scheduler.Handle with utils.WithDurableDelivery(ctx), the durable-port
+// branch must skip and dispatch normally — otherwise we'd re-persist the
+// same signal infinitely.
+func TestDurablePort_DeliveryContextSkipsPersist(t *testing.T) {
+	mgr := &recordingManager{}
+	s := newDurableTestSchedule(t, mgr)
+
+	cmp := &durableSinkComponent{name: "dsink"}
+	if err := s.Install(cmp); err != nil {
+		t.Fatalf("Install: %v", err)
+	}
+	installNode(t, s, "dsink-1", "dsink")
+
+	// Use a non-FromSignal From (mimics real durable-port persist) but
+	// mark the context as durable delivery (mimics reconciler replay).
+	ctx := utils.WithDurableDelivery(context.Background())
+	_, err := s.Handle(ctx, &runner.Msg{
+		From: "source-node:source-port",
+		To:   "dsink-1:in",
+		Data: []byte(`{}`),
+	})
+	if err != nil {
+		t.Fatalf("Handle: %v", err)
+	}
+
+	if calls := mgr.recordedCalls(); len(calls) != 0 {
+		t.Errorf("durable-delivery context should suppress persist; got %d calls", len(calls))
+	}
+	if cmp.hits() != 1 {
+		t.Errorf("expected Component.Handle to be invoked once; got %d", cmp.hits())
+	}
+}
+
 // TestDurablePort_NonDurablePortStillFastPaths verifies that ports without
 // Durable=true preserve the existing blocking-I/O fast path: no persist
 // call, Handle invoked synchronously.

pkg/utils/ctx.go

@@ -5,10 +5,28 @@ import "context"
 type contextKey string
 
 const (
-	leaderKey     contextKey = "leader"
-	sourceNodeKey contextKey = "sourceNode"
+	leaderKey           contextKey = "leader"
+	sourceNodeKey       contextKey = "sourceNode"
+	durableDeliveryKey  contextKey = "durableDelivery"
 )
 
+// WithDurableDelivery marks the context as a TinySignalReconciler-driven
+// delivery of a previously-persisted durable-port signal. Scheduler.Handle
+// uses this to skip the durable-port persist step and instead route the
+// message through normal MsgHandler dispatch (so edge config evaluation
+// runs at delivery time, mapping the source's raw output into the target
+// port's expected shape).
+func WithDurableDelivery(ctx context.Context) context.Context {
+	return context.WithValue(ctx, durableDeliveryKey, true)
+}
+
+// IsDurableDelivery reports whether the current call is the reconciler
+// delivering an already-persisted durable signal.
+func IsDurableDelivery(ctx context.Context) bool {
+	v, _ := ctx.Value(durableDeliveryKey).(bool)
+	return v
+}
+
 func WithLeader(ctx context.Context, isLeader bool) context.Context {
 	return context.WithValue(ctx, leaderKey, isLeader)
 }