package catalogsnapshot

import (
	"bytes"
	"context"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"

	"github.com/nexuscrm/shared/pkg/constants"
	sharedcontract "github.com/nexuscrm/shared/pkg/contract"
	"github.com/nexuscrm/shared/pkg/models"
)

const (
	testRevision   CatalogRevision = 41
	testInstallA                   = "01900000-0000-7000-8000-000000000001"
	testInstallB                   = "01900000-0000-7000-8000-000000000002"
	testCapability                 = "runtime.records.read-selected-root.v1"
)

func TestCatalogSnapshotDeterminismAndSemanticDigest(t *testing.T) {
	input := neutralCompileInput()
	first := mustCompile(t, input)

	reordered := neutralCompileInput()
	reverseObjects(reordered.Objects)
	for i := range reordered.Objects {
		reverseFields(reordered.Objects[i].Metadata.Fields)
		for fieldIndex := range reordered.Objects[i].Metadata.Fields {
			field := &reordered.Objects[i].Metadata.Fields[fieldIndex]
			if field.APIName == "WorkspaceId" {
				field.ReferenceTo = append(field.ReferenceTo, "workspace")
			}
		}
	}
	reverseRelationships(reordered.Relationships)
	reverseRules(reordered.SharingRules)
	reverseDefinitions(reordered.Definitions)
	reverseProjections(reordered.Projections)
	for i := range reordered.Projections {
		reverseStrings(reordered.Projections[i].ObjectAPINames)
		reverseStrings(reordered.Projections[i].DefinitionComponentIDs)
	}
	second := mustCompile(t, reordered)
	if !bytes.Equal(first.Bytes(), second.Bytes()) || first.Digest() != second.Digest() {
		t.Fatal("semantic input ordering changed canonical bytes or digest")
	}

	changed := neutralCompileInput()
	changed.Objects[0].Metadata.Fields[1].PhysicalName = "subject_v2"
	third := mustCompile(t, changed)
	if third.Digest() == first.Digest() || bytes.Equal(third.Bytes(), first.Bytes()) {
		t.Fatal("behavior-affecting physical-name change did not change content address")
	}

	// Compiler ownership is deep: later caller mutation cannot alter the
	// already-published artifact.
	input.Objects[0].Metadata.Fields[0].APIName = "Mutated"
	input.Projections[0].ObjectAPINames[0] = "Mutated"
	if !bytes.Equal(first.Bytes(), mustCompile(t, neutralCompileInput()).Bytes()) {
		t.Fatal("artifact changed after input mutation")
	}
}

func TestRuntimeMetadataViewProjectionIsScopedImmutableAndRevisionBound(t *testing.T) {
	snapshotArtifact := mustCompile(t, neutralCompileInput())
	snapshot := mustOpenSnapshot(t, snapshotArtifact)
	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testCapability,
	})
	if err != nil {
		t.Fatalf("project Runtime view: %v", err)
	}
	view := mustOpenView(t, viewArtifact)
	if err := view.VerifySource(snapshot); err != nil {
		t.Fatalf("verify source: %v", err)
	}
	if err := view.VerifyCompatibility(Compatibility{SchemaGeneration: 12, EngineGeneration: 5}); err != nil {
		t.Fatalf("verify exact compatibility: %v", err)
	}
	if err := view.VerifyCompatibility(Compatibility{SchemaGeneration: 13, EngineGeneration: 5}); err != nil {
		t.Fatalf("newer physical schema should satisfy the snapshot floor: %v", err)
	}
	if err := view.VerifyCompatibility(Compatibility{SchemaGeneration: 11, EngineGeneration: 5}); !errors.Is(err, ErrIncompatible) {
		t.Fatalf("old physical schema error = %v, want incompatibility", err)
	}
	if err := view.VerifyCompatibility(Compatibility{SchemaGeneration: 12, EngineGeneration: 6}); !errors.Is(err, ErrIncompatible) {
		t.Fatalf("different engine generation error = %v, want incompatibility", err)
	}
	if view.Lineage().CatalogRevision != testRevision ||
		view.Lineage().SchemaGeneration != SchemaGeneration(12) ||
		view.Lineage().EngineGeneration != EngineGeneration(5) {
		t.Fatalf("lineage axes collapsed or changed: %#v", view.Lineage())
	}
	if view.AppInstallationID() != testInstallA || view.CapabilityID() != testCapability {
		t.Fatalf("projection scope = %q/%q", view.AppInstallationID(), view.CapabilityID())
	}
	for _, forbiddenKey := range [][]byte{
		[]byte(`"label":`), []byte(`"pluralLabel":`), []byte(`"createable":`),
		[]byte(`"updateable":`), []byte(`"defaultedOnCreate":`),
		[]byte(`"writableReferenceTo":`), []byte(`"childRelationshipName":`),
		[]byte(`"id":`), []byte(`"name":`), []byte(`"relationshipName":`),
		[]byte(`"cascadeDelete":`),
		[]byte(`"restrictedDelete":`), []byte(`"relatedListLabel":`),
		[]byte(`"relatedListFields":`),
	} {
		if bytes.Contains(viewArtifact.Bytes(), forbiddenKey) {
			t.Fatalf("least-privilege Runtime view contains unrelated key %s", forbiddenKey)
		}
	}
	if _, ok := view.Object("Unrelated"); ok {
		t.Fatal("view leaked an object outside the AppInstallation/capability scope")
	}
	if _, _, ok := view.Definition("definition.unrelated"); ok {
		t.Fatal("selected-root view leaked an unrelated executable definition")
	}
	workItem, ok := view.Object("workitem")
	if !ok {
		t.Fatal("case-folded object lookup failed")
	}
	if workItem.FieldsMap["amount"].PhysicalName != "amount_exact" || workItem.FieldsMap["amount"].Scale == nil || *workItem.FieldsMap["amount"].Scale != 4 {
		t.Fatalf("exact/physical field projection lost metadata: %#v", workItem.FieldsMap["amount"])
	}
	workspace, ok := view.Object("Workspace")
	if !ok || workspace.FieldsMap["computedname"].Formula == nil ||
		workspace.FieldsMap["computedname"].FormulaMode == nil ||
		*workspace.FieldsMap["computedname"].FormulaMode != "active" {
		t.Fatalf("request-dependent formula routing facts were not preserved: %#v", workspace)
	}
	if workItem.AccessPolicy == nil || workItem.AccessPolicy.ObjectRead == nil || workItem.AccessPolicy.RecordMutation != nil {
		t.Fatalf("read projection is incomplete or includes mutation policy: %#v", workItem.AccessPolicy)
	}
	if _, ok := view.SharingParentRelationship("WorkItem"); !ok {
		t.Fatal("sharing-parent relationship was not projected")
	}
	if got := view.RelationshipsForChild("WorkItem"); len(got) != 1 || !got[0].IsSharingParent {
		t.Fatalf("ordinary lookup relationships entered the selected-root Runtime view: %#v", got)
	}
	if got := view.SharingRulesForObject("WorkItem"); len(got) != 1 || got[0].ID != "" || got[0].Name != "" ||
		got[0].ShareWithGroupID == nil || *got[0].ShareWithGroupID != "group-readers" {
		t.Fatalf("sharing rules = %#v", got)
	}

	workItem.APIName = "Changed"
	workItem.Fields[0].APIName = "Changed"
	workItem.FieldsMap["amount"].PhysicalName = "changed"
	again, ok := view.Object("WorkItem")
	if !ok || again.APIName != "WorkItem" || again.FieldsMap["amount"].PhysicalName != "amount_exact" {
		t.Fatal("mutable getter result changed the immutable view")
	}

	otherArtifact := mustCompile(t, withRevision(neutralCompileInput(), 42))
	otherSnapshot := mustOpenSnapshot(t, otherArtifact)
	if err := view.VerifySource(otherSnapshot); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("mismatched source error = %v, want corruption", err)
	}
}

