package callmemory

import (
	"context"
	"data-terminal-cloud/internal/deviceauth"
	"data-terminal-cloud/internal/identity"
	"data-terminal-cloud/internal/outbox"
	"data-terminal-cloud/internal/privacy"
	"encoding/json"
	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"
	"github.com/redis/go-redis/v9"
	"os"
	"strings"
	"testing"
	"time"
)

type fixture struct {
	s                                    *Service
	p                                    *pgxpool.Pool
	principal                            identity.Principal
	elder, subject, tenant, device, boot string
}

func setup(t *testing.T) fixture {
	t.Helper()
	url := os.Getenv("TEST_DATABASE_URL")
	if url == "" {
		t.Skip("TEST_DATABASE_URL required")
	}
	p, e := pgxpool.New(context.Background(), url)
	if e != nil {
		t.Fatal(e)
	}
	t.Cleanup(p.Close)
	f := fixture{p: p, s: NewService(p, nil), elder: uuid.NewString(), subject: uuid.NewString(), tenant: "10000000-0000-0000-0000-000000000001", device: uuid.NewString(), boot: uuid.NewString(), principal: identity.Principal{Issuer: "http://localhost:58180/realms/voice-local", Subject: "00000000-0000-0000-0000-000000000101"}}
	f.exec(t, `INSERT INTO subject_profile(id,public_id,tenant_id,household_id,preferred_name,status) VALUES($1::uuid,$2::uuid,$3::uuid,'10000000-0000-0000-0000-000000000201','call-test','active')`, f.subject, f.elder, f.tenant)
	f.exec(t, `INSERT INTO consent_record(tenant_id,subject_id,consent_type,scope,policy_version,notice_digest,actor_user_id,evidence_ref) SELECT tenant_id,$1::uuid,consent_type,scope,policy_version,notice_digest,actor_user_id,evidence_ref FROM consent_record WHERE subject_id='10000000-0000-0000-0000-000000000401' AND withdrawn_at IS NULL`, f.subject)
	f.exec(t, `INSERT INTO consent_snapshot(tenant_id,subject_id,snapshot_version,effective_consents,source_consent_ids) VALUES($1::uuid,$2::uuid,1,'{"voice_chat":true}','[]')`, f.tenant, f.subject)
	f.exec(t, `INSERT INTO device(id,tenant_id,serial_digest,model,hardware_revision,status,credential_ref,minimum_security_version,minimum_firmware_version,onboarding_boot_id,binding_generation) VALUES($1::uuid,$2::uuid,$3,'test','test','registered','local-dev://call-test','1','1',$4::uuid,0)`, f.device, f.tenant, hash([]byte(f.device)), f.boot)
	f.exec(t, `INSERT INTO device_binding(tenant_id,device_id,household_id,subject_id,bound_by,reason,generation) VALUES($1::uuid,$2::uuid,'10000000-0000-0000-0000-000000000201',$3::uuid,'10000000-0000-0000-0000-000000000101','test',1)`, f.tenant, f.device, f.subject)
	f.exec(t, `INSERT INTO memory_user_state(subject_id,tenant_id,legacy_projection_pending) VALUES($1::uuid,$2::uuid,false)`, f.subject, f.tenant)
	return f
}
func (f fixture) exec(t *testing.T, q string, args ...any) {
	t.Helper()
	if _, e := f.p.Exec(context.Background(), q, args...); e != nil {
		t.Fatal(e)
	}
}
func (f fixture) request(t *testing.T, op string, in any) map[string]any {
	t.Helper()
	out, e := f.s.MemoryRequest(context.Background(), f.principal, op, jsonBytes(in))
	if e != nil {
		t.Fatalf("%s: %v", op, e)
	}
	var m map[string]any
	json.Unmarshal(jsonBytes(out), &m)
	return m
}
func (f fixture) add(t *testing.T) (string, map[string]any) {
	t.Helper()
	content := "喜欢喝淡茶"
	out := f.request(t, "mutations", MutationInput{ID: uuid.NewString(), ElderID: f.elder, Action: "ADD", Targets: []string{}, Expected: map[string]int64{}, Patch: MemoryPatch{Content: &content}})
	return out["targets"].([]any)[0].(string), out
}
func (f fixture) call(t *testing.T) (Manifest, CallKey, deviceauth.Identity) {
	t.Helper()
	m := testManifest()
	m.BootID = f.boot
	m.DeviceID = f.device
	m.PrivacyEpoch = 0
	b, mac, e := EncodeManifest([]byte(strings.Repeat("k", 32)), m)
	if e != nil {
		t.Fatal(e)
	}
	f.exec(t, `INSERT INTO call_session(id,tenant_id,subject_id,device_id,boot_id,hook_event_id,call_generation,binding_generation,manifest_id,manifest_b64,manifest_mac,manifest_status,memory_mode,expires_at,lease_expires_at) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$5::uuid,$6,1,1,$7::uuid,$8,$9,'ISSUED','HOT_ONLY',$10,$11)`, m.SessionID, f.tenant, f.subject, f.device, f.boot, m.HookID, m.ManifestID, b, mac, time.UnixMilli(m.ExpiresAt), time.UnixMilli(m.LeaseExpiresAt))
	return m, CallKey{SessionID: &m.SessionID, BootID: f.boot, HookID: m.HookID, Generation: 1, Binding: 1}, deviceauth.Identity{ID: f.device, TenantID: f.tenant, EnrollmentMode: "physical_confirm", BindingGeneration: 1}
}
func TestMutationVersionFenceAndConfirmationReplay(t *testing.T) {
	f := setup(t)
	id, _ := f.add(t)
	m, _, _ := f.call(t)
	out := f.request(t, "mutations", MutationInput{ID: uuid.NewString(), ElderID: f.elder, Action: "DELETE", Targets: []string{}, Expected: map[string]int64{}, Patch: MemoryPatch{}})
	if out["mutation_state"] != "NEEDS_CONFIRMATION" {
		t.Fatal(out)
	}
	resolved := f.request(t, "mutations/resolve", ResolveInput{ID: out["mutation_id"].(string), Revision: int64(out["mutation_revision"].(float64)), Targets: []string{id}, Expected: map[string]int64{id: 1}, Patch: MemoryPatch{}})
	ci := ConfirmInput{ID: out["mutation_id"].(string), Revision: int64(resolved["mutation_revision"].(float64)), Digest: resolved["proposal_digest"].(string), Token: resolved["confirmation_token"].(string), Decision: "CONFIRM"}
	done := f.request(t, "mutations/confirm", ci)
	if done["mutation_state"] != "PROPAGATING" {
		t.Fatal(done)
	}
	ci.Decision = "CANCEL"
	replay := f.request(t, "mutations/confirm", ci)
	if replay["mutation_state"] != "PROPAGATING" {
		t.Fatal("cancel rolled back committed mutation")
	}
	var status string
	var epoch int64
	if e := f.p.QueryRow(context.Background(), `SELECT manifest_status FROM call_session WHERE id=$1::uuid`, m.SessionID).Scan(&status); e != nil {
		t.Fatal(e)
	}
	if status != "REVOKED" {
		t.Fatal("issued manifest survived deletion")
	}
	f.p.QueryRow(context.Background(), `SELECT privacy_epoch FROM memory_user_state WHERE subject_id=$1::uuid`, f.subject).Scan(&epoch)
	if epoch != 1 {
		t.Fatal(epoch)
	}
	if e := f.s.Maintain(context.Background()); e != nil {
		t.Fatal(e)
	}
	progress := f.request(t, "mutations/status", map[string]any{"mutation_id": ci.ID})
	stages := progress["stage_states"].(map[string]any)
	if stages["active_context"] != "PENDING" || stages["backup_cleanup"] != "PENDING" || progress["mutation_state"] == "COMPLETED" {
		t.Fatal(progress)
	}
	f.exec(t, `UPDATE call_session SET local_audio_off=true,sdk_stopped=true,remote_state='UNKNOWN' WHERE id=$1::uuid`, m.SessionID)
	if e := f.s.Maintain(context.Background()); e != nil {
		t.Fatal(e)
	}
	progress = f.request(t, "mutations/status", map[string]any{"mutation_id": ci.ID})
	if progress["stage_states"].(map[string]any)["active_context"] != "PENDING" {
		t.Fatal("local SDK evidence falsely proved supplier context cleanup")
	}
}
func TestLeaseExpiresAndCannotResurrect(t *testing.T) {
	f := setup(t)
	m, k, d := f.call(t)
	in := HeartbeatInput{CallKey: k, ManifestID: m.ManifestID, AttemptID: uuid.NewString(), Runtime: "STARTING"}
	req := uuid.NewString()
	first, e := f.s.DeviceRequest(context.Background(), d, nil, req, "heartbeat", jsonBytes(in))
	if e != nil {
		t.Fatal(e)
	}
	if first.(map[string]any)["allowed"] != true {
		var details []byte
		f.p.QueryRow(context.Background(), `SELECT jsonb_build_object('device',to_jsonb(d)-'credential_ref'-'serial_digest','binding',to_jsonb(b),'consents',(SELECT jsonb_agg(cr.consent_type) FROM consent_record cr WHERE cr.subject_id=b.subject_id)) FROM device d JOIN device_binding b ON b.device_id=d.id WHERE d.id=$1::uuid`, f.device).Scan(&details)
		t.Log(string(details))
		t.Fatal(first)
	}
	if first.(map[string]any)["transcript_upload_allowed"] != true {
		t.Fatal("granted transcript upload not advertised")
	}
	f.exec(t, `UPDATE call_session SET lease_expires_at=now()-interval '1 second' WHERE id=$1::uuid`, m.SessionID)
	late, e := f.s.DeviceRequest(context.Background(), d, nil, req, "heartbeat", jsonBytes(in))
	if e != nil {
		t.Fatal(e)
	}
	if late.(map[string]any)["allowed"] != false {
		t.Fatal("cached heartbeat revived expired lease", late)
	}
}
func TestTextRevisionDrainingAndConsent(t *testing.T) {
	f := setup(t)
	m, k, d := f.call(t)
	attempt := uuid.NewString()
	send := func(seq int64, rev int64, text string) map[string]any {
		ev := Event{ID: uuid.NewString(), Seq: seq, Type: "TRANSCRIPT_REVISION", Occurred: time.Now().UnixMilli(), AttemptID: attempt, Payload: map[string]any{"utterance_id": "u1", "source_revision": rev, "speaker": "USER", "source_final": true, "text": text}}
		out, e := f.s.DeviceRequest(context.Background(), d, nil, uuid.NewString(), "events", jsonBytes(EventsInput{CallKey: k, Events: []Event{ev}}))
		if e != nil {
			t.Fatal(e)
		}
		return out.(map[string]any)
	}
	first := send(1, 1, "明天看牙")
	if len(first["accepted_event_ids"].([]string)) != 1 {
		t.Fatal(first)
	}
	_, e := f.s.DeviceRequest(context.Background(), d, nil, uuid.NewString(), "end", jsonBytes(EndInput{CallKey: k, Reason: "ON_HOOK", AudioStopped: true, SDKStatus: "STOPPED_OBSERVED"}))
	if e != nil {
		t.Fatal(e)
	}
	f.exec(t, `UPDATE call_session SET drain_started_at=now()-interval '20 seconds',last_text_at=now()-interval '20 seconds' WHERE id=$1::uuid`, m.SessionID)
	if e = f.s.Maintain(context.Background()); e != nil {
		t.Fatal(e)
	}
	send(2, 2, "明天不去看牙")
	var state, text string
	var revision int64
	e = f.p.QueryRow(context.Background(), `SELECT c.transcript_state,c.transcript_revision,u.text_content FROM call_session c JOIN call_utterance u ON u.session_id=c.id WHERE c.id=$1::uuid`, m.SessionID).Scan(&state, &revision, &text)
	if e != nil {
		t.Fatal(e)
	}
	if state != "DRAINING" || revision != 2 || text != "明天不去看牙" {
		t.Fatal(state, revision, text)
	}
	f.exec(t, `UPDATE consent_record SET withdrawn_at=now() WHERE subject_id=$1::uuid AND consent_type='transcript_storage'`, f.subject)
	rejected := send(3, 3, "不得保存")
	if len(rejected["accepted_event_ids"].([]string)) != 0 {
		t.Fatal(rejected)
	}
}
func TestMemoryGrantIndependentOfDeviceRole(t *testing.T) {
	f := setup(t)
	f.principal.Subject = "00000000-0000-0000-0000-000000000102"
	_, e := f.s.MemoryRequest(context.Background(), f.principal, "query", jsonBytes(QueryInput{ElderID: f.elder}))
	if e == nil {
		t.Fatal("editor inherited memory permission")
	}
}

