fix(sync): accept cloud relation mutations (#379)

Alan-TheGentleman · web-flow · commit 436c03f79616 · 2026-05-14T13:39:29.000+02:00
diff --git a/internal/cloud/chunkcodec/chunkcodec.go b/internal/cloud/chunkcodec/chunkcodec.go
@@ -260,6 +260,24 @@ type mutationPromptPayload struct {
 	HardDelete bool    `json:"hard_delete,omitempty"`
 }
 
+type mutationRelationPayload struct {
+	SyncID         string   `json:"sync_id"`
+	SourceID       string   `json:"source_id"`
+	TargetID       string   `json:"target_id"`
+	Relation       string   `json:"relation"`
+	Reason         *string  `json:"reason,omitempty"`
+	Evidence       *string  `json:"evidence,omitempty"`
+	Confidence     *float64 `json:"confidence,omitempty"`
+	JudgmentStatus string   `json:"judgment_status"`
+	MarkedByActor  *string  `json:"marked_by_actor,omitempty"`
+	MarkedByKind   *string  `json:"marked_by_kind,omitempty"`
+	MarkedByModel  *string  `json:"marked_by_model,omitempty"`
+	SessionID      *string  `json:"session_id,omitempty"`
+	Project        string   `json:"project"`
+	CreatedAt      string   `json:"created_at,omitempty"`
+	UpdatedAt      string   `json:"updated_at,omitempty"`
+}
+
 func normalizeChunkMutation(raw map[string]any, project string) (map[string]any, error) {
 	mutationJSON, err := json.Marshal(raw)
 	if err != nil {
@@ -321,6 +339,11 @@ func validateSupportedMutation(entity, op string) error {
 			return fmt.Errorf("unsupported mutation %q/%q", entity, op)
 		}
 		return nil
+	case store.SyncEntityRelation:
+		if op != store.SyncOpUpsert {
+			return fmt.Errorf("unsupported mutation %q/%q", entity, op)
+		}
+		return nil
 	default:
 		return fmt.Errorf("unsupported mutation %q/%q", entity, op)
 	}
@@ -425,6 +448,51 @@ func normalizeMutationPayload(entity, op, payload, project string) (normalizedPa
 			return "", "", fmt.Errorf("encode mutation payload: %w", err)
 		}
 		return string(encoded), body.SyncID, nil
+	case store.SyncEntityRelation:
+		var body mutationRelationPayload
+		if err := DecodeSyncMutationPayload(payload, &body); err != nil {
+			return "", "", fmt.Errorf("decode mutation payload: %w", err)
+		}
+		body.SyncID = strings.TrimSpace(body.SyncID)
+		body.SourceID = strings.TrimSpace(body.SourceID)
+		body.TargetID = strings.TrimSpace(body.TargetID)
+		body.Relation = strings.TrimSpace(body.Relation)
+		body.JudgmentStatus = strings.TrimSpace(body.JudgmentStatus)
+		if body.MarkedByActor != nil {
+			trimmed := strings.TrimSpace(*body.MarkedByActor)
+			body.MarkedByActor = &trimmed
+		}
+		if body.MarkedByKind != nil {
+			trimmed := strings.TrimSpace(*body.MarkedByKind)
+			body.MarkedByKind = &trimmed
+		}
+		if body.SyncID == "" {
+			return "", "", fmt.Errorf("relation payload sync_id is required for upsert")
+		}
+		if body.SourceID == "" {
+			return "", "", fmt.Errorf("relation payload source_id is required for upsert")
+		}
+		if body.TargetID == "" {
+			return "", "", fmt.Errorf("relation payload target_id is required for upsert")
+		}
+		if body.Relation == "" {
+			return "", "", fmt.Errorf("relation payload relation is required for upsert")
+		}
+		if body.JudgmentStatus == "" {
+			return "", "", fmt.Errorf("relation payload judgment_status is required for upsert")
+		}
+		if body.MarkedByActor == nil || *body.MarkedByActor == "" {
+			return "", "", fmt.Errorf("relation payload marked_by_actor is required for upsert")
+		}
+		if body.MarkedByKind == nil || *body.MarkedByKind == "" {
+			return "", "", fmt.Errorf("relation payload marked_by_kind is required for upsert")
+		}
+		body.Project = project
+		encoded, err := json.Marshal(body)
+		if err != nil {
+			return "", "", fmt.Errorf("encode mutation payload: %w", err)
+		}
+		return string(encoded), body.SyncID, nil
 	default:
 		return "", "", fmt.Errorf("unsupported mutation %q/%q", entity, op)
 	}
diff --git a/internal/cloud/chunkcodec/chunkcodec_test.go b/internal/cloud/chunkcodec/chunkcodec_test.go
@@ -2,6 +2,7 @@ package chunkcodec
 
 import (
 	"encoding/json"
+	"strings"
 	"testing"
 
 	"github.com/Gentleman-Programming/engram/internal/store"
@@ -72,6 +73,82 @@ func TestCanonicalizeForProjectPreservesMutationMetadataPayloadFields(t *testing
 	assertPayloadField(2, "created_at", "2026-04-08T09:00:00Z")
 }
 
+func TestCanonicalizeForProjectAcceptsRelationUpsertMutation(t *testing.T) {
+	raw := []byte(`{
+		"mutations": [
+			{
+				"entity": "relation",
+				"entity_key": "rel-1",
+				"op": "upsert",
+				"project": "wrong",
+				"payload": "{\"sync_id\":\"rel-1\",\"source_id\":\"obs-a\",\"target_id\":\"obs-b\",\"relation\":\"conflicts_with\",\"reason\":\"different decisions\",\"judgment_status\":\"judged\",\"marked_by_actor\":\"agent-a\",\"marked_by_kind\":\"agent\",\"marked_by_model\":\"model-a\",\"project\":\"wrong\",\"created_at\":\"2026-05-04T01:00:00Z\",\"updated_at\":\"2026-05-04T01:01:00Z\"}"
+			}
+		]
+	}`)
+
+	normalized, err := CanonicalizeForProject(raw, "proj-a")
+	if err != nil {
+		t.Fatalf("canonicalize relation mutation: %v", err)
+	}
+
+	var chunk struct {
+		Mutations []store.SyncMutation `json:"mutations"`
+	}
+	if err := json.Unmarshal(normalized, &chunk); err != nil {
+		t.Fatalf("decode canonicalized chunk: %v", err)
+	}
+	if len(chunk.Mutations) != 1 {
+		t.Fatalf("expected 1 mutation, got %d", len(chunk.Mutations))
+	}
+	mutation := chunk.Mutations[0]
+	if mutation.Entity != store.SyncEntityRelation || mutation.Op != store.SyncOpUpsert || mutation.EntityKey != "rel-1" || mutation.Project != "proj-a" {
+		t.Fatalf("expected canonical relation/upsert mutation, got %+v", mutation)
+	}
+
+	var payload map[string]any
+	if err := json.Unmarshal([]byte(mutation.Payload), &payload); err != nil {
+		t.Fatalf("decode canonical relation payload: %v", err)
+	}
+	if payload["project"] != "proj-a" {
+		t.Fatalf("expected relation payload project rewritten to proj-a, got %#v", payload["project"])
+	}
+	for _, field := range []string{"sync_id", "source_id", "target_id", "relation", "judgment_status", "marked_by_actor", "marked_by_kind"} {
+		if payload[field] == "" || payload[field] == nil {
+			t.Fatalf("expected relation payload field %q to be preserved, got %#v", field, payload)
+		}
+	}
+}
+
+func TestCanonicalizeForProjectRejectsInvalidRelationMutation(t *testing.T) {
+	raw := []byte(`{
+		"mutations": [
+			{
+				"entity": "relation",
+				"entity_key": "rel-1",
+				"op": "upsert",
+				"payload": "{\"sync_id\":\"rel-1\",\"source_id\":\"obs-a\",\"target_id\":\"\",\"judgment_status\":\"judged\",\"marked_by_actor\":\"agent-a\",\"marked_by_kind\":\"agent\"}"
+			}
+		]
+	}`)
+
+	_, err := CanonicalizeForProject(raw, "proj-a")
+	if err == nil {
+		t.Fatal("expected invalid relation mutation to fail")
+	}
+	if got := err.Error(); got == "" || !containsAll(got, []string{"relation", "target_id"}) {
+		t.Fatalf("expected relation target_id validation error, got %q", got)
+	}
+}
+
+func containsAll(s string, parts []string) bool {
+	for _, part := range parts {
+		if !strings.Contains(s, part) {
+			return false
+		}
+	}
+	return true
+}
+
 func TestCanonicalizeForProjectPreservesClosureOnlyDirectSessionOwnership(t *testing.T) {
 	raw := []byte(`{
 		"sessions": [
diff --git a/internal/cloud/cloudserver/mutations.go b/internal/cloud/cloudserver/mutations.go
@@ -170,7 +170,7 @@ func (s *CloudServer) handleMutationPush(w http.ResponseWriter, r *http.Request)
 	}
 
 	// REQ-006 / REQ-008: Validate each entry's payload before storage.
-	// Relation entries are strictly validated (all 6 required fields).
+	// Relation entries are strictly validated (all required fields).
 	// Legacy entities (session, observation, prompt) use the lenient floor only.
 	// Any failure rejects the ENTIRE batch (atomic — no partial inserts).
 	var invalid []map[string]any
@@ -289,20 +289,21 @@ func (s *CloudServer) handleMutationPull(w http.ResponseWriter, r *http.Request)
 
 // ─── REQ-006 / REQ-008: Per-entity payload validation ────────────────────────
 
-// relationRequiredFields lists the 6 fields that MUST be present and non-empty
+// relationRequiredFields lists the fields that MUST be present and non-empty
 // in every relation mutation payload (REQ-006). This list is the stable
 // validation contract — Phase 3 MUST NOT remove or rename these fields without
 // a wire-format version bump.
 var relationRequiredFields = []string{
 	"sync_id",
 	"source_id",
 	"target_id",
+	"relation",
 	"judgment_status",
 	"marked_by_actor",
 	"marked_by_kind",
 }
 
-// validateRelationPayload checks that all 6 required relation fields are present
+// validateRelationPayload checks that all required relation fields are present
 // and non-empty in the decoded payload map.
 // Returns (missingField, false) when any required field is absent or empty,
 // or ("", true) when all required fields are present.
diff --git a/internal/cloud/cloudserver/mutations_test.go b/internal/cloud/cloudserver/mutations_test.go
@@ -1016,7 +1016,7 @@ func (a *simpleProjectAuth) AuthorizeProject(_ string) error {
 // ─── REQ-006, REQ-008: Relation payload validation tests (Phase D) ───────────
 
 // TestHandleMutationPush_ValidRelation_Returns200 (D.1a) verifies REQ-006 happy
-// path: a complete relation payload with all 6 required fields returns HTTP 200.
+// path: a complete relation payload with all required fields returns HTTP 200.
 func TestHandleMutationPush_ValidRelation_Returns200(t *testing.T) {
 	ms := newFakeMutationStore()
 	srv := newMutationTestServer(ms, "secret", []string{"proj-a"})
@@ -1025,6 +1025,7 @@ func TestHandleMutationPush_ValidRelation_Returns200(t *testing.T) {
 		"sync_id":         "rel-001",
 		"source_id":       "obs-src-001",
 		"target_id":       "obs-tgt-001",
+		"relation":        "conflicts_with",
 		"judgment_status": "judged",
 		"marked_by_actor": "alice",
 		"marked_by_kind":  "human",
@@ -1063,6 +1064,7 @@ func TestHandleMutationPush_RelationMissingEachRequiredField(t *testing.T) {
 			payload: json.RawMessage(`{
 				"source_id":       "obs-src",
 				"target_id":       "obs-tgt",
+				"relation":        "conflicts_with",
 				"judgment_status": "judged",
 				"marked_by_actor": "alice",
 				"marked_by_kind":  "human"
@@ -1073,6 +1075,7 @@ func TestHandleMutationPush_RelationMissingEachRequiredField(t *testing.T) {
 			payload: json.RawMessage(`{
 				"sync_id":         "rel-001",
 				"target_id":       "obs-tgt",
+				"relation":        "conflicts_with",
 				"judgment_status": "judged",
 				"marked_by_actor": "alice",
 				"marked_by_kind":  "human"
@@ -1083,6 +1086,18 @@ func TestHandleMutationPush_RelationMissingEachRequiredField(t *testing.T) {
 			payload: json.RawMessage(`{
 				"sync_id":         "rel-001",
 				"source_id":       "obs-src",
+				"relation":        "conflicts_with",
+				"judgment_status": "judged",
+				"marked_by_actor": "alice",
+				"marked_by_kind":  "human"
+			}`),
+		},
+		{
+			name: "relation",
+			payload: json.RawMessage(`{
+				"sync_id":         "rel-001",
+				"source_id":       "obs-src",
+				"target_id":       "obs-tgt",
 				"judgment_status": "judged",
 				"marked_by_actor": "alice",
 				"marked_by_kind":  "human"
@@ -1094,6 +1109,7 @@ func TestHandleMutationPush_RelationMissingEachRequiredField(t *testing.T) {
 				"sync_id":         "rel-001",
 				"source_id":       "obs-src",
 				"target_id":       "obs-tgt",
+				"relation":        "conflicts_with",
 				"marked_by_actor": "alice",
 				"marked_by_kind":  "human"
 			}`),
@@ -1104,6 +1120,7 @@ func TestHandleMutationPush_RelationMissingEachRequiredField(t *testing.T) {
 				"sync_id":         "rel-001",
 				"source_id":       "obs-src",
 				"target_id":       "obs-tgt",
+				"relation":        "conflicts_with",
 				"judgment_status": "judged",
 				"marked_by_kind":  "human"
 			}`),
@@ -1114,6 +1131,7 @@ func TestHandleMutationPush_RelationMissingEachRequiredField(t *testing.T) {
 				"sync_id":         "rel-001",
 				"source_id":       "obs-src",
 				"target_id":       "obs-tgt",
+				"relation":        "conflicts_with",
 				"judgment_status": "judged",
 				"marked_by_actor": "alice"
 			}`),
@@ -1145,7 +1163,7 @@ func TestHandleMutationPush_RelationMissingEachRequiredField(t *testing.T) {
 			}
 			// Verify the missing field name appears in the response
 			var resp struct {
-				Error   string       `json:"error"`
+				Error   string `json:"error"`
 				Invalid []struct {
 					Field  string `json:"field"`
 					Index  int    `json:"index"`
@@ -1175,6 +1193,7 @@ func TestHandleMutationPush_PartialBatch_Atomic(t *testing.T) {
 		"sync_id":         "rel-001",
 		"source_id":       "obs-src-001",
 		"target_id":       "obs-tgt-001",
+		"relation":        "conflicts_with",
 		"judgment_status": "judged",
 		"marked_by_actor": "alice",
 		"marked_by_kind":  "human"
@@ -1183,6 +1202,7 @@ func TestHandleMutationPush_PartialBatch_Atomic(t *testing.T) {
 	invalidPayload := json.RawMessage(`{
 		"sync_id":         "rel-002",
 		"source_id":       "obs-src-002",
+		"relation":        "conflicts_with",
 		"judgment_status": "judged",
 		"marked_by_actor": "bob",
 		"marked_by_kind":  "human"
@@ -1576,6 +1596,7 @@ func TestRelationSync_ServerValidation_MissingField(t *testing.T) {
 		"sync_id":         "rel-g4-001",
 		"source_id":       "obs-src-g4",
 		"target_id":       "obs-tgt-g4",
+		"relation":        "conflicts_with",
 		"marked_by_actor": "agent:test",
 		"marked_by_kind":  "agent"
 	}`)
diff --git a/internal/cloud/cloudstore/cloudstore.go b/internal/cloud/cloudstore/cloudstore.go
@@ -1048,7 +1048,7 @@ func materializedMutationBatchChunk(batch []MutationEntry) ([]byte, chunkSummary
 
 func isChunkMaterializableMutationEntity(entity string) bool {
 	switch strings.TrimSpace(entity) {
-	case store.SyncEntitySession, store.SyncEntityObservation, store.SyncEntityPrompt:
+	case store.SyncEntitySession, store.SyncEntityObservation, store.SyncEntityPrompt, store.SyncEntityRelation:
 		return true
 	default:
 		return false
diff --git a/internal/cloud/cloudstore/cloudstore_test.go b/internal/cloud/cloudstore/cloudstore_test.go
@@ -140,6 +140,47 @@ func TestMaterializedMutationBatchChunkIncludesObservationAlongsidePromptAndSess
 	}
 }
 
+func TestMaterializedMutationBatchChunkCarriesRelationMutationWithoutTypedRows(t *testing.T) {
+	relationPayload := json.RawMessage(`{
+		"sync_id":"rel-04081be99000bdf5",
+		"source_id":"obs-a",
+		"target_id":"obs-b",
+		"relation":"conflicts_with",
+		"judgment_status":"judged",
+		"marked_by_actor":"agent-a",
+		"marked_by_kind":"agent",
+		"marked_by_model":"model-a",
+		"project":"sias-app",
+		"created_at":"2026-05-04T01:49:52Z",
+		"updated_at":"2026-05-04T01:50:00Z"
+	}`)
+
+	payload, counts, err := materializedMutationBatchChunk([]MutationEntry{
+		{Project: "sias-app", Entity: store.SyncEntityRelation, EntityKey: "rel-04081be99000bdf5", Op: store.SyncOpUpsert, Payload: relationPayload},
+	})
+	if err != nil {
+		t.Fatalf("materializedMutationBatchChunk: %v", err)
+	}
+	if counts.sessions != 0 || counts.observations != 0 || counts.prompts != 0 {
+		t.Fatalf("relation-only mutation chunk must not increment typed counts, got %+v", counts)
+	}
+
+	var chunk engramsync.ChunkData
+	if err := json.Unmarshal(payload, &chunk); err != nil {
+		t.Fatalf("decode materialized chunk: %v", err)
+	}
+	if len(chunk.Sessions) != 0 || len(chunk.Observations) != 0 || len(chunk.Prompts) != 0 {
+		t.Fatalf("relation-only mutation must not create typed rows, got sessions=%d observations=%d prompts=%d", len(chunk.Sessions), len(chunk.Observations), len(chunk.Prompts))
+	}
+	if len(chunk.Mutations) != 1 {
+		t.Fatalf("expected relation mutation retained for replay, got %d", len(chunk.Mutations))
+	}
+	mutation := chunk.Mutations[0]
+	if mutation.Entity != store.SyncEntityRelation || mutation.EntityKey != "rel-04081be99000bdf5" || mutation.Project != "sias-app" {
+		t.Fatalf("expected canonical relation mutation, got %+v", mutation)
+	}
+}
+
 func TestMaterializedMutationBatchChunksKeepProjectsSeparate(t *testing.T) {
 	chunks, err := materializedMutationBatchChunks([]MutationEntry{
 		{Project: "proj-a", Entity: store.SyncEntityPrompt, EntityKey: "prompt-a", Op: store.SyncOpUpsert, Payload: json.RawMessage(`{"sync_id":"prompt-a","session_id":"sess-a","content":"a"}`)},
diff --git a/internal/store/diagnostic.go b/internal/store/diagnostic.go
@@ -203,6 +203,8 @@ func ValidateSyncMutationPayload(entity, op, payload, entityKey string) SyncMuta
 			require("target_id")
 			require("relation")
 			require("judgment_status")
+			require("marked_by_actor")
+			require("marked_by_kind")
 			require("project")
 		}
 	default:
diff --git a/internal/store/store_test.go b/internal/store/store_test.go

Original file line number	Diff line number	Diff line change
`@@ -203,6 +203,8 @@ func ValidateSyncMutationPayload(entity, op, payload, entityKey string) SyncMuta`
`203`	`203`	`require("target_id")`
`204`	`204`	`require("relation")`
`205`	`205`	`require("judgment_status")`
	`206`	`+ require("marked_by_actor")`
	`207`	`+ require("marked_by_kind")`
`206`	`208`	`require("project")`
`207`	`209`	`}`
`208`	`210`	`default:`