fix: preserve durable replay fidelity under opt-in trace observability

Signed-off-by: Joshua Temple <joshua.temple@stablekernel.com>

Author: joshua-temple

durable/runner.go

@@ -246,7 +246,12 @@ func (r *Runner[S, E, C]) Start(ctx context.Context, id InstanceID, input C, opt
 	// (WithClock); the scheduler reads it when arming and ticking timers.
 	buf := make([]state.JournalEntry, 0)
 	recClock := newRecordingClock(r.cfg.clock, &buf)
-	castOpts := append([]state.CastOption[S]{state.WithClock[S](recClock)}, opts...)
+	// Force unbounded full-trace history: the runner journals each step's Trace
+	// (notably the EventPayload) and derives both the step ordinal and the checkpoint
+	// snapshots from the instance's retained Trace history, so a recover can replay it
+	// byte-identically. WithUnboundedHistory elevates to full trace and retains every
+	// step; it is prepended so a caller's opts cannot downgrade it.
+	castOpts := append([]state.CastOption[S]{state.WithClock[S](recClock), state.WithUnboundedHistory[S]()}, opts...)
 	inst := r.machine.Cast(input, castOpts...)
 
 	// Persist a baseline checkpoint at baselineStep (below the first fired step)
@@ -530,7 +535,7 @@ func (r *Runner[S, E, C]) recover(ctx context.Context, id InstanceID) (*Handle[S
 	recClock := newRecordingClock(r.cfg.clock, &buf)
 	repClock := newReplayClock(clockReadings(tail), recClock)
 
-	inst, err := r.machine.Restore(snap, state.WithRestoreClock[S](repClock))
+	inst, err := r.machine.Restore(snap, state.WithRestoreClock[S](repClock), state.WithRestoreUnboundedHistory[S]())
 	if err != nil {
 		return nil, fmt.Errorf("durable: restoring checkpoint for %q: %w", id, err)
 	}

durable/runner_test.go

@@ -95,7 +95,10 @@ func parallelMachine() *state.Machine[string, string, *runCtx] {
 // recovered instance must match byte-for-byte.
 func liveSnapshotBytes[C any](t *testing.T, m *state.Machine[string, string, C], initial string, entity C, events []string) []byte {
 	t.Helper()
-	inst := m.Cast(entity, state.WithInitialState(initial))
+	// The Runner records in full-trace, unbounded-history mode, so the never-durable
+	// reference must run in the same mode for a byte-identical snapshot comparison;
+	// a plain (lite) cast would omit the recorded Trace history.
+	inst := m.Cast(entity, state.WithInitialState(initial), state.WithUnboundedHistory[string]())
 	ctx := context.Background()
 	for _, ev := range events {
 		if res := inst.Fire(ctx, ev); res.Err != nil {

durable/timetravel.go

@@ -210,7 +210,10 @@ func restoreBaseline[S comparable, E comparable, C any](
 	if err != nil {
 		return nil, fmt.Errorf("unmarshaling baseline snapshot: %w", err)
 	}
-	inst, err := m.Restore(snap, state.WithRestoreClock[S](repClock))
+	// Reconstruct in the same full-trace, unbounded-history mode the runner records
+	// in, so a time-travel reconstruction's Trace history is byte-identical to the
+	// live run it replays.
+	inst, err := m.Restore(snap, state.WithRestoreClock[S](repClock), state.WithRestoreUnboundedHistory[S]())
 	if err != nil {
 		return nil, fmt.Errorf("restoring baseline snapshot: %w", err)
 	}

state/actor.go

@@ -294,6 +294,19 @@ func NewActor[S comparable, E comparable, C any](inst *Instance[S, E, C], output
 	return a
 }
 
+// inheritObservability puts the actor's backing child instance into the parent's
+// trace mode: full trace, and (for a durable parent) unbounded history retention so
+// the actor's own snapshot round-trips its Traces. The ActorSystem calls it (via an
+// optional-interface assertion) when its parent runs that way, so a journal/replay
+// host that drives the parent — the durable runner — records each actor's rich
+// per-step trace and EventPayload too. It is an internal observability gate, not
+// part of the ActorInstance contract, so a host's own ActorInstance implementation
+// is unaffected.
+func (a *actorAdapter[S, E, C]) inheritObservability(full, unbounded bool) {
+	a.inst.traceFull = full
+	a.inst.histUnbounded = unbounded
+}
+
 // isActorEffect reports whether an effect is one the ActorSystem acts on: a
 // lifecycle effect (SpawnActor / StopActor) or an actor-communication send effect
 // (SendTo / SendParent / RespondToSender / ForwardEvent). Other effects (e.g. a

state/actor_snapshot.go

@@ -97,7 +97,17 @@ func (a *actorAdapter[S, E, C]) RestoreJSON(b []byte) error {
 	if err := json.Unmarshal(b, &snap); err != nil {
 		return &SnapshotError{Op: "unmarshal", Reason: "actor child decode failed: " + err.Error()}
 	}
-	restored, err := a.inst.machine.Restore(snap, WithRestoreClock[S](a.inst.clock))
+	restoreOpts := []RestoreOption[S]{WithRestoreClock[S](a.inst.clock)}
+	// Preserve the actor's observability mode across an in-place restore so a
+	// journal/replay host keeps recording the child's rich trace, and (under a
+	// durable parent) keeps retaining its history so the child still round-trips.
+	switch {
+	case a.inst.histUnbounded:
+		restoreOpts = append(restoreOpts, WithRestoreUnboundedHistory[S]())
+	case a.inst.traceFull:
+		restoreOpts = append(restoreOpts, WithRestoreFullTrace[S]())
+	}
+	restored, err := a.inst.machine.Restore(snap, restoreOpts...)
 	if err != nil {
 		return err
 	}
@@ -245,6 +255,7 @@ func (s *ActorSystem[S, E, C]) restoreActor(ctx context.Context, snap actorSnaps
 	if err != nil {
 		return &SnapshotError{Op: "restore", State: snap.ID, Reason: "actor re-spawn failed: " + err.Error()}
 	}
+	s.propagateTrace(inst)
 
 	ra := &runningActor[E]{
 		inst:  inst,

state/actor_system.go

@@ -287,6 +287,22 @@ func (s *ActorSystem[S, E, C]) absorb(ctx context.Context, effects []Effect, cur
 // spawn creates and registers a running actor from a SpawnActor effect. On an
 // unbound Src it fires the parent's onError (when usable) so completion still
 // routes.
+// propagateTrace puts a freshly spawned or restored child actor into full trace
+// mode when the parent instance runs in full mode, so the parent's observability
+// choice (notably a journal/replay host such as the durable runner) reaches the
+// whole actor subtree. It uses an optional-interface assertion so a host's own
+// ActorInstance implementation is left untouched.
+func (s *ActorSystem[S, E, C]) propagateTrace(inst ActorInstance) {
+	if s.parent == nil || !s.parent.traceFull {
+		return
+	}
+	if ft, ok := inst.(interface {
+		inheritObservability(full, unbounded bool)
+	}); ok {
+		ft.inheritObservability(s.parent.traceFull, s.parent.histUnbounded)
+	}
+}
+
 func (s *ActorSystem[S, E, C]) spawn(ctx context.Context, e SpawnActor) {
 	s.mu.Lock()
 	behavior, ok := s.palette[e.Src.Name]
@@ -305,6 +321,7 @@ func (s *ActorSystem[S, E, C]) spawn(ctx context.Context, e SpawnActor) {
 		s.routeError(ctx, e, err)
 		return
 	}
+	s.propagateTrace(inst)
 
 	ra := &runningActor[E]{
 		inst:  inst,

state/observability_test.go

@@ -395,6 +395,69 @@ func TestHistory_DefaultReturnsNil(t *testing.T) {
 	}
 }
 
+// ---------------------------------------------------------------------------
+// 5b. Restore trace mode: WithRestoreFullTrace / WithRestoreUnboundedHistory.
+// ---------------------------------------------------------------------------
+
+// TestRestore_TraceModeOptIn asserts a restored instance is lite by default (a
+// subsequent Fire omits rich fields and does not retain new traces), that
+// WithRestoreFullTrace restores full per-step traces without continued retention,
+// and that WithRestoreUnboundedHistory restores full traces AND keeps retaining
+// them — the mode the durable runner relies on for byte-identical replay. (Restore
+// always seeds history from the snapshot's traces, so the distinguishing behavior
+// is whether later Fires are rich and whether they keep accumulating.)
+func TestRestore_TraceModeOptIn(t *testing.T) {
+	m := buildToggleMachine()
+	ctx := context.Background()
+
+	src := m.Cast(nil, state.WithInitialState[toggleState](toggleA), state.WithUnboundedHistory[toggleState]())
+	src.Fire(ctx, toggleGo)
+	snap := src.Snapshot()
+
+	// Default restore: lite Fire (no rich fields) and no further retention.
+	lite, err := m.Restore(snap)
+	if err != nil {
+		t.Fatalf("Restore (default): %v", err)
+	}
+	base := len(lite.History())
+	r := lite.Fire(ctx, toggleGo)
+	if len(r.Trace.EnteredStates) != 0 || len(r.Trace.EffectsEmitted) != 0 {
+		t.Errorf("default restore should fire lite, got rich fields: %+v", r.Trace)
+	}
+	if got := len(lite.History()); got != base {
+		t.Errorf("default restore should not retain new traces: %d -> %d", base, got)
+	}
+
+	// WithRestoreFullTrace: rich Fire, but still no continued retention.
+	full, err := m.Restore(snap, state.WithRestoreFullTrace[toggleState]())
+	if err != nil {
+		t.Fatalf("Restore (full trace): %v", err)
+	}
+	fbase := len(full.History())
+	rf := full.Fire(ctx, toggleGo)
+	if len(rf.Trace.EnteredStates) == 0 || len(rf.Trace.EffectsEmitted) == 0 {
+		t.Errorf("WithRestoreFullTrace should populate rich fields, got %+v", rf.Trace)
+	}
+	if got := len(full.History()); got != fbase {
+		t.Errorf("WithRestoreFullTrace should not retain new traces: %d -> %d", fbase, got)
+	}
+
+	// WithRestoreUnboundedHistory: rich Fire AND continued unbounded retention.
+	dur, err := m.Restore(snap, state.WithRestoreUnboundedHistory[toggleState]())
+	if err != nil {
+		t.Fatalf("Restore (unbounded): %v", err)
+	}
+	dbase := len(dur.History())
+	dur.Fire(ctx, toggleGo)
+	dur.Fire(ctx, toggleGo)
+	if got := len(dur.History()); got != dbase+2 {
+		t.Errorf("WithRestoreUnboundedHistory should keep retaining: %d -> %d (want +2)", dbase, got)
+	}
+	if h := dur.History(); len(h[len(h)-1].EnteredStates) == 0 {
+		t.Errorf("retained trace under unbounded restore should be full, got %+v", h[len(h)-1])
+	}
+}
+
 // ---------------------------------------------------------------------------
 // 6. Label cache: m.label correctness.
 // ---------------------------------------------------------------------------

state/options.go

@@ -301,6 +301,8 @@ type RestoreOption[S comparable] func(*restoreConfig[S])
 type restoreConfig[S comparable] struct {
 	clock                Clock
 	rejectMachineVersion bool
+	traceFull            bool
+	histUnbounded        bool
 }
 
 // RejectMachineVersionMismatch makes Restore enforce the machine DEFINITION
@@ -321,6 +323,27 @@ func WithRestoreClock[S comparable](c Clock) RestoreOption[S] {
 	return func(cfg *restoreConfig[S]) { cfg.clock = c }
 }
 
+// WithRestoreFullTrace restores an instance in full trace mode, mirroring
+// WithFullTrace at Cast. A restored instance is lite by default like a freshly
+// cast one; a journal/replay consumer that reconstructs from the recorded Trace
+// (the durable runner) restores in full mode so the rich per-step fields and the
+// EventPayload are produced on every replayed Fire.
+func WithRestoreFullTrace[S comparable]() RestoreOption[S] {
+	return func(cfg *restoreConfig[S]) { cfg.traceFull = true }
+}
+
+// WithRestoreUnboundedHistory restores an instance in full trace mode with
+// unbounded history retention, mirroring WithUnboundedHistory at Cast. The durable
+// runner uses it on recover: it reconstructs the step ordinal and checkpoint
+// snapshots from the retained Trace history, so post-recovery Fires must keep
+// retaining every trace exactly as a freshly started durable instance does.
+func WithRestoreUnboundedHistory[S comparable]() RestoreOption[S] {
+	return func(cfg *restoreConfig[S]) {
+		cfg.traceFull = true
+		cfg.histUnbounded = true
+	}
+}
+
 // SnapshotCodecOption configures MarshalSnapshot / UnmarshalSnapshot.
 type SnapshotCodecOption[C any] func(*snapshotCodecConfig[C])
 

state/snapshot.go

@@ -465,6 +465,8 @@ func (m *Machine[S, E, C]) Restore(snap Snapshot[S, E, C], opts ...RestoreOption
 		historyShallow: copyMap(snap.HistoryShallow),
 		historyDeep:    copyLeafMap(snap.HistoryDeep),
 		clock:          clock,
+		traceFull:      rcfg.traceFull,
+		histUnbounded:  rcfg.histUnbounded,
 	}
 	return inst, nil
 }