diff --git a/internal/postgres/raw_ingest_custody_pgtest_test.go b/internal/postgres/raw_ingest_custody_pgtest_test.go new file mode 100644 index 0000000000..7e5066789e --- /dev/null +++ b/internal/postgres/raw_ingest_custody_pgtest_test.go @@ -0,0 +1,272 @@ +//go:build pgtest + +package postgres + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "io" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.kenn.io/agentsview/internal/artifact" + "go.kenn.io/agentsview/internal/parser" + "go.kenn.io/agentsview/internal/rawsync" +) + +func TestRawCustodyEndToEnd(t *testing.T) { + pgURL := testPGURL(t) + cleanSchemaTestPG(t, pgURL) + t.Cleanup(func() { cleanSchemaTestPG(t, pgURL) }) + pg, err := Open(pgURL, schemaTestSchema, true) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, pg.Close()) }) + require.NoError(t, EnsureSchema(t.Context(), pg, schemaTestSchema)) + + repository, err := artifact.OpenRepository(t.Context(), t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, repository.Close()) }) + objects, err := rawsync.NewArtifactObjectStore(repository.Content()) + require.NoError(t, err) + metadata, err := NewRawIngestStore(pg) + require.NoError(t, err) + service, err := rawsync.NewService( + objects, metadata, rawsync.DefaultManifestLimits(), "parser-data-17", + ) + require.NoError(t, err) + + identity, err := rawsync.NewAuthIdentity("tenant-a", "device-a") + require.NoError(t, err) + firstBody := []byte("{\"type\":\"user\"}\n") + secondBody := []byte("{\"type\":\"assistant\"}\n") + firstRef := rawCustodyObjectRef(t, firstBody) + secondRef := rawCustodyObjectRef(t, secondBody) + missingManifest := rawCustodyManifest( + "capture-missing", "", rawIngestCapturedAt(), firstRef, + ) + _, err = service.CommitManifest(t.Context(), identity, missingManifest) + assert.ErrorIs(t, err, rawsync.ErrMissingObject) + assert.Equal(t, rawIngestCounts{}, readRawIngestCounts(t, pg)) + + missing, err := service.MissingObjects( + t.Context(), identity, parser.AgentCodex, []rawsync.ObjectRef{firstRef, secondRef}, + ) + require.NoError(t, err) + assert.Equal(t, []rawsync.ObjectRef{firstRef, secondRef}, missing) + firstPut, err := service.FinalizeObject( + t.Context(), identity, parser.AgentCodex, firstRef, bytes.NewReader(firstBody), + ) + require.NoError(t, err) + assert.True(t, firstPut.Created) + firstRetry, err := service.FinalizeObject( + t.Context(), identity, parser.AgentCodex, firstRef, bytes.NewReader(firstBody), + ) + require.NoError(t, err) + assert.False(t, firstRetry.Created) + + firstManifest := rawCustodyManifest( + "capture-a", "", rawIngestCapturedAt(), firstRef, + ) + firstCanonical, err := rawsync.ValidateAndCanonicalize( + identity, firstManifest, rawsync.DefaultManifestLimits(), + ) + require.NoError(t, err) + firstAccepted, err := service.CommitManifest(t.Context(), identity, firstManifest) + require.NoError(t, err) + assert.True(t, firstAccepted.Created) + assert.Equal(t, int64(1), firstAccepted.Generation) + + firstAcceptedRetry, err := service.CommitManifest(t.Context(), identity, firstManifest) + require.NoError(t, err) + assert.False(t, firstAcceptedRetry.Created) + assert.Equal(t, firstAccepted.Receipt, firstAcceptedRetry.Receipt) + assert.Equal(t, firstAccepted.Generation, firstAcceptedRetry.Generation) + + _, err = service.FinalizeObject( + t.Context(), identity, parser.AgentCodex, secondRef, bytes.NewReader(secondBody), + ) + require.NoError(t, err) + secondManifest := rawCustodyManifest( + "capture-b", firstAccepted.Receipt, + rawIngestCapturedAt().Add(time.Minute), firstRef, secondRef, + ) + secondCanonical, err := rawsync.ValidateAndCanonicalize( + identity, secondManifest, rawsync.DefaultManifestLimits(), + ) + require.NoError(t, err) + secondAccepted, err := service.CommitManifest(t.Context(), identity, secondManifest) + require.NoError(t, err) + assert.True(t, secondAccepted.Created) + assert.Equal(t, int64(2), secondAccepted.Generation) + + assert.Equal(t, + rawIngestCounts{Manifests: 2, Entries: 2, Objects: 3, Heads: 1, Jobs: 2}, + readRawIngestCounts(t, pg), + ) + assert.Equal(t, 2, rawIngestTableCount(t, pg, "raw_objects")) + var storedCanonical []byte + var storedParent, storedCapture, storedKind string + var storedGeneration int64 + require.NoError(t, pg.QueryRowContext(t.Context(), ` + SELECT canonical_json, parent_receipt, capture_id, kind, generation + FROM raw_manifests + WHERE tenant_id = $1 AND manifest_id = $2`, + identity.TenantID, secondCanonical.ManifestID, + ).Scan( + &storedCanonical, &storedParent, &storedCapture, &storedKind, &storedGeneration, + )) + assert.Equal(t, secondCanonical.CanonicalJSON, storedCanonical) + assert.Equal(t, firstAccepted.Receipt, storedParent) + assert.Equal(t, "capture-b", storedCapture) + assert.Equal(t, "snapshot", storedKind) + assert.Equal(t, int64(2), storedGeneration) + + var exactJobs int + require.NoError(t, pg.QueryRowContext(t.Context(), ` + SELECT count(*) FROM raw_ingest_jobs + WHERE stage = 'parse' AND state = 'ready' + AND processing_version = 'parser-data-17'`, + ).Scan(&exactJobs)) + assert.Equal(t, 2, exactJobs) + + var storedPath, storedEntryType string + var storedLength int64 + require.NoError(t, pg.QueryRowContext(t.Context(), ` + SELECT path, entry_type, size_bytes + FROM raw_manifest_entries + WHERE tenant_id = $1 AND manifest_id = $2 AND entry_index = 0`, + identity.TenantID, secondCanonical.ManifestID, + ).Scan(&storedPath, &storedEntryType, &storedLength)) + assert.Equal(t, "session.jsonl", storedPath) + assert.Equal(t, "file", storedEntryType) + assert.Equal(t, firstRef.Length+secondRef.Length, storedLength) + + rows, err := pg.QueryContext(t.Context(), ` + SELECT sha256, size_bytes FROM raw_manifest_objects + WHERE tenant_id = $1 AND manifest_id = $2 + ORDER BY entry_index, object_index`, + identity.TenantID, secondCanonical.ManifestID, + ) + require.NoError(t, err) + storedObjects := make([]rawsync.ObjectRef, 0, 2) + for rows.Next() { + var object rawsync.ObjectRef + require.NoError(t, rows.Scan(&object.SHA256, &object.Length)) + storedObjects = append(storedObjects, object) + } + require.NoError(t, rows.Err()) + require.NoError(t, rows.Close()) + assert.Equal(t, []rawsync.ObjectRef{firstRef, secondRef}, storedObjects) + + assert.Equal(t, firstBody, readRawCustodyObject(t, objects, identity, firstRef)) + assert.Equal(t, secondBody, readRawCustodyObject(t, objects, identity, secondRef)) + assert.Equal(t, firstCanonical.CanonicalJSON, + readRawCustodyManifest(t, objects, identity, firstCanonical.ManifestID)) + assert.Equal(t, secondCanonical.CanonicalJSON, + readRawCustodyManifest(t, objects, identity, secondCanonical.ManifestID)) + + staleManifest := rawCustodyManifest( + "capture-c", firstAccepted.Receipt, + rawIngestCapturedAt().Add(2*time.Minute), secondRef, + ) + staleCanonical, err := rawsync.ValidateAndCanonicalize( + identity, staleManifest, rawsync.DefaultManifestLimits(), + ) + require.NoError(t, err) + _, err = service.CommitManifest(t.Context(), identity, staleManifest) + var headConflict *rawsync.HeadConflictError + require.ErrorAs(t, err, &headConflict) + assert.ErrorIs(t, err, rawsync.ErrConflict) + assert.Equal(t, secondAccepted.Receipt, headConflict.CurrentReceipt) + assert.Equal(t, int64(2), headConflict.CurrentGeneration) + assert.Equal(t, + rawIngestCounts{Manifests: 2, Entries: 2, Objects: 3, Heads: 1, Jobs: 2}, + readRawIngestCounts(t, pg), + ) + assert.Equal(t, staleCanonical.CanonicalJSON, + readRawCustodyManifest(t, objects, identity, staleCanonical.ManifestID), + "a finalized manifest remains available for reconciliation after a head conflict") + + var headManifest, headReceipt string + var headGeneration int64 + require.NoError(t, pg.QueryRowContext(t.Context(), ` + SELECT manifest_id, receipt, generation FROM raw_source_heads`, + ).Scan(&headManifest, &headReceipt, &headGeneration)) + assert.Equal(t, secondCanonical.ManifestID, headManifest) + assert.Equal(t, secondAccepted.Receipt, headReceipt) + assert.Equal(t, int64(2), headGeneration) +} + +func rawCustodyObjectRef(t *testing.T, body []byte) rawsync.ObjectRef { + t.Helper() + digest := sha256.Sum256(body) + object, err := rawsync.NewObjectRef(hex.EncodeToString(digest[:]), int64(len(body))) + require.NoError(t, err) + return object +} + +func rawCustodyManifest( + captureID string, + parentReceipt string, + capturedAt time.Time, + objects ...rawsync.ObjectRef, +) rawsync.Manifest { + var length int64 + for _, object := range objects { + length += object.Length + } + return rawsync.Manifest{ + SchemaVersion: rawsync.ManifestSchemaVersion, + Provider: parser.AgentCodex, + ConfiguredRootID: "root-a", + SourceKey: "sessions/demo.jsonl#main", + ExpectedParentReceipt: parentReceipt, + CaptureID: captureID, + CapturedAt: capturedAt, + Kind: rawsync.ManifestSnapshot, + Entries: []rawsync.Entry{{ + Path: "session.jsonl", Type: "file", Length: length, Objects: objects, + }}, + } +} + +func readRawCustodyObject( + t *testing.T, + objects rawsync.ObjectStore, + identity rawsync.AuthIdentity, + object rawsync.ObjectRef, +) []byte { + t.Helper() + info, reader, err := objects.OpenObject(t.Context(), identity.TenantID, object) + require.NoError(t, err) + require.Equal(t, object, info.Ref) + return readVerifiedRawCustodyBytes(t, reader) +} + +func readRawCustodyManifest( + t *testing.T, + objects rawsync.ObjectStore, + identity rawsync.AuthIdentity, + manifestID string, +) []byte { + t.Helper() + info, reader, err := objects.OpenManifest(t.Context(), identity, manifestID) + require.NoError(t, err) + require.Equal(t, manifestID, info.Ref.SHA256) + return readVerifiedRawCustodyBytes(t, reader) +} + +func readVerifiedRawCustodyBytes( + t *testing.T, + reader rawsync.VerifiedObjectReader, +) []byte { + t.Helper() + body, err := io.ReadAll(reader) + require.NoError(t, err) + require.NoError(t, reader.Verify()) + require.NoError(t, reader.Close()) + return body +} diff --git a/internal/postgres/raw_ingest_schema.go b/internal/postgres/raw_ingest_schema.go new file mode 100644 index 0000000000..3b49e26bda --- /dev/null +++ b/internal/postgres/raw_ingest_schema.go @@ -0,0 +1,209 @@ +package postgres + +import ( + "context" + "database/sql" + "errors" + "fmt" + "log" + "strings" + + "github.com/jackc/pgx/v5/pgconn" +) + +// Source keys and entry paths may be up to 4096 bytes, which exceeds the +// PostgreSQL B-tree index entry limit (about 2704 bytes on 8 kB pages) once +// combined with the other key columns. Composite keys therefore use fixed-size +// SHA-256 digests of those values; the full text is stored beside them. +const rawIngestDDL = ` +CREATE TABLE IF NOT EXISTS raw_objects ( + tenant_id TEXT NOT NULL, + sha256 TEXT NOT NULL CHECK (sha256 ~ '^[0-9a-f]{64}$'), + size_bytes BIGINT NOT NULL CHECK (size_bytes >= 0), + verified_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY (tenant_id, sha256), + UNIQUE (tenant_id, sha256, size_bytes) +); + +CREATE TABLE IF NOT EXISTS raw_manifests ( + tenant_id TEXT NOT NULL, + manifest_id TEXT NOT NULL CHECK (manifest_id ~ '^[0-9a-f]{64}$'), + device_id TEXT NOT NULL, + provider TEXT NOT NULL, + configured_root_id TEXT NOT NULL, + source_key TEXT NOT NULL, + source_key_sha256 TEXT NOT NULL CHECK (source_key_sha256 ~ '^[0-9a-f]{64}$'), + capture_id TEXT NOT NULL, + parent_receipt TEXT NOT NULL DEFAULT '' + CHECK (parent_receipt = '' OR parent_receipt ~ '^[0-9a-f]{64}$'), + receipt TEXT NOT NULL CHECK (receipt ~ '^[0-9a-f]{64}$'), + generation BIGINT NOT NULL CHECK (generation > 0), + kind TEXT NOT NULL CHECK (kind IN ('snapshot', 'tombstone')), + captured_at TIMESTAMPTZ NOT NULL, + accepted_at TIMESTAMPTZ NOT NULL DEFAULT now(), + canonical_json BYTEA NOT NULL, + PRIMARY KEY (tenant_id, manifest_id), + UNIQUE (tenant_id, receipt), + UNIQUE ( + tenant_id, device_id, provider, configured_root_id, source_key_sha256, + generation + ), + UNIQUE ( + tenant_id, device_id, provider, configured_root_id, source_key_sha256, + capture_id + ) +); + +CREATE TABLE IF NOT EXISTS raw_manifest_entries ( + tenant_id TEXT NOT NULL, + manifest_id TEXT NOT NULL, + entry_index INTEGER NOT NULL CHECK (entry_index >= 0), + path TEXT NOT NULL, + path_sha256 TEXT NOT NULL CHECK (path_sha256 ~ '^[0-9a-f]{64}$'), + entry_type TEXT NOT NULL CHECK (entry_type = 'file'), + size_bytes BIGINT NOT NULL CHECK (size_bytes >= 0), + PRIMARY KEY (tenant_id, manifest_id, entry_index), + UNIQUE (tenant_id, manifest_id, path_sha256), + FOREIGN KEY (tenant_id, manifest_id) + REFERENCES raw_manifests (tenant_id, manifest_id) ON DELETE RESTRICT +); + +CREATE TABLE IF NOT EXISTS raw_manifest_objects ( + tenant_id TEXT NOT NULL, + manifest_id TEXT NOT NULL, + entry_index INTEGER NOT NULL, + object_index INTEGER NOT NULL CHECK (object_index >= 0), + sha256 TEXT NOT NULL, + size_bytes BIGINT NOT NULL CHECK (size_bytes >= 0), + PRIMARY KEY (tenant_id, manifest_id, entry_index, object_index), + FOREIGN KEY (tenant_id, manifest_id, entry_index) + REFERENCES raw_manifest_entries (tenant_id, manifest_id, entry_index) + ON DELETE RESTRICT, + FOREIGN KEY (tenant_id, sha256, size_bytes) + REFERENCES raw_objects (tenant_id, sha256, size_bytes) + ON DELETE RESTRICT +); + +CREATE TABLE IF NOT EXISTS raw_source_heads ( + tenant_id TEXT NOT NULL, + device_id TEXT NOT NULL, + provider TEXT NOT NULL, + configured_root_id TEXT NOT NULL, + source_key TEXT NOT NULL, + source_key_sha256 TEXT NOT NULL CHECK (source_key_sha256 ~ '^[0-9a-f]{64}$'), + manifest_id TEXT, + receipt TEXT CHECK (receipt IS NULL OR receipt ~ '^[0-9a-f]{64}$'), + generation BIGINT NOT NULL DEFAULT 0 CHECK (generation >= 0), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + PRIMARY KEY ( + tenant_id, device_id, provider, configured_root_id, source_key_sha256 + ), + CHECK ( + (generation = 0 AND manifest_id IS NULL AND receipt IS NULL) + OR (generation > 0 AND manifest_id IS NOT NULL AND receipt IS NOT NULL) + ), + FOREIGN KEY (tenant_id, manifest_id) + REFERENCES raw_manifests (tenant_id, manifest_id) ON DELETE RESTRICT +); + +CREATE TABLE IF NOT EXISTS raw_ingest_jobs ( + id BIGSERIAL PRIMARY KEY, + tenant_id TEXT NOT NULL, + manifest_id TEXT NOT NULL, + stage TEXT NOT NULL CHECK (stage IN ('parse')), + processing_version TEXT NOT NULL, + state TEXT NOT NULL DEFAULT 'ready' + CHECK ( + state IN ( + 'ready', 'leased', 'retrying', 'complete', 'failed', + 'superseded' + ) + ), + attempt_count INTEGER NOT NULL DEFAULT 0 CHECK (attempt_count >= 0), + available_at TIMESTAMPTZ NOT NULL DEFAULT now(), + lease_owner TEXT NOT NULL DEFAULT '', + lease_expires_at TIMESTAMPTZ, + last_error_class TEXT NOT NULL DEFAULT '', + last_error TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + UNIQUE (tenant_id, manifest_id, stage, processing_version), + FOREIGN KEY (tenant_id, manifest_id) + REFERENCES raw_manifests (tenant_id, manifest_id) ON DELETE RESTRICT +); + +CREATE INDEX IF NOT EXISTS idx_raw_ingest_jobs_ready + ON raw_ingest_jobs (state, available_at, id) + WHERE state IN ('ready', 'retrying'); +CREATE INDEX IF NOT EXISTS idx_raw_ingest_jobs_lease + ON raw_ingest_jobs (lease_expires_at, id) + WHERE state = 'leased'; +` + +const rawIngestAppendOnlyDDL = ` +CREATE OR REPLACE FUNCTION raw_ingest_reject_accepted_mutation() +RETURNS trigger +LANGUAGE plpgsql +AS $raw_ingest_immutable$ +BEGIN + RAISE EXCEPTION 'accepted raw custody metadata is append-only' + USING ERRCODE = '55000'; +END; +$raw_ingest_immutable$; + +DO $raw_ingest_triggers$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_trigger + WHERE tgname = 'raw_manifests_append_only' + AND tgrelid = 'raw_manifests'::regclass + ) THEN + CREATE TRIGGER raw_manifests_append_only + BEFORE UPDATE OR DELETE ON raw_manifests + FOR EACH ROW EXECUTE FUNCTION raw_ingest_reject_accepted_mutation(); + END IF; + IF NOT EXISTS ( + SELECT 1 FROM pg_trigger + WHERE tgname = 'raw_manifest_entries_append_only' + AND tgrelid = 'raw_manifest_entries'::regclass + ) THEN + CREATE TRIGGER raw_manifest_entries_append_only + BEFORE UPDATE OR DELETE ON raw_manifest_entries + FOR EACH ROW EXECUTE FUNCTION raw_ingest_reject_accepted_mutation(); + END IF; + IF NOT EXISTS ( + SELECT 1 FROM pg_trigger + WHERE tgname = 'raw_manifest_objects_append_only' + AND tgrelid = 'raw_manifest_objects'::regclass + ) THEN + CREATE TRIGGER raw_manifest_objects_append_only + BEFORE UPDATE OR DELETE ON raw_manifest_objects + FOR EACH ROW EXECUTE FUNCTION raw_ingest_reject_accepted_mutation(); + END IF; +END; +$raw_ingest_triggers$; +` + +func ensureRawIngestSchemaPG(ctx context.Context, db *sql.DB) error { + if _, err := db.ExecContext(ctx, rawIngestDDL); err != nil { + return fmt.Errorf("creating raw ingest schema: %w", err) + } + if _, err := db.ExecContext(ctx, rawIngestAppendOnlyDDL); err != nil { + if !rawIngestAppendOnlyUnsupported(err) { + return fmt.Errorf("installing raw ingest append-only guards: %w", err) + } + log.Printf( + "pg schema: raw custody append-only triggers unsupported; " + + "immutability remains application-enforced", + ) + } + return nil +} + +func rawIngestAppendOnlyUnsupported(err error) bool { + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) { + return pgErr.Code == "0A000" + } + return strings.Contains(strings.ToUpper(err.Error()), "SQLSTATE 0A000") +} diff --git a/internal/postgres/raw_ingest_schema_pgtest_test.go b/internal/postgres/raw_ingest_schema_pgtest_test.go new file mode 100644 index 0000000000..68ba6a6531 --- /dev/null +++ b/internal/postgres/raw_ingest_schema_pgtest_test.go @@ -0,0 +1,277 @@ +//go:build pgtest + +package postgres + +import ( + "net/url" + "strings" + "testing" + "time" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestEnsureSchemaCreatesRawIngestCustodyTables(t *testing.T) { + pgURL := testPGURL(t) + cleanSchemaTestPG(t, pgURL) + t.Cleanup(func() { cleanSchemaTestPG(t, pgURL) }) + pg, err := Open(pgURL, schemaTestSchema, true) + require.NoError(t, err) + defer pg.Close() + + require.NoError(t, EnsureSchema(t.Context(), pg, schemaTestSchema)) + require.NoError(t, EnsureSchema(t.Context(), pg, schemaTestSchema)) + + for _, table := range []string{ + "raw_objects", + "raw_manifests", + "raw_manifest_entries", + "raw_manifest_objects", + "raw_source_heads", + "raw_ingest_jobs", + } { + var exists bool + err := pg.QueryRowContext(t.Context(), ` + SELECT EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_schema = $1 AND table_name = $2 + )`, schemaTestSchema, table).Scan(&exists) + require.NoError(t, err) + assert.True(t, exists, table) + } + + for _, index := range []string{ + "idx_raw_ingest_jobs_ready", + "idx_raw_ingest_jobs_lease", + } { + var exists bool + err := pg.QueryRowContext(t.Context(), ` + SELECT EXISTS ( + SELECT 1 FROM pg_indexes + WHERE schemaname = $1 AND indexname = $2 + )`, schemaTestSchema, index).Scan(&exists) + require.NoError(t, err) + assert.True(t, exists, index) + } +} + +func TestRawIngestSchemaUsesTenantScopedKeys(t *testing.T) { + pgURL := testPGURL(t) + cleanSchemaTestPG(t, pgURL) + t.Cleanup(func() { cleanSchemaTestPG(t, pgURL) }) + pg, err := Open(pgURL, schemaTestSchema, true) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, pg.Close()) }) + require.NoError(t, EnsureSchema(t.Context(), pg, schemaTestSchema)) + + wantDefinitions := map[string][]string{ + "raw_objects": { + "PRIMARY KEY (tenant_id, sha256)", + }, + "raw_manifests": { + "PRIMARY KEY (tenant_id, manifest_id)", + "UNIQUE (tenant_id, device_id, provider, configured_root_id, source_key_sha256, generation)", + "UNIQUE (tenant_id, device_id, provider, configured_root_id, source_key_sha256, capture_id)", + }, + "raw_manifest_entries": { + "PRIMARY KEY (tenant_id, manifest_id, entry_index)", + "UNIQUE (tenant_id, manifest_id, path_sha256)", + "FOREIGN KEY (tenant_id, manifest_id)", + }, + "raw_manifest_objects": { + "PRIMARY KEY (tenant_id, manifest_id, entry_index, object_index)", + "FOREIGN KEY (tenant_id, manifest_id, entry_index)", + "FOREIGN KEY (tenant_id, sha256, size_bytes)", + }, + "raw_source_heads": { + "PRIMARY KEY (tenant_id, device_id, provider, configured_root_id, source_key_sha256)", + "FOREIGN KEY (tenant_id, manifest_id)", + }, + "raw_ingest_jobs": { + "UNIQUE (tenant_id, manifest_id, stage, processing_version)", + "FOREIGN KEY (tenant_id, manifest_id)", + }, + } + for table, wanted := range wantDefinitions { + rows, err := pg.QueryContext(t.Context(), ` + SELECT pg_get_constraintdef(oid) + FROM pg_constraint + WHERE conrelid = $1::regclass`, schemaTestSchema+"."+table) + require.NoError(t, err) + var definitions []string + for rows.Next() { + var definition string + require.NoError(t, rows.Scan(&definition)) + definitions = append(definitions, definition) + } + require.NoError(t, rows.Err()) + require.NoError(t, rows.Close()) + joined := strings.Join(definitions, "\n") + for _, definition := range wanted { + assert.Contains(t, joined, definition, table) + } + } +} + +func TestRawIngestSchemaEnforcesCaptureAndObjectCustody(t *testing.T) { + pgURL := testPGURL(t) + cleanSchemaTestPG(t, pgURL) + t.Cleanup(func() { cleanSchemaTestPG(t, pgURL) }) + pg, err := Open(pgURL, schemaTestSchema, true) + require.NoError(t, err) + defer pg.Close() + require.NoError(t, EnsureSchema(t.Context(), pg, schemaTestSchema)) + + insertManifest := ` + INSERT INTO raw_manifests ( + tenant_id, manifest_id, device_id, provider, configured_root_id, + source_key, source_key_sha256, capture_id, parent_receipt, receipt, + generation, kind, captured_at, canonical_json + ) VALUES ($1, $2, 'device-a', 'codex', 'root-a', 'source-a', $6, + 'capture-a', '', $3, 1, 'snapshot', $4, $5)` + firstManifest := strings.Repeat("a", 64) + sourceKeyDigest := rawIngestKeyDigest("source-a") + _, err = pg.ExecContext(t.Context(), insertManifest, + "tenant-a", firstManifest, repeatedHex("b"), + time.Date(2026, 8, 13, 12, 0, 0, 0, time.UTC), []byte("{}"), sourceKeyDigest, + ) + require.NoError(t, err) + + _, duplicateErr := pg.ExecContext(t.Context(), insertManifest, + "tenant-a", repeatedHex("c"), repeatedHex("d"), + time.Date(2026, 8, 13, 12, 0, 1, 0, time.UTC), []byte("{}"), sourceKeyDigest, + ) + assert.Error(t, duplicateErr, + "one authenticated source capture id must identify only one manifest") + + _, err = pg.ExecContext(t.Context(), ` + INSERT INTO raw_manifest_entries ( + tenant_id, manifest_id, entry_index, path, path_sha256, entry_type, + size_bytes + ) VALUES ($1, $2, 0, 'session.jsonl', $3, 'file', 7)`, + "tenant-a", firstManifest, rawIngestKeyDigest("session.jsonl")) + require.NoError(t, err) + _, err = pg.ExecContext(t.Context(), ` + INSERT INTO raw_manifest_objects ( + tenant_id, manifest_id, entry_index, object_index, sha256, size_bytes + ) VALUES ($1, $2, 0, 0, $3, 7)`, + "tenant-a", firstManifest, repeatedHex("e")) + assert.Error(t, err, + "an accepted manifest reference cannot point at an unverified raw object") +} + +func TestRawIngestSchemaMakesAcceptedManifestGraphAppendOnly(t *testing.T) { + pg, store := newRawIngestTestStore(t) + identity := rawIngestIdentity(t, "tenant-a") + object := rawIngestObject(t, "a", 7) + require.NoError(t, store.RecordVerifiedObject(t.Context(), identity, object)) + manifest := rawIngestManifest( + t, identity, "capture-a", "", rawIngestCapturedAt(), object, + ) + _, err := store.CommitManifest(t.Context(), manifest, "parser-data-17") + require.NoError(t, err) + + mutations := []string{ + `UPDATE raw_manifests SET canonical_json = '{}'`, + `DELETE FROM raw_manifests`, + `UPDATE raw_manifest_entries SET path = 'changed.jsonl'`, + `DELETE FROM raw_manifest_entries`, + `UPDATE raw_manifest_objects SET object_index = 1`, + `DELETE FROM raw_manifest_objects`, + } + for _, mutation := range mutations { + _, err := pg.ExecContext(t.Context(), mutation) + var pgErr *pgconn.PgError + require.ErrorAs(t, err, &pgErr, mutation) + assert.Equal(t, "55000", pgErr.Code, mutation) + } + assert.Equal(t, + rawIngestCounts{Manifests: 1, Entries: 1, Objects: 1, Heads: 1, Jobs: 1}, + readRawIngestCounts(t, pg), + ) +} + +func TestSyncEnsureSchemaCreatesRawCustodyOnLegacyFastPath(t *testing.T) { + pgURL := testPGURL(t) + cleanSchemaTestPG(t, pgURL) + t.Cleanup(func() { cleanSchemaTestPG(t, pgURL) }) + pg, err := Open(pgURL, schemaTestSchema, true) + require.NoError(t, err) + defer pg.Close() + require.NoError(t, EnsureSchema(t.Context(), pg, schemaTestSchema)) + + _, err = pg.ExecContext(t.Context(), ` + DROP TABLE raw_ingest_jobs, raw_source_heads, raw_manifest_objects, + raw_manifest_entries, raw_manifests, raw_objects`) + require.NoError(t, err) + require.True(t, pushSchemaCurrent(t.Context(), pg), + "fixture must exercise the schema-current sync fast path") + + syncer := &Sync{pg: pg, schema: schemaTestSchema} + require.NoError(t, syncer.EnsureSchema(t.Context())) + + var exists bool + require.NoError(t, pg.QueryRowContext(t.Context(), ` + SELECT EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_schema = $1 AND table_name = 'raw_objects' + )`, schemaTestSchema).Scan(&exists)) + assert.True(t, exists, + "schema-current sync must still install newly introduced custody tables") +} + +// TestSyncEnsureSchemaFastPathToleratesRestrictedRole pins the restricted-role +// push path: a privileged role provisioned the schema, including the raw +// custody tables, but the push role cannot CREATE. PostgreSQL still checks +// CREATE for CREATE TABLE IF NOT EXISTS, so the fast path must skip raw +// custody DDL on SQLSTATE 42501 rather than fail every push. +func TestSyncEnsureSchemaFastPathToleratesRestrictedRole(t *testing.T) { + pgURL := testPGURL(t) + const schema = "agentsview_raw_privilege_test" + const role = "agentsview_raw_restricted" + const rolePassword = "agentsview_raw_restricted_pw" + + admin, err := Open(pgURL, schema, true) + require.NoError(t, err, "Open admin") + t.Cleanup(func() { _ = admin.Close() }) + _, err = admin.Exec(`DROP SCHEMA IF EXISTS ` + schema + ` CASCADE`) + require.NoError(t, err, "drop schema") + require.NoError(t, EnsureSchema(t.Context(), admin, schema)) + require.True(t, pushSchemaCurrent(t.Context(), admin), + "fixture must exercise the schema-current sync fast path") + + _, _ = admin.Exec(`DROP OWNED BY ` + role) + _, _ = admin.Exec(`DROP ROLE IF EXISTS ` + role) + _, err = admin.Exec(`CREATE ROLE ` + role + ` LOGIN PASSWORD '` + rolePassword + `'`) + require.NoError(t, err, "create restricted role") + t.Cleanup(func() { + _, _ = admin.Exec(`DROP SCHEMA IF EXISTS ` + schema + ` CASCADE`) + _, _ = admin.Exec(`DROP OWNED BY ` + role) + _, _ = admin.Exec(`DROP ROLE IF EXISTS ` + role) + }) + for _, grant := range []string{ + `GRANT USAGE ON SCHEMA ` + schema + ` TO ` + role, + `GRANT SELECT, INSERT, UPDATE, DELETE ON ALL TABLES IN SCHEMA ` + schema + ` TO ` + role, + } { + _, err = admin.Exec(grant) + require.NoError(t, err, grant) + } + + restrictedURL, err := url.Parse(pgURL) + require.NoError(t, err) + restrictedURL.User = url.UserPassword(role, rolePassword) + restricted, err := Open(restrictedURL.String(), schema, true) + require.NoError(t, err, "Open restricted") + t.Cleanup(func() { _ = restricted.Close() }) + + syncer := &Sync{pg: restricted, schema: schema} + require.NoError(t, syncer.EnsureSchema(t.Context()), + "restricted push role must not fail on raw custody DDL") + assert.True(t, syncer.schemaDone) +} + +func repeatedHex(value string) string { + return strings.Repeat(value, 64) +} diff --git a/internal/postgres/raw_ingest_schema_test.go b/internal/postgres/raw_ingest_schema_test.go new file mode 100644 index 0000000000..0f646ead17 --- /dev/null +++ b/internal/postgres/raw_ingest_schema_test.go @@ -0,0 +1,25 @@ +package postgres + +import ( + "errors" + "fmt" + "testing" + + "github.com/jackc/pgx/v5/pgconn" + "github.com/stretchr/testify/assert" +) + +func TestRawIngestAppendOnlyUnsupportedClassifiesFeatureErrors(t *testing.T) { + t.Parallel() + + unsupported := &pgconn.PgError{Code: "0A000"} + assert.True(t, rawIngestAppendOnlyUnsupported(unsupported)) + assert.True(t, rawIngestAppendOnlyUnsupported( + fmt.Errorf("installing trigger: %w", unsupported), + )) + assert.True(t, rawIngestAppendOnlyUnsupported(errors.New( + "ERROR: unimplemented PL/pgSQL (SQLSTATE 0A000)", + ))) + assert.False(t, rawIngestAppendOnlyUnsupported(&pgconn.PgError{Code: "42501"})) + assert.False(t, rawIngestAppendOnlyUnsupported(errors.New("connection closed"))) +} diff --git a/internal/postgres/raw_ingest_store.go b/internal/postgres/raw_ingest_store.go new file mode 100644 index 0000000000..4ab0cbde73 --- /dev/null +++ b/internal/postgres/raw_ingest_store.go @@ -0,0 +1,540 @@ +package postgres + +import ( + "context" + "crypto/rand" + "crypto/sha256" + "database/sql" + "encoding/hex" + "errors" + "fmt" + "math" + "strings" + "unicode" + "unicode/utf8" + + "go.kenn.io/agentsview/internal/rawsync" +) + +const rawIngestBatchRows = 256 + +// RawIngestStore implements raw custody metadata over PostgreSQL. +type RawIngestStore struct { + db *sql.DB + newReceipt func() (string, error) +} + +// NewRawIngestStore constructs a PostgreSQL raw custody metadata store. +func NewRawIngestStore(db *sql.DB) (*RawIngestStore, error) { + if db == nil { + return nil, fmt.Errorf("%w: PostgreSQL connection is required", rawsync.ErrInvalid) + } + return &RawIngestStore{db: db, newReceipt: generateRawIngestReceipt}, nil +} + +// RecordVerifiedObject records an object only after physical verification. +func (s *RawIngestStore) RecordVerifiedObject( + ctx context.Context, + identity rawsync.AuthIdentity, + object rawsync.ObjectRef, +) error { + return s.RecordVerifiedObjects(ctx, identity, []rawsync.ObjectRef{object}) +} + +// RecordVerifiedObjects records physically verified objects in bounded batches. +func (s *RawIngestStore) RecordVerifiedObjects( + ctx context.Context, + identity rawsync.AuthIdentity, + objects []rawsync.ObjectRef, +) error { + if err := validateRawIngestIdentity(identity); err != nil { + return err + } + unique, err := uniqueRawIngestObjects(objects) + if err != nil { + return err + } + for start := 0; start < len(unique); start += rawIngestBatchRows { + end := min(start+rawIngestBatchRows, len(unique)) + var query strings.Builder + query.WriteString(`INSERT INTO raw_objects (tenant_id, sha256, size_bytes) VALUES `) + args := make([]any, 0, 3*(end-start)) + for i, object := range unique[start:end] { + if i > 0 { + query.WriteByte(',') + } + argument := i*3 + 1 + fmt.Fprintf(&query, "($%d,$%d,$%d)", argument, argument+1, argument+2) + args = append(args, identity.TenantID, object.SHA256, object.Length) + } + query.WriteString(` ON CONFLICT (tenant_id, sha256) DO UPDATE + SET verified_at = now() + WHERE raw_objects.size_bytes = EXCLUDED.size_bytes`) + result, err := s.db.ExecContext(ctx, query.String(), args...) + if err != nil { + return fmt.Errorf("recording verified raw objects: %w", err) + } + accepted, err := result.RowsAffected() + if err != nil { + return fmt.Errorf("checking verified raw object registration: %w", err) + } + if accepted != int64(end-start) { + return fmt.Errorf( + "recording verified raw objects: %w: digest length differs", + rawsync.ErrConflict, + ) + } + } + return nil +} + +// MissingObjects returns absent verified-object metadata in request order. +func (s *RawIngestStore) MissingObjects( + ctx context.Context, + identity rawsync.AuthIdentity, + objects []rawsync.ObjectRef, +) ([]rawsync.ObjectRef, error) { + if err := validateRawIngestIdentity(identity); err != nil { + return nil, err + } + unique, err := uniqueRawIngestObjects(objects) + if err != nil { + return nil, err + } + present, err := loadPresentRawObjects(ctx, s.db, identity.TenantID, unique) + if err != nil { + return nil, fmt.Errorf("querying verified raw objects: %w", err) + } + missing := make([]rawsync.ObjectRef, 0) + for _, object := range unique { + if !present[object] { + missing = append(missing, object) + } + } + return missing, nil +} + +// CommitManifest atomically records a manifest, advances its head, and queues parsing. +func (s *RawIngestStore) CommitManifest( + ctx context.Context, + manifest rawsync.CanonicalManifest, + processingVersion string, +) (rawsync.CommitResult, error) { + if err := validateRawIngestProcessingVersion(processingVersion); err != nil { + return rawsync.CommitResult{}, err + } + if err := validateRawIngestCanonicalManifest(manifest); err != nil { + return rawsync.CommitResult{}, err + } + tx, err := s.db.BeginTx(ctx, nil) + if err != nil { + return rawsync.CommitResult{}, fmt.Errorf("beginning raw manifest commit: %w", err) + } + committed := false + defer func() { + if !committed { + _ = tx.Rollback() + } + }() + + if result, found, err := lookupRawIngestCapture(ctx, tx, manifest); err != nil { + return rawsync.CommitResult{}, err + } else if found { + return result, nil + } + if err := ensureRawIngestHead(ctx, tx, manifest); err != nil { + return rawsync.CommitResult{}, err + } + head, err := lockRawIngestHead(ctx, tx, manifest) + if err != nil { + return rawsync.CommitResult{}, err + } + // A concurrent first commit may have become visible while this transaction + // waited for the source-head lock. Recheck capture idempotency under the lock. + if result, found, err := lookupRawIngestCapture(ctx, tx, manifest); err != nil { + return rawsync.CommitResult{}, err + } else if found { + return result, nil + } + if manifest.Manifest.ExpectedParentReceipt != head.Receipt { + return rawsync.CommitResult{}, &rawsync.HeadConflictError{ + CurrentManifestID: head.ManifestID, + CurrentReceipt: head.Receipt, + CurrentGeneration: head.Generation, + } + } + present, err := loadPresentRawObjects( + ctx, tx, manifest.Identity.TenantID, manifest.Objects, + ) + if err != nil { + return rawsync.CommitResult{}, fmt.Errorf("verifying raw manifest objects: %w", err) + } + for _, object := range manifest.Objects { + if !present[object] { + return rawsync.CommitResult{}, fmt.Errorf( + "%w: %s", rawsync.ErrMissingObject, object.SHA256, + ) + } + } + if head.Generation == math.MaxInt64 { + return rawsync.CommitResult{}, fmt.Errorf("raw source generation exhausted") + } + generation := head.Generation + 1 + receipt, err := s.newReceipt() + if err != nil { + return rawsync.CommitResult{}, fmt.Errorf("generating raw manifest receipt: %w", err) + } + if _, err := tx.ExecContext(ctx, ` + INSERT INTO raw_manifests ( + tenant_id, manifest_id, device_id, provider, configured_root_id, + source_key, source_key_sha256, capture_id, parent_receipt, receipt, + generation, kind, captured_at, canonical_json + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14)`, + manifest.Identity.TenantID, + manifest.ManifestID, + manifest.Identity.DeviceID, + string(manifest.Manifest.Provider), + manifest.Manifest.ConfiguredRootID, + manifest.Manifest.SourceKey, + rawIngestKeyDigest(manifest.Manifest.SourceKey), + manifest.Manifest.CaptureID, + manifest.Manifest.ExpectedParentReceipt, + receipt, + generation, + string(manifest.Manifest.Kind), + manifest.Manifest.CapturedAt, + manifest.CanonicalJSON, + ); err != nil { + return rawsync.CommitResult{}, fmt.Errorf("inserting raw manifest: %w", err) + } + if err := insertRawManifestEntries(ctx, tx, manifest); err != nil { + return rawsync.CommitResult{}, err + } + if err := insertRawManifestObjects(ctx, tx, manifest); err != nil { + return rawsync.CommitResult{}, err + } + if _, err := tx.ExecContext(ctx, ` + INSERT INTO raw_ingest_jobs ( + tenant_id, manifest_id, stage, processing_version, state + ) VALUES ($1, $2, 'parse', $3, 'ready')`, + manifest.Identity.TenantID, manifest.ManifestID, processingVersion, + ); err != nil { + return rawsync.CommitResult{}, fmt.Errorf("enqueuing raw parse job: %w", err) + } + updated, err := tx.ExecContext(ctx, ` + UPDATE raw_source_heads + SET manifest_id = $6, receipt = $7, generation = $8, updated_at = now() + WHERE tenant_id = $1 AND device_id = $2 AND provider = $3 + AND configured_root_id = $4 AND source_key_sha256 = $5 AND generation = $9`, + manifest.Identity.TenantID, + manifest.Identity.DeviceID, + string(manifest.Manifest.Provider), + manifest.Manifest.ConfiguredRootID, + rawIngestKeyDigest(manifest.Manifest.SourceKey), + manifest.ManifestID, + receipt, + generation, + head.Generation, + ) + if err != nil { + return rawsync.CommitResult{}, fmt.Errorf("advancing raw source head: %w", err) + } + affected, err := updated.RowsAffected() + if err != nil { + return rawsync.CommitResult{}, fmt.Errorf("checking raw source head advance: %w", err) + } + if affected != 1 { + return rawsync.CommitResult{}, fmt.Errorf("advancing raw source head affected %d rows", affected) + } + if err := tx.Commit(); err != nil { + return rawsync.CommitResult{}, fmt.Errorf("committing raw manifest: %w", err) + } + committed = true + return rawsync.CommitResult{ + ManifestID: manifest.ManifestID, + Receipt: receipt, + Generation: generation, + Created: true, + }, nil +} + +type rawIngestHead struct { + ManifestID string + Receipt string + Generation int64 +} + +func lookupRawIngestCapture( + ctx context.Context, + tx *sql.Tx, + manifest rawsync.CanonicalManifest, +) (rawsync.CommitResult, bool, error) { + var stored rawsync.CommitResult + err := tx.QueryRowContext(ctx, ` + SELECT manifest_id, receipt, generation + FROM raw_manifests + WHERE tenant_id = $1 AND device_id = $2 AND provider = $3 + AND configured_root_id = $4 AND source_key_sha256 = $5 AND capture_id = $6`, + manifest.Identity.TenantID, + manifest.Identity.DeviceID, + string(manifest.Manifest.Provider), + manifest.Manifest.ConfiguredRootID, + rawIngestKeyDigest(manifest.Manifest.SourceKey), + manifest.Manifest.CaptureID, + ).Scan(&stored.ManifestID, &stored.Receipt, &stored.Generation) + if errors.Is(err, sql.ErrNoRows) { + return rawsync.CommitResult{}, false, nil + } + if err != nil { + return rawsync.CommitResult{}, false, fmt.Errorf("checking raw capture idempotency: %w", err) + } + if stored.ManifestID != manifest.ManifestID { + return rawsync.CommitResult{}, false, fmt.Errorf( + "raw capture identifier reused: %w", rawsync.ErrConflict, + ) + } + stored.Created = false + return stored, true, nil +} + +func ensureRawIngestHead( + ctx context.Context, + tx *sql.Tx, + manifest rawsync.CanonicalManifest, +) error { + _, err := tx.ExecContext(ctx, ` + INSERT INTO raw_source_heads ( + tenant_id, device_id, provider, configured_root_id, source_key, + source_key_sha256 + ) VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT ( + tenant_id, device_id, provider, configured_root_id, source_key_sha256 + ) DO NOTHING`, + manifest.Identity.TenantID, + manifest.Identity.DeviceID, + string(manifest.Manifest.Provider), + manifest.Manifest.ConfiguredRootID, + manifest.Manifest.SourceKey, + rawIngestKeyDigest(manifest.Manifest.SourceKey), + ) + if err != nil { + return fmt.Errorf("ensuring raw source head: %w", err) + } + return nil +} + +func lockRawIngestHead( + ctx context.Context, + tx *sql.Tx, + manifest rawsync.CanonicalManifest, +) (rawIngestHead, error) { + var head rawIngestHead + err := tx.QueryRowContext(ctx, ` + SELECT COALESCE(manifest_id, ''), COALESCE(receipt, ''), generation + FROM raw_source_heads + WHERE tenant_id = $1 AND device_id = $2 AND provider = $3 + AND configured_root_id = $4 AND source_key_sha256 = $5 + FOR UPDATE`, + manifest.Identity.TenantID, + manifest.Identity.DeviceID, + string(manifest.Manifest.Provider), + manifest.Manifest.ConfiguredRootID, + rawIngestKeyDigest(manifest.Manifest.SourceKey), + ).Scan(&head.ManifestID, &head.Receipt, &head.Generation) + if err != nil { + return rawIngestHead{}, fmt.Errorf("locking raw source head: %w", err) + } + return head, nil +} + +type rawObjectQueryer interface { + QueryContext(context.Context, string, ...any) (*sql.Rows, error) +} + +func loadPresentRawObjects( + ctx context.Context, + queryer rawObjectQueryer, + tenantID string, + objects []rawsync.ObjectRef, +) (map[rawsync.ObjectRef]bool, error) { + present := make(map[rawsync.ObjectRef]bool, len(objects)) + for start := 0; start < len(objects); start += rawIngestBatchRows { + end := min(start+rawIngestBatchRows, len(objects)) + var query strings.Builder + query.WriteString(`SELECT sha256, size_bytes FROM raw_objects WHERE tenant_id = $1 AND (sha256, size_bytes) IN (`) + args := make([]any, 1, 1+2*(end-start)) + args[0] = tenantID + for i, object := range objects[start:end] { + if i > 0 { + query.WriteByte(',') + } + argument := 2 + i*2 + fmt.Fprintf(&query, "($%d,$%d)", argument, argument+1) + args = append(args, object.SHA256, object.Length) + } + query.WriteByte(')') + rows, err := queryer.QueryContext(ctx, query.String(), args...) + if err != nil { + return nil, err + } + for rows.Next() { + var object rawsync.ObjectRef + if err := rows.Scan(&object.SHA256, &object.Length); err != nil { + _ = rows.Close() + return nil, err + } + present[object] = true + } + if err := rows.Err(); err != nil { + _ = rows.Close() + return nil, err + } + if err := rows.Close(); err != nil { + return nil, err + } + } + return present, nil +} + +func insertRawManifestEntries( + ctx context.Context, + tx *sql.Tx, + manifest rawsync.CanonicalManifest, +) error { + entries := manifest.Manifest.Entries + for start := 0; start < len(entries); start += rawIngestBatchRows { + end := min(start+rawIngestBatchRows, len(entries)) + var query strings.Builder + query.WriteString(`INSERT INTO raw_manifest_entries (tenant_id, manifest_id, entry_index, path, path_sha256, entry_type, size_bytes) VALUES `) + args := make([]any, 0, 7*(end-start)) + for i, entry := range entries[start:end] { + if i > 0 { + query.WriteByte(',') + } + argument := i*7 + 1 + fmt.Fprintf(&query, "($%d,$%d,$%d,$%d,$%d,$%d,$%d)", + argument, argument+1, argument+2, argument+3, argument+4, argument+5, + argument+6, + ) + args = append(args, + manifest.Identity.TenantID, manifest.ManifestID, start+i, + entry.Path, rawIngestKeyDigest(entry.Path), entry.Type, entry.Length, + ) + } + if _, err := tx.ExecContext(ctx, query.String(), args...); err != nil { + return fmt.Errorf("inserting raw manifest entries: %w", err) + } + } + return nil +} + +type rawManifestObjectRow struct { + EntryIndex int + ObjectIndex int + Object rawsync.ObjectRef +} + +func insertRawManifestObjects( + ctx context.Context, + tx *sql.Tx, + manifest rawsync.CanonicalManifest, +) error { + rows := make([]rawManifestObjectRow, 0, len(manifest.Objects)) + for entryIndex, entry := range manifest.Manifest.Entries { + for objectIndex, object := range entry.Objects { + rows = append(rows, rawManifestObjectRow{ + EntryIndex: entryIndex, ObjectIndex: objectIndex, Object: object, + }) + } + } + for start := 0; start < len(rows); start += rawIngestBatchRows { + end := min(start+rawIngestBatchRows, len(rows)) + var query strings.Builder + query.WriteString(`INSERT INTO raw_manifest_objects (tenant_id, manifest_id, entry_index, object_index, sha256, size_bytes) VALUES `) + args := make([]any, 0, 6*(end-start)) + for i, row := range rows[start:end] { + if i > 0 { + query.WriteByte(',') + } + argument := i*6 + 1 + fmt.Fprintf(&query, "($%d,$%d,$%d,$%d,$%d,$%d)", + argument, argument+1, argument+2, argument+3, argument+4, argument+5, + ) + args = append(args, + manifest.Identity.TenantID, manifest.ManifestID, + row.EntryIndex, row.ObjectIndex, row.Object.SHA256, row.Object.Length, + ) + } + if _, err := tx.ExecContext(ctx, query.String(), args...); err != nil { + return fmt.Errorf("inserting raw manifest object references: %w", err) + } + } + return nil +} + +func validateRawIngestCanonicalManifest(manifest rawsync.CanonicalManifest) error { + return rawsync.ValidateCanonicalManifest(manifest) +} + +func validateRawIngestIdentity(identity rawsync.AuthIdentity) error { + validated, err := rawsync.NewAuthIdentity(identity.TenantID, identity.DeviceID) + if err != nil || validated != identity { + return fmt.Errorf("%w: authenticated identity is not canonical", rawsync.ErrInvalid) + } + return nil +} + +func validateRawIngestObject(object rawsync.ObjectRef) error { + validated, err := rawsync.NewObjectRef(object.SHA256, object.Length) + if err != nil || validated != object { + return fmt.Errorf("%w: raw object reference is not canonical", rawsync.ErrInvalid) + } + return nil +} + +func uniqueRawIngestObjects(objects []rawsync.ObjectRef) ([]rawsync.ObjectRef, error) { + seen := make(map[string]rawsync.ObjectRef, len(objects)) + unique := make([]rawsync.ObjectRef, 0, len(objects)) + for _, object := range objects { + if err := validateRawIngestObject(object); err != nil { + return nil, err + } + if previous, ok := seen[object.SHA256]; ok { + if previous.Length != object.Length { + return nil, fmt.Errorf("%w: digest has conflicting lengths", rawsync.ErrConflict) + } + continue + } + seen[object.SHA256] = object + unique = append(unique, object) + } + return unique, nil +} + +func validateRawIngestProcessingVersion(value string) error { + if value == "" || len(value) > 128 || !utf8.ValidString(value) || + strings.TrimSpace(value) != value { + return fmt.Errorf("%w: processing version is not canonical", rawsync.ErrInvalid) + } + for _, r := range value { + if unicode.IsControl(r) { + return fmt.Errorf("%w: processing version contains a control character", rawsync.ErrInvalid) + } + } + return nil +} + +// rawIngestKeyDigest returns the fixed-size key used in composite indexes for +// values whose full text may exceed the PostgreSQL B-tree entry limit. +func rawIngestKeyDigest(value string) string { + sum := sha256.Sum256([]byte(value)) + return hex.EncodeToString(sum[:]) +} + +func generateRawIngestReceipt() (string, error) { + var value [32]byte + if _, err := rand.Read(value[:]); err != nil { + return "", err + } + return hex.EncodeToString(value[:]), nil +} diff --git a/internal/postgres/raw_ingest_store_pgtest_test.go b/internal/postgres/raw_ingest_store_pgtest_test.go new file mode 100644 index 0000000000..f63c79af91 --- /dev/null +++ b/internal/postgres/raw_ingest_store_pgtest_test.go @@ -0,0 +1,482 @@ +//go:build pgtest + +package postgres + +import ( + "crypto/sha256" + "database/sql" + "encoding/hex" + "errors" + "fmt" + "math/rand/v2" + "regexp" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.kenn.io/agentsview/internal/parser" + "go.kenn.io/agentsview/internal/rawsync" +) + +func TestRawIngestStoreObjectRegistry(t *testing.T) { + pg, store := newRawIngestTestStore(t) + _, err := NewRawIngestStore(nil) + assert.ErrorIs(t, err, rawsync.ErrInvalid) + identity := rawIngestIdentity(t, "tenant-a") + otherTenant := rawIngestIdentity(t, "tenant-b") + first := rawIngestObject(t, "a", 7) + second := rawIngestObject(t, "b", 11) + + missing, err := store.MissingObjects(t.Context(), identity, []rawsync.ObjectRef{ + second, first, second, + }) + require.NoError(t, err) + assert.Equal(t, []rawsync.ObjectRef{second, first}, missing) + + require.NoError(t, store.RecordVerifiedObject(t.Context(), identity, first)) + require.NoError(t, store.RecordVerifiedObject(t.Context(), identity, first)) + missing, err = store.MissingObjects(t.Context(), identity, []rawsync.ObjectRef{second, first}) + require.NoError(t, err) + assert.Equal(t, []rawsync.ObjectRef{second}, missing) + missing, err = store.MissingObjects(t.Context(), otherTenant, []rawsync.ObjectRef{first}) + require.NoError(t, err) + assert.Equal(t, []rawsync.ObjectRef{first}, missing, + "verified-object metadata must never deduplicate across tenants") + + conflictingLength := rawsync.ObjectRef{SHA256: first.SHA256, Length: first.Length + 1} + err = store.RecordVerifiedObject(t.Context(), identity, conflictingLength) + assert.ErrorIs(t, err, rawsync.ErrConflict) + assert.Equal(t, 1, rawIngestTableCount(t, pg, "raw_objects")) +} + +func TestRawIngestStoreBatchesVerifiedObjectRegistration(t *testing.T) { + pg, store := newRawIngestTestStore(t) + identity := rawIngestIdentity(t, "tenant-a") + objects := make([]rawsync.ObjectRef, 0, rawIngestBatchRows+1) + for i := range rawIngestBatchRows + 1 { + sum := sha256.Sum256([]byte(fmt.Sprintf("verified-object-%03d", i))) + object, err := rawsync.NewObjectRef(hex.EncodeToString(sum[:]), int64(i+1)) + require.NoError(t, err) + objects = append(objects, object) + } + + require.NoError(t, store.RecordVerifiedObjects(t.Context(), identity, objects)) + assert.Equal(t, len(objects), rawIngestTableCount(t, pg, "raw_objects")) + require.NoError(t, store.RecordVerifiedObjects(t.Context(), identity, objects)) + assert.Equal(t, len(objects), rawIngestTableCount(t, pg, "raw_objects")) + + objects[len(objects)-1].Length++ + err := store.RecordVerifiedObjects(t.Context(), identity, objects) + assert.ErrorIs(t, err, rawsync.ErrConflict) +} + +func TestRawIngestStoreCommitIsAtomicFencedAndIdempotent(t *testing.T) { + pg, store := newRawIngestTestStore(t) + identity := rawIngestIdentity(t, "tenant-a") + object := rawIngestObject(t, "a", 7) + require.NoError(t, store.RecordVerifiedObject(t.Context(), identity, object)) + firstManifest := rawIngestManifest( + t, identity, "capture-a", "", rawIngestCapturedAt(), object, + ) + + first, err := store.CommitManifest(t.Context(), firstManifest, "parser-data-17") + require.NoError(t, err) + assert.True(t, first.Created) + assert.Equal(t, firstManifest.ManifestID, first.ManifestID) + assert.Equal(t, int64(1), first.Generation) + assert.Regexp(t, regexp.MustCompile(`^[0-9a-f]{64}$`), first.Receipt) + assert.Equal(t, rawIngestCounts{Manifests: 1, Entries: 1, Objects: 1, Heads: 1, Jobs: 1}, + readRawIngestCounts(t, pg)) + + retried, err := store.CommitManifest(t.Context(), firstManifest, "parser-data-17") + require.NoError(t, err) + assert.False(t, retried.Created) + assert.Equal(t, first.ManifestID, retried.ManifestID) + assert.Equal(t, first.Receipt, retried.Receipt) + assert.Equal(t, first.Generation, retried.Generation) + assert.Equal(t, rawIngestCounts{Manifests: 1, Entries: 1, Objects: 1, Heads: 1, Jobs: 1}, + readRawIngestCounts(t, pg)) + + reusedCapture := rawIngestManifest( + t, identity, "capture-a", "", rawIngestCapturedAt().Add(time.Second), object, + ) + _, err = store.CommitManifest(t.Context(), reusedCapture, "parser-data-17") + assert.ErrorIs(t, err, rawsync.ErrConflict) + assert.Equal(t, rawIngestCounts{Manifests: 1, Entries: 1, Objects: 1, Heads: 1, Jobs: 1}, + readRawIngestCounts(t, pg)) + + secondManifest := rawIngestManifest( + t, identity, "capture-b", first.Receipt, rawIngestCapturedAt().Add(time.Minute), object, + ) + second, err := store.CommitManifest(t.Context(), secondManifest, "parser-data-17") + require.NoError(t, err) + assert.True(t, second.Created) + assert.Equal(t, int64(2), second.Generation) + assert.Equal(t, rawIngestCounts{Manifests: 2, Entries: 2, Objects: 2, Heads: 1, Jobs: 2}, + readRawIngestCounts(t, pg)) + + staleManifest := rawIngestManifest( + t, identity, "capture-c", first.Receipt, rawIngestCapturedAt().Add(2*time.Minute), object, + ) + _, err = store.CommitManifest(t.Context(), staleManifest, "parser-data-17") + var headConflict *rawsync.HeadConflictError + require.ErrorAs(t, err, &headConflict) + assert.ErrorIs(t, err, rawsync.ErrConflict) + assert.Equal(t, second.ManifestID, headConflict.CurrentManifestID) + assert.Equal(t, second.Receipt, headConflict.CurrentReceipt) + assert.Equal(t, int64(2), headConflict.CurrentGeneration) + assert.Equal(t, rawIngestCounts{Manifests: 2, Entries: 2, Objects: 2, Heads: 1, Jobs: 2}, + readRawIngestCounts(t, pg)) + + var headManifest, headReceipt string + var generation int64 + require.NoError(t, pg.QueryRowContext(t.Context(), ` + SELECT manifest_id, receipt, generation FROM raw_source_heads`, + ).Scan(&headManifest, &headReceipt, &generation)) + assert.Equal(t, second.ManifestID, headManifest) + assert.Equal(t, second.Receipt, headReceipt) + assert.Equal(t, int64(2), generation) +} + +func TestRawIngestStoreMissingObjectChangesNoAcceptanceState(t *testing.T) { + pg, store := newRawIngestTestStore(t) + identity := rawIngestIdentity(t, "tenant-a") + manifest := rawIngestManifest( + t, identity, "capture-a", "", rawIngestCapturedAt(), rawIngestObject(t, "a", 7), + ) + + _, err := store.CommitManifest(t.Context(), manifest, " ") + assert.ErrorIs(t, err, rawsync.ErrInvalid) + assert.Equal(t, rawIngestCounts{}, readRawIngestCounts(t, pg)) + + _, err = store.CommitManifest(t.Context(), manifest, "parser-data-17") + assert.ErrorIs(t, err, rawsync.ErrMissingObject) + assert.Equal(t, rawIngestCounts{}, readRawIngestCounts(t, pg)) +} + +func TestRawIngestStoreConcurrentIdenticalRetryConverges(t *testing.T) { + pg, store := newRawIngestTestStore(t) + identity := rawIngestIdentity(t, "tenant-a") + object := rawIngestObject(t, "a", 7) + require.NoError(t, store.RecordVerifiedObject(t.Context(), identity, object)) + manifest := rawIngestManifest( + t, identity, "capture-a", "", rawIngestCapturedAt(), object, + ) + + start := make(chan struct{}) + outcomes := make(chan rawIngestOutcome, 2) + var workers sync.WaitGroup + for range 2 { + workers.Add(1) + go func() { + defer workers.Done() + <-start + result, err := store.CommitManifest(t.Context(), manifest, "parser-data-17") + outcomes <- rawIngestOutcome{result: result, err: err} + }() + } + close(start) + workers.Wait() + close(outcomes) + + results := make([]rawsync.CommitResult, 0, 2) + for got := range outcomes { + require.NoError(t, got.err) + results = append(results, got.result) + } + require.Len(t, results, 2) + assert.Equal(t, results[0].Receipt, results[1].Receipt) + assert.Equal(t, int64(1), results[0].Generation) + assert.Equal(t, int64(1), results[1].Generation) + assert.NotEqual(t, results[0].Created, results[1].Created) + assert.Equal(t, rawIngestCounts{Manifests: 1, Entries: 1, Objects: 1, Heads: 1, Jobs: 1}, + readRawIngestCounts(t, pg)) +} + +func TestRawIngestStoreBatchesManifestReferences(t *testing.T) { + pg, store := newRawIngestTestStore(t) + identity := rawIngestIdentity(t, "tenant-a") + const objectCount = rawIngestBatchRows + 1 + objects := make([]rawsync.ObjectRef, 0, objectCount) + for i := range objectCount { + sum := sha256.Sum256([]byte(fmt.Sprintf("object-%03d", i))) + object, err := rawsync.NewObjectRef(hex.EncodeToString(sum[:]), 1) + require.NoError(t, err) + require.NoError(t, store.RecordVerifiedObject(t.Context(), identity, object)) + objects = append(objects, object) + } + manifest, err := rawsync.ValidateAndCanonicalize(identity, rawsync.Manifest{ + SchemaVersion: rawsync.ManifestSchemaVersion, + Provider: parser.AgentCodex, + ConfiguredRootID: "root-a", + SourceKey: "sessions/batched.jsonl", + CaptureID: "capture-a", + CapturedAt: rawIngestCapturedAt(), + Kind: rawsync.ManifestSnapshot, + Entries: []rawsync.Entry{{ + Path: "session.jsonl", Type: "file", Length: int64(objectCount), Objects: objects, + }}, + }, rawsync.DefaultManifestLimits()) + require.NoError(t, err) + + result, err := store.CommitManifest(t.Context(), manifest, "parser-data-17") + require.NoError(t, err) + assert.True(t, result.Created) + assert.Equal(t, objectCount, rawIngestTableCount(t, pg, "raw_manifest_objects")) + var first, last string + require.NoError(t, pg.QueryRowContext(t.Context(), ` + SELECT + (SELECT sha256 FROM raw_manifest_objects WHERE object_index = 0), + (SELECT sha256 FROM raw_manifest_objects WHERE object_index = $1)`, + objectCount-1, + ).Scan(&first, &last)) + assert.Equal(t, objects[0].SHA256, first) + assert.Equal(t, objects[objectCount-1].SHA256, last) +} + +func TestRawIngestStoreJobFailureRollsBackManifestAndHead(t *testing.T) { + pg, store := newRawIngestTestStore(t) + identity := rawIngestIdentity(t, "tenant-a") + object := rawIngestObject(t, "a", 7) + require.NoError(t, store.RecordVerifiedObject(t.Context(), identity, object)) + _, err := pg.ExecContext(t.Context(), ` + CREATE FUNCTION reject_raw_ingest_job() RETURNS trigger + LANGUAGE plpgsql AS $$ + BEGIN + RAISE EXCEPTION 'injected raw ingest job failure'; + END; + $$; + CREATE TRIGGER reject_raw_ingest_job + BEFORE INSERT ON raw_ingest_jobs + FOR EACH ROW EXECUTE FUNCTION reject_raw_ingest_job()`) + require.NoError(t, err) + manifest := rawIngestManifest( + t, identity, "capture-a", "", rawIngestCapturedAt(), object, + ) + + _, err = store.CommitManifest(t.Context(), manifest, "parser-data-17") + require.Error(t, err) + assert.False(t, errors.Is(err, rawsync.ErrConflict)) + assert.Equal(t, rawIngestCounts{}, readRawIngestCounts(t, pg)) +} + +func TestRawIngestStoreConcurrentHeadAdvance(t *testing.T) { + pg, store := newRawIngestTestStore(t) + identity := rawIngestIdentity(t, "tenant-a") + object := rawIngestObject(t, "a", 7) + require.NoError(t, store.RecordVerifiedObject(t.Context(), identity, object)) + initialManifest := rawIngestManifest( + t, identity, "capture-a", "", rawIngestCapturedAt(), object, + ) + initial, err := store.CommitManifest(t.Context(), initialManifest, "parser-data-17") + require.NoError(t, err) + + candidates := []rawsync.CanonicalManifest{ + rawIngestManifest(t, identity, "capture-b", initial.Receipt, rawIngestCapturedAt().Add(time.Minute), object), + rawIngestManifest(t, identity, "capture-c", initial.Receipt, rawIngestCapturedAt().Add(2*time.Minute), object), + } + start := make(chan struct{}) + outcomes := make(chan rawIngestOutcome, len(candidates)) + var workers sync.WaitGroup + for _, candidate := range candidates { + candidate := candidate + workers.Add(1) + go func() { + defer workers.Done() + <-start + result, err := store.CommitManifest(t.Context(), candidate, "parser-data-17") + outcomes <- rawIngestOutcome{result: result, err: err} + }() + } + close(start) + workers.Wait() + close(outcomes) + + var winners, conflicts int + for got := range outcomes { + if got.err == nil { + winners++ + assert.Equal(t, int64(2), got.result.Generation) + continue + } + var conflict *rawsync.HeadConflictError + require.ErrorAs(t, got.err, &conflict) + assert.Equal(t, int64(2), conflict.CurrentGeneration) + conflicts++ + } + assert.Equal(t, 1, winners) + assert.Equal(t, 1, conflicts) + assert.Equal(t, rawIngestCounts{Manifests: 2, Entries: 2, Objects: 2, Heads: 1, Jobs: 2}, + readRawIngestCounts(t, pg)) +} + +type rawIngestOutcome struct { + result rawsync.CommitResult + err error +} + +func newRawIngestTestStore(t *testing.T) (*sql.DB, *RawIngestStore) { + t.Helper() + pgURL := testPGURL(t) + cleanSchemaTestPG(t, pgURL) + t.Cleanup(func() { cleanSchemaTestPG(t, pgURL) }) + pg, err := Open(pgURL, schemaTestSchema, true) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, pg.Close()) }) + require.NoError(t, EnsureSchema(t.Context(), pg, schemaTestSchema)) + store, err := NewRawIngestStore(pg) + require.NoError(t, err) + return pg, store +} + +func rawIngestIdentity(t *testing.T, tenant string) rawsync.AuthIdentity { + t.Helper() + identity, err := rawsync.NewAuthIdentity(tenant, "device-a") + require.NoError(t, err) + return identity +} + +func rawIngestObject(t *testing.T, digit string, length int64) rawsync.ObjectRef { + t.Helper() + object, err := rawsync.NewObjectRef(repeatedHex(digit), length) + require.NoError(t, err) + return object +} + +func rawIngestManifest( + t *testing.T, + identity rawsync.AuthIdentity, + captureID string, + parentReceipt string, + capturedAt time.Time, + object rawsync.ObjectRef, +) rawsync.CanonicalManifest { + t.Helper() + manifest, err := rawsync.ValidateAndCanonicalize(identity, rawsync.Manifest{ + SchemaVersion: rawsync.ManifestSchemaVersion, + Provider: parser.AgentCodex, + ConfiguredRootID: "root-a", + SourceKey: "sessions/demo.jsonl#main", + ExpectedParentReceipt: parentReceipt, + CaptureID: captureID, + CapturedAt: capturedAt, + Kind: rawsync.ManifestSnapshot, + Entries: []rawsync.Entry{{ + Path: "session.jsonl", + Type: "file", + Length: object.Length, + Objects: []rawsync.ObjectRef{object}, + }}, + }, rawsync.DefaultManifestLimits()) + require.NoError(t, err) + return manifest +} + +func rawIngestCapturedAt() time.Time { + return time.Date(2026, 8, 13, 12, 0, 0, 0, time.UTC) +} + +type rawIngestCounts struct { + Manifests int + Entries int + Objects int + Heads int + Jobs int +} + +func readRawIngestCounts(t *testing.T, pg *sql.DB) rawIngestCounts { + t.Helper() + var counts rawIngestCounts + require.NoError(t, pg.QueryRowContext(t.Context(), ` + SELECT + (SELECT count(*) FROM raw_manifests), + (SELECT count(*) FROM raw_manifest_entries), + (SELECT count(*) FROM raw_manifest_objects), + (SELECT count(*) FROM raw_source_heads), + (SELECT count(*) FROM raw_ingest_jobs)`, + ).Scan(&counts.Manifests, &counts.Entries, &counts.Objects, &counts.Heads, &counts.Jobs)) + return counts +} + +func rawIngestTableCount(t *testing.T, pg *sql.DB, table string) int { + t.Helper() + var count int + require.NoError(t, pg.QueryRowContext(t.Context(), + "SELECT count(*) FROM "+table, + ).Scan(&count)) + return count +} + +func TestRawIngestStoreAcceptsMaximumLengthIncompressibleKeys(t *testing.T) { + pg, store := newRawIngestTestStore(t) + limits := rawsync.DefaultManifestLimits() + sourceKey := rawIngestIncompressibleText(t, 1, 4096, sourceKeyAlphabet) + entryPath := rawIngestIncompressibleText(t, 2, limits.MaxPathBytes, entryPathAlphabet) + object := rawIngestObject(t, "a", 7) + commit := func(identity rawsync.AuthIdentity, captureID, parent string) rawsync.CommitResult { + t.Helper() + require.NoError(t, store.RecordVerifiedObject(t.Context(), identity, object)) + manifest, err := rawsync.ValidateAndCanonicalize(identity, rawsync.Manifest{ + SchemaVersion: rawsync.ManifestSchemaVersion, + Provider: parser.AgentCodex, + ConfiguredRootID: "root-a", + SourceKey: sourceKey, + ExpectedParentReceipt: parent, + CaptureID: captureID, + CapturedAt: rawIngestCapturedAt(), + Kind: rawsync.ManifestSnapshot, + Entries: []rawsync.Entry{{ + Path: entryPath, + Type: "file", + Length: object.Length, + Objects: []rawsync.ObjectRef{object}, + }}, + }, limits) + require.NoError(t, err) + result, err := store.CommitManifest(t.Context(), manifest, "parser-data-17") + require.NoError(t, err) + return result + } + + identity := rawIngestIdentity(t, "tenant-a") + first := commit(identity, "capture-a", "") + assert.True(t, first.Created) + second := commit(identity, "capture-b", first.Receipt) + assert.Equal(t, int64(2), second.Generation) + otherTenant := commit(rawIngestIdentity(t, "tenant-b"), "capture-a", "") + assert.Equal(t, int64(1), otherTenant.Generation, + "identical long source keys must stay independent across tenants") + assert.Equal(t, rawIngestCounts{Manifests: 3, Entries: 3, Objects: 3, Heads: 2, Jobs: 3}, + readRawIngestCounts(t, pg)) + + var storedSourceKey, storedPath string + require.NoError(t, pg.QueryRowContext(t.Context(), ` + SELECT m.source_key, e.path + FROM raw_manifests m + JOIN raw_manifest_entries e USING (tenant_id, manifest_id) + WHERE m.manifest_id = $1`, first.ManifestID, + ).Scan(&storedSourceKey, &storedPath)) + assert.Equal(t, sourceKey, storedSourceKey) + assert.Equal(t, entryPath, storedPath) +} + +const ( + sourceKeyAlphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" + + "!\"#$%&'()*+,-./:;<=>?@[\\]^_`{|}~ " + entryPathAlphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789" + + "!\"#$%&'()*+,-;<=>?@[]^_`{|}~" +) + +// rawIngestIncompressibleText returns deterministic pseudo-random text that +// pglz cannot shrink, so index entries carry its full byte length. +func rawIngestIncompressibleText(t *testing.T, seed uint64, length int, alphabet string) string { + t.Helper() + source := rand.New(rand.NewPCG(seed, seed+1)) + text := make([]byte, length) + for i := range text { + text[i] = alphabet[source.IntN(len(alphabet))] + } + return string(text) +} diff --git a/internal/postgres/schema.go b/internal/postgres/schema.go index ce2e8f0e6b..48b082031b 100644 --- a/internal/postgres/schema.go +++ b/internal/postgres/schema.go @@ -904,6 +904,9 @@ func EnsureSchema( if _, err := db.ExecContext(ctx, coreDDL); err != nil { return fmt.Errorf("creating pg tables: %w", err) } + if err := ensureRawIngestSchemaPG(ctx, db); err != nil { + return err + } log.Printf( "pg schema: core DDL step completed in %s", time.Since(step).Round(time.Millisecond), diff --git a/internal/postgres/schema_test.go b/internal/postgres/schema_test.go index 20ad7f936f..25b67c5398 100644 --- a/internal/postgres/schema_test.go +++ b/internal/postgres/schema_test.go @@ -580,7 +580,7 @@ func TestEnsureSchemaChecksDataVersionBeforeDDL(t *testing.T) { "EnsureSchema must not mutate PG before data-version refusal") } -func TestSyncEnsureSchemaSkipsDDLWhenSchemaCompatible(t *testing.T) { +func TestSyncEnsureSchemaSkipsLegacyDDLWhenSchemaCompatible(t *testing.T) { pg, state := newSchemaProbeDB(t, nil) state.existingTables = map[string]bool{ "model_pricing": true, @@ -602,14 +602,18 @@ func TestSyncEnsureSchemaSkipsDDLWhenSchemaCompatible(t *testing.T) { require.NoError(t, syncer.EnsureSchema(context.Background())) executed := strings.ToLower(state.executedSQL()) - assert.NotContains(t, executed, "create index", - "compatible PG schema must skip index DDL") + assert.NotContains(t, executed, "create table if not exists sessions", + "compatible PG schema must skip legacy table DDL") + assert.NotContains(t, executed, "create index if not exists idx_sessions_parent", + "compatible PG schema must skip legacy index DDL") assert.NotContains(t, executed, "alter index", - "compatible PG schema must skip index DDL") - assert.NotContains(t, executed, "create table", - "compatible PG schema must skip table DDL") + "compatible PG schema must skip legacy index migrations") assert.Equal(t, 0, state.alterTableExecCount(), "compatible PG schema must not run column migrations") + assert.Contains(t, executed, "create table if not exists raw_objects", + "raw custody tables must be bootstrapped independently") + assert.Contains(t, executed, "create index if not exists idx_raw_ingest_jobs_ready", + "raw custody indexes must be bootstrapped independently") assert.Contains(t, executed, "insert into sync_metadata", "compatible PG schema must still run row-level data repairs") } diff --git a/internal/postgres/sync.go b/internal/postgres/sync.go index 4545d0bc5b..031682dc18 100644 --- a/internal/postgres/sync.go +++ b/internal/postgres/sync.go @@ -433,6 +433,16 @@ func (s *Sync) ensureSchemaLocked(ctx context.Context) error { if _, err := ensureVectorBaseSchemaPG(ctx, s.pg); err != nil { log.Printf("pg schema: vector schema setup failed: %v", err) } + // A restricted push role may lack CREATE on a schema that a + // privileged role provisioned. Raw custody is server-side only, + // so skip it here rather than failing every push; the full + // EnsureSchema bootstrap path still requires it. + if err := ensureRawIngestSchemaPG(ctx, s.pg); err != nil { + if !isInsufficientPrivilege(err) { + return err + } + log.Printf("pg schema: raw custody schema skipped, insufficient privilege: %v", err) + } s.schemaDone = true return nil } diff --git a/internal/rawsync/manifest.go b/internal/rawsync/manifest.go new file mode 100644 index 0000000000..41631415b4 --- /dev/null +++ b/internal/rawsync/manifest.go @@ -0,0 +1,483 @@ +// Package rawsync defines the authenticated raw-ingest custody domain. +package rawsync + +import ( + "bytes" + "cmp" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "path" + "slices" + "strings" + "time" + "unicode" + "unicode/utf8" + + "go.kenn.io/agentsview/internal/parser" +) + +var ( + ErrInvalid = errors.New("invalid raw sync value") + ErrNotFound = errors.New("raw sync object not found") + ErrConflict = errors.New("raw sync conflict") + ErrMissingObject = errors.New("raw sync manifest references missing object") +) + +const ( + ManifestSchemaVersion = 1 + maxOpaqueIDBytes = 128 + maxSourceKeyBytes = 4096 +) + +// AuthIdentity is the tenant and device identity supplied by authentication. +type AuthIdentity struct { + TenantID string + DeviceID string +} + +// NewAuthIdentity validates authenticated tenant and device identifiers. +func NewAuthIdentity(tenantID, deviceID string) (AuthIdentity, error) { + if err := validateOpaqueID("tenant", tenantID); err != nil { + return AuthIdentity{}, err + } + if err := validateOpaqueID("device", deviceID); err != nil { + return AuthIdentity{}, err + } + return AuthIdentity{TenantID: tenantID, DeviceID: deviceID}, nil +} + +// ObjectRef is the semantic identity of one immutable source object. +type ObjectRef struct { + SHA256 string `json:"sha256"` + Length int64 `json:"length"` +} + +// NewObjectRef validates and constructs an immutable object reference. +func NewObjectRef(sha256 string, length int64) (ObjectRef, error) { + if !isCanonicalSHA256(sha256) { + return ObjectRef{}, fmt.Errorf("%w: object digest must be lowercase SHA-256", ErrInvalid) + } + if length < 0 { + return ObjectRef{}, fmt.Errorf("%w: object length must not be negative", ErrInvalid) + } + return ObjectRef{SHA256: sha256, Length: length}, nil +} + +// ManifestKind identifies whether a manifest captures source files or removal. +type ManifestKind string + +const ( + ManifestSnapshot ManifestKind = "snapshot" + ManifestTombstone ManifestKind = "tombstone" +) + +// Entry describes one logical source file and its ordered object slices. +type Entry struct { + Path string `json:"path"` + Type string `json:"type"` + Length int64 `json:"length"` + Objects []ObjectRef `json:"objects"` +} + +// Manifest declares one complete logical provider-source generation. +type Manifest struct { + SchemaVersion int `json:"schema_version"` + Provider parser.AgentType `json:"provider"` + ConfiguredRootID string `json:"configured_root_id"` + SourceKey string `json:"source_key"` + ExpectedParentReceipt string `json:"expected_parent_receipt,omitempty"` + CaptureID string `json:"capture_id"` + CapturedAt time.Time `json:"captured_at"` + Kind ManifestKind `json:"kind"` + Entries []Entry `json:"entries,omitempty"` +} + +// ManifestLimits bounds work and retained metadata before object-store access. +type ManifestLimits struct { + MaxCanonicalBytes int + MaxEntries int + MaxObjects int + MaxPathBytes int + MaxFileBytes int64 +} + +// DefaultManifestLimits returns the production manifest validation bounds. +func DefaultManifestLimits() ManifestLimits { + return ManifestLimits{ + MaxCanonicalBytes: 1 << 20, + MaxEntries: 4096, + MaxObjects: 16384, + MaxPathBytes: 4096, + MaxFileBytes: 16 << 30, + } +} + +// CanonicalManifest is an authenticated, validated manifest ready for custody. +type CanonicalManifest struct { + Identity AuthIdentity + Manifest Manifest + ManifestID string + CanonicalJSON []byte + Objects []ObjectRef +} + +type canonicalEnvelope struct { + SchemaVersion int `json:"schema_version"` + TenantID string `json:"tenant_id"` + DeviceID string `json:"device_id"` + Provider parser.AgentType `json:"provider"` + ConfiguredRootID string `json:"configured_root_id"` + SourceKey string `json:"source_key"` + ExpectedParentReceipt string `json:"expected_parent_receipt,omitempty"` + CaptureID string `json:"capture_id"` + CapturedAt time.Time `json:"captured_at"` + Kind ManifestKind `json:"kind"` + Entries []Entry `json:"entries,omitempty"` +} + +// ValidateAndCanonicalize binds authentication to validated canonical bytes. +func ValidateAndCanonicalize( + identity AuthIdentity, + manifest Manifest, + limits ManifestLimits, +) (CanonicalManifest, error) { + canonicalIdentity, err := NewAuthIdentity(identity.TenantID, identity.DeviceID) + if err != nil || canonicalIdentity != identity { + return CanonicalManifest{}, fmt.Errorf("%w: authenticated identity is not canonical", ErrInvalid) + } + if err := validateManifestLimits(limits); err != nil { + return CanonicalManifest{}, err + } + if err := validateManifestHeader(manifest); err != nil { + return CanonicalManifest{}, err + } + if err := validateManifestCardinality(manifest, limits); err != nil { + return CanonicalManifest{}, err + } + + canonical := manifest + canonical.CapturedAt = manifest.CapturedAt.UTC() + canonical.Entries = cloneEntries(manifest.Entries) + slices.SortFunc(canonical.Entries, func(a, b Entry) int { + return strings.Compare(a.Path, b.Path) + }) + objects, err := validateEntries(canonical, limits) + if err != nil { + return CanonicalManifest{}, err + } + + envelope := canonicalEnvelope{ + SchemaVersion: canonical.SchemaVersion, + TenantID: identity.TenantID, + DeviceID: identity.DeviceID, + Provider: canonical.Provider, + ConfiguredRootID: canonical.ConfiguredRootID, + SourceKey: canonical.SourceKey, + ExpectedParentReceipt: canonical.ExpectedParentReceipt, + CaptureID: canonical.CaptureID, + CapturedAt: canonical.CapturedAt, + Kind: canonical.Kind, + Entries: canonical.Entries, + } + encoded, err := json.Marshal(envelope) + if err != nil { + return CanonicalManifest{}, fmt.Errorf("encoding canonical raw manifest: %w", err) + } + encoded = append(encoded, '\n') + if len(encoded) > limits.MaxCanonicalBytes { + return CanonicalManifest{}, fmt.Errorf( + "%w: canonical manifest exceeds %d bytes", ErrInvalid, limits.MaxCanonicalBytes, + ) + } + digest := sha256.Sum256(encoded) + return CanonicalManifest{ + Identity: identity, + Manifest: canonical, + ManifestID: hex.EncodeToString(digest[:]), + CanonicalJSON: encoded, + Objects: objects, + }, nil +} + +// ValidateCanonicalManifest verifies canonical integrity without imposing a +// deployment's policy limits a second time. +func ValidateCanonicalManifest(manifest CanonicalManifest) error { + limits := integrityManifestLimits(manifest) + validated, err := ValidateAndCanonicalize(manifest.Identity, manifest.Manifest, limits) + if err != nil || validated.ManifestID != manifest.ManifestID || + !bytes.Equal(validated.CanonicalJSON, manifest.CanonicalJSON) || + !slices.Equal(validated.Objects, manifest.Objects) || + !manifestsEqual(validated.Manifest, manifest.Manifest) { + return fmt.Errorf("%w: canonical raw manifest is inconsistent", ErrInvalid) + } + return nil +} + +// manifestsEqual reports whether two manifests are identical, including entry +// order and the captured-at location, so a noncanonical struct cannot ride +// along with canonical bytes. +func manifestsEqual(a, b Manifest) bool { + return a.SchemaVersion == b.SchemaVersion && + a.Provider == b.Provider && + a.ConfiguredRootID == b.ConfiguredRootID && + a.SourceKey == b.SourceKey && + a.ExpectedParentReceipt == b.ExpectedParentReceipt && + a.CaptureID == b.CaptureID && + a.CapturedAt.Equal(b.CapturedAt) && + a.CapturedAt.Location() == b.CapturedAt.Location() && + a.Kind == b.Kind && + slices.EqualFunc(a.Entries, b.Entries, entriesEqual) +} + +func entriesEqual(a, b Entry) bool { + return a.Path == b.Path && a.Type == b.Type && a.Length == b.Length && + slices.Equal(a.Objects, b.Objects) +} + +func validateManifestLimits(limits ManifestLimits) error { + if limits.MaxCanonicalBytes <= 0 || limits.MaxEntries <= 0 || + limits.MaxObjects <= 0 || limits.MaxPathBytes <= 0 || limits.MaxFileBytes <= 0 { + return fmt.Errorf("%w: manifest limits must be positive", ErrInvalid) + } + return nil +} + +func validateManifestCardinality(manifest Manifest, limits ManifestLimits) error { + if len(manifest.Entries) > limits.MaxEntries { + return fmt.Errorf("%w: manifest entry limit exceeded", ErrInvalid) + } + objectCount := 0 + for _, entry := range manifest.Entries { + if len(entry.Objects) > limits.MaxObjects-objectCount { + return fmt.Errorf("%w: manifest object limit exceeded", ErrInvalid) + } + objectCount += len(entry.Objects) + } + return nil +} + +func integrityManifestLimits(manifest CanonicalManifest) ManifestLimits { + limits := ManifestLimits{ + MaxCanonicalBytes: max(1, len(manifest.CanonicalJSON)), + MaxEntries: max(1, len(manifest.Manifest.Entries)), + MaxObjects: 1, + MaxPathBytes: 1, + MaxFileBytes: 1, + } + for _, entry := range manifest.Manifest.Entries { + limits.MaxObjects += len(entry.Objects) + limits.MaxPathBytes = max(limits.MaxPathBytes, len(entry.Path)) + limits.MaxFileBytes = max(limits.MaxFileBytes, entry.Length) + } + return limits +} + +func validateManifestHeader(manifest Manifest) error { + if manifest.SchemaVersion != ManifestSchemaVersion { + return fmt.Errorf("%w: unsupported manifest schema version %d", ErrInvalid, manifest.SchemaVersion) + } + if err := validateProvider(manifest.Provider); err != nil { + return err + } + if err := validateOpaqueID("configured root", manifest.ConfiguredRootID); err != nil { + return err + } + if err := validateSourceKey(manifest.SourceKey); err != nil { + return err + } + if manifest.ExpectedParentReceipt != "" && !isCanonicalSHA256(manifest.ExpectedParentReceipt) { + return fmt.Errorf("%w: expected parent receipt must be lowercase hexadecimal", ErrInvalid) + } + if err := validateOpaqueID("capture", manifest.CaptureID); err != nil { + return err + } + if manifest.CapturedAt.IsZero() { + return fmt.Errorf("%w: capture time is required", ErrInvalid) + } + switch manifest.Kind { + case ManifestSnapshot: + if len(manifest.Entries) == 0 { + return fmt.Errorf("%w: snapshot manifest requires entries", ErrInvalid) + } + case ManifestTombstone: + if len(manifest.Entries) != 0 { + return fmt.Errorf("%w: tombstone manifest cannot contain entries", ErrInvalid) + } + default: + return fmt.Errorf("%w: unsupported manifest kind %q", ErrInvalid, manifest.Kind) + } + return nil +} + +func validateSourceKey(value string) error { + if value == "" || len(value) > maxSourceKeyBytes || !utf8.ValidString(value) { + return fmt.Errorf("%w: source key is missing, oversized, or invalid UTF-8", ErrInvalid) + } + for _, r := range value { + if unicode.IsControl(r) { + return fmt.Errorf("%w: source key contains a control character", ErrInvalid) + } + } + return nil +} + +func validateEntries(manifest Manifest, limits ManifestLimits) ([]ObjectRef, error) { + if len(manifest.Entries) > limits.MaxEntries { + return nil, fmt.Errorf("%w: manifest entry limit exceeded", ErrInvalid) + } + byDigest := make(map[string]ObjectRef) + objectCount := 0 + previousPath := "" + for _, entry := range manifest.Entries { + if err := validateEntryPath(entry.Path, limits.MaxPathBytes); err != nil { + return nil, err + } + if entry.Path == previousPath { + return nil, fmt.Errorf("%w: duplicate manifest path %q", ErrInvalid, entry.Path) + } + previousPath = entry.Path + if entry.Type != "file" { + return nil, fmt.Errorf("%w: unsupported entry type %q", ErrInvalid, entry.Type) + } + if entry.Length < 0 || entry.Length > limits.MaxFileBytes { + return nil, fmt.Errorf("%w: entry length is outside configured limits", ErrInvalid) + } + if len(entry.Objects) == 0 { + return nil, fmt.Errorf("%w: file entry requires at least one object", ErrInvalid) + } + objectCount += len(entry.Objects) + if objectCount > limits.MaxObjects { + return nil, fmt.Errorf("%w: manifest object limit exceeded", ErrInvalid) + } + var total int64 + for _, object := range entry.Objects { + validated, objectErr := NewObjectRef(object.SHA256, object.Length) + if objectErr != nil || validated != object { + return nil, fmt.Errorf("%w: invalid object reference", ErrInvalid) + } + if object.Length > entry.Length-total { + return nil, fmt.Errorf("%w: object lengths exceed entry length", ErrInvalid) + } + total += object.Length + if previous, ok := byDigest[object.SHA256]; ok && previous.Length != object.Length { + return nil, fmt.Errorf("%w: digest has conflicting lengths", ErrInvalid) + } + byDigest[object.SHA256] = object + } + if total != entry.Length { + return nil, fmt.Errorf("%w: object lengths do not equal entry length", ErrInvalid) + } + } + objects := make([]ObjectRef, 0, len(byDigest)) + for _, object := range byDigest { + objects = append(objects, object) + } + slices.SortFunc(objects, func(a, b ObjectRef) int { + if byHash := strings.Compare(a.SHA256, b.SHA256); byHash != 0 { + return byHash + } + return cmp.Compare(a.Length, b.Length) + }) + return objects, nil +} + +func validateEntryPath(value string, maxBytes int) error { + if value == "" || len(value) > maxBytes || !utf8.ValidString(value) || + path.IsAbs(value) || path.Clean(value) != value || value == "." || + value == ".." || strings.HasPrefix(value, "../") || + strings.ContainsRune(value, '\\') || isPlatformUnsafeEntryPath(value) { + return fmt.Errorf("%w: entry path is not a canonical relative path", ErrInvalid) + } + for _, r := range value { + if unicode.IsControl(r) { + return fmt.Errorf("%w: entry path contains a control character", ErrInvalid) + } + } + return nil +} + +func isPlatformUnsafeEntryPath(value string) bool { + if strings.ContainsRune(value, ':') { + return true + } + for component := range strings.SplitSeq(value, "/") { + if strings.HasSuffix(component, ".") || strings.HasSuffix(component, " ") { + return true + } + base, _, _ := strings.Cut(component, ".") + upper := strings.ToUpper(base) + switch upper { + case "CON", "PRN", "AUX", "NUL", "CLOCK$", "CONIN$", "CONOUT$": + return true + } + if len(upper) == 4 && upper[3] >= '1' && upper[3] <= '9' && + (upper[:3] == "COM" || upper[:3] == "LPT") { + return true + } + } + return false +} + +// validateProvider fails closed for providers the server cannot classify and +// for providers whose raw source trees must never leave the device. +func validateProvider(provider parser.AgentType) error { + if err := validateOpaqueID("provider", string(provider)); err != nil { + return err + } + def, ok := parser.AgentByType(provider) + if !ok { + return fmt.Errorf("%w: unknown provider %q", ErrInvalid, provider) + } + if def.RemoteSyncExcluded { + return fmt.Errorf("%w: provider %q is excluded from remote sync", ErrInvalid, provider) + } + return nil +} + +func cloneEntries(source []Entry) []Entry { + if len(source) == 0 { + return nil + } + cloned := make([]Entry, len(source)) + for i, entry := range source { + cloned[i] = entry + cloned[i].Objects = append([]ObjectRef(nil), entry.Objects...) + } + return cloned +} + +func validateOpaqueID(name, value string) error { + if value == "" { + return fmt.Errorf("%w: %s identifier is required", ErrInvalid, name) + } + if len(value) > maxOpaqueIDBytes { + return fmt.Errorf("%w: %s identifier exceeds %d bytes", ErrInvalid, name, maxOpaqueIDBytes) + } + if !utf8.ValidString(value) || strings.TrimSpace(value) != value { + return fmt.Errorf("%w: %s identifier is not canonical UTF-8", ErrInvalid, name) + } + if strings.ContainsAny(value, `/\`) { + return fmt.Errorf("%w: %s identifier contains a path separator", ErrInvalid, name) + } + for _, r := range value { + if unicode.IsControl(r) { + return fmt.Errorf("%w: %s identifier contains a control character", ErrInvalid, name) + } + } + return nil +} + +func isCanonicalSHA256(value string) bool { + if len(value) != 64 { + return false + } + for _, char := range value { + if (char < '0' || char > '9') && (char < 'a' || char > 'f') { + return false + } + } + return true +} diff --git a/internal/rawsync/manifest_test.go b/internal/rawsync/manifest_test.go new file mode 100644 index 0000000000..f63d071175 --- /dev/null +++ b/internal/rawsync/manifest_test.go @@ -0,0 +1,296 @@ +package rawsync + +import ( + "crypto/sha256" + "encoding/hex" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.kenn.io/agentsview/internal/parser" +) + +func TestNewAuthIdentityValidatesOpaqueIDs(t *testing.T) { + t.Parallel() + + identity, err := NewAuthIdentity("tenant-a", "device-a") + require.NoError(t, err) + assert.Equal(t, AuthIdentity{TenantID: "tenant-a", DeviceID: "device-a"}, identity) + + for _, tc := range []struct { + name string + tenant string + device string + }{ + {name: "missing tenant", device: "device-a"}, + {name: "missing device", tenant: "tenant-a"}, + {name: "leading whitespace", tenant: " tenant-a", device: "device-a"}, + {name: "trailing whitespace", tenant: "tenant-a", device: "device-a "}, + {name: "control character", tenant: "tenant-a", device: "bad\nvalue"}, + {name: "forward slash", tenant: "tenant/a", device: "device-a"}, + {name: "backslash", tenant: "tenant-a", device: `device\a`}, + {name: "oversized", tenant: strings.Repeat("x", 129), device: "device-a"}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + _, err := NewAuthIdentity(tc.tenant, tc.device) + assert.ErrorIs(t, err, ErrInvalid) + }) + } +} + +func TestNewObjectRefRequiresCanonicalSHA256AndNonNegativeLength(t *testing.T) { + t.Parallel() + + ref, err := NewObjectRef(strings.Repeat("a", 64), 12) + require.NoError(t, err) + assert.Equal(t, ObjectRef{SHA256: strings.Repeat("a", 64), Length: 12}, ref) + + empty, err := NewObjectRef(strings.Repeat("b", 64), 0) + require.NoError(t, err) + assert.Zero(t, empty.Length) + + for _, tc := range []struct { + name string + hash string + length int64 + }{ + {name: "short hash", hash: "abcd", length: 1}, + {name: "uppercase hash", hash: strings.Repeat("A", 64), length: 1}, + {name: "non-hex hash", hash: strings.Repeat("g", 64), length: 1}, + {name: "negative length", hash: strings.Repeat("a", 64), length: -1}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + _, err := NewObjectRef(tc.hash, tc.length) + assert.ErrorIs(t, err, ErrInvalid) + }) + } +} + +func TestValidateAndCanonicalizeProducesAuthenticatedStableEnvelope(t *testing.T) { + t.Parallel() + + manifest := validManifest() + identity, err := NewAuthIdentity("tenant-a", "device-a") + require.NoError(t, err) + + got, err := ValidateAndCanonicalize(identity, manifest, DefaultManifestLimits()) + require.NoError(t, err) + + wantJSON := `{"schema_version":1,"tenant_id":"tenant-a","device_id":"device-a","provider":"codex","configured_root_id":"root-a","source_key":"sessions/demo.jsonl#main","capture_id":"capture-a","captured_at":"2026-08-13T12:34:56Z","kind":"snapshot","entries":[{"path":"a.jsonl","type":"file","length":3,"objects":[{"sha256":"bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb","length":3}]},{"path":"z.jsonl","type":"file","length":8,"objects":[{"sha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","length":4},{"sha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","length":4}]}]}` + "\n" + assert.Equal(t, wantJSON, string(got.CanonicalJSON)) + wantSum := sha256.Sum256([]byte(wantJSON)) + assert.Equal(t, hex.EncodeToString(wantSum[:]), got.ManifestID) + assert.Equal(t, []string{"a.jsonl", "z.jsonl"}, []string{ + got.Manifest.Entries[0].Path, + got.Manifest.Entries[1].Path, + }) + assert.Equal(t, []ObjectRef{ + {SHA256: strings.Repeat("a", 64), Length: 4}, + {SHA256: strings.Repeat("b", 64), Length: 3}, + }, got.Objects) + assert.Len(t, got.Manifest.Entries[1].Objects, 2, + "repeated chunks must remain in reconstruction order") + assert.Equal(t, "z.jsonl", manifest.Entries[0].Path, + "canonicalization must not mutate the caller's manifest") +} + +func TestValidateAndCanonicalizeBindsAuthenticatedIdentity(t *testing.T) { + t.Parallel() + + manifest := validManifest() + a, err := NewAuthIdentity("tenant-a", "device-a") + require.NoError(t, err) + b, err := NewAuthIdentity("tenant-b", "device-a") + require.NoError(t, err) + + first, err := ValidateAndCanonicalize(a, manifest, DefaultManifestLimits()) + require.NoError(t, err) + again, err := ValidateAndCanonicalize(a, manifest, DefaultManifestLimits()) + require.NoError(t, err) + otherTenant, err := ValidateAndCanonicalize(b, manifest, DefaultManifestLimits()) + require.NoError(t, err) + + assert.Equal(t, first.ManifestID, again.ManifestID) + assert.Equal(t, first.CanonicalJSON, again.CanonicalJSON) + assert.NotEqual(t, first.ManifestID, otherTenant.ManifestID) +} + +func TestValidateAndCanonicalizeNormalizesCapturedInstantToUTC(t *testing.T) { + t.Parallel() + + identity, err := NewAuthIdentity("tenant-a", "device-a") + require.NoError(t, err) + utc := validManifest() + offset := cloneManifest(utc) + offset.CapturedAt = utc.CapturedAt.In(time.FixedZone("offset", 2*60*60)) + + first, err := ValidateAndCanonicalize(identity, utc, DefaultManifestLimits()) + require.NoError(t, err) + second, err := ValidateAndCanonicalize(identity, offset, DefaultManifestLimits()) + require.NoError(t, err) + + assert.Equal(t, first.ManifestID, second.ManifestID) + assert.Equal(t, first.CanonicalJSON, second.CanonicalJSON) +} + +func TestValidateAndCanonicalizeRejectsMalformedManifest(t *testing.T) { + t.Parallel() + + identity, err := NewAuthIdentity("tenant-a", "device-a") + require.NoError(t, err) + valid := validManifest() + + for _, tc := range []struct { + name string + mutate func(*Manifest, *ManifestLimits) + }{ + {name: "unsupported schema", mutate: func(m *Manifest, _ *ManifestLimits) { m.SchemaVersion = 2 }}, + {name: "missing provider", mutate: func(m *Manifest, _ *ManifestLimits) { m.Provider = "" }}, + {name: "unknown provider", mutate: func(m *Manifest, _ *ManifestLimits) { m.Provider = "not-an-agent" }}, + {name: "remote sync excluded provider", mutate: func(m *Manifest, _ *ManifestLimits) { m.Provider = parser.AgentOmnigent }}, + {name: "missing root", mutate: func(m *Manifest, _ *ManifestLimits) { m.ConfiguredRootID = "" }}, + {name: "missing source", mutate: func(m *Manifest, _ *ManifestLimits) { m.SourceKey = "" }}, + {name: "source control", mutate: func(m *Manifest, _ *ManifestLimits) { m.SourceKey = "bad\nsource" }}, + {name: "missing capture", mutate: func(m *Manifest, _ *ManifestLimits) { m.CaptureID = "" }}, + {name: "short parent receipt", mutate: func(m *Manifest, _ *ManifestLimits) { m.ExpectedParentReceipt = "abcd" }}, + {name: "uppercase parent receipt", mutate: func(m *Manifest, _ *ManifestLimits) { m.ExpectedParentReceipt = strings.Repeat("A", 64) }}, + {name: "zero captured time", mutate: func(m *Manifest, _ *ManifestLimits) { m.CapturedAt = time.Time{} }}, + {name: "absolute path", mutate: func(m *Manifest, _ *ManifestLimits) { m.Entries[0].Path = "/escape" }}, + {name: "parent path", mutate: func(m *Manifest, _ *ManifestLimits) { m.Entries[0].Path = "../escape" }}, + {name: "backslash path", mutate: func(m *Manifest, _ *ManifestLimits) { m.Entries[0].Path = `dir\file` }}, + {name: "alternate data path", mutate: func(m *Manifest, _ *ManifestLimits) { m.Entries[0].Path = "session.jsonl:stream" }}, + {name: "drive relative path", mutate: func(m *Manifest, _ *ManifestLimits) { m.Entries[0].Path = "C:session.jsonl" }}, + {name: "windows device path", mutate: func(m *Manifest, _ *ManifestLimits) { m.Entries[0].Path = "dir/CON.jsonl" }}, + {name: "windows trailing dot", mutate: func(m *Manifest, _ *ManifestLimits) { m.Entries[0].Path = "session.jsonl." }}, + {name: "windows trailing space", mutate: func(m *Manifest, _ *ManifestLimits) { m.Entries[0].Path = "session.jsonl " }}, + {name: "duplicate path", mutate: func(m *Manifest, _ *ManifestLimits) { m.Entries[1].Path = m.Entries[0].Path }}, + {name: "unsupported entry type", mutate: func(m *Manifest, _ *ManifestLimits) { m.Entries[0].Type = "directory" }}, + {name: "empty object list", mutate: func(m *Manifest, _ *ManifestLimits) { m.Entries[0].Objects = nil }}, + {name: "object sum mismatch", mutate: func(m *Manifest, _ *ManifestLimits) { m.Entries[0].Length++ }}, + {name: "invalid embedded object", mutate: func(m *Manifest, _ *ManifestLimits) { m.Entries[0].Objects[0].SHA256 = "bad" }}, + {name: "tombstone with entries", mutate: func(m *Manifest, _ *ManifestLimits) { m.Kind = ManifestTombstone }}, + {name: "snapshot without entries", mutate: func(m *Manifest, _ *ManifestLimits) { m.Entries = nil }}, + {name: "entry limit", mutate: func(_ *Manifest, limits *ManifestLimits) { limits.MaxEntries = 1 }}, + {name: "object limit", mutate: func(_ *Manifest, limits *ManifestLimits) { limits.MaxObjects = 2 }}, + {name: "path limit", mutate: func(_ *Manifest, limits *ManifestLimits) { limits.MaxPathBytes = 2 }}, + {name: "file limit", mutate: func(_ *Manifest, limits *ManifestLimits) { limits.MaxFileBytes = 7 }}, + {name: "canonical limit", mutate: func(_ *Manifest, limits *ManifestLimits) { limits.MaxCanonicalBytes = 10 }}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + manifest := cloneManifest(valid) + limits := DefaultManifestLimits() + tc.mutate(&manifest, &limits) + _, err := ValidateAndCanonicalize(identity, manifest, limits) + assert.ErrorIs(t, err, ErrInvalid) + }) + } +} + +func TestValidateAndCanonicalizeAcceptsEmptyTombstone(t *testing.T) { + t.Parallel() + + identity, err := NewAuthIdentity("tenant-a", "device-a") + require.NoError(t, err) + manifest := validManifest() + manifest.Kind = ManifestTombstone + manifest.Entries = nil + + got, err := ValidateAndCanonicalize(identity, manifest, DefaultManifestLimits()) + require.NoError(t, err) + assert.Empty(t, got.Objects) + assert.Empty(t, got.Manifest.Entries) +} + +func validManifest() Manifest { + return Manifest{ + SchemaVersion: ManifestSchemaVersion, + Provider: parser.AgentCodex, + ConfiguredRootID: "root-a", + SourceKey: "sessions/demo.jsonl#main", + CaptureID: "capture-a", + CapturedAt: time.Date(2026, 8, 13, 12, 34, 56, 0, time.UTC), + Kind: ManifestSnapshot, + Entries: []Entry{ + { + Path: "z.jsonl", + Type: "file", + Length: 8, + Objects: []ObjectRef{ + {SHA256: strings.Repeat("a", 64), Length: 4}, + {SHA256: strings.Repeat("a", 64), Length: 4}, + }, + }, + { + Path: "a.jsonl", + Type: "file", + Length: 3, + Objects: []ObjectRef{ + {SHA256: strings.Repeat("b", 64), Length: 3}, + }, + }, + }, + } +} + +func cloneManifest(source Manifest) Manifest { + cloned := source + cloned.Entries = make([]Entry, len(source.Entries)) + for i, entry := range source.Entries { + cloned.Entries[i] = entry + cloned.Entries[i].Objects = append([]ObjectRef(nil), entry.Objects...) + } + return cloned +} + +func TestValidateCanonicalManifestRejectsNoncanonicalValues(t *testing.T) { + t.Parallel() + + identity, err := NewAuthIdentity("tenant-a", "device-a") + require.NoError(t, err) + canonical, err := ValidateAndCanonicalize(identity, validManifest(), DefaultManifestLimits()) + require.NoError(t, err) + require.NoError(t, ValidateCanonicalManifest(canonical)) + + for _, tc := range []struct { + name string + mutate func(*CanonicalManifest) + }{ + {name: "unsorted entries", mutate: func(m *CanonicalManifest) { + m.Manifest.Entries[0], m.Manifest.Entries[1] = m.Manifest.Entries[1], m.Manifest.Entries[0] + }}, + {name: "non-utc captured time", mutate: func(m *CanonicalManifest) { + m.Manifest.CapturedAt = m.Manifest.CapturedAt.In(time.FixedZone("offset", 2*60*60)) + }}, + {name: "wrong manifest id", mutate: func(m *CanonicalManifest) { + m.ManifestID = strings.Repeat("0", 64) + }}, + {name: "tampered canonical json", mutate: func(m *CanonicalManifest) { + m.CanonicalJSON = append([]byte(nil), m.CanonicalJSON...) + m.CanonicalJSON[len(m.CanonicalJSON)-2] = 'x' + }}, + {name: "extra object", mutate: func(m *CanonicalManifest) { + m.Objects = append(m.Objects, ObjectRef{SHA256: strings.Repeat("c", 64), Length: 1}) + }}, + {name: "wrong identity", mutate: func(m *CanonicalManifest) { + m.Identity.DeviceID = "device-b" + }}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + mutated := canonical + mutated.Manifest = cloneManifest(canonical.Manifest) + mutated.Objects = append([]ObjectRef(nil), canonical.Objects...) + tc.mutate(&mutated) + assert.ErrorIs(t, ValidateCanonicalManifest(mutated), ErrInvalid) + }) + } +} diff --git a/internal/rawsync/metadata_store.go b/internal/rawsync/metadata_store.go new file mode 100644 index 0000000000..398b80dcfb --- /dev/null +++ b/internal/rawsync/metadata_store.go @@ -0,0 +1,41 @@ +package rawsync + +import ( + "context" + "fmt" +) + +// CommitResult is the durable receipt for one accepted source generation. +type CommitResult struct { + ManifestID string + Receipt string + Generation int64 + Created bool +} + +// HeadConflictError reports the current accepted source head. +type HeadConflictError struct { + CurrentManifestID string + CurrentReceipt string + CurrentGeneration int64 +} + +func (e *HeadConflictError) Error() string { + if e == nil { + return "raw sync source head conflict" + } + return fmt.Sprintf( + "raw sync source head conflict at generation %d", e.CurrentGeneration, + ) +} + +// Unwrap makes source-head conflicts discoverable through ErrConflict. +func (e *HeadConflictError) Unwrap() error { return ErrConflict } + +// MetadataStore records verified custody and atomically accepts manifests. +type MetadataStore interface { + RecordVerifiedObject(context.Context, AuthIdentity, ObjectRef) error + RecordVerifiedObjects(context.Context, AuthIdentity, []ObjectRef) error + MissingObjects(context.Context, AuthIdentity, []ObjectRef) ([]ObjectRef, error) + CommitManifest(context.Context, CanonicalManifest, string) (CommitResult, error) +} diff --git a/internal/rawsync/object_store.go b/internal/rawsync/object_store.go new file mode 100644 index 0000000000..1e63c7a20e --- /dev/null +++ b/internal/rawsync/object_store.go @@ -0,0 +1,37 @@ +package rawsync + +import ( + "context" + "io" + "time" +) + +// ObjectInfo describes one immutable semantic custody object. +type ObjectInfo struct { + Ref ObjectRef + Modified time.Time +} + +// PutResult distinguishes a new immutable object from an identical retry. +type PutResult struct { + Info ObjectInfo + Created bool +} + +// VerifiedObjectReader verifies the complete semantic object on demand. +type VerifiedObjectReader interface { + io.ReadCloser + Verify() error +} + +// ObjectStore owns immutable source objects and canonical manifest envelopes. +type ObjectStore interface { + PutObject(context.Context, string, ObjectRef, io.Reader) (PutResult, error) + StatObject(context.Context, string, ObjectRef) (ObjectInfo, error) + OpenObject(context.Context, string, ObjectRef) (ObjectInfo, VerifiedObjectReader, error) + MissingObjects(context.Context, string, []ObjectRef) ([]ObjectRef, error) + // VerifyObjects requires every supplied semantic identity to exist exactly. + VerifyObjects(context.Context, string, []ObjectRef) error + PutManifest(context.Context, CanonicalManifest) (PutResult, error) + OpenManifest(context.Context, AuthIdentity, string) (ObjectInfo, VerifiedObjectReader, error) +} diff --git a/internal/rawsync/object_store_artifact.go b/internal/rawsync/object_store_artifact.go new file mode 100644 index 0000000000..68e1e4ce3d --- /dev/null +++ b/internal/rawsync/object_store_artifact.go @@ -0,0 +1,303 @@ +package rawsync + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "sync" + + "go.kenn.io/agentsview/internal/artifact" +) + +const rawObjectStatConcurrency = 32 + +type artifactObjectStore struct { + store artifact.ArtifactStore +} + +// NewArtifactObjectStore adapts the verified artifact ledger for raw custody. +func NewArtifactObjectStore(store artifact.ArtifactStore) (ObjectStore, error) { + if store == nil { + return nil, fmt.Errorf("%w: artifact store is required", ErrInvalid) + } + return &artifactObjectStore{store: store}, nil +} + +func (s *artifactObjectStore) PutObject( + ctx context.Context, + tenantID string, + object ObjectRef, + body io.Reader, +) (PutResult, error) { + if body == nil { + return PutResult{}, fmt.Errorf("%w: raw object body is required", ErrInvalid) + } + ref, identity, err := rawArtifactCoordinates(tenantID, object) + if err != nil { + return PutResult{}, err + } + created, err := s.store.Create(ctx, ref, identity, "application/octet-stream", body) + if err != nil { + return PutResult{}, mapArtifactCreateError("putting raw object", err) + } + return putResultFromArtifact(created), nil +} + +func (s *artifactObjectStore) StatObject( + ctx context.Context, + tenantID string, + object ObjectRef, +) (ObjectInfo, error) { + ref, expected, err := rawArtifactCoordinates(tenantID, object) + if err != nil { + return ObjectInfo{}, err + } + entry, err := s.store.Stat(ctx, ref) + if err != nil { + return ObjectInfo{}, mapArtifactError("stating raw object", err) + } + if entry.Identity != expected { + return ObjectInfo{}, fmt.Errorf( + "stating raw object: %w: semantic identity differs", ErrConflict, + ) + } + return objectInfoFromArtifact(entry), nil +} + +func (s *artifactObjectStore) OpenObject( + ctx context.Context, + tenantID string, + object ObjectRef, +) (ObjectInfo, VerifiedObjectReader, error) { + ref, expected, err := rawArtifactCoordinates(tenantID, object) + if err != nil { + return ObjectInfo{}, nil, err + } + entry, reader, err := s.store.Open(ctx, ref) + if err != nil { + return ObjectInfo{}, nil, mapArtifactError("opening raw object", err) + } + if entry.Identity != expected { + _ = reader.Close() + return ObjectInfo{}, nil, fmt.Errorf( + "opening raw object: %w: semantic identity differs", ErrConflict, + ) + } + return objectInfoFromArtifact(entry), reader, nil +} + +func (s *artifactObjectStore) MissingObjects( + ctx context.Context, + tenantID string, + objects []ObjectRef, +) ([]ObjectRef, error) { + if err := validateOpaqueID("tenant", tenantID); err != nil { + return nil, err + } + seen := make(map[string]ObjectRef, len(objects)) + unique := make([]ObjectRef, 0, len(objects)) + for _, object := range objects { + validated, err := NewObjectRef(object.SHA256, object.Length) + if err != nil || validated != object { + return nil, fmt.Errorf("%w: invalid missing-object reference", ErrInvalid) + } + if previous, ok := seen[object.SHA256]; ok { + if previous.Length != object.Length { + return nil, fmt.Errorf("%w: digest has conflicting lengths", ErrConflict) + } + continue + } + seen[object.SHA256] = object + unique = append(unique, object) + } + workCtx, cancel := context.WithCancel(ctx) + defer cancel() + jobs := make(chan int) + missingAt := make([]bool, len(unique)) + var firstErr error + var errorOnce sync.Once + var workers sync.WaitGroup + workerCount := min(rawObjectStatConcurrency, len(unique)) + for range workerCount { + workers.Go(func() { + for index := range jobs { + _, err := s.StatObject(workCtx, tenantID, unique[index]) + switch { + case errors.Is(err, ErrNotFound): + missingAt[index] = true + case err != nil: + errorOnce.Do(func() { + firstErr = err + cancel() + }) + } + } + }) + } + for index := range unique { + select { + case jobs <- index: + case <-workCtx.Done(): + close(jobs) + workers.Wait() + if firstErr != nil { + return nil, firstErr + } + return nil, workCtx.Err() + } + } + close(jobs) + workers.Wait() + if firstErr != nil { + return nil, firstErr + } + missing := make([]ObjectRef, 0) + for index, object := range unique { + if missingAt[index] { + missing = append(missing, object) + } + } + return missing, nil +} + +func (s *artifactObjectStore) VerifyObjects( + ctx context.Context, + tenantID string, + objects []ObjectRef, +) error { + missing, err := s.MissingObjects(ctx, tenantID, objects) + if err != nil { + return err + } + if len(missing) != 0 { + return fmt.Errorf("%w: %s", ErrMissingObject, missing[0].SHA256) + } + return nil +} + +func (s *artifactObjectStore) PutManifest( + ctx context.Context, + manifest CanonicalManifest, +) (PutResult, error) { + if err := ValidateCanonicalManifest(manifest); err != nil { + return PutResult{}, err + } + origin, err := tenantArtifactOrigin(manifest.Identity.TenantID) + if err != nil { + return PutResult{}, err + } + ref, err := artifact.NewRef(origin, artifact.KindManifests, manifest.ManifestID+".json") + if err != nil { + return PutResult{}, mapArtifactError("constructing manifest reference", err) + } + identity, err := artifact.NewIdentity(manifest.ManifestID, int64(len(manifest.CanonicalJSON))) + if err != nil { + return PutResult{}, mapArtifactError("constructing manifest identity", err) + } + created, err := s.store.Create( + ctx, ref, identity, "application/json", bytes.NewReader(manifest.CanonicalJSON), + ) + if err != nil { + return PutResult{}, mapArtifactCreateError("putting canonical manifest", err) + } + return putResultFromArtifact(created), nil +} + +func (s *artifactObjectStore) OpenManifest( + ctx context.Context, + identity AuthIdentity, + manifestID string, +) (ObjectInfo, VerifiedObjectReader, error) { + canonical, err := NewAuthIdentity(identity.TenantID, identity.DeviceID) + if err != nil || canonical != identity || !isCanonicalSHA256(manifestID) { + return ObjectInfo{}, nil, fmt.Errorf("%w: invalid manifest identity", ErrInvalid) + } + origin, err := tenantArtifactOrigin(identity.TenantID) + if err != nil { + return ObjectInfo{}, nil, err + } + ref, err := artifact.NewRef(origin, artifact.KindManifests, manifestID+".json") + if err != nil { + return ObjectInfo{}, nil, mapArtifactError("constructing manifest reference", err) + } + entry, reader, err := s.store.Open(ctx, ref) + if err != nil { + return ObjectInfo{}, nil, mapArtifactError("opening canonical manifest", err) + } + return objectInfoFromArtifact(entry), reader, nil +} + +func rawArtifactCoordinates( + tenantID string, + object ObjectRef, +) (artifact.Ref, artifact.Identity, error) { + origin, err := tenantArtifactOrigin(tenantID) + if err != nil { + return artifact.Ref{}, artifact.Identity{}, err + } + validated, err := NewObjectRef(object.SHA256, object.Length) + if err != nil || validated != object { + return artifact.Ref{}, artifact.Identity{}, fmt.Errorf("%w: invalid object reference", ErrInvalid) + } + ref, err := artifact.NewRef(origin, artifact.KindRaw, object.SHA256) + if err != nil { + return artifact.Ref{}, artifact.Identity{}, mapArtifactError("constructing raw reference", err) + } + identity, err := artifact.NewIdentity(object.SHA256, object.Length) + if err != nil { + return artifact.Ref{}, artifact.Identity{}, mapArtifactError("constructing raw identity", err) + } + return ref, identity, nil +} + +func tenantArtifactOrigin(tenantID string) (string, error) { + if err := validateOpaqueID("tenant", tenantID); err != nil { + return "", err + } + sum := sha256.Sum256([]byte(tenantID)) + return "tenant-" + hex.EncodeToString(sum[:]), nil +} + +func objectInfoFromArtifact(entry artifact.Entry) ObjectInfo { + return ObjectInfo{ + Ref: ObjectRef{ + SHA256: entry.Identity.SHA256, + Length: entry.Identity.Size, + }, + Modified: entry.Modified, + } +} + +func putResultFromArtifact(result artifact.CreateResult) PutResult { + return PutResult{ + Info: objectInfoFromArtifact(result.Entry), + Created: result.Created, + } +} + +func mapArtifactError(operation string, err error) error { + switch { + case errors.Is(err, artifact.ErrArtifactNotFound): + return fmt.Errorf("%s: %w: %w", operation, ErrNotFound, err) + case errors.Is(err, artifact.ErrArtifactInvalid): + return fmt.Errorf("%s: %w: %w", operation, ErrInvalid, err) + case errors.Is(err, artifact.ErrArtifactConflict), + errors.Is(err, artifact.ErrArtifactCorrupt): + return fmt.Errorf("%s: %w: %w", operation, ErrConflict, err) + default: + return fmt.Errorf("%s: %w", operation, err) + } +} + +func mapArtifactCreateError(operation string, err error) error { + if errors.Is(err, artifact.ErrArtifactInvalid) || + errors.Is(err, artifact.ErrArtifactConflict) || + errors.Is(err, artifact.ErrArtifactCorrupt) { + return fmt.Errorf("%s: %w: %w", operation, ErrConflict, err) + } + return mapArtifactError(operation, err) +} diff --git a/internal/rawsync/object_store_artifact_test.go b/internal/rawsync/object_store_artifact_test.go new file mode 100644 index 0000000000..c1044c782d --- /dev/null +++ b/internal/rawsync/object_store_artifact_test.go @@ -0,0 +1,172 @@ +package rawsync + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "io" + "strings" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.kenn.io/agentsview/internal/artifact" +) + +func TestArtifactObjectStoreContract(t *testing.T) { + t.Parallel() + + repository, err := artifact.OpenRepository(t.Context(), t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, repository.Close()) }) + store, err := NewArtifactObjectStore(repository.Content()) + require.NoError(t, err) + identity, err := NewAuthIdentity("tenant-a", "device-a") + require.NoError(t, err) + body := []byte("raw bytes") + ref := objectRefForBytes(t, body) + + created, err := store.PutObject(t.Context(), identity.TenantID, ref, bytes.NewReader(body)) + require.NoError(t, err) + assert.True(t, created.Created) + assert.Equal(t, ref, created.Info.Ref) + retried, err := store.PutObject(t.Context(), identity.TenantID, ref, bytes.NewReader(body)) + require.NoError(t, err) + assert.False(t, retried.Created) + + stat, err := store.StatObject(t.Context(), identity.TenantID, ref) + require.NoError(t, err) + assert.Equal(t, ref, stat.Ref) + info, reader, err := store.OpenObject(t.Context(), identity.TenantID, ref) + require.NoError(t, err) + got, err := io.ReadAll(reader) + require.NoError(t, err) + require.NoError(t, reader.Verify()) + require.NoError(t, reader.Close()) + assert.Equal(t, body, got) + assert.Equal(t, ref, info.Ref) + wrongLength := ObjectRef{SHA256: ref.SHA256, Length: ref.Length + 1} + _, err = store.StatObject(t.Context(), identity.TenantID, wrongLength) + assert.ErrorIs(t, err, ErrConflict) + _, wrongReader, err := store.OpenObject(t.Context(), identity.TenantID, wrongLength) + assert.ErrorIs(t, err, ErrConflict) + assert.Nil(t, wrongReader) + if wrongReader != nil { + _ = wrongReader.Close() + } + _, err = store.MissingObjects( + t.Context(), identity.TenantID, []ObjectRef{wrongLength}, + ) + assert.ErrorIs(t, err, ErrConflict) + + missing, err := store.MissingObjects(t.Context(), identity.TenantID, []ObjectRef{ + ref, + {SHA256: strings.Repeat("c", 64), Length: 7}, + {SHA256: strings.Repeat("c", 64), Length: 7}, + }) + require.NoError(t, err) + assert.Equal(t, []ObjectRef{{SHA256: strings.Repeat("c", 64), Length: 7}}, missing) + + otherTenantMissing, err := store.MissingObjects(t.Context(), "tenant-b", []ObjectRef{ref}) + require.NoError(t, err) + assert.Equal(t, []ObjectRef{ref}, otherTenantMissing) + _, _, err = store.OpenObject(t.Context(), "tenant-b", ref) + assert.ErrorIs(t, err, ErrNotFound) +} + +func TestArtifactObjectStoreRejectsInvalidWritesAndRequests(t *testing.T) { + t.Parallel() + + repository, err := artifact.OpenRepository(t.Context(), t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, repository.Close()) }) + store, err := NewArtifactObjectStore(repository.Content()) + require.NoError(t, err) + ref := objectRefForBytes(t, []byte("expected")) + + _, err = store.PutObject(t.Context(), "tenant-a", ref, bytes.NewReader([]byte("corrupt"))) + assert.ErrorIs(t, err, ErrConflict) + _, err = store.PutObject(t.Context(), "tenant-a", ref, bytes.NewReader([]byte("EXPected"))) + assert.ErrorIs(t, err, ErrConflict) + _, err = store.PutObject(t.Context(), "bad/tenant", ref, bytes.NewReader([]byte("expected"))) + assert.ErrorIs(t, err, ErrInvalid) + _, err = store.MissingObjects(t.Context(), "tenant-a", []ObjectRef{ + ref, + {SHA256: ref.SHA256, Length: ref.Length + 1}, + }) + assert.ErrorIs(t, err, ErrConflict) + + ctx, cancel := context.WithCancel(t.Context()) + cancel() + _, err = store.PutObject(ctx, "tenant-a", ref, bytes.NewReader([]byte("expected"))) + assert.ErrorIs(t, err, context.Canceled) + _, err = NewArtifactObjectStore(nil) + assert.ErrorIs(t, err, ErrInvalid) +} + +func TestArtifactObjectStoreRetainsCanonicalManifestEnvelope(t *testing.T) { + t.Parallel() + + repository, err := artifact.OpenRepository(t.Context(), t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, repository.Close()) }) + store, err := NewArtifactObjectStore(repository.Content()) + require.NoError(t, err) + identity, err := NewAuthIdentity("tenant-a", "device-a") + require.NoError(t, err) + manifest, err := ValidateAndCanonicalize(identity, validManifest(), DefaultManifestLimits()) + require.NoError(t, err) + + created, err := store.PutManifest(t.Context(), manifest) + require.NoError(t, err) + assert.True(t, created.Created) + assert.Equal(t, ObjectRef{SHA256: manifest.ManifestID, Length: int64(len(manifest.CanonicalJSON))}, created.Info.Ref) + retried, err := store.PutManifest(t.Context(), manifest) + require.NoError(t, err) + assert.False(t, retried.Created) + + info, reader, err := store.OpenManifest(t.Context(), identity, manifest.ManifestID) + require.NoError(t, err) + got, err := io.ReadAll(reader) + require.NoError(t, err) + require.NoError(t, reader.Verify()) + require.NoError(t, reader.Close()) + assert.Equal(t, manifest.CanonicalJSON, got) + assert.Equal(t, created.Info.Ref, info.Ref) + + other, err := NewAuthIdentity("tenant-b", "device-a") + require.NoError(t, err) + _, _, err = store.OpenManifest(t.Context(), other, manifest.ManifestID) + assert.ErrorIs(t, err, ErrNotFound) +} + +func TestArtifactObjectStoreAcceptsCanonicalManifestBeyondDefaultPolicy(t *testing.T) { + t.Parallel() + + repository, err := artifact.OpenRepository(t.Context(), t.TempDir()) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, repository.Close()) }) + store, err := NewArtifactObjectStore(repository.Content()) + require.NoError(t, err) + identity, err := NewAuthIdentity("tenant-a", "device-a") + require.NoError(t, err) + manifest := validManifest() + manifest.Entries[0].Path = strings.Repeat("p", DefaultManifestLimits().MaxPathBytes+1) + limits := DefaultManifestLimits() + limits.MaxPathBytes++ + canonical, err := ValidateAndCanonicalize(identity, manifest, limits) + require.NoError(t, err) + + result, err := store.PutManifest(t.Context(), canonical) + require.NoError(t, err) + assert.True(t, result.Created) +} + +func objectRefForBytes(t *testing.T, body []byte) ObjectRef { + t.Helper() + sum := sha256.Sum256(body) + ref, err := NewObjectRef(hex.EncodeToString(sum[:]), int64(len(body))) + require.NoError(t, err) + return ref +} diff --git a/internal/rawsync/service.go b/internal/rawsync/service.go new file mode 100644 index 0000000000..8cc456535a --- /dev/null +++ b/internal/rawsync/service.go @@ -0,0 +1,212 @@ +package rawsync + +import ( + "context" + "fmt" + "io" + "reflect" + "strings" + "unicode" + "unicode/utf8" + + "go.kenn.io/agentsview/internal/parser" +) + +// Service coordinates physical custody with durable metadata acceptance. +type Service struct { + objects ObjectStore + metadata MetadataStore + limits ManifestLimits + processingVersion string +} + +// NewService constructs the sole raw-custody boundary used by upload APIs. +func NewService( + objects ObjectStore, + metadata MetadataStore, + limits ManifestLimits, + processingVersion string, +) (*Service, error) { + if isNilServiceDependency(objects) { + return nil, fmt.Errorf("%w: object store is required", ErrInvalid) + } + if isNilServiceDependency(metadata) { + return nil, fmt.Errorf("%w: metadata store is required", ErrInvalid) + } + if err := validateManifestLimits(limits); err != nil { + return nil, err + } + if err := validateServiceProcessingVersion(processingVersion); err != nil { + return nil, err + } + return &Service{ + objects: objects, + metadata: metadata, + limits: limits, + processingVersion: processingVersion, + }, nil +} + +// FinalizeObject verifies immutable physical custody before metadata registration. +// The provider is the upload's declared source and gates custody so excluded +// providers never persist bytes; objects remain content-addressed per tenant. +func (s *Service) FinalizeObject( + ctx context.Context, + identity AuthIdentity, + provider parser.AgentType, + object ObjectRef, + body io.Reader, +) (PutResult, error) { + if err := validateServiceIdentity(identity); err != nil { + return PutResult{}, err + } + if err := validateProvider(provider); err != nil { + return PutResult{}, err + } + if err := s.validateServiceObject(object); err != nil { + return PutResult{}, err + } + if body == nil { + return PutResult{}, fmt.Errorf("%w: raw object body is required", ErrInvalid) + } + if err := ctx.Err(); err != nil { + return PutResult{}, err + } + result, err := s.objects.PutObject(ctx, identity.TenantID, object, body) + if err != nil { + return PutResult{}, fmt.Errorf("finalizing raw object custody: %w", err) + } + if result.Info.Ref != object { + return PutResult{}, fmt.Errorf("raw object custody returned a different identity: %w", ErrConflict) + } + if err := ctx.Err(); err != nil { + return PutResult{}, err + } + if err := s.metadata.RecordVerifiedObject(ctx, identity, object); err != nil { + return PutResult{}, fmt.Errorf("registering verified raw object: %w", err) + } + return result, nil +} + +// MissingObjects checks physical custody, which is authoritative for upload resumption. +func (s *Service) MissingObjects( + ctx context.Context, + identity AuthIdentity, + provider parser.AgentType, + objects []ObjectRef, +) ([]ObjectRef, error) { + if err := validateServiceIdentity(identity); err != nil { + return nil, err + } + if err := validateProvider(provider); err != nil { + return nil, err + } + for _, object := range objects { + if err := s.validateServiceObject(object); err != nil { + return nil, err + } + } + if err := ctx.Err(); err != nil { + return nil, err + } + missing, err := s.objects.MissingObjects(ctx, identity.TenantID, objects) + if err != nil { + return nil, fmt.Errorf("checking raw object custody: %w", err) + } + return missing, nil +} + +// CommitManifest verifies custody, stores the canonical envelope, then accepts metadata. +func (s *Service) CommitManifest( + ctx context.Context, + identity AuthIdentity, + manifest Manifest, +) (CommitResult, error) { + canonical, err := ValidateAndCanonicalize(identity, manifest, s.limits) + if err != nil { + return CommitResult{}, err + } + if err := ctx.Err(); err != nil { + return CommitResult{}, err + } + if err := s.objects.VerifyObjects(ctx, identity.TenantID, canonical.Objects); err != nil { + return CommitResult{}, fmt.Errorf("verifying manifest object custody: %w", err) + } + if err := s.metadata.RecordVerifiedObjects(ctx, identity, canonical.Objects); err != nil { + return CommitResult{}, fmt.Errorf("registering manifest object custody: %w", err) + } + if err := ctx.Err(); err != nil { + return CommitResult{}, err + } + manifestResult, err := s.objects.PutManifest(ctx, canonical) + if err != nil { + return CommitResult{}, fmt.Errorf("finalizing canonical manifest custody: %w", err) + } + expectedManifestRef := ObjectRef{ + SHA256: canonical.ManifestID, + Length: int64(len(canonical.CanonicalJSON)), + } + if manifestResult.Info.Ref != expectedManifestRef { + return CommitResult{}, fmt.Errorf( + "canonical manifest custody returned a different identity: %w", ErrConflict, + ) + } + if err := ctx.Err(); err != nil { + return CommitResult{}, err + } + result, err := s.metadata.CommitManifest(ctx, canonical, s.processingVersion) + if err != nil { + return CommitResult{}, fmt.Errorf("accepting canonical raw manifest: %w", err) + } + return result, nil +} + +func validateServiceIdentity(identity AuthIdentity) error { + canonical, err := NewAuthIdentity(identity.TenantID, identity.DeviceID) + if err != nil || canonical != identity { + return fmt.Errorf("%w: authenticated identity is not canonical", ErrInvalid) + } + return nil +} + +// validateServiceObject rejects noncanonical references and objects no valid +// manifest could reference, so oversized uploads never become orphans. +func (s *Service) validateServiceObject(object ObjectRef) error { + canonical, err := NewObjectRef(object.SHA256, object.Length) + if err != nil || canonical != object { + return fmt.Errorf("%w: raw object reference is not canonical", ErrInvalid) + } + if object.Length > s.limits.MaxFileBytes { + return fmt.Errorf( + "%w: raw object exceeds the %d byte file limit", ErrInvalid, s.limits.MaxFileBytes, + ) + } + return nil +} + +func validateServiceProcessingVersion(value string) error { + if value == "" || len(value) > 128 || !utf8.ValidString(value) || + strings.TrimSpace(value) != value { + return fmt.Errorf("%w: processing version is not canonical", ErrInvalid) + } + for _, r := range value { + if unicode.IsControl(r) { + return fmt.Errorf("%w: processing version contains a control character", ErrInvalid) + } + } + return nil +} + +func isNilServiceDependency(value any) bool { + if value == nil { + return true + } + reflected := reflect.ValueOf(value) + switch reflected.Kind() { + case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, + reflect.Pointer, reflect.Slice: + return reflected.IsNil() + default: + return false + } +} diff --git a/internal/rawsync/service_test.go b/internal/rawsync/service_test.go new file mode 100644 index 0000000000..edf3eea927 --- /dev/null +++ b/internal/rawsync/service_test.go @@ -0,0 +1,570 @@ +package rawsync + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "io" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.kenn.io/agentsview/internal/parser" +) + +func TestServiceConstructorRejectsInvalidDependenciesAndConfiguration(t *testing.T) { + t.Parallel() + + objects := &recordingObjectStore{} + metadata := &recordingMetadataStore{} + limits := DefaultManifestLimits() + var typedNilObjects *recordingObjectStore + var typedNilMetadata *recordingMetadataStore + + tests := []struct { + name string + objects ObjectStore + metadata MetadataStore + limits ManifestLimits + processing string + }{ + {name: "nil object store", metadata: metadata, limits: limits, processing: "parser-data-17"}, + {name: "typed nil object store", objects: typedNilObjects, metadata: metadata, limits: limits, processing: "parser-data-17"}, + {name: "nil metadata store", objects: objects, limits: limits, processing: "parser-data-17"}, + {name: "typed nil metadata store", objects: objects, metadata: typedNilMetadata, limits: limits, processing: "parser-data-17"}, + {name: "invalid limits", objects: objects, metadata: metadata, limits: ManifestLimits{}, processing: "parser-data-17"}, + {name: "blank processing version", objects: objects, metadata: metadata, limits: limits}, + {name: "noncanonical processing version", objects: objects, metadata: metadata, limits: limits, processing: " parser-data-17"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + service, err := NewService(tt.objects, tt.metadata, tt.limits, tt.processing) + assert.Nil(t, service) + assert.ErrorIs(t, err, ErrInvalid) + }) + } +} + +func TestServiceFinalizeObjectOrdersCustodyBeforeRegistration(t *testing.T) { + t.Parallel() + + t.Run("success", func(t *testing.T) { + t.Parallel() + events := []string{} + object := serviceObject("body") + objects := &recordingObjectStore{ + events: &events, + putObjectResult: PutResult{ + Info: ObjectInfo{Ref: object}, Created: true, + }, + } + metadata := &recordingMetadataStore{events: &events} + service := newTestService(t, objects, metadata) + identity, _ := validServiceInput(t) + + result, err := service.FinalizeObject( + t.Context(), identity, parser.AgentCodex, object, bytes.NewBufferString("body"), + ) + require.NoError(t, err) + assert.True(t, result.Created) + assert.Equal(t, []string{"put-object", "record-object"}, events) + assert.Equal(t, identity.TenantID, objects.putObjectTenant) + assert.Equal(t, object, metadata.recordedObjects[0]) + }) + + t.Run("physical write failure", func(t *testing.T) { + t.Parallel() + events := []string{} + putErr := errors.New("object backend unavailable") + objects := &recordingObjectStore{events: &events, putObjectErr: putErr} + metadata := &recordingMetadataStore{events: &events} + service := newTestService(t, objects, metadata) + identity, object := validServiceInput(t) + + _, err := service.FinalizeObject( + t.Context(), identity, parser.AgentCodex, object.Objects[0], bytes.NewBufferString("body"), + ) + assert.ErrorIs(t, err, putErr) + assert.Equal(t, []string{"put-object"}, events) + }) + + t.Run("metadata failure leaves an unreferenced immutable object", func(t *testing.T) { + t.Parallel() + events := []string{} + metadataErr := errors.New("metadata backend unavailable") + objects := &recordingObjectStore{events: &events} + metadata := &recordingMetadataStore{events: &events, recordErr: metadataErr} + service := newTestService(t, objects, metadata) + identity, manifest := validServiceInput(t) + + _, err := service.FinalizeObject( + t.Context(), identity, parser.AgentCodex, manifest.Objects[0], bytes.NewBufferString("body"), + ) + assert.ErrorIs(t, err, metadataErr) + assert.Equal(t, []string{"put-object", "record-object"}, events) + }) +} + +func TestServiceRejectsExcludedProvidersBeforeCustody(t *testing.T) { + t.Parallel() + + for _, provider := range []parser.AgentType{"", "not-an-agent", parser.AgentOmnigent, parser.AgentTrae} { + t.Run(string(provider), func(t *testing.T) { + t.Parallel() + events := []string{} + objects := &recordingObjectStore{events: &events} + service := newTestService(t, objects, &recordingMetadataStore{events: &events}) + identity, manifest := validServiceInput(t) + + _, err := service.FinalizeObject( + t.Context(), identity, provider, manifest.Objects[0], bytes.NewBufferString("body"), + ) + assert.ErrorIs(t, err, ErrInvalid) + _, err = service.MissingObjects(t.Context(), identity, provider, manifest.Objects) + assert.ErrorIs(t, err, ErrInvalid) + assert.Empty(t, events, "excluded providers must never reach custody") + }) + } +} + +func TestServiceRejectsObjectsLargerThanAnyManifestFile(t *testing.T) { + t.Parallel() + + events := []string{} + objects := &recordingObjectStore{events: &events} + service := newTestService(t, objects, &recordingMetadataStore{events: &events}) + identity, _ := validServiceInput(t) + oversized := ObjectRef{ + SHA256: strings.Repeat("c", 64), Length: DefaultManifestLimits().MaxFileBytes + 1, + } + + _, err := service.FinalizeObject( + t.Context(), identity, parser.AgentCodex, oversized, bytes.NewBufferString("body"), + ) + assert.ErrorIs(t, err, ErrInvalid) + _, err = service.MissingObjects(t.Context(), identity, parser.AgentCodex, []ObjectRef{oversized}) + assert.ErrorIs(t, err, ErrInvalid) + assert.Empty(t, events, "unreferenceable objects must never reach custody") +} + +func TestServiceMissingObjectsUsesPhysicalCustody(t *testing.T) { + t.Parallel() + + events := []string{} + identity, manifest := validServiceInput(t) + objects := &recordingObjectStore{ + events: &events, + missing: []ObjectRef{manifest.Objects[0]}, + } + service := newTestService(t, objects, &recordingMetadataStore{events: &events}) + + missing, err := service.MissingObjects(t.Context(), identity, parser.AgentCodex, manifest.Objects) + require.NoError(t, err) + assert.Equal(t, manifest.Objects, missing) + assert.Equal(t, []string{"missing-objects"}, events) + assert.Equal(t, identity.TenantID, objects.missingTenant) +} + +func TestServiceCommitVerifiesEveryObjectBeforeMetadataCommit(t *testing.T) { + t.Parallel() + + events := []string{} + identity, manifest := validServiceInput(t) + objects := &recordingObjectStore{events: &events} + metadata := &recordingMetadataStore{ + events: &events, + commitResult: CommitResult{ + ManifestID: manifest.ManifestID, Receipt: serviceObject("receipt").SHA256, + Generation: 1, Created: true, + }, + } + service := newTestService(t, objects, metadata) + + result, err := service.CommitManifest(t.Context(), identity, manifest.Manifest) + require.NoError(t, err) + assert.NotEmpty(t, result.Receipt) + assert.Equal(t, []string{ + "verify-objects", "record-objects", "put-manifest", "commit", + }, events) + assert.Equal(t, manifest.Objects, objects.verifiedObjects) + require.Len(t, metadata.committed, 1) + assert.Equal(t, manifest.ManifestID, metadata.committed[0].ManifestID) + assert.Equal(t, manifest.CanonicalJSON, metadata.committed[0].CanonicalJSON) + assert.Equal(t, "parser-data-17", metadata.processingVersion) +} + +func TestServiceCommitUsesBoundedCallsAtMaximumObjectCardinality(t *testing.T) { + t.Parallel() + + identity, err := NewAuthIdentity("tenant-a", "device-a") + require.NoError(t, err) + limits := DefaultManifestLimits() + limits.MaxCanonicalBytes = 4 << 20 + objects := make([]ObjectRef, 0, limits.MaxObjects) + for i := range limits.MaxObjects { + digest := sha256.Sum256([]byte{byte(i >> 8), byte(i)}) + objects = append(objects, ObjectRef{ + SHA256: hex.EncodeToString(digest[:]), Length: 1, + }) + } + manifest := Manifest{ + SchemaVersion: ManifestSchemaVersion, + Provider: parser.AgentCodex, + ConfiguredRootID: "root-a", + SourceKey: "sessions/maximum.jsonl", + CaptureID: "capture-a", + CapturedAt: time.Date(2026, 8, 13, 12, 0, 0, 0, time.UTC), + Kind: ManifestSnapshot, + Entries: []Entry{{ + Path: "session.jsonl", Type: "file", + Length: int64(len(objects)), Objects: objects, + }}, + } + canonical, err := ValidateAndCanonicalize(identity, manifest, limits) + require.NoError(t, err) + events := []string{} + physical := &recordingObjectStore{events: &events} + metadata := &recordingMetadataStore{events: &events} + service, err := NewService(physical, metadata, limits, "parser-data-17") + require.NoError(t, err) + + _, err = service.CommitManifest(t.Context(), identity, manifest) + require.NoError(t, err) + assert.Equal(t, []string{ + "verify-objects", "record-objects", "put-manifest", "commit", + }, events) + assert.Equal(t, canonical.Objects, physical.verifiedObjects) + assert.Len(t, metadata.recordedObjects, limits.MaxObjects) +} + +func TestServiceCommitRejectsBeforeAcceptanceBoundary(t *testing.T) { + t.Parallel() + + t.Run("invalid manifest is rejected before custody access", func(t *testing.T) { + t.Parallel() + events := []string{} + identity, manifest := validServiceInput(t) + manifest.Manifest.Entries = nil + service := newTestService( + t, &recordingObjectStore{events: &events}, &recordingMetadataStore{events: &events}, + ) + + _, err := service.CommitManifest(t.Context(), identity, manifest.Manifest) + assert.ErrorIs(t, err, ErrInvalid) + assert.Empty(t, events) + }) + + t.Run("missing object never finalizes or commits manifest", func(t *testing.T) { + t.Parallel() + events := []string{} + identity, manifest := validServiceInput(t) + objects := &recordingObjectStore{events: &events, verifyErr: ErrMissingObject} + service := newTestService(t, objects, &recordingMetadataStore{events: &events}) + + _, err := service.CommitManifest(t.Context(), identity, manifest.Manifest) + assert.ErrorIs(t, err, ErrMissingObject) + assert.Equal(t, []string{"verify-objects"}, events) + }) + + t.Run("physical verification conflict never registers object", func(t *testing.T) { + t.Parallel() + events := []string{} + identity, manifest := validServiceInput(t) + objects := &recordingObjectStore{events: &events, verifyErr: ErrConflict} + service := newTestService(t, objects, &recordingMetadataStore{events: &events}) + + _, err := service.CommitManifest(t.Context(), identity, manifest.Manifest) + assert.ErrorIs(t, err, ErrConflict) + assert.Equal(t, []string{"verify-objects"}, events) + }) + + t.Run("physical verification failure never registers object", func(t *testing.T) { + t.Parallel() + events := []string{} + identity, manifest := validServiceInput(t) + verifyErr := errors.New("custody backend unavailable") + objects := &recordingObjectStore{events: &events, verifyErr: verifyErr} + service := newTestService(t, objects, &recordingMetadataStore{events: &events}) + + _, err := service.CommitManifest(t.Context(), identity, manifest.Manifest) + assert.ErrorIs(t, err, verifyErr) + assert.Equal(t, []string{"verify-objects"}, events) + }) + + t.Run("manifest custody failure never commits metadata", func(t *testing.T) { + t.Parallel() + events := []string{} + identity, manifest := validServiceInput(t) + putErr := errors.New("manifest backend unavailable") + objects := &recordingObjectStore{events: &events, putManifestErr: putErr} + service := newTestService(t, objects, &recordingMetadataStore{events: &events}) + + _, err := service.CommitManifest(t.Context(), identity, manifest.Manifest) + assert.ErrorIs(t, err, putErr) + assert.Equal(t, []string{ + "verify-objects", "record-objects", "put-manifest", + }, events) + }) + + t.Run("manifest semantic mismatch never commits metadata", func(t *testing.T) { + t.Parallel() + events := []string{} + identity, manifest := validServiceInput(t) + objects := &recordingObjectStore{ + events: &events, + putManifestResult: PutResult{ + Info: ObjectInfo{Ref: serviceObject("different manifest")}, + }, + } + service := newTestService(t, objects, &recordingMetadataStore{events: &events}) + + _, err := service.CommitManifest(t.Context(), identity, manifest.Manifest) + assert.ErrorIs(t, err, ErrConflict) + assert.Equal(t, []string{ + "verify-objects", "record-objects", "put-manifest", + }, events) + }) +} + +func TestServiceCommitPreservesMetadataConflictAfterManifestCustody(t *testing.T) { + t.Parallel() + + events := []string{} + identity, manifest := validServiceInput(t) + conflict := &HeadConflictError{CurrentGeneration: 4} + objects := &recordingObjectStore{events: &events} + metadata := &recordingMetadataStore{events: &events, commitErr: conflict} + service := newTestService(t, objects, metadata) + + _, err := service.CommitManifest(t.Context(), identity, manifest.Manifest) + assert.ErrorIs(t, err, ErrConflict) + var headConflict *HeadConflictError + assert.ErrorAs(t, err, &headConflict) + assert.Equal(t, int64(4), headConflict.CurrentGeneration) + assert.Equal(t, []string{ + "verify-objects", "record-objects", "put-manifest", "commit", + }, events) +} + +func TestServiceCommitStopsBetweenBoundedObjectChecks(t *testing.T) { + t.Parallel() + + events := []string{} + identity, manifest := validServiceInput(t) + second := serviceObject("second") + manifest.Manifest.Entries[0].Objects = append(manifest.Manifest.Entries[0].Objects, second) + manifest.Manifest.Entries[0].Length += second.Length + canonical, err := ValidateAndCanonicalize(identity, manifest.Manifest, DefaultManifestLimits()) + require.NoError(t, err) + ctx, cancel := context.WithCancel(t.Context()) + objects := &recordingObjectStore{events: &events} + metadata := &recordingMetadataStore{events: &events, afterRecordObjects: cancel} + service := newTestService(t, objects, metadata) + + _, err = service.CommitManifest(ctx, identity, canonical.Manifest) + assert.ErrorIs(t, err, context.Canceled) + assert.Equal(t, []string{"verify-objects", "record-objects"}, events) +} + +func newTestService( + t *testing.T, + objects ObjectStore, + metadata MetadataStore, +) *Service { + t.Helper() + service, err := NewService(objects, metadata, DefaultManifestLimits(), "parser-data-17") + require.NoError(t, err) + return service +} + +func validServiceInput(t *testing.T) (AuthIdentity, CanonicalManifest) { + t.Helper() + identity, err := NewAuthIdentity("tenant-a", "device-a") + require.NoError(t, err) + object := serviceObject("body") + manifest, err := ValidateAndCanonicalize(identity, Manifest{ + SchemaVersion: ManifestSchemaVersion, + Provider: parser.AgentCodex, + ConfiguredRootID: "root-a", + SourceKey: "sessions/session.jsonl", + CaptureID: "capture-a", + CapturedAt: time.Date(2026, 8, 13, 12, 0, 0, 0, time.UTC), + Kind: ManifestSnapshot, + Entries: []Entry{{ + Path: "session.jsonl", Type: "file", Length: object.Length, + Objects: []ObjectRef{object}, + }}, + }, DefaultManifestLimits()) + require.NoError(t, err) + return identity, manifest +} + +func serviceObject(body string) ObjectRef { + digest := sha256.Sum256([]byte(body)) + return ObjectRef{SHA256: hex.EncodeToString(digest[:]), Length: int64(len(body))} +} + +type recordingObjectStore struct { + events *[]string + putObjectResult PutResult + putObjectErr error + putObjectTenant string + statInfo ObjectInfo + statErr error + missing []ObjectRef + missingErr error + missingTenant string + verifyErr error + verifiedObjects []ObjectRef + putManifestResult PutResult + putManifestErr error +} + +func (s *recordingObjectStore) VerifyObjects( + _ context.Context, + _ string, + objects []ObjectRef, +) error { + s.event("verify-objects") + s.verifiedObjects = append([]ObjectRef(nil), objects...) + return s.verifyErr +} + +func (s *recordingObjectStore) event(value string) { + if s.events != nil { + *s.events = append(*s.events, value) + } +} + +func (s *recordingObjectStore) PutObject( + _ context.Context, + tenantID string, + object ObjectRef, + _ io.Reader, +) (PutResult, error) { + s.event("put-object") + s.putObjectTenant = tenantID + if s.putObjectResult.Info.Ref == (ObjectRef{}) && s.putObjectErr == nil { + s.putObjectResult.Info.Ref = object + } + return s.putObjectResult, s.putObjectErr +} + +func (s *recordingObjectStore) StatObject( + _ context.Context, + _ string, + object ObjectRef, +) (ObjectInfo, error) { + s.event("stat-object") + if s.statInfo == (ObjectInfo{}) { + return ObjectInfo{Ref: object}, s.statErr + } + return s.statInfo, s.statErr +} + +func (s *recordingObjectStore) OpenObject( + context.Context, + string, + ObjectRef, +) (ObjectInfo, VerifiedObjectReader, error) { + return ObjectInfo{}, nil, errors.New("unexpected OpenObject call") +} + +func (s *recordingObjectStore) MissingObjects( + _ context.Context, + tenantID string, + _ []ObjectRef, +) ([]ObjectRef, error) { + s.event("missing-objects") + s.missingTenant = tenantID + return append([]ObjectRef(nil), s.missing...), s.missingErr +} + +func (s *recordingObjectStore) PutManifest( + _ context.Context, + manifest CanonicalManifest, +) (PutResult, error) { + s.event("put-manifest") + if s.putManifestResult.Info.Ref == (ObjectRef{}) && s.putManifestErr == nil { + s.putManifestResult.Info.Ref = ObjectRef{ + SHA256: manifest.ManifestID, Length: int64(len(manifest.CanonicalJSON)), + } + } + return s.putManifestResult, s.putManifestErr +} + +func (s *recordingObjectStore) OpenManifest( + context.Context, + AuthIdentity, + string, +) (ObjectInfo, VerifiedObjectReader, error) { + return ObjectInfo{}, nil, errors.New("unexpected OpenManifest call") +} + +type recordingMetadataStore struct { + events *[]string + recordErr error + recordedObjects []ObjectRef + afterRecord func() + afterRecordObjects func() + commitResult CommitResult + commitErr error + committed []CanonicalManifest + processingVersion string +} + +func (s *recordingMetadataStore) RecordVerifiedObjects( + _ context.Context, + _ AuthIdentity, + objects []ObjectRef, +) error { + s.event("record-objects") + s.recordedObjects = append(s.recordedObjects, objects...) + if s.afterRecordObjects != nil { + s.afterRecordObjects() + } + return s.recordErr +} + +func (s *recordingMetadataStore) event(value string) { + if s.events != nil { + *s.events = append(*s.events, value) + } +} + +func (s *recordingMetadataStore) RecordVerifiedObject( + _ context.Context, + _ AuthIdentity, + object ObjectRef, +) error { + s.event("record-object") + s.recordedObjects = append(s.recordedObjects, object) + if s.afterRecord != nil { + s.afterRecord() + } + return s.recordErr +} + +func (s *recordingMetadataStore) MissingObjects( + context.Context, + AuthIdentity, + []ObjectRef, +) ([]ObjectRef, error) { + return nil, errors.New("metadata MissingObjects must not be called") +} + +func (s *recordingMetadataStore) CommitManifest( + _ context.Context, + manifest CanonicalManifest, + processingVersion string, +) (CommitResult, error) { + s.event("commit") + s.committed = append(s.committed, manifest) + s.processingVersion = processingVersion + return s.commitResult, s.commitErr +}