func TestRecordMutationProjectionCompilesAssignmentPolicyPhysicalReference(t *testing.T) {
	input := neutralCompileInput()
	workItem := input.Objects[0].Metadata
	workspace := input.Objects[1].Metadata
	workspace.ReadSource = &models.ObjectReadSourceConfig{
		Kind: "physical_table", PhysicalTable: "workspace_rows",
	}
	workItem.AccessPolicy.Lifecycle = &models.RecordLifecycleAccessPolicy{
		Hooks: []models.RecordLifecycleHookPolicy{
			{Phase: string(sharedcontract.RecordLifecycleAfterCreateInTx), Capability: sharedcontract.AssignmentPolicyEligibilityLifecycleCapability},
			{Phase: string(sharedcontract.RecordLifecycleAfterUpdateInTx), Capability: sharedcontract.AssignmentPolicyEligibilityLifecycleCapability},
			{Phase: string(sharedcontract.RecordLifecycleAfterDeleteInTx), Capability: sharedcontract.AssignmentPolicyEligibilityLifecycleCapability},
		},
		AssignmentPolicyEligibility: &models.AssignmentPolicyEligibilityLifecyclePolicy{
			Role:           models.AssignmentPolicyEligibilitySourceRole(" SUBJECT_POPULATION "),
			SubjectIDField: "Id",
			EnabledWhen: []models.AssignmentPolicyEligibilityBoolCondition{
				{Field: "Done", Equals: false},
			},
			Selector: models.AssignmentPolicyEligibilitySelector{Reference: &models.AssignmentPolicyEligibilityReferenceSelector{
				SourceField: "WorkspaceId", TargetObject: "Workspace", TargetField: "Name",
			}},
		},
	}
	workspace.AccessPolicy = &models.ObjectAccessPolicy{Lifecycle: &models.RecordLifecycleAccessPolicy{
		Hooks: []models.RecordLifecycleHookPolicy{
			{Phase: string(sharedcontract.RecordLifecycleAfterCreateInTx), Capability: sharedcontract.AssignmentPolicyEligibilityLifecycleCapability},
			{Phase: string(sharedcontract.RecordLifecycleAfterUpdateInTx), Capability: sharedcontract.AssignmentPolicyEligibilityLifecycleCapability},
			{Phase: string(sharedcontract.RecordLifecycleAfterDeleteInTx), Capability: sharedcontract.AssignmentPolicyEligibilityLifecycleCapability},
		},
		AssignmentPolicyEligibility: &models.AssignmentPolicyEligibilityLifecyclePolicy{
			Role:        models.AssignmentPolicyEligibilitySourceRole("SELECTOR_AUTHORITY"),
			EnabledWhen: []models.AssignmentPolicyEligibilityBoolCondition{{Field: "Active", Equals: true}},
			Selector:    models.AssignmentPolicyEligibilitySelector{Field: "Name"},
		},
	}}
	workspace.Fields = append(workspace.Fields, models.FieldMetadata{
		ID: "field-workspace-active", APIName: "Active", Label: "Active", Type: constants.FieldTypeBoolean, PhysicalName: "is_active",
	})
	input.Projections = []RuntimeProjectionInput{{
		CatalogRevision:   testRevision,
		AppInstallationID: testInstallA,
		CapabilityID:      "runtime.records.mutation.v1",
		Profile:           ProjectionRecordMutationV1,
		ObjectAPINames:    []string{"Unrelated", "Workspace", "WorkItem"},
	}}

	snapshot := mustOpenSnapshot(t, mustCompile(t, input))
	artifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: "runtime.records.mutation.v1",
	})
	if err != nil {
		t.Fatal(err)
	}
	view := mustOpenView(t, artifact)
	projected, ok := view.Object("WorkItem")
	if !ok || projected.AccessPolicy == nil || projected.AccessPolicy.Lifecycle == nil ||
		projected.AccessPolicy.Lifecycle.AssignmentPolicyEligibility == nil {
		t.Fatalf("projected assignment-policy lifecycle = %#v", projected.AccessPolicy)
	}
	reference := projected.AccessPolicy.Lifecycle.AssignmentPolicyEligibility.Selector.Reference
	eligibility := projected.AccessPolicy.Lifecycle.AssignmentPolicyEligibility
	if eligibility.ResolvedSourceTable != "work_items" ||
		eligibility.Role != models.AssignmentPolicyEligibilitySubjectPopulation ||
		eligibility.ResolvedSourceIDColumn != "id" ||
		eligibility.ResolvedSubjectIDColumn != "id" ||
		eligibility.EnabledWhen[0].ResolvedColumn != "is_done" ||
		reference == nil || reference.ResolvedSourceFieldColumn != "workspace_id" ||
		reference.ResolvedTargetTable != "workspace_rows" ||
		reference.ResolvedTargetIDColumn != "id" ||
		reference.ResolvedTargetFieldColumn != "display_name" {
		t.Fatalf("compiled assignment-policy storage = eligibility=%#v reference=%#v", eligibility, reference)
	}
	authority, ok := view.Object("Workspace")
	if !ok || authority.AccessPolicy == nil || authority.AccessPolicy.Lifecycle == nil ||
		authority.AccessPolicy.Lifecycle.AssignmentPolicyEligibility == nil {
		t.Fatalf("projected selector authority = %#v", authority.AccessPolicy)
	}
	authorityEligibility := authority.AccessPolicy.Lifecycle.AssignmentPolicyEligibility
	if authorityEligibility.ResolvedSourceTable != "workspace_rows" ||
		authorityEligibility.Role != models.AssignmentPolicyEligibilitySelectorAuthority ||
		authorityEligibility.ResolvedSourceIDColumn != "id" ||
		authorityEligibility.EnabledWhen[0].ResolvedColumn != "is_active" ||
		authorityEligibility.Selector.ResolvedFieldColumn != "display_name" {
		t.Fatalf("compiled selector authority storage = %#v", authorityEligibility)
	}
}

