story/internal/callmemory/source_deletion_integration_test.go

package callmemory

import (
	"context"
	"testing"

	"github.com/google/uuid"
)

func TestSourceDeletionIncludesBothAutomaticAndReviewedMemories(t *testing.T) {
	for _, origin := range []string{"automatic", "reviewed"} {
		t.Run(origin, func(t *testing.T) {
			f := setup(t)
			ctx := context.Background()
			automatic, _ := f.add(t)
			reviewed, _ := f.add(t)
			unrelated, _ := f.add(t)
			call, _, _ := f.call(t)
			f.exec(t, `UPDATE call_session SET manifest_status='DISCARDED',call_state='ENDED',local_audio_off=true,sdk_stopped=true,remote_state='STOPPED_OBSERVED',ended_at=now() WHERE id=$1::uuid`, call.SessionID)
			for _, utterance := range []string{"story-source", "unrelated-source"} {
				f.exec(t, `INSERT INTO call_utterance(session_id,utterance_id,source_revision,speaker,text_content,interrupted,source_at) VALUES($1::uuid,$2,1,'USER',$2,false,now())`, call.SessionID, utterance)
			}
			story, _ := addCallSourceStory(t, f, call.SessionID, "story-source", 1)
			f.exec(t, `UPDATE memory_version SET source_refs=$2::jsonb WHERE memory_id=(SELECT id FROM curated_memory WHERE public_id=$1::uuid)`, automatic, jsonBytes([]map[string]any{{"type": "provider_memory", "session_id": call.SessionID, "provider_id": "provider-test"}}))
			f.exec(t, `UPDATE memory_version SET source_refs=$2::jsonb WHERE memory_id=(SELECT id FROM curated_memory WHERE public_id=$1::uuid)`, reviewed, jsonBytes([]map[string]any{{"type": "story", "story_id": story}, {"session_id": call.SessionID, "utterance_id": "story-source", "source_revision": 1}}))
			target, derived := automatic, reviewed
			if origin == "reviewed" {
				target, derived = reviewed, automatic
			}
			mutation := f.request(t, "mutations", MutationInput{ID: uuid.NewString(), ElderID: f.elder, Action: "DELETE", Targets: []string{target}, Expected: map[string]int64{target: 1}, Scope: "SOURCE_AND_DERIVED"})
			targets := mutation["targets"].([]any)
			if len(targets) != 2 || targets[0] != target || targets[1] != derived {
				t.Fatalf("both source-derived memories were not tracked: %v", targets)
			}
			var derivedScope string
			if err := f.p.QueryRow(ctx, `SELECT scope FROM memory_deletion_ledger WHERE memory_id=(SELECT id FROM curated_memory WHERE public_id=$1::uuid)`, derived).Scan(&derivedScope); err != nil || derivedScope != "MEMORY_AND_DERIVED" {
				t.Fatalf("derived memory enlarged original source scope: scope=%s err=%v", derivedScope, err)
			}
			if err := f.s.Maintain(ctx); err != nil {
				t.Fatal(err)
			}
			var remaining, unrelatedStatus, keptSource string
			if err := f.p.QueryRow(ctx, `SELECT count(*)::text FROM memory_version mv JOIN curated_memory cm ON cm.id=mv.memory_id WHERE cm.public_id=ANY($1::uuid[]) AND mv.content<>'[deleted]'`, []string{automatic, reviewed}).Scan(&remaining); err != nil || remaining != "0" {
				t.Fatalf("one derived memory retained plaintext: count=%s err=%v", remaining, err)
			}
			if err := f.p.QueryRow(ctx, `SELECT status FROM curated_memory WHERE public_id=$1::uuid`, unrelated).Scan(&unrelatedStatus); err != nil || unrelatedStatus != "ACTIVE" {
				t.Fatalf("unrelated memory changed: status=%s err=%v", unrelatedStatus, err)
			}
			if err := f.p.QueryRow(ctx, `SELECT text_content FROM call_utterance WHERE session_id=$1::uuid AND utterance_id='unrelated-source'`, call.SessionID).Scan(&keptSource); err != nil {
				t.Fatal(err)
			}
			if origin == "reviewed" && keptSource != "unrelated-source" {
				t.Fatal("deleting the derived automatic summary erased unrelated original speech")
			}
			if origin == "automatic" && keptSource != "[deleted]" {
				t.Fatal("session-wide deletion left an original source")
			}
			f.exec(t, `UPDATE memory_projection_job SET state='VISIBLE' WHERE subject_id=$1::uuid`, f.subject)
			f.exec(t, `UPDATE memory_user_state SET viking_visible_revision=memory_revision WHERE subject_id=$1::uuid`, f.subject)
			f.exec(t, `UPDATE memory_deletion_ledger SET backup_cleanup_verified_at=now(),evidence_ref='synthetic test' WHERE memory_id=(SELECT id FROM curated_memory WHERE public_id=$1::uuid)`, target)
			if err := f.s.Maintain(ctx); err != nil {
				t.Fatal(err)
			}
			progress := f.request(t, "mutations/status", map[string]any{"mutation_id": mutation["mutation_id"]})
			if progress["mutation_state"] == "COMPLETED" || progress["stage_states"].(map[string]any)["backup_cleanup"] != "PENDING" {
				t.Fatal("deletion completed before the derived memory backup was cleared", progress)
			}
			f.exec(t, `UPDATE memory_deletion_ledger SET backup_cleanup_verified_at=now(),evidence_ref='synthetic test' WHERE memory_id=(SELECT id FROM curated_memory WHERE public_id=$1::uuid)`, derived)
			if err := f.s.Maintain(ctx); err != nil {
				t.Fatal(err)
			}
			progress = f.request(t, "mutations/status", map[string]any{"mutation_id": mutation["mutation_id"]})
			if progress["mutation_state"] != "COMPLETED" {
				t.Fatal("both memories were reconciled but deletion remained open", progress)
			}
			var deleted int
			if err := f.p.QueryRow(ctx, `SELECT count(*) FROM curated_memory WHERE public_id=ANY($1::uuid[]) AND status='DELETED'`, []string{automatic, reviewed}).Scan(&deleted); err != nil || deleted != 2 {
				t.Fatalf("both memories did not reach DELETED: count=%d err=%v", deleted, err)
			}
		})
	}
}