215 lines
11 KiB
Go
215 lines
11 KiB
Go
|
package callmemory
|
||
|
|
|
||
|
|
import (
|
||
|
|
"context"
|
||
|
|
"data-terminal-cloud/internal/identity"
|
||
|
|
"encoding/base64"
|
||
|
|
"errors"
|
||
|
|
"reflect"
|
||
|
|
"sort"
|
||
|
|
"strings"
|
||
|
|
"testing"
|
||
|
|
"time"
|
||
|
|
|
||
|
|
"github.com/google/uuid"
|
||
|
|
)
|
||
|
|
|
||
|
|
func queryMemoryIDs(t *testing.T, result map[string]any) []string {
|
||
|
|
t.Helper()
|
||
|
|
ids := []string{}
|
||
|
|
for _, raw := range result["items"].([]any) {
|
||
|
|
item := raw.(map[string]any)
|
||
|
|
ids = append(ids, item["memory_id"].(string))
|
||
|
|
if _, err := time.Parse(time.RFC3339Nano, item["created_at"].(string)); err != nil {
|
||
|
|
t.Fatalf("invalid created_at: %v", item)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
return ids
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestMemoryQueryCreatedAtPagination(t *testing.T) {
|
||
|
|
f := setup(t)
|
||
|
|
base := time.Date(2026, 10, 2, 10, 0, 0, 123456000, time.UTC)
|
||
|
|
ids := make([]string, 5)
|
||
|
|
for i := range ids {
|
||
|
|
ids[i], _ = f.add(t)
|
||
|
|
// The first three share a microsecond timestamp. The older records have
|
||
|
|
// newer updated_at values, which must not move them to the first page.
|
||
|
|
created := base
|
||
|
|
if i >= 3 {
|
||
|
|
created = base.Add(-time.Duration(i) * time.Hour)
|
||
|
|
}
|
||
|
|
f.exec(t, `UPDATE curated_memory SET created_at=$2,updated_at=$3 WHERE public_id=$1::uuid`, ids[i], created, base.Add(time.Duration(i)*time.Hour))
|
||
|
|
}
|
||
|
|
want := append([]string{}, ids[:3]...)
|
||
|
|
sort.Sort(sort.Reverse(sort.StringSlice(want)))
|
||
|
|
want = append(want, ids[3:]...)
|
||
|
|
in := QueryInput{ElderID: f.elder, Limit: 2}
|
||
|
|
first := f.request(t, "query", in)
|
||
|
|
got := queryMemoryIDs(t, first)
|
||
|
|
if !reflect.DeepEqual(got, want[:2]) || !strings.HasPrefix(first["next_cursor"].(string), "v1.") {
|
||
|
|
t.Fatalf("first page=%v want=%v cursor=%v", got, want[:2], first["next_cursor"])
|
||
|
|
}
|
||
|
|
// The legacy UUID anchor resolves within the authorized list and gives the
|
||
|
|
// same chronological continuation as the new timestamp/ID cursor.
|
||
|
|
in.Cursor = got[1]
|
||
|
|
legacy := f.request(t, "query", in)
|
||
|
|
in.Cursor = first["next_cursor"].(string)
|
||
|
|
second := f.request(t, "query", in)
|
||
|
|
if !reflect.DeepEqual(queryMemoryIDs(t, legacy), queryMemoryIDs(t, second)) {
|
||
|
|
t.Fatalf("legacy=%v new=%v", legacy, second)
|
||
|
|
}
|
||
|
|
// A new arrival and an edit cannot shift the saved page boundary.
|
||
|
|
newID, _ := f.add(t)
|
||
|
|
f.exec(t, `UPDATE curated_memory SET created_at=$2 WHERE public_id=$1::uuid`, newID, base.Add(time.Hour))
|
||
|
|
f.exec(t, `UPDATE curated_memory SET updated_at=$2 WHERE public_id=$1::uuid`, ids[4], base.Add(24*time.Hour))
|
||
|
|
// A self-contained cursor also survives removal of its anchor from the list.
|
||
|
|
f.exec(t, `UPDATE curated_memory SET status='DELETED' WHERE public_id=$1::uuid`, got[1])
|
||
|
|
second = f.request(t, "query", in)
|
||
|
|
got = append(got, queryMemoryIDs(t, second)...)
|
||
|
|
in.Cursor = second["next_cursor"].(string)
|
||
|
|
if in.Cursor == "" {
|
||
|
|
t.Fatal("missing second-page cursor")
|
||
|
|
}
|
||
|
|
last := f.request(t, "query", in)
|
||
|
|
got = append(got, queryMemoryIDs(t, last)...)
|
||
|
|
if !reflect.DeepEqual(got, want) || last["next_cursor"] != "" {
|
||
|
|
t.Fatalf("pages=%v want=%v last=%v", got, want, last)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestMemoryQueryCursorScopeAndValidation(t *testing.T) {
|
||
|
|
f, other := setup(t), setup(t)
|
||
|
|
id, _ := f.add(t)
|
||
|
|
foreignID, _ := other.add(t)
|
||
|
|
base := time.Date(2026, 10, 2, 0, 0, 0, 0, time.UTC)
|
||
|
|
f.exec(t, `UPDATE curated_memory SET created_at=$2 WHERE public_id=$1::uuid`, id, base)
|
||
|
|
bad := []QueryInput{
|
||
|
|
{ElderID: f.elder, Cursor: foreignID},
|
||
|
|
{ElderID: f.elder, Cursor: uuid.NewString()},
|
||
|
|
{ElderID: f.elder, Cursor: id, Type: "RELATIONSHIP"},
|
||
|
|
{ElderID: f.elder, Cursor: id, Status: "DISABLED"},
|
||
|
|
{ElderID: f.elder, Cursor: "invalid"},
|
||
|
|
{ElderID: f.elder, Cursor: "v1.invalid"},
|
||
|
|
{ElderID: f.elder, Cursor: "v1." + base64.RawURLEncoding.EncodeToString([]byte(`{"created_at":"invalid","id":"`+id+`"}`))},
|
||
|
|
{ElderID: f.elder, Cursor: "v1." + base64.RawURLEncoding.EncodeToString(jsonBytes(memoryQueryCursor{CreatedAt: base, ID: "invalid"}))},
|
||
|
|
{ElderID: f.elder, Cursor: strings.Repeat("x", 1025)},
|
||
|
|
}
|
||
|
|
for _, in := range bad {
|
||
|
|
_, err := f.s.MemoryRequest(context.Background(), f.principal, "query", jsonBytes(in))
|
||
|
|
var typed *Error
|
||
|
|
if !errors.As(err, &typed) || typed.Code != "INVALID_CURSOR" {
|
||
|
|
t.Fatalf("cursor=%q type=%q status=%q error=%v", in.Cursor, in.Type, in.Status, err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
// Even a valid timestamp/ID boundary cannot return another elder's records.
|
||
|
|
in := QueryInput{ElderID: f.elder, Cursor: "v1." + base64.RawURLEncoding.EncodeToString(jsonBytes(memoryQueryCursor{CreatedAt: base.Add(time.Hour), ID: foreignID}))}
|
||
|
|
if got := queryMemoryIDs(t, f.request(t, "query", in)); !reflect.DeepEqual(got, []string{id}) {
|
||
|
|
t.Fatalf("cross-elder cursor returned %v", got)
|
||
|
|
}
|
||
|
|
_, err := f.s.MemoryRequest(context.Background(), identity.Principal{Issuer: f.principal.Issuer, Subject: uuid.NewString()}, "query", jsonBytes(QueryInput{ElderID: f.elder, Cursor: id}))
|
||
|
|
var typed *Error
|
||
|
|
if !errors.As(err, &typed) || typed.Code != "MEMORY_ACCESS_DENIED" {
|
||
|
|
t.Fatalf("unrelated account query error=%v", err)
|
||
|
|
}
|
||
|
|
// A delegated reader cannot use a hidden sensitive record as a legacy anchor.
|
||
|
|
reader := identity.Principal{Issuer: f.principal.Issuer, Subject: uuid.NewString()}
|
||
|
|
user := uuid.NewString()
|
||
|
|
f.exec(t, `INSERT INTO user_account(id,tenant_id,oidc_subject,oidc_issuer) VALUES($1::uuid,$2::uuid,$3,$4)`, user, f.tenant, reader.Subject, reader.Issuer)
|
||
|
|
f.exec(t, `INSERT INTO household_member(tenant_id,household_id,user_id,role,status) VALUES($1::uuid,'10000000-0000-0000-0000-000000000201',$2::uuid,'viewer','active')`, f.tenant, user)
|
||
|
|
f.exec(t, `INSERT INTO memory_access_grant(subject_id,user_id,can_read,can_write,sensitive_read) VALUES($1::uuid,$2::uuid,true,false,false)`, f.subject, user)
|
||
|
|
f.exec(t, `UPDATE curated_memory SET sensitivity='sensitive' WHERE public_id=$1::uuid`, id)
|
||
|
|
_, err = f.s.MemoryRequest(context.Background(), reader, "query", jsonBytes(QueryInput{ElderID: f.elder, Cursor: id}))
|
||
|
|
if !errors.As(err, &typed) || typed.Code != "INVALID_CURSOR" {
|
||
|
|
t.Fatalf("hidden sensitive cursor error=%v", err)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func addStatusCall(t *testing.T, f fixture, ended bool, epoch int64, turns ...[2]string) string {
|
||
|
|
t.Helper()
|
||
|
|
sid := uuid.NewString()
|
||
|
|
f.exec(t, `INSERT INTO call_session(id,tenant_id,subject_id,device_id,boot_id,hook_event_id,call_generation,binding_generation,call_state,transcript_state,transcript_revision,memory_mode,privacy_epoch,ended_at,budget_snapshot) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$1::uuid,$1,1,1,CASE WHEN $5::boolean THEN 'ENDED' ELSE 'TALKING' END,'PARTIAL',$7,'HOT_ONLY',$6,CASE WHEN $5::boolean THEN now() END,'{"protocol":"DIRECT-CHAT-1"}')`, sid, f.tenant, f.subject, f.device, ended, epoch, len(turns))
|
||
|
|
for _, turn := range turns {
|
||
|
|
f.exec(t, `INSERT INTO call_utterance(session_id,utterance_id,source_revision,speaker,text_content,interrupted,source_at) VALUES($1::uuid,$2,0,$3,$4,false,now())`, sid, uuid.NewString(), turn[0], turn[1])
|
||
|
|
}
|
||
|
|
return sid
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestMemorySyncStatusInsufficientTranscript(t *testing.T) {
|
||
|
|
f, other := setup(t), setup(t)
|
||
|
|
f.s.ConfigureManagedSync(func(context.Context) error { return nil })
|
||
|
|
memoryID, _ := f.add(t)
|
||
|
|
f.exec(t, `UPDATE memory_user_state SET privacy_epoch=3 WHERE subject_id=$1::uuid`, f.subject)
|
||
|
|
user, assistant := [2]string{"USER", "今天走了两公里"}, [2]string{"ASSISTANT", "散步后记得休息"}
|
||
|
|
empty := addStatusCall(t, f, true, 3)
|
||
|
|
f.exec(t, `UPDATE call_session SET transcript_state='EMPTY' WHERE id=$1::uuid`, empty)
|
||
|
|
missingUser := addStatusCall(t, f, true, 3, assistant, [2]string{"USER", " \t\n "})
|
||
|
|
addStatusCall(t, f, true, 3, user)
|
||
|
|
addStatusCall(t, f, true, 3, [2]string{"UNKNOWN", "无法归属的文字"})
|
||
|
|
addStatusCall(t, f, true, 3, assistant, [2]string{"USER", "[deleted]"})
|
||
|
|
// Missing calls are counted once whether no job exists or several transcript
|
||
|
|
// revisions were skipped; SKIPPED itself is not the classification signal.
|
||
|
|
f.exec(t, `INSERT INTO call_memory_sync(session_id,transcript_revision,state) VALUES($1::uuid,0,'SKIPPED'),($2::uuid,1,'SKIPPED'),($2::uuid,2,'SKIPPED')`, empty, missingUser)
|
||
|
|
completeSkipped := addStatusCall(t, f, true, 3, user, assistant)
|
||
|
|
f.exec(t, `INSERT INTO call_memory_sync(session_id,transcript_revision,state) VALUES($1::uuid,2,'SKIPPED')`, completeSkipped)
|
||
|
|
addStatusCall(t, f, false, 3, assistant)
|
||
|
|
draining := addStatusCall(t, f, true, 3, assistant)
|
||
|
|
f.exec(t, `UPDATE call_session SET transcript_state='DRAINING',drain_started_at=now() WHERE id=$1::uuid`, draining)
|
||
|
|
collecting := addStatusCall(t, f, true, 3)
|
||
|
|
f.exec(t, `UPDATE call_session SET transcript_state='COLLECTING' WHERE id=$1::uuid`, collecting)
|
||
|
|
addStatusCall(t, f, true, 2, assistant) // an old privacy epoch
|
||
|
|
addStatusCall(t, other, true, 0, assistant)
|
||
|
|
legacy := addStatusCall(t, f, true, 3, assistant)
|
||
|
|
f.exec(t, `UPDATE call_session SET budget_snapshot='{"protocol":"CALL-MEMORY-1.1"}' WHERE id=$1::uuid`, legacy)
|
||
|
|
off := addStatusCall(t, f, true, 3, assistant)
|
||
|
|
f.exec(t, `UPDATE call_session SET memory_mode='OFF' WHERE id=$1::uuid`, off)
|
||
|
|
suppressed := addStatusCall(t, f, true, 3, assistant)
|
||
|
|
f.exec(t, `INSERT INTO memory_source_suppression(subject_id,memory_id,source_refs,privacy_epoch) SELECT $1::uuid,id,jsonb_build_array(jsonb_build_object('session_id',$3::text)),3 FROM curated_memory WHERE public_id=$2::uuid`, f.subject, memoryID, suppressed)
|
||
|
|
for _, state := range []string{"PENDING", "UPLOADED", "SYNCED"} {
|
||
|
|
sid := addStatusCall(t, f, true, 3, user, assistant)
|
||
|
|
f.exec(t, `INSERT INTO call_memory_sync(session_id,transcript_revision,state) VALUES($1::uuid,1,$2),($1::uuid,2,$2)`, sid, state)
|
||
|
|
}
|
||
|
|
status := f.request(t, "sync/status", map[string]any{"elder_id": f.elder})
|
||
|
|
for key, want := range map[string]float64{"insufficient_transcript_count": 5, "upload_pending_count": 1, "extracting_count": 1, "synced_call_count": 1} {
|
||
|
|
if status[key] != want {
|
||
|
|
t.Fatalf("%s=%v want=%v status=%v", key, status[key], want, status)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
// Late valid text fixes the classification without needing a worker or retry.
|
||
|
|
f.exec(t, `INSERT INTO call_utterance(session_id,utterance_id,source_revision,speaker,text_content,interrupted,source_at) VALUES($1::uuid,'late-user',0,'USER','我也去散步了',false,now())`, missingUser)
|
||
|
|
status = f.request(t, "sync/status", map[string]any{"elder_id": f.elder})
|
||
|
|
if status["insufficient_transcript_count"] != float64(4) {
|
||
|
|
t.Fatalf("late text status=%v", status)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func TestMemorySyncStatusInsufficientRespectsAuthorization(t *testing.T) {
|
||
|
|
for _, consent := range []string{"personalized_memory", "transcript_storage"} {
|
||
|
|
for _, expire := range []bool{false, true} {
|
||
|
|
name := consent + "/withdrawn"
|
||
|
|
if expire {
|
||
|
|
name = consent + "/expired"
|
||
|
|
}
|
||
|
|
t.Run(name, func(t *testing.T) {
|
||
|
|
f := setup(t)
|
||
|
|
addStatusCall(t, f, true, 0)
|
||
|
|
in := map[string]any{"elder_id": f.elder}
|
||
|
|
if status := f.request(t, "sync/status", in); status["insufficient_transcript_count"] != float64(1) {
|
||
|
|
t.Fatalf("authorized status=%v", status)
|
||
|
|
}
|
||
|
|
if expire {
|
||
|
|
f.exec(t, `UPDATE consent_record SET granted_at=now()-interval '1 day',expires_at=now()-interval '1 second' WHERE subject_id=$1::uuid AND consent_type=$2`, f.subject, consent)
|
||
|
|
} else {
|
||
|
|
f.exec(t, `UPDATE consent_record SET withdrawn_at=now() WHERE subject_id=$1::uuid AND consent_type=$2`, f.subject, consent)
|
||
|
|
}
|
||
|
|
// Match the new epoch so authorization, independently of epoch filtering,
|
||
|
|
// must suppress these calls from the missing-transcript count.
|
||
|
|
f.exec(t, `UPDATE call_session SET privacy_epoch=(SELECT privacy_epoch FROM memory_user_state WHERE subject_id=$1::uuid) WHERE subject_id=$1::uuid`, f.subject)
|
||
|
|
if status := f.request(t, "sync/status", in); status["insufficient_transcript_count"] != float64(0) {
|
||
|
|
t.Fatalf("unauthorized status=%v", status)
|
||
|
|
}
|
||
|
|
})
|
||
|
|
}
|
||
|
|
}
|
||
|
|
}
|