func TestRecordMutationProjectionRejectsPartialAssignmentPolicyRoleSet(t *testing.T) {
	input := neutralCompileInput()
	workspace := input.Objects[1].Metadata
	workspace.AccessPolicy = &models.ObjectAccessPolicy{Lifecycle: &models.RecordLifecycleAccessPolicy{
		AssignmentPolicyEligibility: &models.AssignmentPolicyEligibilityLifecyclePolicy{
			Role:        models.AssignmentPolicyEligibilitySelectorAuthority,
			EnabledWhen: []models.AssignmentPolicyEligibilityBoolCondition{{Field: "Active", Equals: true}},
			Selector:    models.AssignmentPolicyEligibilitySelector{Field: "Name"},
		},
	}}
	workspace.Fields = append(workspace.Fields, models.FieldMetadata{
		ID: "field-workspace-active", APIName: "Active", Label: "Active", Type: constants.FieldTypeBoolean, PhysicalName: "is_active",
	})
	input.Projections = []RuntimeProjectionInput{{
		CatalogRevision: testRevision, AppInstallationID: testInstallA,
		CapabilityID: "runtime.records.mutation.v1", Profile: ProjectionRecordMutationV1,
		ObjectAPINames: []string{"Unrelated", "Workspace", "WorkItem"},
	}}
	snapshot := mustOpenSnapshot(t, mustCompile(t, input))
	_, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: "runtime.records.mutation.v1",
	})
	if err == nil || !strings.Contains(err.Error(), "exactly one selector authority and at least one subject population") {
		t.Fatalf("partial assignment-policy role set error = %v", err)
	}
}

func TestRecordQueryProjectionCarriesCompleteRelationshipClosureWithoutCatalogPresentation(t *testing.T) {
	input := neutralCompileInput()
	input.Projections = []RuntimeProjectionInput{{
		CatalogRevision: testRevision, AppInstallationID: testInstallA,
		CapabilityID: "runtime.records.query.v1", Profile: ProjectionRecordQueryV1,
		ObjectAPINames: []string{"Unrelated", "WorkItem", "Workspace"},
	}}
	snapshot := mustOpenSnapshot(t, mustCompile(t, input))
	artifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: "runtime.records.query.v1",
	})
	if err != nil {
		t.Fatal(err)
	}
	view := mustOpenView(t, artifact)
	relationships := view.RelationshipsForChild("workitem")
	if len(relationships) != 2 || relationships[0].RelationshipName == "" || relationships[1].RelationshipName == "" {
		t.Fatalf("query relationships=%#v", relationships)
	}
	if got := view.RelationshipsForParent("Workspace"); len(got) != 1 || got[0].ChildObjectAPIName != "WorkItem" {
		t.Fatalf("children of Workspace=%#v", got)
	}
	workItem, ok := view.Object("WorkItem")
	var workspaceField *models.FieldMetadata
	if ok {
		workspaceField = workItem.FieldsMap["workspaceid"]
	}
	if workspaceField == nil || workspaceField.ChildRelationshipName == nil ||
		*workspaceField.ChildRelationshipName != "WorkItems" {
		t.Fatalf("query child relationship authority=%#v", workItem)
	}
	if bytes.Contains(artifact.Bytes(), []byte(`"relatedListLabel"`)) ||
		bytes.Contains(artifact.Bytes(), []byte(`"relatedListFields"`)) ||
		bytes.Contains(artifact.Bytes(), []byte(`"cascadeDelete"`)) {
		t.Fatalf("Catalog presentation or mutation facts entered record-query view: %s", artifact.Bytes())
	}
	if report := CompareRuntimeView(view, parityInputForProjection(input, []string{"Unrelated", "WorkItem", "Workspace"})); !report.Equivalent() {
		t.Fatalf("record-query parity=%#v", report.Differences)
	}

	changed := neutralCompileInput()
	changed.Projections = input.Projections
	changed.Relationships[1].Metadata.RelationshipName = "DifferentQueryPath"
	changedView, err := ProjectRuntimeMetadataView(mustOpenSnapshot(t, mustCompile(t, changed)), ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: "runtime.records.query.v1",
	})
	if err != nil {
		t.Fatal(err)
	}
	if changedView.Digest() == artifact.Digest() {
		t.Fatal("query relationship behavior changed without changing the view digest")
	}
}

func TestParentControlledObjectWithoutSharingParentRemainsLocallyFailClosed(t *testing.T) {
	input := neutralCompileInput()
	malformed := &models.ObjectMetadata{
		ID: "obj-malformed-child", APIName: "MalformedChild", Label: "Malformed Child", PluralLabel: "Malformed Children",
		SharingModel: constants.SharingModelControlledByParent,
		Fields: []models.FieldMetadata{
			{ID: "field-malformed-id", APIName: "Id", Label: "ID", Type: constants.FieldTypeString, PhysicalName: "id", IsSystem: true},
			{ID: "field-malformed-owner", APIName: "OwnerId", Label: "Owner", Type: constants.FieldTypeReference, PhysicalName: "owner_id", IsSystem: true},
			{ID: "field-malformed-parent", APIName: "WorkspaceId", Label: "Workspace", Type: constants.FieldTypeReference, PhysicalName: "workspace_id", ReferenceTo: []string{"Workspace"}},
		},
	}
	input.Objects = append(input.Objects, ObjectInput{CatalogRevision: testRevision, Metadata: malformed})
	input.Projections = []RuntimeProjectionInput{{
		CatalogRevision: testRevision, AppInstallationID: testInstallA,
		CapabilityID: "runtime.records.evaluate-access.v1", Profile: ProjectionRecordAccessV1,
		ObjectAPINames: []string{"MalformedChild", "Unrelated", "WorkItem", "Workspace"},
	}}

	snapshot := mustOpenSnapshot(t, mustCompile(t, input))
	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: "runtime.records.evaluate-access.v1",
	})
	if err != nil {
		t.Fatalf("project locally malformed object: %v", err)
	}
	view := mustOpenView(t, viewArtifact)
	if object, ok := view.Object("MalformedChild"); !ok || models.RowVisibilityFor(object).Kind != models.RowVisibilityParentControlled {
		t.Fatalf("malformed object fact was not preserved: %#v, %v", object, ok)
	}
	if relationship, ok := view.SharingParentRelationship("MalformedChild"); ok || relationship != nil {
		t.Fatalf("missing sharing-parent fact was invented: %#v, %v", relationship, ok)
	}
	if report := CompareRuntimeView(view, parityInputForProjection(input, []string{"MalformedChild", "Unrelated", "WorkItem", "Workspace"})); !report.Equivalent() {
		t.Fatalf("locally malformed object changed parity: %#v", report.Differences)
	}
}

