diff --git a/CHANGELOG.md b/CHANGELOG.md index 7eb1a5a..252ea14 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -165,6 +165,8 @@ The project publishes 0.x prerelease versions; a stable release line is not yet ### Fixed +- Enforce text-chunk uniqueness, index nullable memory references, cascade memory relations with workspace deletion, batch generation listings, and paginate relation listings with opaque cursors. Refs #178. + - Follow the shared design language for reading, numeric and action alignment; generate the existing Web color variables from a pinned design-system token snapshot, and use a single consistent empty-state pattern. Refs #211. - Improve web caption and status contrast in both themes, including tinted danger diff --git a/docs/MIGRATION_SEQUENCE.md b/docs/MIGRATION_SEQUENCE.md index df3c8d1..e02332c 100644 --- a/docs/MIGRATION_SEQUENCE.md +++ b/docs/MIGRATION_SEQUENCE.md @@ -6,30 +6,31 @@ These draft changes are cumulative, not independently deployable: | --- | --- | --- | --- | | 1 | #194 / #183 | 0024 file lexical lane | released/main schema 23 (merged) | | 2 | #173 HNSW completion (supersedes #197 HOLD) | 0025 HNSW indexes + text continuation | #194, schema 24 | -| 3 | #195 / #185 | 0026 data-plane hygiene | schema 25 | +| 3 | #195 / #185 | 0027 data-plane hygiene | schema 26 | -The PR base chain is `main` → `codex/fix-pr-183` → `codex/fix-pr-180` -→ `codex/fix-pr-185`. Successor branches must include their predecessor schema and source. Local -repair branches are rebuilt on current main and replay the original authored -changes; published commit identities remain in the original PR history. -Keep this order when retargeting after a predecessor merges. +Migrations 0024–0026 are now on `main`. PR #195 must integrate that exact +history before adding data-plane hygiene as 0027. Local repair branches retain +the original authored commits and add a current-main integration commit rather +than rewriting published history. -On 2026-09-10, main `2986fe38175f54d99f15dd38a498708c6ecd88cd` and published -tags `v0.1.0` / `v0.1.1` contain only migrations 0001–0023. This does not prove -that a private deployment never applied a draft. Consequently migration -numbers and SQL identities are retained, not renumbered on an assumption. +The hygiene migration was originally reviewed as draft 0026, but main now owns +0026 for `memory_producer_agent`. The unmerged hygiene migration therefore +moves to 0027. An operator who privately applied the old draft under version 26 +must stop and obtain a recovery plan; do not rewrite Goose history to make the +new main sequence appear valid. Migration 0025 creates the three cosine HNSW indexes. The shipping text route no longer uses `DISTINCT ON (f.id) ORDER BY f.id` as its primary plan: it walks cosine-ordered candidates and falls back to that exact query only when a bounded scan underfills. Visual cosine-order already matched HNSW. Face DDL is not a face-query speedup. #195 still requires predecessor schema 25. Recall and live -latency remain `#175`, not this migration. +latency remain `#175`, not this migration. #195 now requires predecessor schema +26. This document does not waive a review gate or authorize deployment. Goose startup remains strict: no `WithAllowMissing` or equivalent option is -enabled. A database that already applied 26 while omitting 24/25 will correctly +enabled. A database that already applied 27 while omitting 24/25/26 will correctly fail startup against the cumulative schema. Stop and obtain an operator-owned recovery plan for such a database; do not edit its migration history, renumber its SQL, or apply lower versions out of order to manufacture a pass. @@ -55,9 +56,10 @@ an operator-owned recovery; do not mark an unverified partial schema applied. `scripts/verify.sh integration` creates a separate, owned `_test` database and runs `TestMigrationUpgradeSequence`. It applies real Goose migrations to 23, seeds a file with duplicate text chunks, then advances one version at a time -to the branch's declared head (24, 25, or 26). Each step checks full applied +to the branch's declared head (24 through 27). Each step checks full applied history and preserved data; subsequent steps check lexical backfill, valid -HNSW DDL, and deduplication/unique rejection. Finally the ordinary production +HNSW DDL, producer-agent indexing, and hygiene deduplication/unique rejection. +Finally the ordinary production `DB.Migrate` startup path must accept the resulting history unchanged. The dedicated test uses `MEM_MIGRATION_SEQUENCE_TEST_DB`, refuses a database diff --git a/scripts/verify.sh b/scripts/verify.sh index 85dbdd3..8f5a0ac 100755 --- a/scripts/verify.sh +++ b/scripts/verify.sh @@ -4,7 +4,7 @@ set -euo pipefail REPO_ROOT="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd)" MODE="${1:-unit}" -EXPECTED_MIGRATION_HEAD=26 +EXPECTED_MIGRATION_HEAD=27 MIGRATION_ROLLBACK_TARGET=11 MODEL_TEXT_CANONICAL_BASE=15 WORKSPACE_AI_PROFILE_BASE=16 diff --git a/server/internal/api/api.go b/server/internal/api/api.go index d7edd63..a6762db 100644 --- a/server/internal/api/api.go +++ b/server/internal/api/api.go @@ -78,7 +78,7 @@ type MemoryService interface { Restore(context.Context, memory.LifecycleCommand) (*memory.MutationResult, error) Forget(context.Context, memory.ForgetCommand) (*memory.ForgetResult, error) CreateRelation(context.Context, memory.CreateRelationCommand) (*memory.CreateRelationResult, error) - ListRelations(context.Context, memory.ListRelationsQuery) ([]memory.Relation, error) + ListRelations(context.Context, memory.ListRelationsQuery) (*memory.ListRelationsResult, error) } // DurableContextService is the scoped durable-context port (mem#70). Handlers diff --git a/server/internal/api/handlers_memory.go b/server/internal/api/handlers_memory.go index 879e1d4..4020bb3 100644 --- a/server/internal/api/handlers_memory.go +++ b/server/internal/api/handlers_memory.go @@ -763,18 +763,21 @@ func (s *Server) handleListMemoryRelations(w http.ResponseWriter, r *http.Reques } tok := r.Context().Value(ctxToken).(*auth.Token) - relations, err := s.Memory.ListRelations(r.Context(), memory.ListRelationsQuery{ + result, err := s.Memory.ListRelations(r.Context(), memory.ListRelationsQuery{ WorkspaceID: currentWorkspace(r).ID, MemoryID: id, Direction: direction, RelationType: relationType, AllowedPaths: tok.Paths, Limit: limit, + Cursor: r.URL.Query().Get("cursor"), }) if err != nil { switch { case errors.Is(err, memory.ErrInvalidCommand): writeError(w, http.StatusBadRequest, "invalid_relation_query", err.Error()) + case errors.Is(err, memory.ErrInvalidCursor): + writeError(w, http.StatusBadRequest, "invalid_cursor", err.Error()) case errors.Is(err, memory.ErrNotFound): writeError(w, http.StatusNotFound, "not_found", "memory not found") case errors.Is(err, memory.ErrForgotten): @@ -789,8 +792,12 @@ func (s *Server) handleListMemoryRelations(w http.ResponseWriter, r *http.Reques } return } - if relations == nil { - relations = []memory.Relation{} + resp := map[string]any{"relations": result.Relations} + if result.Relations == nil { + resp["relations"] = []memory.Relation{} } - writeJSON(w, http.StatusOK, map[string]any{"relations": relations}) + if result.NextCursor != "" { + resp["next_cursor"] = result.NextCursor + } + writeJSON(w, http.StatusOK, resp) } diff --git a/server/internal/api/handlers_memory_test.go b/server/internal/api/handlers_memory_test.go index 54887a4..0fc2e6d 100644 --- a/server/internal/api/handlers_memory_test.go +++ b/server/internal/api/handlers_memory_test.go @@ -107,10 +107,10 @@ func (s *memoryServiceStub) CreateRelation( func (s *memoryServiceStub) ListRelations( _ context.Context, q memory.ListRelationsQuery, -) ([]memory.Relation, error) { +) (*memory.ListRelationsResult, error) { s.calls++ s.listRelationsQuery = q - return s.relations, s.controlErr + return &memory.ListRelationsResult{Relations: s.relations}, s.controlErr } func memoryHandlerContext(req *http.Request, paths []string) (*http.Request, uuid.UUID, uuid.UUID, uuid.UUID) { diff --git a/server/internal/db/migration_sequence_test.go b/server/internal/db/migration_sequence_test.go index b1fa7a0..00cefda 100644 --- a/server/internal/db/migration_sequence_test.go +++ b/server/internal/db/migration_sequence_test.go @@ -3,6 +3,7 @@ package db import ( "context" "database/sql" + "errors" "os" "strconv" "strings" @@ -11,6 +12,7 @@ import ( "github.com/google/uuid" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgconn" "github.com/pressly/goose/v3" ) @@ -107,8 +109,12 @@ func TestMigrationUpgradeSequence(t *testing.T) { } } var chunks int - if err := sqldb.QueryRowContext(ctx, "SELECT count(*) FROM embeddings_text WHERE file_id=$1", fileID).Scan(&chunks); err != nil || chunks != 2 { - t.Fatalf("preserved chunks=%d, want=2, err=%v", chunks, err) + wantChunks := 2 + if version >= 27 { + wantChunks = 1 + } + if err := sqldb.QueryRowContext(ctx, "SELECT count(*) FROM embeddings_text WHERE file_id=$1", fileID).Scan(&chunks); err != nil || chunks != wantChunks { + t.Fatalf("preserved chunks=%d, want=%d, err=%v", chunks, wantChunks, err) } if version >= 26 { var producerIdx int @@ -118,6 +124,13 @@ func TestMigrationUpgradeSequence(t *testing.T) { } t.Logf("PASS: strict Goose upgrade to %d; complete history and populated data preserved", version) } + if head >= 27 { + _, err := sqldb.ExecContext(ctx, "INSERT INTO embeddings_text(file_id,chunk_index,chunk_text) VALUES($1,0,'duplicate')", fileID) + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) || pgErr.Code != "23505" { + t.Fatalf("expected duplicate rejection 23505, got %v", err) + } + } // The real startup path must accept the upgraded history unchanged. if err := (&DB{url: dsn}).Migrate(ctx); err != nil { t.Fatalf("production startup after sequential upgrade: %v", err) diff --git a/server/internal/db/migrations/0027_data_plane_hygiene.sql b/server/internal/db/migrations/0027_data_plane_hygiene.sql new file mode 100644 index 0000000..bfed4bc --- /dev/null +++ b/server/internal/db/migrations/0027_data_plane_hygiene.sql @@ -0,0 +1,85 @@ +-- +goose Up +-- Data-plane hygiene from the index audit (#178), sequenced after main's 0026. +-- Three independent fixes bundled into one migration because each is a single +-- DDL statement and none warrants its own schema version. + +-- Item 1: embeddings_text uniqueness on (file_id, chunk_index). +-- The write path (indexer.go) DELETEs all chunks for a file before re-inserting, +-- so duplicates should never exist in practice. Deduplicate defensively before +-- adding the constraint: if any duplicates survived, keep the row with the +-- smallest UUID for deterministic selection (UUID order is not insert order). +-- +goose StatementBegin +DELETE FROM embeddings_text + WHERE id NOT IN ( + SELECT DISTINCT ON (file_id, chunk_index) id + FROM embeddings_text + ORDER BY file_id, chunk_index, id + ); +-- +goose StatementEnd + +-- +goose StatementBegin +ALTER TABLE embeddings_text + ADD CONSTRAINT uq_embeddings_text_file_chunk UNIQUE (file_id, chunk_index); +-- +goose StatementEnd + +-- Item 2: partial indexes for ON DELETE SET NULL lookups on memories. +-- Without these, every file or user delete takes a RowExclusiveLock on memories +-- and performs a sequential scan to find the rows to null out. +-- +goose StatementBegin +CREATE INDEX IF NOT EXISTS idx_memories_source_file_id + ON memories (source_file_id) WHERE source_file_id IS NOT NULL; +-- +goose StatementEnd + +-- +goose StatementBegin +CREATE INDEX IF NOT EXISTS idx_memories_created_by_user_id + ON memories (created_by_user_id) WHERE created_by_user_id IS NOT NULL; +-- +goose StatementEnd + +-- Item 3: memory_relations FKs must cascade with memories. +-- memories.workspace_id is ON DELETE CASCADE, so a workspace delete removes +-- memories rows. Without matching cascade on memory_relations, the delete then +-- fails on any edge touching those memories. Align the referential actions. +-- +goose StatementBegin +ALTER TABLE memory_relations + DROP CONSTRAINT IF EXISTS memory_relations_workspace_id_fkey, + DROP CONSTRAINT IF EXISTS memory_relations_source_id_fkey, + DROP CONSTRAINT IF EXISTS memory_relations_target_id_fkey; +-- +goose StatementEnd + +-- +goose StatementBegin +ALTER TABLE memory_relations + ADD CONSTRAINT memory_relations_workspace_id_fkey + FOREIGN KEY (workspace_id) REFERENCES workspaces(id) ON DELETE CASCADE, + ADD CONSTRAINT memory_relations_source_id_fkey + FOREIGN KEY (source_id) REFERENCES memories(id) ON DELETE CASCADE, + ADD CONSTRAINT memory_relations_target_id_fkey + FOREIGN KEY (target_id) REFERENCES memories(id) ON DELETE CASCADE; +-- +goose StatementEnd + +-- +goose Down +-- +goose StatementBegin +ALTER TABLE memory_relations + DROP CONSTRAINT IF EXISTS memory_relations_workspace_id_fkey, + DROP CONSTRAINT IF EXISTS memory_relations_source_id_fkey, + DROP CONSTRAINT IF EXISTS memory_relations_target_id_fkey; +-- +goose StatementEnd + +-- +goose StatementBegin +ALTER TABLE memory_relations + ADD CONSTRAINT memory_relations_workspace_id_fkey + FOREIGN KEY (workspace_id) REFERENCES workspaces(id), + ADD CONSTRAINT memory_relations_source_id_fkey + FOREIGN KEY (source_id) REFERENCES memories(id), + ADD CONSTRAINT memory_relations_target_id_fkey + FOREIGN KEY (target_id) REFERENCES memories(id); +-- +goose StatementEnd + +-- +goose StatementBegin +DROP INDEX IF EXISTS idx_memories_created_by_user_id; +DROP INDEX IF EXISTS idx_memories_source_file_id; +-- +goose StatementEnd + +-- +goose StatementBegin +ALTER TABLE embeddings_text + DROP CONSTRAINT IF EXISTS uq_embeddings_text_file_chunk; +-- +goose StatementEnd diff --git a/server/internal/db/migrations_test.go b/server/internal/db/migrations_test.go new file mode 100644 index 0000000..a4c1338 --- /dev/null +++ b/server/internal/db/migrations_test.go @@ -0,0 +1,109 @@ +package db + +import ( + "context" + "errors" + "os" + "strings" + "testing" + "time" + + "github.com/google/uuid" + "github.com/jackc/pgx/v5/pgconn" + "github.com/jackc/pgx/v5/pgxpool" +) + +func TestEmbeddingsTextUniqueConstraint(t *testing.T) { + dsn := os.Getenv("MEM_TEST_DB") + if dsn == "" { + t.Skip("MEM_TEST_DB not set; skipping DB integration test") + } + config, err := pgxpool.ParseConfig(dsn) + if err != nil { + t.Fatalf("parse MEM_TEST_DB: %v", err) + } + if !strings.HasSuffix(config.ConnConfig.Database, "_test") { + t.Fatalf("refusing to modify non-test database %q", config.ConnConfig.Database) + } + + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + pool, err := pgxpool.NewWithConfig(ctx, config) + if err != nil { + t.Fatalf("connect: %v", err) + } + defer pool.Close() + + db := &DB{Pool: pool, url: dsn} + if err := db.Migrate(ctx); err != nil { + t.Fatalf("migrate: %v", err) + } + tx, err := pool.Begin(ctx) + if err != nil { + t.Fatal(err) + } + defer tx.Rollback(context.Background()) + + var userID, workspaceID, fileID uuid.UUID + if err := tx.QueryRow(ctx, ` + INSERT INTO users (email, password_hash) VALUES ($1, 'x') + RETURNING id + `, uuid.NewString()+"@example.test").Scan(&userID); err != nil { + t.Fatalf("seed user: %v", err) + } + if err := tx.QueryRow(ctx, ` + INSERT INTO workspaces (name, resource_owner_user_id) + VALUES ('unique-ctest-ws', $1) + ON CONFLICT (resource_owner_user_id) DO UPDATE SET name = EXCLUDED.name + RETURNING id + `, userID).Scan(&workspaceID); err != nil { + t.Fatalf("seed workspace: %v", err) + } + if err := tx.QueryRow(ctx, ` + INSERT INTO files (user_id, name, path, size, sha256, mime, storage_key) + VALUES ($1, 'unique-ctest.txt', '/unique-ctest.txt', 0, '', 'text/plain', 'test://unique') + RETURNING id + `, userID).Scan(&fileID); err != nil { + t.Fatalf("seed file: %v", err) + } + if _, err := tx.Exec(ctx, ` + INSERT INTO embeddings_text (file_id, chunk_index, chunk_text, provider) + VALUES ($1, 0, 'chunk zero', 'test') + `, fileID); err != nil { + t.Fatalf("first insert: %v", err) + } + // Replay the actual migration over populated, pre-constraint data inside + // this rollback-only transaction. A fresh-schema migrate alone misses this. + if _, err := tx.Exec(ctx, `ALTER TABLE embeddings_text DROP CONSTRAINT uq_embeddings_text_file_chunk`); err != nil { + t.Fatal(err) + } + if _, err := tx.Exec(ctx, `INSERT INTO embeddings_text (file_id, chunk_index, chunk_text) + VALUES ($1, 0, 'legacy duplicate')`, fileID); err != nil { + t.Fatal(err) + } + migration, err := migrationsFS.ReadFile("migrations/0027_data_plane_hygiene.sql") + if err != nil { + t.Fatal(err) + } + up := strings.SplitN(string(migration), "-- +goose Down", 2)[0] + if _, err := tx.Exec(ctx, up); err != nil { + t.Fatalf("migrate populated table: %v", err) + } + var survivors int + if err := tx.QueryRow(ctx, `SELECT count(*) FROM embeddings_text WHERE file_id=$1`, fileID).Scan(&survivors); err != nil || survivors != 1 { + t.Fatalf("deduplicated rows = %d, err=%v", survivors, err) + } + + _, err = tx.Exec(ctx, ` + INSERT INTO embeddings_text (file_id, chunk_index, chunk_text, provider) + VALUES ($1, 0, 'duplicate chunk', 'test') + `, fileID) + if err == nil { + t.Fatal("expected duplicate (file_id, chunk_index) to be rejected") + } + var pgErr *pgconn.PgError + if !errors.As(err, &pgErr) || pgErr.Code != "23505" { + t.Fatalf("expected unique violation 23505, got %v", err) + } +} diff --git a/server/internal/indexgeneration/service.go b/server/internal/indexgeneration/service.go index a8bb35b..4220936 100644 --- a/server/internal/indexgeneration/service.go +++ b/server/internal/indexgeneration/service.go @@ -287,19 +287,33 @@ func (s *Service) List(ctx context.Context, workspaceID uuid.UUID, limit int) ([ } defer rows.Close() out := make([]Build, 0, limit) + buildIDs := make([]uuid.UUID, 0, limit) for rows.Next() { build, err := scanBuild(rows) if err != nil { return nil, fmt.Errorf("%w: scan build: %v", ErrUnavailable, err) } - generations, err := listGenerations(ctx, s.pool, workspaceID, build.ID) - if err != nil { - return nil, err - } - build.Generations = generations out = append(out, *build) + buildIDs = append(buildIDs, build.ID) } - return out, rows.Err() + if err := rows.Err(); err != nil { + return nil, err + } + if len(buildIDs) == 0 { + return out, nil + } + generations, err := listGenerationsForBuilds(ctx, s.pool, workspaceID, buildIDs) + if err != nil { + return nil, err + } + byBuild := make(map[uuid.UUID][]Generation, len(buildIDs)) + for _, g := range generations { + byBuild[g.BuildID] = append(byBuild[g.BuildID], g) + } + for i := range out { + out[i].Generations = byBuild[out[i].ID] + } + return out, nil } func (s *Service) Cancel( diff --git a/server/internal/indexgeneration/service_integration_test.go b/server/internal/indexgeneration/service_integration_test.go index 7d812ff..07417f0 100644 --- a/server/internal/indexgeneration/service_integration_test.go +++ b/server/internal/indexgeneration/service_integration_test.go @@ -8,10 +8,12 @@ import ( "os" "reflect" "strings" + "sync/atomic" "testing" "time" "github.com/google/uuid" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" "github.com/jackc/pgx/v5/pgxpool" @@ -418,6 +420,37 @@ func TestIndexGenerationPostgres(t *testing.T) { if err != nil { t.Fatalf("create replacement generation: %v", err) } + listConfig, err := pgxpool.ParseConfig(dsn) + if err != nil { + t.Fatal(err) + } + tracer := &generationListQueryCounter{} + listConfig.MaxConns = 1 + listConfig.ConnConfig.Tracer = tracer + listPool, err := pgxpool.NewWithConfig(ctx, listConfig) + if err != nil { + t.Fatal(err) + } + defer listPool.Close() + listCtx, listCancel := context.WithTimeout(ctx, 5*time.Second) + builds, err := New(listPool).List(listCtx, ws.ID, 100) + listCancel() + if err != nil || len(builds) != 3 { + t.Fatalf("batch list builds = %#v, err=%v", builds, err) + } + if got := tracer.count.Load(); got != 2 { + t.Fatalf("List used %d queries for 3 builds; want 2", got) + } + for _, build := range builds { + if len(build.Generations) == 0 { + t.Fatalf("build %s lost its generations", build.ID) + } + for _, generation := range build.Generations { + if generation.BuildID != build.ID || generation.WorkspaceID != ws.ID { + t.Fatalf("generation attached to wrong build: %#v", generation) + } + } + } failing, err := service.ClaimTarget(ctx, ws.ID, managed.ID) if err != nil { t.Fatal(err) @@ -555,6 +588,16 @@ func TestIndexGenerationPostgres(t *testing.T) { } } +type generationListQueryCounter struct{ count atomic.Int64 } + +func (c *generationListQueryCounter) TraceQueryStart(ctx context.Context, _ *pgx.Conn, _ pgx.TraceQueryStartData) context.Context { + c.count.Add(1) + return ctx +} + +func (c *generationListQueryCounter) TraceQueryEnd(context.Context, *pgx.Conn, pgx.TraceQueryEndData) { +} + func insertGenerationTestFile( t *testing.T, ctx context.Context, diff --git a/server/internal/indexgeneration/store.go b/server/internal/indexgeneration/store.go index 245236e..8e06ba2 100644 --- a/server/internal/indexgeneration/store.go +++ b/server/internal/indexgeneration/store.go @@ -106,6 +106,42 @@ func listGenerations( return out, rows.Err() } +func listGenerationsForBuilds( + ctx context.Context, + q queryer, + workspaceID uuid.UUID, + buildIDs []uuid.UUID, +) ([]Generation, error) { + rows, err := q.Query(ctx, ` + SELECT id, build_id, workspace_id, route_kind, provider, + model_revision, output_dimension, pipeline_revision, + profile_id, profile_revision, state, created_at, updated_at + FROM index_generations + WHERE workspace_id = $1 AND build_id = ANY($2::uuid[]) + ORDER BY build_id, route_kind + `, workspaceID, buildIDs) + if err != nil { + return nil, fmt.Errorf("%w: list route generations batch: %v", ErrUnavailable, err) + } + defer rows.Close() + var out []Generation + for rows.Next() { + var generation Generation + if err := rows.Scan( + &generation.ID, &generation.BuildID, &generation.WorkspaceID, + &generation.RouteKind, &generation.Provider, + &generation.ModelRevision, &generation.OutputDimension, + &generation.PipelineRevision, &generation.ProfileID, + &generation.ProfileRevision, &generation.State, + &generation.CreatedAt, &generation.UpdatedAt, + ); err != nil { + return nil, fmt.Errorf("%w: scan route generation: %v", ErrUnavailable, err) + } + out = append(out, generation) + } + return out, rows.Err() +} + func getInflightIdentity( ctx context.Context, q queryer, diff --git a/server/internal/memory/memory_integration_test.go b/server/internal/memory/memory_integration_test.go index e27bc0c..87af545 100644 --- a/server/internal/memory/memory_integration_test.go +++ b/server/internal/memory/memory_integration_test.go @@ -1270,37 +1270,68 @@ func TestMemoryPostgres(t *testing.T) { t.Fatalf("superseded markers = %+v", superseded) } - outbound, err := service.ListRelations(ctx, ListRelationsQuery{ + outboundResult, err := service.ListRelations(ctx, ListRelationsQuery{ WorkspaceID: workspaceA, MemoryID: fresh.Memory.ID, Direction: "source", AllowedPaths: []string{scope}, }) - if err != nil || len(outbound) != 1 || - outbound[0].TargetID != old.Memory.ID || - outbound[0].RelationType != RelSupersedes || - outbound[0].Reason != "decision updated" { - t.Fatalf("outbound relations = %+v err=%v", outbound, err) + if err != nil || len(outboundResult.Relations) != 1 || + outboundResult.Relations[0].TargetID != old.Memory.ID || + outboundResult.Relations[0].RelationType != RelSupersedes || + outboundResult.Relations[0].Reason != "decision updated" { + t.Fatalf("outbound relations = %+v err=%v", outboundResult, err) } - inbound, err := service.ListRelations(ctx, ListRelationsQuery{ + inboundResult, err := service.ListRelations(ctx, ListRelationsQuery{ WorkspaceID: workspaceA, MemoryID: old.Memory.ID, Direction: "target", RelationType: RelSupersedes, AllowedPaths: []string{scope}, }) - if err != nil || len(inbound) != 1 || inbound[0].SourceID != fresh.Memory.ID { - t.Fatalf("inbound relations = %+v err=%v", inbound, err) + if err != nil || len(inboundResult.Relations) != 1 || inboundResult.Relations[0].SourceID != fresh.Memory.ID { + t.Fatalf("inbound relations = %+v err=%v", inboundResult, err) } - typed, err := service.ListRelations(ctx, ListRelationsQuery{ + typedResult, err := service.ListRelations(ctx, ListRelationsQuery{ WorkspaceID: workspaceA, MemoryID: old.Memory.ID, Direction: "target", RelationType: RelCorrects, AllowedPaths: []string{scope}, }) - if err != nil || len(typed) != 0 { - t.Fatalf("type-filtered relations = %+v err=%v", typed, err) + if err != nil || len(typedResult.Relations) != 0 { + t.Fatalf("type-filtered relations = %+v err=%v", typedResult, err) + } + + // Keyset pages must preserve equal-timestamp ties without duplicates. + for _, target := range []uuid.UUID{old.Memory.ID, newest.Memory.ID} { + if _, err := service.CreateRelation(ctx, CreateRelationCommand{ + WorkspaceID: workspaceA, SourceID: fresh.Memory.ID, + TargetID: target, RelationType: RelOccurrenceOf, AllowedPaths: []string{scope}, + }); err != nil { + t.Fatal(err) + } + } + if _, err := database.Pool.Exec(ctx, `UPDATE memory_relations SET created_at = '2026-01-01T00:00:00Z' + WHERE workspace_id = $1 AND source_id = $2 AND relation_type = $3`, + workspaceA, fresh.Memory.ID, RelOccurrenceOf); err != nil { + t.Fatal(err) + } + pageQuery := ListRelationsQuery{WorkspaceID: workspaceA, MemoryID: fresh.Memory.ID, + RelationType: RelOccurrenceOf, AllowedPaths: []string{scope}, Limit: 1} + firstPage, err := service.ListRelations(ctx, pageQuery) + if err != nil || len(firstPage.Relations) != 1 || firstPage.NextCursor == "" { + t.Fatalf("first relation page = %+v, err=%v", firstPage, err) + } + pageQuery.Cursor = firstPage.NextCursor + secondPage, err := service.ListRelations(ctx, pageQuery) + if err != nil || len(secondPage.Relations) != 1 || secondPage.NextCursor != "" || + secondPage.Relations[0].ID == firstPage.Relations[0].ID { + t.Fatalf("second relation page = %+v, err=%v", secondPage, err) + } + pageQuery.RelationType = RelSupersedes + if _, err := service.ListRelations(ctx, pageQuery); err == nil { + t.Fatal("relation cursor reused with a different filter") } // Anchors the caller cannot read are hidden as not found. diff --git a/server/internal/memory/relation.go b/server/internal/memory/relation.go index b9c238f..c8fbcda 100644 --- a/server/internal/memory/relation.go +++ b/server/internal/memory/relation.go @@ -2,6 +2,9 @@ package memory import ( "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" "errors" "fmt" "strings" @@ -79,6 +82,13 @@ type ListRelationsQuery struct { RelationType string // optional filter AllowedPaths []string Limit int + Cursor string // opaque keyset cursor from a prior ListRelationsResult +} + +// ListRelationsResult is a keyset-paginated page of relations. +type ListRelationsResult struct { + Relations []Relation `json:"relations"` + NextCursor string `json:"next_cursor,omitempty"` } // validateCreateRelationCommand checks the command fields without touching the database. @@ -235,8 +245,9 @@ func validateListRelationsQuery(q ListRelationsQuery) error { } // ListRelations returns relations for a given memory. Direction controls -// whether MemoryID is matched as source or target. -func (s *Service) ListRelations(ctx context.Context, q ListRelationsQuery) ([]Relation, error) { +// whether MemoryID is matched as source or target. Results are keyset-paginated +// using the same cursor shape as memory/list.go. +func (s *Service) ListRelations(ctx context.Context, q ListRelationsQuery) (*ListRelationsResult, error) { if s == nil || s.pool == nil { return nil, fmt.Errorf("memory service is not configured") } @@ -259,9 +270,6 @@ func (s *Service) ListRelations(ctx context.Context, q ListRelationsQuery) ([]Re return nil, err } - // The anchor memory must exist, be visible under the caller's allowed - // paths, and not be forgotten — mirroring Get — so listing relations never - // leaks the existence of a memory the caller cannot read. args := []any{q.WorkspaceID, q.MemoryID} where := []string{"m.workspace_id = $1", "m.id = $2"} args, where = appendPathFilters(args, where, "m.path", "/", allowed) @@ -280,6 +288,20 @@ func (s *Service) ListRelations(ctx context.Context, q ListRelationsQuery) ([]Re return nil, ErrForgotten } + filterHash, err := relationFilterHash(q.WorkspaceID, q.MemoryID, direction, q.RelationType) + if err != nil { + return nil, fmt.Errorf("list relations: filter hash: %w", err) + } + + var cursor *decodedListCursor + if strings.TrimSpace(q.Cursor) != "" { + decoded, err := decodeListCursor(q.Cursor, filterHash) + if err != nil { + return nil, err + } + cursor = &decoded + } + args = []any{q.WorkspaceID, q.MemoryID} var dirColumn string if direction == "source" { @@ -296,7 +318,15 @@ func (s *Service) ListRelations(ctx context.Context, q ListRelationsQuery) ([]Re args = append(args, relType) where = append(where, fmt.Sprintf("r.relation_type = $%d", len(args))) } - args = append(args, q.Limit) + if cursor != nil { + args = append(args, cursor.createdAt, cursor.id) + timeArg, idArg := len(args)-1, len(args) + where = append(where, fmt.Sprintf( + "(r.created_at < $%d OR (r.created_at = $%d AND r.id > $%d))", + timeArg, timeArg, idArg, + )) + } + args = append(args, q.Limit+1) limitIdx := len(args) sql := fmt.Sprintf(` @@ -313,7 +343,7 @@ func (s *Service) ListRelations(ctx context.Context, q ListRelationsQuery) ([]Re } defer rows.Close() - out := make([]Relation, 0, q.Limit) + out := make([]Relation, 0, q.Limit+1) for rows.Next() { var rel Relation if err := rows.Scan(&rel.ID, &rel.WorkspaceID, &rel.SourceID, &rel.TargetID, @@ -322,7 +352,43 @@ func (s *Service) ListRelations(ctx context.Context, q ListRelationsQuery) ([]Re } out = append(out, rel) } - return out, rows.Err() + if err := rows.Err(); err != nil { + return nil, err + } + + result := &ListRelationsResult{Relations: out} + if len(out) <= q.Limit { + return result, nil + } + result.Relations = out[:q.Limit] + last := result.Relations[len(result.Relations)-1] + result.NextCursor, err = encodeListCursor(last.CreatedAt, last.ID, filterHash) + if err != nil { + return nil, fmt.Errorf("list relations: encode cursor: %w", err) + } + return result, nil +} + +type relationFilterFingerprint struct { + WorkspaceID string `json:"workspace_id"` + MemoryID string `json:"memory_id"` + Direction string `json:"direction"` + RelationType string `json:"relation_type"` +} + +func relationFilterHash(workspaceID, memoryID uuid.UUID, direction, relationType string) (string, error) { + payload := relationFilterFingerprint{ + WorkspaceID: workspaceID.String(), + MemoryID: memoryID.String(), + Direction: direction, + RelationType: strings.ToLower(strings.TrimSpace(relationType)), + } + encoded, err := json.Marshal(payload) + if err != nil { + return "", err + } + sum := sha256.Sum256(encoded) + return hex.EncodeToString(sum[:]), nil } // IsSuperseded returns true if the given memory has been superseded or corrected @@ -410,8 +476,6 @@ func (s *Service) loadMemoryInTx(ctx context.Context, tx pgx.Tx, workspaceID, me // from target and looks for source. The exact edge being written is excluded // so an idempotent replay of an existing edge is not reported as a cycle. func (s *Service) wouldCycle(ctx context.Context, tx pgx.Tx, workspaceID, sourceID, targetID uuid.UUID) (bool, error) { - // BFS from target following supersedes/corrects edges in their forward - // direction: if we ever reach sourceID, there is a cycle. var found bool err := tx.QueryRow(ctx, ` WITH RECURSIVE chain(id) AS (