func TestPrepareIdempotenceCancelAndFirstHeartbeat(t *testing.T) {
	f := setup(t)
	ctx := context.Background()
	profileID := "test-" + uuid.NewString()
	caps, evidence, mapping, template := map[string]bool{}, map[string]string{}, map[string]string{}, map[string]any{}
	for _, g := range []string{"H01", "H02", "H04", "H05", "H06", "H07", "H09", "H11"} {
		caps[g] = true
		evidence[g] = "synthetic integration test, no hardware claim"
	}
	for _, field := range []string{"memory_enabled", "hot_context", "welcome_text", "user_id", "assistant_id", "dialog_id"} {
		mapping[field] = "/" + field
		template[field] = ""
	}
	f.exec(t, `INSERT INTO call_provider_profile(id,protocol_version,sdk_version,bot_id,model_version,transport,audio_profile,vendor_template,field_mapping,capabilities,evidence,published) VALUES($1,'CALL-MEMORY-1.1','test-sdk','test-bot','test-model','RTC','{"id":"test-audio"}',$2::jsonb,$3::jsonb,$4::jsonb,$5::jsonb,true)`, profileID, jsonBytes(template), jsonBytes(mapping), jsonBytes(caps), jsonBytes(evidence))
	f.exec(t, `UPDATE device SET activation_state='ACTIVE' WHERE id=$1::uuid`, f.device)
	f.exec(t, `INSERT INTO device_activation_job(device_id,tenant_id,provider_device_name,status,binding_generation,config_version,config_ack_at,engine_ready_at,sdk_registered_at,config_expires_at,ready_binding_generation) VALUES($1::uuid,$2::uuid,$3,'READY',1,1,now(),now(),now(),now()+interval '1 hour',1)`, f.device, f.tenant, "test-"+f.device)
	d := deviceauth.Identity{ID: f.device, PublicID: f.device, TenantID: f.tenant, EnrollmentMode: "physical_confirm", BindingGeneration: 1, ActivationState: "ACTIVE"}
	key := []byte(strings.Repeat("k", 32))
	if _, e := f.s.DeviceRequest(ctx, d, key, uuid.NewString(), "warmup", jsonBytes(WarmupInput{Contract: ContractVersion, KnownVersion: 999, BootID: f.boot, Binding: 1, SDK: "test-sdk", Firmware: "test", AudioID: "test-audio"})); e != nil {
		t.Fatal(e)
	}
	k := CallKey{BootID: f.boot, HookID: "cancel-hook", Generation: 1, Binding: 1}
	if _, e := f.s.DeviceRequest(ctx, d, key, uuid.NewString(), "end", jsonBytes(EndInput{CallKey: k, Reason: "CANCEL_BEFORE_START", AudioStopped: true, SDKStatus: "NOT_STARTED"})); e != nil {
		t.Fatal(e)
	}
	if _, e := f.s.DeviceRequest(ctx, d, key, uuid.NewString(), "prepare", jsonBytes(PrepareInput{CallKey: k, AudioID: "test-audio", Purpose: "NORMAL"})); e == nil {
		t.Fatal("end-before-prepare was resurrected")
	}
	k.HookID = "new-hook"
	k.Generation = 2
	input := jsonBytes(PrepareInput{CallKey: k, AudioID: "test-audio", Purpose: "NORMAL"})
	first, e := f.s.DeviceRequest(ctx, d, key, uuid.NewString(), "prepare", input)
	if e != nil {
		t.Fatal(e)
	}
	second, e := f.s.DeviceRequest(ctx, d, key, uuid.NewString(), "prepare", input)
	if e != nil {
		t.Fatal(e)
	}
	if string(jsonBytes(first)) != string(jsonBytes(second)) {
		t.Fatal("prepare created a second manifest")
	}
	signed := first.(map[string]any)
	m, e := DecodeManifest(key, signed["manifest_b64"].(string), signed["manifest_mac"].(string))
	if e != nil {
		t.Fatal(e)
	}
	k.SessionID = &m.SessionID
	attempt := uuid.NewString()
	ev := EventsInput{CallKey: k, Events: []Event{{ID: uuid.NewString(), Seq: 1, Type: "START_INTENT", AttemptID: attempt, Occurred: time.Now().UnixMilli(), Payload: map[string]any{"manifest_id": m.ManifestID}}}}
	out, e := f.s.DeviceRequest(ctx, d, key, uuid.NewString(), "events", jsonBytes(ev))
	if e != nil {
		t.Fatal(e)
	}
	if len(out.(map[string]any)["accepted_event_ids"].([]string)) != 0 {
		t.Fatal("start accepted without first heartbeat")
	}
	if _, e = f.s.DeviceRequest(ctx, d, key, uuid.NewString(), "heartbeat", jsonBytes(HeartbeatInput{CallKey: k, ManifestID: m.ManifestID, AttemptID: attempt, Runtime: "STARTING"})); e != nil {
		t.Fatal(e)
	}
	out, e = f.s.DeviceRequest(ctx, d, key, uuid.NewString(), "events", jsonBytes(ev))
	if e != nil {
		t.Fatal(e)
	}
	if len(out.(map[string]any)["accepted_event_ids"].([]string)) != 1 {
		t.Fatal(out)
	}
}