func TestRuntimeMetadataViewContentExcludesCatalogOnlyPresentationAndRemainsSourceBound(t *testing.T) {
	baselineSnapshot := mustOpenSnapshot(t, mustCompile(t, neutralCompileInput()))
	baseline, err := ProjectRuntimeMetadataView(baselineSnapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testCapability,
	})
	if err != nil {
		t.Fatalf("project baseline Runtime view: %v", err)
	}

	changed := neutralCompileInput()
	changed.Relationships[0].Metadata.ID = "different-catalog-row"
	changed.Relationships[0].Metadata.RelationshipName = "DifferentDisplayName"
	changed.Relationships[0].Metadata.CascadeDelete = true
	changed.Relationships[0].Metadata.RestrictedDelete = true
	changed.Relationships[0].Metadata.RelatedListLabel = stringPointer("Different label")
	changed.Relationships[0].Metadata.RelatedListFields = stringPointer("Name,OwnerId")
	changed.SharingRules[0].Metadata.ID = "different-rule-row"
	changed.SharingRules[0].Metadata.Name = "Different display name"
	changedSnapshotArtifact := mustCompile(t, changed)
	if changedSnapshotArtifact.Digest() == baselineSnapshot.Digest() {
		t.Fatal("complete Catalog snapshot ignored authoring changes")
	}
	changedView, err := ProjectRuntimeMetadataView(mustOpenSnapshot(t, changedSnapshotArtifact), ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testCapability,
	})
	if err != nil {
		t.Fatalf("project changed Runtime view: %v", err)
	}
	if changedView.Digest() == baseline.Digest() {
		t.Fatal("Runtime view lost its exact source-snapshot binding")
	}
	baselineDoc := mustOpenView(t, baseline).doc
	changedDoc := mustOpenView(t, changedView).doc
	baselineDoc.SourceSnapshotDigest = ""
	changedDoc.SourceSnapshotDigest = ""
	baselineContent, _ := marshalCanonical(baselineDoc)
	changedContent, _ := marshalCanonical(changedDoc)
	if !bytes.Equal(changedContent, baselineContent) {
		t.Fatal("Catalog-only relationship/rule identity or presentation entered least-privilege Runtime content")
	}
}

func TestCompilerFailsClosedOnMissingOrMixedRevision(t *testing.T) {
	tests := []struct {
		name   string
		mutate func(*CompileInput)
		kind   ConflictKind
	}{
		{"missing top-level revision", func(in *CompileInput) { in.Lineage.CatalogRevision = 0 }, ConflictMissingRevision},
		{"missing schema generation", func(in *CompileInput) { in.Lineage.SchemaGeneration = 0 }, ConflictMissingRevision},
		{"missing engine generation", func(in *CompileInput) { in.Lineage.EngineGeneration = 0 }, ConflictMissingRevision},
		{"mixed object revision", func(in *CompileInput) { in.Objects[1].CatalogRevision++ }, ConflictMixedRevision},
		{"mixed relationship revision", func(in *CompileInput) { in.Relationships[0].CatalogRevision++ }, ConflictMixedRevision},
		{"mixed rule revision", func(in *CompileInput) { in.SharingRules[0].CatalogRevision++ }, ConflictMixedRevision},
		{"mixed projection revision", func(in *CompileInput) { in.Projections[0].CatalogRevision++ }, ConflictMixedRevision},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			input := neutralCompileInput()
			tc.mutate(&input)
			_, err := CompileCatalogSnapshot(input)
			var conflict *ConflictError
			if !errors.As(err, &conflict) || conflict.Kind != tc.kind {
				t.Fatalf("error = %T %v, want conflict %s", err, err, tc.kind)
			}
		})
	}
}

func TestCatalogSnapshotDoesNotRequireA1cProjectionAuthority(t *testing.T) {
	input := neutralCompileInput()
	input.Projections = nil
	snapshot := mustOpenSnapshot(t, mustCompile(t, input))
	if keys := snapshot.ProjectionKeys(); len(keys) != 0 {
		t.Fatalf("projection keys = %#v, want none before A1c publication", keys)
	}
	_, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testCapability,
	})
	var conflict *ConflictError
	if !errors.As(err, &conflict) || conflict.Kind != ConflictInvalidProjection {
		t.Fatalf("missing A1c projection error = %T %v", err, err)
	}
}

func TestTargetRecordReadProjectionRequiresExplicitFullSameRevisionSchemaClosure(t *testing.T) {
	incomplete := neutralCompileInput()
	workItem := incomplete.Objects[0].Metadata
	if workItem.APIName != "WorkItem" {
		t.Fatalf("fixture object[0] = %q, want WorkItem", workItem.APIName)
	}
	workItem.AccessPolicy.RowRead = &models.RowReadPolicy{
		Kind: "targetRecordRead", TargetObjectField: "Visibility", TargetRecordField: "WorkspaceId",
	}
	_, err := CompileCatalogSnapshot(incomplete)
	var projectionConflict *ConflictError
	if !errors.As(err, &projectionConflict) || projectionConflict.Kind != ConflictInvalidProjection ||
		!strings.Contains(err.Error(), "Unrelated") {
		t.Fatalf("incomplete targetRecordRead closure error = %T %v", err, err)
	}

	complete := neutralCompileInput()
	complete.Objects[0].Metadata.AccessPolicy.RowRead = &models.RowReadPolicy{
		Kind: "targetRecordRead", TargetObjectField: "Visibility", TargetRecordField: "WorkspaceId",
	}
	complete.Projections[0].ObjectAPINames = append(complete.Projections[0].ObjectAPINames, "Unrelated")
	completeSnapshotArtifact := mustCompile(t, complete)
	snapshot := mustOpenSnapshot(t, completeSnapshotArtifact)
	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testCapability,
	})
	if err != nil {
		t.Fatalf("project complete targetRecordRead closure: %v", err)
	}
	view := mustOpenView(t, viewArtifact)
	if objects := view.Objects(); len(objects) != len(complete.Objects) {
		t.Fatalf("targetRecordRead view objects = %d, want complete closure %d", len(objects), len(complete.Objects))
	}
	if report := CompareRuntimeView(view, parityInputForProjection(complete, complete.Projections[0].ObjectAPINames)); !report.Equivalent() {
		t.Fatalf("targetRecordRead full-closure parity differences: %#v", report.Differences)
	}

	var tampered catalogDocument
	if err := decodeStrict(completeSnapshotArtifact.Bytes(), &tampered); err != nil {
		t.Fatalf("decode complete artifact for hostile-input test: %v", err)
	}
	tampered.Projections[0].ObjectAPINames = []string{"Workspace", "WorkItem"}
	tamperedBytes, err := marshalCanonical(tampered)
	if err != nil {
		t.Fatalf("encode hostile artifact: %v", err)
	}
	if _, err := OpenCatalogSnapshot(makeArtifact(ArtifactCatalogSnapshot, tamperedBytes)); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("validly digested incomplete targetRecordRead artifact error = %v, want corruption", err)
	}
}

