diff --git a/internal/fold/evaluate.go b/internal/fold/evaluate.go index 49ec283..498a8c6 100644 --- a/internal/fold/evaluate.go +++ b/internal/fold/evaluate.go @@ -15,15 +15,23 @@ import ( // authority scope) is NOT done here — that is the operad layer's responsibility. // fold only enforces structural invariants: existence, immutability, version. func Evaluate(state graph.GraphState, env graph.Envelope) (graph.GraphState, graph.EvalResult, error) { + return EvaluateAt(state, env, time.Now().UTC()) +} + +// EvaluateAt applies one Envelope using the provided timestamp for structural +// CreatedAt fields (node/relation). Runtime applies should pass the exact +// persisted AppliedAt so replay remains bit-deterministic over time. +func EvaluateAt(state graph.GraphState, env graph.Envelope, appliedAt time.Time) (graph.GraphState, graph.EvalResult, error) { if err := validateEnvelopeStructure(env); err != nil { return state, graph.EvalResult{}, err } + ts := appliedAt.UTC() switch env.RewriteType { case graph.ADD: - return applyADD(state, env) + return applyADD(state, env, ts) case graph.LINK: - return applyLINK(state, env) + return applyLINK(state, env, ts) case graph.MUTATE: return applyMUTATE(state, env) case graph.UNLINK: @@ -33,7 +41,7 @@ func Evaluate(state graph.GraphState, env graph.Envelope) (graph.GraphState, gra } } -func applyADD(state graph.GraphState, env graph.Envelope) (graph.GraphState, graph.EvalResult, error) { +func applyADD(state graph.GraphState, env graph.Envelope, createdAt time.Time) (graph.GraphState, graph.EvalResult, error) { if _, exists := state.Nodes[env.NodeURN]; exists { return state, graph.EvalResult{}, fmt.Errorf("%w: %s", ErrNodeExists, env.NodeURN) } @@ -48,7 +56,7 @@ func applyADD(state graph.GraphState, env graph.Envelope) (graph.GraphState, gra URN: env.NodeURN, TypeID: env.TypeID, Properties: props, - CreatedAt: time.Now().UTC(), + CreatedAt: createdAt, Version: 1, } // Maintain the NodesByType index so sweep/t-cone can look up nodes by @@ -58,7 +66,7 @@ func applyADD(state graph.GraphState, env graph.Envelope) (graph.GraphState, gra return next, graph.EvalResult{AffectedNodeURN: env.NodeURN}, nil } -func applyLINK(state graph.GraphState, env graph.Envelope) (graph.GraphState, graph.EvalResult, error) { +func applyLINK(state graph.GraphState, env graph.Envelope, createdAt time.Time) (graph.GraphState, graph.EvalResult, error) { if _, exists := state.Nodes[env.SrcURN]; !exists { return state, graph.EvalResult{}, fmt.Errorf("%w: src %s", ErrNodeNotFound, env.SrcURN) } @@ -81,7 +89,7 @@ func applyLINK(state graph.GraphState, env graph.Envelope) (graph.GraphState, gr TgtURN: env.TgtURN, TgtPort: env.TgtPort, ContractURN: env.ContractURN, - CreatedAt: time.Now().UTC(), + CreatedAt: createdAt, } // Maintain the by-endpoint relation indexes so occupancy walks and // t-cone neighbourhood queries run in O(edges-at-urn) instead of diff --git a/internal/fold/fold_test.go b/internal/fold/fold_test.go index f6af77d..0fe3d27 100644 --- a/internal/fold/fold_test.go +++ b/internal/fold/fold_test.go @@ -2,6 +2,7 @@ package fold_test import ( "errors" + "reflect" "testing" "time" @@ -244,6 +245,31 @@ func TestProgram_Rollback(t *testing.T) { } } +func TestProgramAt_PartialAppliedAtRejected(t *testing.T) { + state := graph.NewGraphState() + + envelopes := []graph.Envelope{ + userNode("urn:moos:user:sam"), + {RewriteType: graph.ADD, Actor: "urn:moos:user:sam", + NodeURN: "urn:moos:workstation:hp-laptop", TypeID: "workstation", + Properties: map[string]graph.Property{"hostname": {Value: "hp-laptop", Mutability: "immutable"}}}, + } + + next, results, err := fold.EvaluateProgramAt(state, envelopes, []time.Time{time.Now().UTC()}) + if err == nil { + t.Fatal("expected error from partial appliedAt slice") + } + if results != nil { + t.Fatal("expected nil results on appliedAt length mismatch") + } + if len(state.Nodes) != 0 { + t.Error("state was modified despite appliedAt validation failure") + } + if len(next.Nodes) != 0 { + t.Error("returned state has nodes despite appliedAt validation failure") + } +} + func TestReplay(t *testing.T) { // Build a log via Evaluate state := graph.NewGraphState() @@ -305,7 +331,8 @@ func TestCI3_IdentityStability(t *testing.T) { // TestCI4_ReplayDeterminism verifies same log → same state always. func TestCI4_ReplayDeterminism(t *testing.T) { env := userNode("urn:moos:user:sam") - log := []graph.PersistedRewrite{{Envelope: env, AppliedAt: time.Now().UTC(), LogSeq: 1}} + appliedAt := time.Date(2026, time.July, 9, 10, 40, 0, 123, time.UTC) + log := []graph.PersistedRewrite{{Envelope: env, AppliedAt: appliedAt, LogSeq: 1}} s1, err1 := fold.Replay(log) s2, err2 := fold.Replay(log) @@ -315,7 +342,25 @@ func TestCI4_ReplayDeterminism(t *testing.T) { } n1 := s1.Nodes["urn:moos:user:sam"] n2 := s2.Nodes["urn:moos:user:sam"] - if n1.URN != n2.URN || n1.TypeID != n2.TypeID { + if !reflect.DeepEqual(n1, n2) { t.Error("CI-4 violated: same log produced different states") } + if !n1.CreatedAt.Equal(appliedAt) { + t.Fatalf("replay did not preserve applied_at for node created_at: got %s want %s", n1.CreatedAt, appliedAt) + } +} + +func TestReplayUsesTimestampFallbackWhenAppliedAtMissing(t *testing.T) { + env := userNode("urn:moos:user:legacy") + ts := time.Date(2026, time.April, 10, 9, 30, 0, 0, time.UTC) + log := []graph.PersistedRewrite{{Envelope: env, Timestamp: ts, LogSeq: 1}} + + s, err := fold.Replay(log) + if err != nil { + t.Fatalf("replay: %v", err) + } + n := s.Nodes["urn:moos:user:legacy"] + if !n.CreatedAt.Equal(ts) { + t.Fatalf("replay did not fallback to timestamp for node created_at: got %s want %s", n.CreatedAt, ts) + } } diff --git a/internal/fold/program.go b/internal/fold/program.go index 1f30e53..98d7528 100644 --- a/internal/fold/program.go +++ b/internal/fold/program.go @@ -12,11 +12,28 @@ import ( // the original state is returned unchanged. // On success, returns the new state and per-envelope results. func EvaluateProgram(state graph.GraphState, envelopes []graph.Envelope) (graph.GraphState, []graph.EvalResult, error) { + return EvaluateProgramAt(state, envelopes, nil) +} + +// EvaluateProgramAt is EvaluateProgram with optional caller-supplied timestamps. +// If appliedAt[i] is present, it is used as the fold-time timestamp for +// envelope i (node/relation CreatedAt on ADD/LINK). Otherwise wall clock is +// used for backwards compatibility. +func EvaluateProgramAt(state graph.GraphState, envelopes []graph.Envelope, appliedAt []time.Time) (graph.GraphState, []graph.EvalResult, error) { + if len(appliedAt) != 0 && len(appliedAt) != len(envelopes) { + return state, nil, fmt.Errorf("program appliedAt length mismatch: got %d, want 0 or %d", len(appliedAt), len(envelopes)) + } + useAppliedAt := len(appliedAt) == len(envelopes) + working := state.Clone() results := make([]graph.EvalResult, 0, len(envelopes)) for i, env := range envelopes { - next, result, err := Evaluate(working, env) + ts := time.Now().UTC() + if useAppliedAt { + ts = appliedAt[i] + } + next, result, err := EvaluateAt(working, env, ts) if err != nil { return state, nil, fmt.Errorf("program step %d (%s): %w", i, env.RewriteType, err) } @@ -28,7 +45,7 @@ func EvaluateProgram(state graph.GraphState, envelopes []graph.Envelope) (graph. // ProgramResult is returned by kernel.Runtime.ApplyProgram. type ProgramResult struct { - Results []graph.EvalResult `json:"results"` + Results []graph.EvalResult `json:"results"` Persisted []graph.PersistedRewrite `json:"persisted"` - AppliedAt time.Time `json:"applied_at"` + AppliedAt time.Time `json:"applied_at"` } diff --git a/internal/fold/replay.go b/internal/fold/replay.go index 25dd2bd..07ae0d1 100644 --- a/internal/fold/replay.go +++ b/internal/fold/replay.go @@ -4,6 +4,7 @@ import ( "errors" "fmt" slog "log" + "time" "moos/kernel/internal/graph" ) @@ -24,7 +25,7 @@ func Replay(entries []graph.PersistedRewrite) (graph.GraphState, error) { state := graph.NewGraphState() for i, pr := range entries { - next, _, err := Evaluate(state, pr.Envelope) + next, _, err := EvaluateAt(state, pr.Envelope, replayAppliedAt(pr)) if err != nil { if isIdempotentSkip(pr.Envelope.RewriteType, err) { slog.Printf("replay: skipping idempotent error at seq %d (index %d, %s): %v", @@ -64,3 +65,15 @@ func (e *ReplayError) Error() string { } func (e *ReplayError) Unwrap() error { return e.Cause } + +func replayAppliedAt(pr graph.PersistedRewrite) time.Time { + if !pr.AppliedAt.IsZero() { + return pr.AppliedAt + } + if !pr.Timestamp.IsZero() { + return pr.Timestamp + } + // Legacy/malformed log lines without any timestamp keep a stable zero time + // so replay remains deterministic and never consults wall clock. + return time.Time{} +} diff --git a/internal/kernel/runtime.go b/internal/kernel/runtime.go index 9d36430..45be4fd 100644 --- a/internal/kernel/runtime.go +++ b/internal/kernel/runtime.go @@ -216,13 +216,13 @@ func (rt *Runtime) applyWithOptions(env graph.Envelope, opts applyOptions) (grap return graph.EvalResult{}, err } - next, result, err := fold.Evaluate(rt.state, env) + now := time.Now().UTC() + next, result, err := fold.EvaluateAt(rt.state, env, now) if err != nil { return graph.EvalResult{}, err } seq := rt.logSeq.Add(1) - now := time.Now().UTC() persisted := graph.PersistedRewrite{ Envelope: env, AppliedAt: now, @@ -292,6 +292,11 @@ func (rt *Runtime) ApplyProgram(envelopes []graph.Envelope) ([]graph.EvalResult, // an index of guarded nodes so we can skip the inner scan when no gates // touch the target set (PR #8 review, Gemini). injected := make([]graph.Envelope, len(envelopes)) + baseNow := time.Now().UTC() + appliedAt := make([]time.Time, len(envelopes)) + for i := range envelopes { + appliedAt[i] = baseNow.Add(time.Duration(i) * time.Nanosecond) + } workingState := rt.state.Clone() for i, env := range envelopes { if env.RewriteType == graph.MUTATE && rt.registry != nil { @@ -331,7 +336,7 @@ func (rt *Runtime) ApplyProgram(envelopes []graph.Envelope) ([]graph.EvalResult, // Advance the working state by applying this envelope before // validating the next one. On failure bail early — ApplyProgram // is all-or-nothing so a later error still rolls back cleanly. - next, _, err := fold.Evaluate(workingState, env) + next, _, err := fold.EvaluateAt(workingState, env, appliedAt[i]) if err != nil { return nil, err } @@ -339,16 +344,15 @@ func (rt *Runtime) ApplyProgram(envelopes []graph.Envelope) ([]graph.EvalResult, } envelopes = injected - nextState, results, err := fold.EvaluateProgram(rt.state, envelopes) + nextState, results, err := fold.EvaluateProgramAt(rt.state, envelopes, appliedAt) if err != nil { return nil, err } - now := time.Now().UTC() persisted := make([]graph.PersistedRewrite, len(envelopes)) for i, env := range envelopes { seq := rt.logSeq.Add(1) - ts := now.Add(time.Duration(i) * time.Nanosecond) + ts := appliedAt[i] persisted[i] = graph.PersistedRewrite{ Envelope: env, AppliedAt: ts, @@ -593,13 +597,13 @@ func (rt *Runtime) applyReactiveLocked(env graph.Envelope) { } } - next, _, err := fold.Evaluate(rt.state, env) + now := time.Now().UTC() + next, _, err := fold.EvaluateAt(rt.state, env, now) if err != nil { return // skip (e.g. ErrNodeExists for idempotent proposals) } seq := rt.logSeq.Add(1) - now := time.Now().UTC() persisted := graph.PersistedRewrite{ Envelope: env, AppliedAt: now, @@ -739,13 +743,13 @@ func (rt *Runtime) bumpSessionLocalT(sessionURN graph.URN) { // Inject PropertySpec for additive MUTATE (local_t may not be on node yet). mutEnv = rt.injectPropertySpec(mutEnv, actorNode) - next, _, err := fold.Evaluate(rt.state, mutEnv) + now := time.Now().UTC() + next, _, err := fold.EvaluateAt(rt.state, mutEnv, now) if err != nil { return // local_t not declared in ontology or other structural issue — skip } seq := rt.logSeq.Add(1) - now := time.Now().UTC() persisted := graph.PersistedRewrite{ Envelope: mutEnv, AppliedAt: now,