diff --git a/internal/cloud/chunkcodec/chunkcodec.go b/internal/cloud/chunkcodec/chunkcodec.go index 231b01e37..36af43b73 100644 --- a/internal/cloud/chunkcodec/chunkcodec.go +++ b/internal/cloud/chunkcodec/chunkcodec.go @@ -278,6 +278,34 @@ type mutationRelationPayload struct { UpdatedAt string `json:"updated_at,omitempty"` } +// ValidateRelationPayload reports the first required relation payload field that +// is absent, not a string, or empty after trimming whitespace. +func ValidateRelationPayload(payload json.RawMessage) (string, bool) { + var fields map[string]any + if err := json.Unmarshal(payload, &fields); err != nil { + return "sync_id", false + } + for _, field := range []string{ + "sync_id", + "source_id", + "target_id", + "relation", + "judgment_status", + "marked_by_actor", + "marked_by_kind", + } { + value, ok := fields[field] + if !ok { + return field, false + } + text, ok := value.(string) + if !ok || strings.TrimSpace(text) == "" { + return field, false + } + } + return "", true +} + func normalizeChunkMutation(raw map[string]any, project string) (map[string]any, error) { mutationJSON, err := json.Marshal(raw) if err != nil { diff --git a/internal/cloud/cloudserver/mutations.go b/internal/cloud/cloudserver/mutations.go index 3cc5b2698..072cdadc8 100644 --- a/internal/cloud/cloudserver/mutations.go +++ b/internal/cloud/cloudserver/mutations.go @@ -10,6 +10,7 @@ import ( "strconv" "strings" + "github.com/Gentleman-Programming/engram/internal/cloud/chunkcodec" "github.com/Gentleman-Programming/engram/internal/cloud/cloudstore" "github.com/Gentleman-Programming/engram/internal/cloud/constants" "github.com/Gentleman-Programming/engram/internal/project" @@ -316,41 +317,12 @@ func (s *CloudServer) handleMutationPull(w http.ResponseWriter, r *http.Request) // ─── REQ-006 / REQ-008: Per-entity payload validation ──────────────────────── -// 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 required relation fields are present -// and non-empty in the decoded payload map. +// and non-empty in the decoded payload map using the canonical chunk validator. // Returns (missingField, false) when any required field is absent or empty, // or ("", true) when all required fields are present. func validateRelationPayload(payload json.RawMessage) (string, bool) { - var fields map[string]any - if err := json.Unmarshal(payload, &fields); err != nil { - // Malformed JSON: treat sync_id as missing (first required field). - return "sync_id", false - } - for _, field := range relationRequiredFields { - v, ok := fields[field] - if !ok { - return field, false - } - s, isStr := v.(string) - if !isStr || strings.TrimSpace(s) == "" { - return field, false - } - } - return "", true + return chunkcodec.ValidateRelationPayload(payload) } // validateLegacyPayload is a no-op for legacy entities (session, observation, diff --git a/internal/cloud/cloudstore/cloudstore.go b/internal/cloud/cloudstore/cloudstore.go index 29e665870..64f9090a6 100644 --- a/internal/cloud/cloudstore/cloudstore.go +++ b/internal/cloud/cloudstore/cloudstore.go @@ -415,6 +415,9 @@ func materializedChunkMutations(project string, chunk engramsync.ChunkData) ([]M if len(payload) == 0 { payload = json.RawMessage("{}") } + if field, ok := chunkcodec.ValidateRelationPayload(payload); !ok { + return nil, fmt.Errorf("cloudstore: materialize chunk: mutations[%d].payload.%s is required for relation", i, field) + } entries = append(entries, MutationEntry{Project: project, Entity: store.SyncEntityRelation, EntityKey: entityKey, Op: op, Payload: payload}) } diff --git a/internal/cloud/cloudstore/cloudstore_test.go b/internal/cloud/cloudstore/cloudstore_test.go index a3897b208..09c4028c6 100644 --- a/internal/cloud/cloudstore/cloudstore_test.go +++ b/internal/cloud/cloudstore/cloudstore_test.go @@ -490,7 +490,7 @@ func TestMaterializedChunkMutationsRejectsMissingSyncIDs(t *testing.T) { func TestMaterializedChunkMutationsCarriesRelationFromChunkMutations(t *testing.T) { project := "proj-materialize-rel" - relationPayload := `{"sync_id":"rel-1","source_id":"obs-a","target_id":"obs-b","relation":"related","project":"proj-materialize-rel"}` + relationPayload := `{"sync_id":"rel-1","source_id":"obs-a","target_id":"obs-b","relation":"related","judgment_status":"judged","marked_by_actor":"agent-a","marked_by_kind":"agent","project":"proj-materialize-rel"}` chunk := engramsync.ChunkData{ Observations: []store.Observation{{SyncID: "obs-a"}, {SyncID: "obs-b"}}, Mutations: []store.SyncMutation{ @@ -545,7 +545,7 @@ func TestWriteChunkMaterializesRelationMutationIntoCloudMutations(t *testing.T) {"sync_id":"obs-b","session_id":"s-1","type":"decision","title":"B","content":"B","scope":"project","created_at":"2026-04-29T10:01:00Z","updated_at":"2026-04-29T10:01:00Z"} ], "mutations":[ - {"entity":"relation","entity_key":"rel-1","op":"upsert","payload":"{\"sync_id\":\"rel-1\",\"source_id\":\"obs-a\",\"target_id\":\"obs-b\",\"relation\":\"related\"}"} + {"entity":"relation","entity_key":"rel-1","op":"upsert","payload":"{\"sync_id\":\"rel-1\",\"source_id\":\"obs-a\",\"target_id\":\"obs-b\",\"relation\":\"related\",\"judgment_status\":\"judged\",\"marked_by_actor\":\"agent-a\",\"marked_by_kind\":\"agent\"}"} ] }`), project) if err != nil { @@ -593,6 +593,32 @@ func TestWriteChunkMaterializesRelationMutationIntoCloudMutations(t *testing.T) } } +func TestWriteChunkRejectsIncompleteRelationBeforePersistence(t *testing.T) { + cs := openTestCloudStore(t) + project := "test-invalid-chunk-relation-" + strings.ReplaceAll(t.Name(), "/", "-") + payload := []byte(`{ + "mutations":[ + {"entity":"relation","entity_key":"rel-1","op":"upsert","payload":"{\"source_id\":\"obs-a\",\"target_id\":\"obs-b\",\"relation\":\"related\",\"judgment_status\":\"judged\",\"marked_by_actor\":\"agent-a\",\"marked_by_kind\":\"agent\"}"} + ] + }`) + + err := cs.WriteChunk(context.Background(), project, chunkIDFromPayload(payload), "tester", "2026-04-29T10:03:00Z", payload) + if err == nil || !strings.Contains(err.Error(), "mutations[0].payload.sync_id is required") { + t.Fatalf("expected missing relation sync_id error, got %v", err) + } + + var chunks, mutations int + if err := cs.db.QueryRowContext(context.Background(), `SELECT COUNT(*) FROM cloud_chunks WHERE project_name = $1`, project).Scan(&chunks); err != nil { + t.Fatalf("count chunks: %v", err) + } + if err := cs.db.QueryRowContext(context.Background(), `SELECT COUNT(*) FROM cloud_mutations WHERE project = $1`, project).Scan(&mutations); err != nil { + t.Fatalf("count mutations: %v", err) + } + if chunks != 0 || mutations != 0 { + t.Fatalf("expected no persisted chunk or mutation after validation failure, got chunks=%d mutations=%d", chunks, mutations) + } +} + func TestWriteChunkMaterializesMutationsAndIsReplayIdempotent(t *testing.T) { cs := openTestCloudStore(t) project := "test-chunk-materialize-" + strings.ReplaceAll(time.Now().UTC().Format("20060102150405.000000000"), ".", "-")