func TestCatalogSnapshotValidatesGroupedTargetRecordReadPolicy(t *testing.T) {
	valid := neutralCompileInput()
	valid.Objects[0].Metadata.AccessPolicy.RowRead = &models.RowReadPolicy{
		Kind: "targetRecordRead",
		TargetAssociations: []models.RowReadTargetAssociation{
			{TargetObjectField: "Visibility", TargetRecordField: "WorkspaceId"},
		},
	}
	valid.Projections[0].ObjectAPINames = append(valid.Projections[0].ObjectAPINames, "Unrelated")
	if _, err := CompileCatalogSnapshot(valid); err != nil {
		t.Fatalf("grouped targetRecordRead policy rejected: %v", err)
	}

	mixed := neutralCompileInput()
	mixed.Objects[0].Metadata.AccessPolicy.RowRead = &models.RowReadPolicy{
		Kind:              "targetRecordRead",
		TargetObjectField: "Visibility",
		TargetRecordField: "WorkspaceId",
		TargetAssociations: []models.RowReadTargetAssociation{
			{TargetObjectField: "Visibility", TargetRecordField: "WorkspaceId"},
		},
	}
	mixed.Projections[0].ObjectAPINames = append(mixed.Projections[0].ObjectAPINames, "Unrelated")
	_, err := CompileCatalogSnapshot(mixed)
	var policyConflict *ConflictError
	if !errors.As(err, &policyConflict) || policyConflict.Kind != ConflictInvalidAccessPolicy ||
		!strings.Contains(err.Error(), "cannot mix") {
		t.Fatalf("mixed targetRecordRead shape error = %T %v, want invalid access policy", err, err)
	}
}

func TestCompilerClassifiesMalformedEffectiveMetadata(t *testing.T) {
	tests := []struct {
		name   string
		mutate func(*CompileInput)
		kind   ConflictKind
	}{
		{"duplicate object", func(in *CompileInput) {
			duplicate := *in.Objects[0].Metadata
			duplicate.APIName = "workitem"
			in.Objects = append(in.Objects, ObjectInput{CatalogRevision: testRevision, Metadata: &duplicate})
		}, ConflictDuplicateObject},
		{"invalid object API name", func(in *CompileInput) {
			in.Objects[0].Metadata.APIName = "Work/Item"
		}, ConflictInvalidIdentity},
		{"duplicate field", func(in *CompileInput) {
			duplicate := in.Objects[0].Metadata.Fields[0]
			duplicate.APIName = "id"
			in.Objects[0].Metadata.Fields = append(in.Objects[0].Metadata.Fields, duplicate)
		}, ConflictDuplicateField},
		{"invalid read source", func(in *CompileInput) {
			in.Objects[0].Metadata.ReadSource = &models.ObjectReadSourceConfig{Kind: "physical_table", PhysicalTable: "bad-name"}
		}, ConflictInvalidReadSource},
		{"missing relationship parent", func(in *CompileInput) { in.Relationships[0].Metadata.ParentObjectAPIName = "Missing" }, ConflictInvalidRelationship},
		{"ambiguous sharing parent", func(in *CompileInput) {
			duplicate := *in.Relationships[0].Metadata
			duplicate.ParentObjectAPIName = "Unrelated"
			duplicate.ID = "rel-2"
			in.Relationships = append(in.Relationships, RelationshipInput{CatalogRevision: testRevision, Metadata: &duplicate})
		}, ConflictAmbiguousSharingParent},
		{"deleted sharing rule", func(in *CompileInput) { in.SharingRules[0].Metadata.IsDeleted = true }, ConflictInvalidSharingRule},
		{"unverified criteria", func(in *CompileInput) { in.SharingRules[0].Metadata.Criteria = `State == "open"` }, ConflictInvalidSharingRule},
		{"invalid definition version", func(in *CompileInput) { in.Definitions[0].DefinitionVersion = "" }, ConflictInvalidDefinition},
		{"noncanonical definition digest", func(in *CompileInput) {
			in.Definitions[0].ContentDigest = Digest(strings.ToUpper(string(in.Definitions[0].ContentDigest)))
		}, ConflictInvalidDefinition},
		{"unknown projection object", func(in *CompileInput) {
			in.Projections[0].ObjectAPINames = append(in.Projections[0].ObjectAPINames, "Missing")
		}, ConflictInvalidProjection},
	}
	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			input := neutralCompileInput()
			tc.mutate(&input)
			_, err := CompileCatalogSnapshot(input)
			var conflict *ConflictError
			if !errors.As(err, &conflict) || conflict.Kind != tc.kind {
				t.Fatalf("error = %T %v, want conflict %s", err, err, tc.kind)
			}
		})
	}
}

func TestCatalogObjectAPINameGrammarFailsClosedAtArtifactBoundaries(t *testing.T) {
	artifact := mustCompile(t, neutralCompileInput())

	var snapshotDoc catalogDocument
	if err := decodeStrict(artifact.bytes, &snapshotDoc); err != nil {
		t.Fatal(err)
	}
	snapshotDoc.Objects[0].Metadata.APIName = "Work/Item"
	invalidSnapshotBytes, err := marshalCanonical(snapshotDoc)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenCatalogSnapshot(makeArtifact(ArtifactCatalogSnapshot, invalidSnapshotBytes)); !errors.Is(err, ErrArtifactCorrupt) || !strings.Contains(err.Error(), "ObjectAPIName must match") {
		t.Fatalf("invalid Catalog object API name error = %v, want corrupt ObjectKey v1 rejection", err)
	}

	snapshot := mustOpenSnapshot(t, artifact)
	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	var viewDoc runtimeViewDocument
	if err := decodeStrict(viewArtifact.bytes, &viewDoc); err != nil {
		t.Fatal(err)
	}
	viewDoc.Objects[0].APIName = "Work/Item"
	invalidViewBytes, err := marshalCanonical(viewDoc)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, invalidViewBytes)); !errors.Is(err, ErrArtifactCorrupt) || !strings.Contains(err.Error(), "ObjectAPIName must match") {
		t.Fatalf("invalid Runtime object API name error = %v, want corrupt ObjectKey v1 rejection", err)
	}
}

func TestArtifactCorruptionCanonicalityAndVersionFailClosed(t *testing.T) {
	artifact := mustCompile(t, neutralCompileInput())
	corrupt := artifact
	corrupt.bytes = append([]byte(nil), artifact.bytes...)
	corrupt.bytes[len(corrupt.bytes)-2] ^= 1
	if _, err := OpenCatalogSnapshot(corrupt); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("corrupt bytes error = %v, want corruption", err)
	}

	nonCanonical := artifact
	nonCanonical.bytes = append([]byte(" "), artifact.bytes...)
	nonCanonical.digest = ComputeDigest(nonCanonical.bytes)
	if _, err := OpenCatalogSnapshot(nonCanonical); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("non-canonical bytes error = %v, want corruption", err)
	}

	var doc catalogDocument
	if err := decodeStrict(artifact.bytes, &doc); err != nil {
		t.Fatal(err)
	}
	doc.Schema = "nexuscrm.catalog-snapshot/v999"
	unsupportedBytes, err := marshalCanonical(doc)
	if err != nil {
		t.Fatal(err)
	}
	unsupported := makeArtifact(ArtifactCatalogSnapshot, unsupportedBytes)
	if _, err := OpenCatalogSnapshot(unsupported); !errors.Is(err, ErrUnsupportedSchema) {
		t.Fatalf("unknown schema error = %v, want explicit version rejection", err)
	}

	snapshot := mustOpenSnapshot(t, artifact)
	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testCapability,
	})
	if err != nil {
		t.Fatal(err)
	}
	var viewDoc runtimeViewDocument
	if err := decodeStrict(viewArtifact.bytes, &viewDoc); err != nil {
		t.Fatal(err)
	}
	viewDoc.Objects[0].ResolvedBaseReadTable = "mismatched_table"
	mismatchedViewBytes, err := marshalCanonical(viewDoc)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, mismatchedViewBytes)); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("mismatched base read table error = %v, want corruption", err)
	}
	if err := decodeStrict(viewArtifact.bytes, &viewDoc); err != nil {
		t.Fatal(err)
	}
	viewDoc.Schema = "nexuscrm.runtime-metadata-view/v999"
	unsupportedViewBytes, err := marshalCanonical(viewDoc)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, unsupportedViewBytes)); !errors.Is(err, ErrUnsupportedSchema) {
		t.Fatalf("unknown Runtime view schema error = %v, want explicit version rejection", err)
	}

	wrongKind := artifact
	wrongKind.kind = ArtifactRuntimeMetadataView
	if _, err := OpenRuntimeMetadataView(wrongKind); err == nil {
		t.Fatal("kind/schema mismatch was accepted")
	}
}

