// Package conformance provides backend-neutral activity replication fixtures. // Projection builders and cloud sinks use separate harnesses because they sit // on opposite sides of the replication boundary. package conformance import ( "context" "encoding/json" "errors" "fmt" "reflect" activityreplication "github.com/tutti-os/tutti/packages/agent/activity-replication" ) // ProjectionBuilder seeds canonical state and builds its wire projection. It // is implemented by the Tutti SQLite adapter and the tsh local builder tests. type ProjectionBuilder interface { Reset(context.Context) error Seed(context.Context, []CanonicalSeed) error Build(context.Context) (activityreplication.ChangeBatch, error) } // CanonicalSeed is source-side canonical state without mutation transport // identity. Adapters persist these snapshots through their native store APIs. type CanonicalSeed struct { WorkspaceID string EntityType activityreplication.EntityType Key activityreplication.EntityKey Target *activityreplication.Target TargetScope *activityreplication.TargetScope Session *activityreplication.Session SessionScope *activityreplication.SessionScope Turn *activityreplication.Turn Interaction *activityreplication.Interaction Message *activityreplication.Message } type ProjectionFixture struct { Name string Canonical []CanonicalSeed WantBatch activityreplication.ChangeBatch } // RunProjection verifies source-side canonical-to-wire projection. Mutation // and transaction IDs are transport identities generated by each builder, so // the comparison checks that they validate but ignores their concrete values. func RunProjection(ctx context.Context, builder ProjectionBuilder, fixture ProjectionFixture) error { if err := builder.Reset(ctx); err != nil { return fmt.Errorf("reset projection builder: %w", err) } if err := builder.Seed(ctx, fixture.Canonical); err != nil { return fmt.Errorf("seed canonical state: %w", err) } got, err := builder.Build(ctx) if err != nil { return fmt.Errorf("build projection: %w", err) } if err := activityreplication.ValidateBatch(got); err != nil { return fmt.Errorf("built batch is invalid: %w", err) } if err := activityreplication.ValidateBatch(fixture.WantBatch); err != nil { return fmt.Errorf("fixture batch is invalid: %w", err) } retry, err := builder.Build(ctx) if err != nil { return fmt.Errorf("rebuild projection after lost response: %w", err) } if err := activityreplication.ValidateBatch(retry); err != nil { return fmt.Errorf("rebuilt batch is invalid: %w", err) } if len(retry.Mutations) != len(got.Mutations) { return fmt.Errorf("rebuilt batch has %d mutations, want %d", len(retry.Mutations), len(got.Mutations)) } for index := range got.Mutations { if retry.Mutations[index].MutationID != got.Mutations[index].MutationID { return fmt.Errorf("rebuilt mutation %d changed mutation ID %q to %q", index, got.Mutations[index].MutationID, retry.Mutations[index].MutationID) } } if !reflect.DeepEqual(normalizeProjectionBatch(retry), normalizeProjectionBatch(got)) { return errors.New("rebuilt batch changed its canonical projection") } got = normalizeProjectionBatch(got) want := normalizeProjectionBatch(fixture.WantBatch) if !reflect.DeepEqual(got, want) { gotJSON, _ := json.Marshal(got) wantJSON, _ := json.Marshal(want) return fmt.Errorf("projection %s, want %s", gotJSON, wantJSON) } return nil } func normalizeProjectionBatch(batch activityreplication.ChangeBatch) activityreplication.ChangeBatch { batch.Mutations = append([]activityreplication.Mutation(nil), batch.Mutations...) for index := range batch.Mutations { batch.Mutations[index].MutationID = "" batch.Mutations[index].TransactionID = "" } return batch } type ApplyReport struct { Result activityreplication.ApplyResult Acknowledgements []activityreplication.MutationAcknowledgement } // Sink is implemented by cloud projection sinks such as tsh-server MySQL. type Sink interface { Reset(context.Context) error Apply(context.Context, activityreplication.ChangeBatch) (ApplyReport, error) Lookup(context.Context, activityreplication.EntityType, activityreplication.EntityKey) (SinkSnapshot, bool, error) } // SinkSnapshot is the persisted read model observed by the sink conformance // harness. Scopes are explicit because acknowledgements and entity payloads // alone cannot prove that routing and authorization attribution was retained. type SinkSnapshot struct { Entity json.RawMessage TargetScope *activityreplication.TargetScope SessionScope *activityreplication.SessionScope } type SinkStep struct { Name string Batch activityreplication.ChangeBatch WantResult activityreplication.ApplyResult WantAcknowledgements []activityreplication.MutationAcknowledgement WantRejection *RejectionExpectation } type RejectionExpectation struct { Kind activityreplication.RejectionKind MutationID string TransactionID string } type SnapshotExpectation struct { EntityType activityreplication.EntityType Key activityreplication.EntityKey Present bool Snapshot SinkSnapshot } type SinkFixture struct { Name string Steps []SinkStep WantSnapshots []SnapshotExpectation } // RunSink verifies sink acknowledgements and final projection state. It checks // each acknowledgement in full, including retry transaction identity and the // original cursor returned for duplicates. func RunSink(ctx context.Context, sink Sink, fixture SinkFixture) error { if err := sink.Reset(ctx); err != nil { return fmt.Errorf("reset sink: %w", err) } for _, step := range fixture.Steps { report, err := sink.Apply(ctx, step.Batch) if step.WantRejection != nil { if rejectionErr := matchRejection(err, *step.WantRejection); rejectionErr != nil { return fmt.Errorf("step %q: %w", step.Name, rejectionErr) } continue } if err != nil { return fmt.Errorf("step %q: apply: %w", step.Name, err) } summarized, err := activityreplication.SummarizeAcknowledgements(report.Acknowledgements) if err != nil { return fmt.Errorf("step %q: summarize acknowledgements: %w", step.Name, err) } if report.Result != summarized { return fmt.Errorf("step %q: result %#v does not summarize acknowledgements as %#v", step.Name, report.Result, summarized) } if report.Result != step.WantResult { return fmt.Errorf("step %q: result %#v, want %#v", step.Name, report.Result, step.WantResult) } if !reflect.DeepEqual(report.Acknowledgements, step.WantAcknowledgements) { return fmt.Errorf("step %q: acknowledgements %#v, want %#v", step.Name, report.Acknowledgements, step.WantAcknowledgements) } } for _, expectation := range fixture.WantSnapshots { got, found, err := sink.Lookup(ctx, expectation.EntityType, expectation.Key) if err != nil { return fmt.Errorf("lookup %s: %w", expectation.EntityType, err) } if found != expectation.Present { return fmt.Errorf("lookup %s: found %t, want %t", expectation.EntityType, found, expectation.Present) } if !expectation.Present { continue } if !jsonEqual(got.Entity, expectation.Snapshot.Entity) { return fmt.Errorf("lookup %s: entity snapshot %s, want %s", expectation.EntityType, got.Entity, expectation.Snapshot.Entity) } if !reflect.DeepEqual(got.TargetScope, expectation.Snapshot.TargetScope) { return fmt.Errorf("lookup %s: target scope %#v, want %#v", expectation.EntityType, got.TargetScope, expectation.Snapshot.TargetScope) } if !reflect.DeepEqual(got.SessionScope, expectation.Snapshot.SessionScope) { return fmt.Errorf("lookup %s: session scope %#v, want %#v", expectation.EntityType, got.SessionScope, expectation.Snapshot.SessionScope) } } return nil } func matchRejection(err error, want RejectionExpectation) error { if err == nil { return errors.New("expected permanent rejection, got nil") } var rejection *activityreplication.PermanentRejection if !errors.As(err, &rejection) { return fmt.Errorf("error %T is not a permanent rejection: %w", err, err) } if rejection.Kind != want.Kind || rejection.MutationID != want.MutationID || rejection.TransactionID != want.TransactionID { return fmt.Errorf("rejection %#v, want %#v", rejection, want) } return nil } func jsonEqual(left, right json.RawMessage) bool { var leftValue any var rightValue any return json.Unmarshal(left, &leftValue) == nil && json.Unmarshal(right, &rightValue) == nil && reflect.DeepEqual(leftValue, rightValue) }