type projectionFunc func(context.Context, ProjectionRecord, string, string, bool) (ProjectionResult, error)

func (f projectionFunc) Project(c context.Context, r ProjectionRecord, k, id string, reconcile bool) (ProjectionResult, error) {
	return f(c, r, k, id, reconcile)
}
func TestProjectionAckIsNotVisibilityAndWatermarkHasNoHoles(t *testing.T) {
	f := setup(t)
	_, one := f.add(t)
	f.add(t)
	f.add(t)
	f.exec(t, `UPDATE memory_projection_job SET next_attempt_at='2000-01-01',state=CASE WHEN memory_revision=3 THEN 'VISIBLE' ELSE state END WHERE subject_id=$1::uuid`, f.subject)
	visible := false
	f.s.ConfigureProjection(projectionFunc(func(_ context.Context, r ProjectionRecord, _, _ string, _ bool) (ProjectionResult, error) {
		return ProjectionResult{Kind: "event", ID: r.ID, Visible: visible}, nil
	}))
	if e := f.s.ProjectNext(context.Background()); e != nil {
		t.Fatal(e)
	}
	status := f.request(t, "sync/status", map[string]any{"elder_id": f.elder})
	if status["viking_visible_revision"].(float64) != 0 {
		t.Fatal("ack counted as visible", status)
	}
	visible = true
	f.exec(t, `UPDATE memory_projection_job SET next_attempt_at='2000-01-01' WHERE subject_id=$1::uuid`, f.subject)
	if e := f.s.ProjectNext(context.Background()); e != nil {
		t.Fatal(e)
	}
	status = f.request(t, "sync/status", map[string]any{"elder_id": f.elder})
	if status["viking_visible_revision"].(float64) != 1 {
		t.Fatal("watermark skipped gap at 2", status)
	}
	if e := f.s.Maintain(context.Background()); e != nil {
		t.Fatal(e)
	}
	done := f.request(t, "mutations/status", map[string]any{"mutation_id": one["mutation_id"]})
	if done["mutation_state"] != "COMPLETED" {
		t.Fatal(done)
	}
}