func TestRuntimeViewRejectsHostileValidlyDigestedProjectionChanges(t *testing.T) {
	snapshot := mustOpenSnapshot(t, mustCompile(t, neutralCompileInput()))
	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{
		AppInstallationID: testInstallA, CapabilityID: testCapability,
	})
	if err != nil {
		t.Fatal(err)
	}

	var incomplete runtimeViewDocument
	if err := decodeStrict(viewArtifact.bytes, &incomplete); err != nil {
		t.Fatal(err)
	}
	removedSubject := false
	for objectIndex := range incomplete.Objects {
		if incomplete.Objects[objectIndex].APIName != "WorkItem" {
			continue
		}
		fields := incomplete.Objects[objectIndex].Fields
		for fieldIndex := range fields {
			if fields[fieldIndex].APIName == "Subject" {
				incomplete.Objects[objectIndex].Fields = append(fields[:fieldIndex], fields[fieldIndex+1:]...)
				removedSubject = true
				break
			}
		}
	}
	if !removedSubject {
		t.Fatal("fixture has no WorkItem.Subject field to remove")
	}
	incompleteBytes, err := marshalCanonical(incomplete)
	if err != nil {
		t.Fatal(err)
	}
	incompleteView, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, incompleteBytes))
	if err != nil {
		t.Fatalf("structurally valid incomplete view should require source verification, got: %v", err)
	}
	if err := incompleteView.VerifySource(snapshot); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("incomplete exact projection verification error = %v, want corruption", err)
	}

	var missingParent runtimeViewDocument
	if err := decodeStrict(viewArtifact.bytes, &missingParent); err != nil {
		t.Fatal(err)
	}
	missingParent.Relationships = nil
	missingParentBytes, err := marshalCanonical(missingParent)
	if err != nil {
		t.Fatal(err)
	}
	missingParentView, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, missingParentBytes))
	if err != nil {
		t.Fatalf("structurally valid missing-parent view should require source verification, got: %v", err)
	}
	if err := missingParentView.VerifySource(snapshot); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("missing-parent exact projection verification error = %v, want corruption", err)
	}

	var externalPolicy runtimeViewDocument
	if err := decodeStrict(viewArtifact.bytes, &externalPolicy); err != nil {
		t.Fatal(err)
	}
	for i := range externalPolicy.Objects {
		if externalPolicy.Objects[i].APIName == "WorkItem" {
			externalPolicy.Objects[i].AccessPolicy.ObjectRead.Object = "Unrelated"
		}
	}
	externalPolicyBytes, err := marshalCanonical(externalPolicy)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, externalPolicyBytes)); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("view with an external policy dependency error = %v, want corruption", err)
	}
}

func TestRuntimeViewRejectsValidlyDigestedOutOfProfileAccessPolicy(t *testing.T) {
	snapshot := mustOpenSnapshot(t, mustCompile(t, neutralCompileInput()))
	viewArtifact := mustProject(t, snapshot, testInstallA, testCapability)
	var doc runtimeViewDocument
	if err := decodeStrict(viewArtifact.Bytes(), &doc); err != nil {
		t.Fatalf("decode selected-root view: %v", err)
	}
	for i := range doc.Objects {
		if doc.Objects[i].APIName != "WorkItem" {
			continue
		}
		doc.Objects[i].AccessPolicy.RecordMutation = &models.RecordMutationAccessPolicy{
			Deny: []models.RecordMutationDenial{{Field: "Status"}},
		}
	}
	encoded, err := marshalCanonical(doc)
	if err != nil {
		t.Fatalf("encode hostile view: %v", err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, encoded)); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("validly digested selected-root view with mutation access policy error = %v, want corruption", err)
	}

	recordAccessSnapshot := mustOpenSnapshot(t, mustCompile(t, recordAccessCompileInput()))
	recordAccessArtifact := mustProject(t, recordAccessSnapshot, testInstallA, recordAccessCapability)
	if err := decodeStrict(recordAccessArtifact.Bytes(), &doc); err != nil {
		t.Fatalf("decode record-access view: %v", err)
	}
	for i := range doc.Objects {
		if doc.Objects[i].APIName == "WorkItem" {
			doc.Objects[i].AccessPolicy.Lifecycle = &models.RecordLifecycleAccessPolicy{}
		}
	}
	encoded, err = marshalCanonical(doc)
	if err != nil {
		t.Fatalf("encode hostile record-access view: %v", err)
	}
	if _, err := OpenRuntimeMetadataView(makeArtifact(ArtifactRuntimeMetadataView, encoded)); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("validly digested record-access view with lifecycle access policy error = %v, want corruption", err)
	}
}

func TestDirectoryStoreIsImmutableContentAddressedAndDetectsCorruption(t *testing.T) {
	root := filepath.Join(t.TempDir(), "new", "nested", "store")
	store, err := NewDirectoryStore(root)
	if err != nil {
		t.Fatal(err)
	}
	if !filepath.IsAbs(store.root) {
		t.Fatalf("store root %q is not absolute", store.root)
	}
	snapshotArtifact := mustCompile(t, neutralCompileInput())
	snapshot := mustOpenSnapshot(t, snapshotArtifact)
	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{AppInstallationID: testInstallA, CapabilityID: testCapability})
	if err != nil {
		t.Fatal(err)
	}

	for _, artifact := range []Artifact{snapshotArtifact, viewArtifact} {
		if err := store.Put(t.Context(), artifact); err != nil {
			t.Fatalf("put %s: %v", artifact.Kind(), err)
		}
		if err := store.Put(t.Context(), artifact); err != nil {
			t.Fatalf("idempotent put %s: %v", artifact.Kind(), err)
		}
		loaded, err := store.Get(t.Context(), artifact.Kind(), artifact.Digest())
		if err != nil {
			t.Fatalf("get %s: %v", artifact.Kind(), err)
		}
		if !bytes.Equal(loaded.Bytes(), artifact.Bytes()) {
			t.Fatalf("stored %s bytes changed", artifact.Kind())
		}
	}

	ctx, cancel := context.WithCancel(t.Context())
	cancel()
	if err := store.Put(ctx, snapshotArtifact); !errors.Is(err, context.Canceled) {
		t.Fatalf("canceled put error = %v", err)
	}
	if _, err := store.Get(ctx, snapshotArtifact.Kind(), snapshotArtifact.Digest()); !errors.Is(err, context.Canceled) {
		t.Fatalf("canceled get error = %v", err)
	}
	if _, err := store.Get(t.Context(), ArtifactCatalogSnapshot, Digest("sha256:"+string(bytes.Repeat([]byte{'0'}, 64)))); !errors.Is(err, ErrArtifactNotFound) {
		t.Fatalf("missing get error = %v", err)
	}

	path, err := store.artifactPath(snapshotArtifact.Kind(), snapshotArtifact.Digest())
	if err != nil {
		t.Fatal(err)
	}
	if err := os.Chmod(path, 0o640); err != nil {
		t.Fatal(err)
	}
	data := snapshotArtifact.Bytes()
	data[len(data)-3] ^= 1
	if err := os.WriteFile(path, data, 0o440); err != nil {
		t.Fatal(err)
	}
	if _, err := store.Get(t.Context(), snapshotArtifact.Kind(), snapshotArtifact.Digest()); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("corrupt store read error = %v, want corruption", err)
	}
	if err := store.Put(t.Context(), snapshotArtifact); !errors.Is(err, ErrArtifactCorrupt) {
		t.Fatalf("immutable overwrite of corrupt digest path error = %v, want corruption", err)
	}
}

func TestDirectoryStoreConcurrentIdenticalPublicationConverges(t *testing.T) {
	store, err := NewDirectoryStore(t.TempDir())
	if err != nil {
		t.Fatal(err)
	}
	artifact := mustCompile(t, neutralCompileInput())
	var wg sync.WaitGroup
	errorsOut := make(chan error, 16)
	for range 16 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			errorsOut <- store.Put(t.Context(), artifact)
		}()
	}
	wg.Wait()
	close(errorsOut)
	for err := range errorsOut {
		if err != nil {
			t.Fatalf("concurrent put: %v", err)
		}
	}
	if _, err := store.Get(t.Context(), artifact.Kind(), artifact.Digest()); err != nil {
		t.Fatalf("get converged artifact: %v", err)
	}
}

func TestSelectedRootMetadataParityOracleDetectsEveryProjectedClass(t *testing.T) {
	input := neutralCompileInput()
	artifact := mustCompile(t, input)
	snapshot := mustOpenSnapshot(t, artifact)
	viewArtifact, err := ProjectRuntimeMetadataView(snapshot, ProjectionKey{AppInstallationID: testInstallA, CapabilityID: testCapability})
	if err != nil {
		t.Fatal(err)
	}
	view := mustOpenView(t, viewArtifact)
	expected := parityInputForProjection(input, []string{"WorkItem", "Workspace"})
	if report := CompareRuntimeView(view, expected); !report.Equivalent() {
		t.Fatalf("parity differences: %#v", report.Differences)
	}

	expected.Objects[0].Fields[1].PhysicalName = "wrong_column"
	report := CompareRuntimeView(view, expected)
	if report.Equivalent() || len(report.Differences) != 1 || report.Differences[0].Kind != "mismatch" {
		t.Fatalf("changed physical metadata report = %#v", report.Differences)
	}
}

func neutralCompileInput() CompileInput {
	workspace := &models.ObjectMetadata{
		ID: "obj-workspace", APIName: "Workspace", Label: "Workspace", PluralLabel: "Workspaces",
		SharingModel: constants.SharingModelPrivate, EnableHierarchySharing: true,
		Fields: []models.FieldMetadata{
			{ID: "field-workspace-id", APIName: "Id", Label: "ID", Type: constants.FieldTypeReference, PhysicalName: "id", IsSystem: true},
			{ID: "field-workspace-name", APIName: "Name", Label: "Name", Type: constants.FieldTypeString, PhysicalName: "display_name"},
			{ID: "field-workspace-computed", APIName: "ComputedName", Label: "Computed Name", Type: constants.FieldTypeFormula, Formula: stringPointer("Name"), FormulaMode: stringPointer("active"), ReturnType: fieldTypePointer(constants.FieldTypeString)},
		},
	}
	workItem := &models.ObjectMetadata{
		ID: "obj-work-item", APIName: "WorkItem", Label: "Work Item", PluralLabel: "Work Items",
		RequiredReadPermission: "ReadWork", SharingModel: constants.SharingModelControlledByParent,
		AccessPolicy: &models.ObjectAccessPolicy{
			Mode:           "derived",
			ObjectRead:     &models.AccessPolicyExpr{Kind: "objectPermission", Object: "Workspace", Operation: constants.PermRead},
			FieldRead:      &models.AccessPolicyExpr{Inherit: "objectRead"},
			RowRead:        &models.RowReadPolicy{Kind: "ownerVisibility", OwnerField: "OwnerId", VisibilityField: "Visibility"},
			RecordMutation: &models.RecordMutationAccessPolicy{Deny: []models.RecordMutationDenial{{Field: "Status"}}},
		},
		ReadSource: &models.ObjectReadSourceConfig{Kind: "physical_table", PhysicalTable: "work_items"},
		Fields: []models.FieldMetadata{
			{ID: "field-id", APIName: "Id", Label: "ID", Type: constants.FieldTypeReference, PhysicalName: "id", IsSystem: true},
			{ID: "field-subject", APIName: "Subject", Label: "Subject", Type: constants.FieldTypeString, PhysicalName: "subject_text", Createable: true, Updateable: true},
			{ID: "field-done", APIName: "Done", Label: "Done", Type: constants.FieldTypeBoolean, PhysicalName: "is_done"},
			{ID: "field-count", APIName: "Count", Label: "Count", Type: constants.SchemaFieldType("integer"), PhysicalName: "item_count"},
			{ID: "field-amount", APIName: "Amount", Label: "Amount", Type: constants.FieldTypeCurrency, PhysicalName: "amount_exact", Precision: intPointer(38), Scale: intPointer(4)},
			{ID: "field-date", APIName: "DueDate", Label: "Due Date", Type: constants.FieldTypeDate, PhysicalName: "due_date"},
			{ID: "field-datetime", APIName: "UpdatedAt", Label: "Updated At", Type: constants.FieldTypeDatetime, PhysicalName: "updated_at"},
			{ID: "field-workspace", APIName: "WorkspaceId", Label: "Workspace", Type: constants.FieldTypeReference, PhysicalName: "workspace_id", ReferenceTo: []string{"Workspace"}, IsMasterDetail: true, ChildRelationshipName: stringPointer("WorkItems")},
			{ID: "field-unrelated", APIName: "UnrelatedId", Label: "Unrelated", Type: constants.FieldTypeReference, PhysicalName: "unrelated_id", ReferenceTo: []string{"Unrelated"}},
			{ID: "field-owner", APIName: "OwnerId", Label: "Owner", Type: constants.FieldTypeReference, PhysicalName: "owner_id", IsSystem: true},
			{ID: "field-visibility", APIName: "Visibility", Label: "Visibility", Type: constants.FieldTypeString, PhysicalName: "visibility_state"},
			{ID: "field-hidden", APIName: "InternalNote", Label: "Internal Note", Type: constants.FieldTypeString, PhysicalName: "internal_note", ReadProtectionPolicy: "never"},
		},
	}
	unrelated := &models.ObjectMetadata{
		ID: "obj-unrelated", APIName: "Unrelated", Label: "Unrelated", PluralLabel: "Unrelated",
		SharingModel: constants.SharingModelPublicReadWrite,
		Fields:       []models.FieldMetadata{{ID: "field-unrelated-id", APIName: "Id", Label: "ID", Type: constants.FieldTypeString}},
	}
	relationship := &models.Relationship{
		ID: "rel-workspace", ChildObjectAPIName: "WorkItem", ParentObjectAPIName: "Workspace",
		FieldAPIName: "WorkspaceId", RelationshipName: "Workspace", RelationshipType: "masterdetail", IsSharingParent: true,
	}
	ordinaryLookup := &models.Relationship{
		ID: "rel-unrelated", ChildObjectAPIName: "WorkItem", ParentObjectAPIName: "Unrelated",
		FieldAPIName: "UnrelatedId", RelationshipName: "Unrelated", RelationshipType: "lookup",
	}
	rule := &models.SharingRule{
		ID: "rule-workspace-readers", ObjectAPIName: "WorkItem", Name: "WorkspaceReaders",
		Criteria: "", AccessLevel: constants.PermRead, ShareWithGroupID: stringPointer("group-readers"),
	}
	return CompileInput{
		OrgID: "org-neutral", Lineage: Lineage{CatalogRevision: testRevision, SchemaGeneration: 12, EngineGeneration: 5},
		Objects: []ObjectInput{
			{CatalogRevision: testRevision, Metadata: workItem},
			{CatalogRevision: testRevision, Metadata: workspace},
			{CatalogRevision: testRevision, Metadata: unrelated},
		},
		Relationships: []RelationshipInput{
			{CatalogRevision: testRevision, Metadata: relationship},
			{CatalogRevision: testRevision, Metadata: ordinaryLookup},
		},
		SharingRules: []SharingRuleInput{{CatalogRevision: testRevision, Metadata: rule}},
		Definitions: []DefinitionInput{
			{CatalogRevision: testRevision, ComponentID: "definition.unrelated", DefinitionVersion: "7", ContentDigest: Digest("sha256:" + string(bytes.Repeat([]byte{'1'}, 64)))},
			{CatalogRevision: testRevision, ComponentID: "definition.other", DefinitionVersion: "2", ContentDigest: Digest("sha256:" + string(bytes.Repeat([]byte{'2'}, 64)))},
		},
		Projections: []RuntimeProjectionInput{
			{CatalogRevision: testRevision, AppInstallationID: testInstallA, CapabilityID: testCapability, Profile: ProjectionSelectedRootReadV1, ObjectAPINames: []string{"Workspace", "WorkItem"}},
			{CatalogRevision: testRevision, AppInstallationID: testInstallB, CapabilityID: "runtime.unrelated.v1", Profile: ProjectionSelectedRootReadV1, ObjectAPINames: []string{"Unrelated"}, DefinitionComponentIDs: []string{"definition.unrelated", "definition.other"}},
		},
	}
}