func TestSensitiveMutationStatusCannotBypassListGrant(t *testing.T) {
	f := setup(t)
	content, sensitivity := "需要权限的内容", "sensitive"
	out := f.request(t, "mutations", MutationInput{ID: uuid.NewString(), ElderID: f.elder, Action: "ADD", Targets: []string{}, Expected: map[string]int64{}, Patch: MemoryPatch{Content: &content, Sensitivity: &sensitivity}})
	f.request(t, "access/grant", AccessInput{Elder: f.elder, User: "10000000-0000-0000-0000-000000000112", Read: true})
	f.principal.Subject = "00000000-0000-0000-0000-000000000102"
	rows := f.request(t, "query", QueryInput{ElderID: f.elder})
	if len(rows["items"].([]any)) != 0 {
		t.Fatal(rows)
	}
	if _, e := f.s.MemoryRequest(context.Background(), f.principal, "mutations/status", jsonBytes(map[string]any{"mutation_id": out["mutation_id"]})); e == nil {
		t.Fatal("sensitive status bypass")
	}
}

func TestProjectionRechecksConsentAndSensitiveState(t *testing.T) {
	f := setup(t)
	id, _ := f.add(t)
	// An expiry is a clock event, not an UPDATE of withdrawn_at.
	f.exec(t, `UPDATE consent_record SET expires_at=now()+interval '1 hour' WHERE subject_id=$1::uuid AND consent_type='personalized_memory'`, f.subject)
	var observed string
	f.s.ConfigureProjection(projectionFunc(func(_ context.Context, r ProjectionRecord, _, _ string, _ bool) (ProjectionResult, error) {
		observed = r.Status
		f.exec(t, `UPDATE consent_record SET withdrawn_at=now() WHERE subject_id=$1::uuid AND consent_type='personalized_memory'`, f.subject)
		return ProjectionResult{Kind: "event", ID: r.ID, Visible: true}, nil
	}))
	if e := f.s.ProjectNext(context.Background()); e != nil {
		t.Fatal(e)
	}
	if observed != "ACTIVE" {
		t.Fatal(observed)
	}
	var visible int64
	if e := f.p.QueryRow(context.Background(), `SELECT viking_visible_revision FROM memory_user_state WHERE subject_id=$1::uuid`, f.subject).Scan(&visible); e != nil {
		t.Fatal(e)
	}
	if visible != 0 {
		t.Fatal("concurrent consent withdrawal advanced visibility")
	}
	f.s.ConfigureProjection(projectionFunc(func(_ context.Context, r ProjectionRecord, _, _ string, _ bool) (ProjectionResult, error) {
		if r.ID == id && r.Status != "DISABLED" {
			t.Fatal("withdrawn consent projected as active")
		}
		return ProjectionResult{Kind: "event", ID: r.ID, Visible: true}, nil
	}))
	if e := f.s.ProjectNext(context.Background()); e != nil {
		t.Fatal(e)
	}
	// A separate subject with a sensitive ALLOWED record must not publish plaintext.
	other := setup(t)
	content, sensitivity := "private medical detail", "sensitive"
	other.request(t, "mutations", MutationInput{ID: uuid.NewString(), ElderID: other.elder, Action: "ADD", Targets: []string{}, Expected: map[string]int64{}, Patch: MemoryPatch{Content: &content, Sensitivity: &sensitivity}})
	f.exec(t, `UPDATE memory_projection_job SET next_attempt_at=now()+interval '1 hour' WHERE subject_id<>$1::uuid`, other.subject)
	other.s.ConfigureProjection(projectionFunc(func(_ context.Context, r ProjectionRecord, _, _ string, _ bool) (ProjectionResult, error) {
		if r.Status != "DISABLED" {
			t.Fatal("sensitive memory eligible for projection")
		}
		return ProjectionResult{Kind: "event", Visible: true}, nil
	}))
	if e := other.s.ProjectNext(context.Background()); e != nil {
		t.Fatal(e)
	}
}

type noObjects struct{}

func (noObjects) Remove(context.Context, string) error { return nil }
func TestPrivacyDeleteIncludesNewCallDataAndWaitsForRuntime(t *testing.T) {
	f := setup(t)
	ctx := context.Background()
	_, mutation := f.add(t)
	m, _, _ := f.call(t)
	f.exec(t, `INSERT INTO call_utterance(session_id,utterance_id,source_revision,speaker,text_content,interrupted,source_at) VALUES($1::uuid,'u1',1,'USER','private final text',false,now())`, m.SessionID)
	story, _ := addCallSourceStory(t, f, m.SessionID, "u1", 1)
	f.exec(t, `UPDATE call_session SET budget_snapshot='{"protocol":"DIRECT-CHAT-1","story_context":{"private":"original words"}}'::jsonb WHERE id=$1::uuid`, m.SessionID)
	f.exec(t, `INSERT INTO outbox_event(tenant_id,aggregate_type,aggregate_id,event_type,payload) VALUES($1::uuid,'call_session',$2::uuid,'call.story_extraction_requested','{"transcript_revision":1}'::jsonb)`, f.tenant, m.SessionID)
	request := uuid.NewString()
	f.exec(t, `INSERT INTO privacy_request(id,tenant_id,subject_id,request_type,scope,requested_by,idempotency_key,request_digest) VALUES($1::uuid,$2::uuid,$3::uuid,'DELETE','{}','10000000-0000-0000-0000-000000000101',$1,repeat('a',64))`, request, f.tenant, f.subject)
	f.exec(t, `INSERT INTO privacy_subtask(privacy_request_id,subsystem,target_type,target_ref) VALUES($1::uuid,'postgresql','subject',$2)`, request, f.subject)
	cache := redis.NewClient(&redis.Options{Addr: "127.0.0.1:1"})
	defer cache.Close()
	processor, e := privacy.NewDeleteProcessor(f.p, cache, noObjects{})
	if e != nil {
		t.Fatal(e)
	}
	if e = processor.Handle(ctx, outbox.Event{AggregateID: request}); e == nil || !outbox.IsRetryable(e) || outbox.RetryDelay(e) < time.Minute {
		t.Fatalf("unproved runtime/projection must hold deletion open with a delayed retry: %v", e)
	}
	var text, patch string
	var manifest *string
	if e = f.p.QueryRow(ctx, `SELECT text_content FROM call_utterance WHERE session_id=$1::uuid`, m.SessionID).Scan(&text); e != nil {
		t.Fatal(e)
	}
	if text != "[deleted]" {
		t.Fatal("new call transcript survived privacy deletion")
	}
	if e = f.p.QueryRow(ctx, `SELECT manifest_b64 FROM call_session WHERE id=$1::uuid`, m.SessionID).Scan(&manifest); e != nil {
		t.Fatal(e)
	}
	if manifest != nil {
		t.Fatal("manifest retained private context")
	}
	if e = f.p.QueryRow(ctx, `SELECT patch::text FROM memory_mutation WHERE id=$1::uuid`, mutation["mutation_id"]).Scan(&patch); e != nil {
		t.Fatal(e)
	}
	if patch != "{}" {
		t.Fatal("mutation retained deleted plaintext")
	}
	var storyStatus, storyContent, extractionStatus string
	var hasContext bool
	if e = f.p.QueryRow(ctx, `SELECT sd.status,sr.content FROM story_draft sd JOIN story_revision sr ON sr.id=sd.current_revision_id WHERE sd.id=$1::uuid`, story).Scan(&storyStatus, &storyContent); e != nil {
		t.Fatal(e)
	}
	if storyStatus != "SUPERSEDED" || storyContent != "[已按隐私请求删除]" {
		t.Fatal("direct-call story survived privacy deletion")
	}
	if e = f.p.QueryRow(ctx, `SELECT budget_snapshot ? 'story_context' FROM call_session WHERE id=$1::uuid`, m.SessionID).Scan(&hasContext); e != nil || hasContext {
		t.Fatalf("direct-call story context survived privacy deletion: %v", e)
	}
	if e = f.p.QueryRow(ctx, `SELECT status FROM outbox_event WHERE aggregate_id=$1::uuid AND event_type='call.story_extraction_requested'`, m.SessionID).Scan(&extractionStatus); e != nil || extractionStatus != "completed" {
		t.Fatalf("direct-call extraction job survived privacy deletion: status=%s err=%v", extractionStatus, e)
	}
	// Synthetic local/remote/backup evidence; this is not hardware acceptance.
	f.exec(t, `UPDATE memory_projection_job SET state='VISIBLE' 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 subject_id=$1::uuid`, f.subject)
	if e = processor.Handle(ctx, outbox.Event{AggregateID: request}); e == nil {
		t.Fatal("missing device stop was ignored")
	}
	f.exec(t, `UPDATE call_session SET local_audio_off=true,sdk_stopped=true,remote_state='STOPPED_OBSERVED' WHERE id=$1::uuid`, m.SessionID)
	if e = processor.Handle(ctx, outbox.Event{AggregateID: request}); e != nil {
		t.Fatal(e)
	}
	var state string
	if e = f.p.QueryRow(ctx, `SELECT status FROM privacy_request WHERE id=$1::uuid`, request).Scan(&state); e != nil {
		t.Fatal(e)
	}
	if state != "COMPLETED" {
		t.Fatal(state)
	}
}