func parityInputForProjection(input CompileInput, objectNames []string) ParityInput {
	included := map[string]bool{}
	for _, name := range objectNames {
		included[foldIdentity(name)] = true
	}
	result := ParityInput{}
	for _, object := range input.Objects {
		if included[foldIdentity(object.Metadata.APIName)] {
			result.Objects = append(result.Objects, object.Metadata)
		}
	}
	for _, relationship := range input.Relationships {
		if included[foldIdentity(relationship.Metadata.ChildObjectAPIName)] && included[foldIdentity(relationship.Metadata.ParentObjectAPIName)] {
			result.Relationships = append(result.Relationships, relationship.Metadata)
		}
	}
	for _, rule := range input.SharingRules {
		if included[foldIdentity(rule.Metadata.ObjectAPIName)] {
			result.SharingRules = append(result.SharingRules, rule.Metadata)
		}
	}
	return result
}

func mustCompile(t *testing.T, input CompileInput) Artifact {
	t.Helper()
	artifact, err := CompileCatalogSnapshot(input)
	if err != nil {
		t.Fatalf("compile Catalog snapshot: %v", err)
	}
	return artifact
}

func mustOpenSnapshot(t *testing.T, artifact Artifact) *CatalogSnapshot {
	t.Helper()
	snapshot, err := OpenCatalogSnapshot(artifact)
	if err != nil {
		t.Fatalf("open Catalog snapshot: %v", err)
	}
	return snapshot
}

func mustOpenView(t *testing.T, artifact Artifact) *RuntimeMetadataView {
	t.Helper()
	view, err := OpenRuntimeMetadataView(artifact)
	if err != nil {
		t.Fatalf("open Runtime metadata view: %v", err)
	}
	return view
}

func withRevision(input CompileInput, revision CatalogRevision) CompileInput {
	input.Lineage.CatalogRevision = revision
	for i := range input.Objects {
		input.Objects[i].CatalogRevision = revision
	}
	for i := range input.Relationships {
		input.Relationships[i].CatalogRevision = revision
	}
	for i := range input.SharingRules {
		input.SharingRules[i].CatalogRevision = revision
	}
	for i := range input.Definitions {
		input.Definitions[i].CatalogRevision = revision
	}
	for i := range input.ReportDefinitions {
		input.ReportDefinitions[i].CatalogRevision = revision
	}
	for i := range input.CommandRegistrations {
		input.CommandRegistrations[i].CatalogRevision = revision
	}
	for i := range input.ApprovalDefinitions {
		input.ApprovalDefinitions[i].CatalogRevision = revision
	}
	for i := range input.MutationDefinitions {
		input.MutationDefinitions[i].CatalogRevision = revision
	}
	for i := range input.AttachmentSourceResolvers {
		input.AttachmentSourceResolvers[i].CatalogRevision = revision
	}
	for i := range input.AttachmentSourceFields {
		input.AttachmentSourceFields[i].CatalogRevision = revision
	}
	for i := range input.Projections {
		input.Projections[i].CatalogRevision = revision
	}
	return input
}

func intPointer(value int) *int                                 { return &value }
func stringPointer(value string) *string                        { return &value }
func fieldTypePointer(value models.FieldType) *models.FieldType { return &value }

func reverseObjects(values []ObjectInput)                { reverseSlice(values) }
func reverseRelationships(values []RelationshipInput)    { reverseSlice(values) }
func reverseRules(values []SharingRuleInput)             { reverseSlice(values) }
func reverseDefinitions(values []DefinitionInput)        { reverseSlice(values) }
func reverseProjections(values []RuntimeProjectionInput) { reverseSlice(values) }
func reverseFields(values []models.FieldMetadata)        { reverseSlice(values) }
func reverseStrings(values []string)                     { reverseSlice(values) }

func reverseSlice[T any](values []T) {
	for left, right := 0, len(values)-1; left < right; left, right = left+1, right-1 {
		values[left], values[right] = values[right], values[left]
	}
}