func TestLateRevisionCannotRestoreDeletedSource(t *testing.T) {
	f := setup(t)
	ctx := context.Background()
	m, k, d := f.call(t)
	id, _ := f.add(t)
	f.exec(t, `INSERT INTO memory_source_suppression(subject_id,memory_id,source_refs,privacy_epoch,scope) SELECT subject_id,id,$2::jsonb,0,'SOURCE_AND_DERIVED' FROM curated_memory WHERE public_id=$1::uuid`, id, jsonBytes([]map[string]any{{"session_id": m.SessionID, "utterance_id": "erased-source"}}))
	event := Event{ID: uuid.NewString(), Seq: 1, Type: "TRANSCRIPT_REVISION", AttemptID: uuid.NewString(), Occurred: time.Now().UnixMilli(), Payload: map[string]any{"utterance_id": "erased-source", "source_revision": 2, "speaker": "USER", "source_final": true, "text": "must stay erased"}}
	result, e := f.s.DeviceRequest(ctx, d, nil, uuid.NewString(), "events", jsonBytes(EventsInput{CallKey: k, Events: []Event{event}}))
	if e != nil {
		t.Fatal(e)
	}
	raw := string(jsonBytes(result))
	if !strings.Contains(raw, "SOURCE_REDACTED") {
		t.Fatal(raw)
	}
	var count int
	if e = f.p.QueryRow(ctx, `SELECT count(*) FROM call_utterance WHERE session_id=$1::uuid`, m.SessionID).Scan(&count); e != nil {
		t.Fatal(e)
	}
	if count != 0 {
		t.Fatal("late final recreated source text")
	}
}

func TestStatusUsesDocumentKeyWithoutRenewingLease(t *testing.T) {
	f := setup(t)
	m, k, d := f.call(t)
	ctx := context.Background()
	var before, after time.Time
	if e := f.p.QueryRow(ctx, `SELECT lease_expires_at FROM call_session WHERE id=$1::uuid`, m.SessionID).Scan(&before); e != nil {
		t.Fatal(e)
	}
	input := StatusInput{SessionID: k.SessionID, BootID: k.BootID, HookID: k.HookID}
	if _, e := f.s.DeviceRequest(ctx, d, nil, uuid.NewString(), "status", jsonBytes(input)); e != nil {
		t.Fatal(e)
	}
	if e := f.p.QueryRow(ctx, `SELECT lease_expires_at FROM call_session WHERE id=$1::uuid`, m.SessionID).Scan(&after); e != nil {
		t.Fatal(e)
	}
	if !before.Equal(after) {
		t.Fatal("status renewed lease")
	}
	input.BootID = uuid.NewString()
	if _, e := f.s.DeviceRequest(ctx, d, nil, uuid.NewString(), "status", jsonBytes(input)); e == nil {
		t.Fatal("status accepted another boot identity")
	}
}