Codex Local
bf1233d5df
ci / verify (push) Canceled after 0s
Add customer and operator login, BLE provisioning, device fleet controls, call memory, memoir generation, and refreshed family and maintenance portals. Include API contracts, migrations, firmware sources, tests, assets, and deployment documentation.
727 lines
32 KiB
Go
727 lines
32 KiB
Go
package callmemory
|
|
|
|
import (
|
|
"context"
|
|
"crypto/hmac"
|
|
"crypto/rand"
|
|
"data-terminal-cloud/internal/identity"
|
|
"data-terminal-cloud/internal/outbox"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"errors"
|
|
"github.com/google/uuid"
|
|
"github.com/jackc/pgx/v5"
|
|
"strings"
|
|
"time"
|
|
"unicode/utf8"
|
|
)
|
|
|
|
type queryer interface {
|
|
QueryRow(context.Context, string, ...any) pgx.Row
|
|
}
|
|
type Access struct {
|
|
Tenant, Subject, User string
|
|
Sensitive bool
|
|
Write bool
|
|
Manage bool
|
|
}
|
|
|
|
// Memory grants are independent of device management. Owners retain access;
|
|
// delegated household roles must receive an explicit memory grant.
|
|
func ResolveAccess(ctx context.Context, q queryer, p identity.Principal, elder string, write bool) (Access, error) {
|
|
var a Access
|
|
var read, edit bool
|
|
if !validUUID(elder) {
|
|
return a, fault(400, "INVALID_ELDER_ID")
|
|
}
|
|
e := q.QueryRow(ctx, `SELECT sp.tenant_id::text,sp.id::text,ua.id::text,(hm.role='owner' OR COALESCE(g.can_read,false)),(hm.role='owner' OR COALESCE(g.can_write,false)),(hm.role='owner' OR COALESCE(g.sensitive_read,false)),hm.role='owner' FROM subject_profile sp JOIN household h ON h.id=sp.household_id AND h.status='active' JOIN household_member hm ON hm.household_id=h.id AND hm.status='active' JOIN user_account ua ON ua.id=hm.user_id AND ua.tenant_id=sp.tenant_id AND ua.status='active' LEFT JOIN memory_access_grant g ON g.subject_id=sp.id AND g.user_id=ua.id WHERE sp.public_id=$1::uuid AND sp.status IN ('active','suspended') AND ua.oidc_issuer=$2 AND ua.oidc_subject=$3`, elder, p.Issuer, p.Subject).Scan(&a.Tenant, &a.Subject, &a.User, &read, &edit, &a.Sensitive, &a.Manage)
|
|
if errors.Is(e, pgx.ErrNoRows) {
|
|
return a, fault(403, "MEMORY_ACCESS_DENIED")
|
|
}
|
|
if e != nil {
|
|
return a, e
|
|
}
|
|
if !read || (write && !edit) {
|
|
return a, fault(403, "MEMORY_ACCESS_DENIED")
|
|
}
|
|
a.Write = edit
|
|
return a, nil
|
|
}
|
|
|
|
type MemoryPatch struct {
|
|
Content *string `json:"content,omitempty"`
|
|
Type *string `json:"memory_type,omitempty"`
|
|
Policy *string `json:"proactive_policy,omitempty"`
|
|
Sensitivity *string `json:"sensitivity,omitempty"`
|
|
EntityKey *string `json:"entity_key,omitempty"`
|
|
Status *string `json:"status,omitempty"`
|
|
ItemStatus *string `json:"item_status,omitempty"`
|
|
PlannedAt *time.Time `json:"planned_at,omitempty"`
|
|
CompletedAt *time.Time `json:"completed_at,omitempty"`
|
|
CompletionBasis json.RawMessage `json:"completion_basis,omitempty"`
|
|
}
|
|
type MutationInput struct {
|
|
ID string `json:"mutation_id"`
|
|
ElderID string `json:"elder_id"`
|
|
Action string `json:"action"`
|
|
Targets []string `json:"targets"`
|
|
Expected map[string]int64 `json:"expected_record_versions"`
|
|
Patch MemoryPatch `json:"patch"`
|
|
Scope string `json:"scope,omitempty"`
|
|
SourceRefs []map[string]any `json:"-"`
|
|
}
|
|
type ResolveInput struct {
|
|
ID string `json:"mutation_id"`
|
|
Revision int64 `json:"expected_mutation_revision"`
|
|
Targets []string `json:"targets"`
|
|
Expected map[string]int64 `json:"expected_record_versions"`
|
|
Patch MemoryPatch `json:"patch"`
|
|
}
|
|
type ConfirmInput struct {
|
|
ID string `json:"mutation_id"`
|
|
Revision int64 `json:"expected_mutation_revision"`
|
|
Digest string `json:"proposal_digest"`
|
|
Token string `json:"confirmation_token"`
|
|
Decision string `json:"decision"`
|
|
}
|
|
type QueryInput struct {
|
|
ElderID string `json:"elder_id"`
|
|
Type string `json:"memory_type,omitempty"`
|
|
Status string `json:"status,omitempty"`
|
|
Cursor string `json:"cursor,omitempty"`
|
|
Limit int `json:"limit,omitempty"`
|
|
}
|
|
type mutation struct {
|
|
ID, ElderID, Subject, Actor, Action, State, Scope, Digest, TokenHash string
|
|
Revision int64
|
|
Targets []string
|
|
Expected map[string]int64
|
|
Patch MemoryPatch
|
|
SourceRefs []map[string]any
|
|
Expires *time.Time
|
|
}
|
|
|
|
func (s *Service) MemoryRequest(ctx context.Context, p identity.Principal, op string, raw []byte) (any, error) {
|
|
if op == "viking/query" || op == "viking/delete" {
|
|
return s.vikingRequest(ctx, p, op, raw)
|
|
}
|
|
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
|
defer cancel()
|
|
tx, e := s.pool.Begin(ctx)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
defer tx.Rollback(context.Background())
|
|
var out any
|
|
switch op {
|
|
case "calls/recordings", "calls/recording-access":
|
|
out, e = s.callRecordings(ctx, tx, p, op, raw)
|
|
case "calls/query", "calls/turns":
|
|
out, e = s.callRecords(ctx, tx, p, op, raw)
|
|
case "access/query", "access/grant":
|
|
out, e = s.accessRequest(ctx, tx, p, op, raw)
|
|
case "query":
|
|
var in QueryInput
|
|
if e = StrictJSON(raw, &in); e == nil {
|
|
out, e = s.query(ctx, tx, p, in)
|
|
}
|
|
case "sync/status":
|
|
var in struct {
|
|
ElderID string `json:"elder_id"`
|
|
}
|
|
if e = StrictJSON(raw, &in); e == nil {
|
|
var a Access
|
|
a, e = ResolveAccess(ctx, tx, p, in.ElderID, false)
|
|
if e == nil {
|
|
out, e = s.syncStatus(ctx, tx, a.Subject)
|
|
}
|
|
}
|
|
case "mutations":
|
|
var in MutationInput
|
|
if e = StrictJSON(raw, &in); e == nil {
|
|
out, e = s.mutate(ctx, tx, p, in, hash(raw))
|
|
}
|
|
case "mutations/status":
|
|
var in struct {
|
|
ID string `json:"mutation_id"`
|
|
}
|
|
if e = StrictJSON(raw, &in); e == nil {
|
|
var m mutation
|
|
m, e = loadMutation(ctx, tx, p, in.ID, false)
|
|
if e == nil {
|
|
out, e = s.mutationStatus(ctx, tx, m.ID)
|
|
}
|
|
}
|
|
case "mutations/resolve":
|
|
var in ResolveInput
|
|
if e = StrictJSON(raw, &in); e == nil {
|
|
out, e = s.resolve(ctx, tx, p, in)
|
|
}
|
|
case "mutations/confirm":
|
|
var in ConfirmInput
|
|
if e = StrictJSON(raw, &in); e == nil {
|
|
out, e = s.confirm(ctx, tx, p, in)
|
|
}
|
|
default:
|
|
e = fault(404, "NOT_FOUND")
|
|
}
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if e = tx.Commit(ctx); e != nil {
|
|
return nil, e
|
|
}
|
|
return out, nil
|
|
}
|
|
func validateMutation(in MutationInput) error {
|
|
if !validUUID(in.ID) || !oneOf(in.Action, "ADD", "CORRECT", "SET_USAGE", "DELETE") || len(in.Targets) > 50 || !oneOf(in.Scope, "", "MEMORY_AND_DERIVED", "SOURCE_AND_DERIVED") {
|
|
return fault(400, "INVALID_MUTATION")
|
|
}
|
|
seen := map[string]bool{}
|
|
for _, id := range in.Targets {
|
|
if !validUUID(id) || seen[id] || in.Expected[id] < 1 {
|
|
return fault(400, "INVALID_TARGET_VERSION")
|
|
}
|
|
seen[id] = true
|
|
}
|
|
p := in.Patch
|
|
if p.Content != nil && (utf8.RuneCountInString(strings.TrimSpace(*p.Content)) < 1 || utf8.RuneCountInString(*p.Content) > 4000) {
|
|
return fault(400, "INVALID_MEMORY_CONTENT")
|
|
}
|
|
if p.Type != nil && !oneOf(*p.Type, "RELATIONSHIP", "PROFILE", "EVENT", "PREFERENCE", "OPEN_ITEM", "STORY", "INTERACTION") {
|
|
return fault(400, "INVALID_MEMORY_TYPE")
|
|
}
|
|
if p.Policy != nil && !oneOf(*p.Policy, "ALLOWED", "RESPOND_ONLY", "NEVER") {
|
|
return fault(400, "INVALID_USAGE_POLICY")
|
|
}
|
|
if p.Sensitivity != nil && !oneOf(*p.Sensitivity, "normal", "sensitive", "high") {
|
|
return fault(400, "INVALID_SENSITIVITY")
|
|
}
|
|
if p.EntityKey != nil && len(*p.EntityKey) > 128 {
|
|
return fault(400, "INVALID_ENTITY_KEY")
|
|
}
|
|
if p.Status != nil && !oneOf(*p.Status, "ACTIVE", "DISABLED") {
|
|
return fault(400, "INVALID_MEMORY_STATUS")
|
|
}
|
|
if p.ItemStatus != nil && !oneOf(*p.ItemStatus, "OPEN", "COMPLETED", "CANCELLED", "UNKNOWN") {
|
|
return fault(400, "INVALID_ITEM_STATUS")
|
|
}
|
|
if p.ItemStatus != nil && *p.ItemStatus == "COMPLETED" && (p.CompletedAt == nil || len(p.CompletionBasis) == 0 || string(p.CompletionBasis) == "null") {
|
|
return fault(400, "COMPLETION_EVIDENCE_REQUIRED")
|
|
}
|
|
if in.Action == "ADD" && (len(in.Targets) > 0 || p.Content == nil) {
|
|
return fault(400, "INVALID_ADD")
|
|
}
|
|
if in.Action == "DELETE" && string(jsonBytes(p)) != "{}" {
|
|
return fault(400, "DELETE_PATCH_FORBIDDEN")
|
|
}
|
|
if in.Action == "CORRECT" && string(jsonBytes(p)) == "{}" {
|
|
return fault(400, "EMPTY_CORRECTION")
|
|
}
|
|
if in.Action == "SET_USAGE" && (p.Policy == nil && p.Status == nil || p.Content != nil || p.Type != nil || p.EntityKey != nil || p.ItemStatus != nil || p.PlannedAt != nil || p.CompletedAt != nil || len(p.CompletionBasis) > 0) {
|
|
return fault(400, "INVALID_USAGE_PATCH")
|
|
}
|
|
return nil
|
|
}
|
|
func (s *Service) mutate(ctx context.Context, tx pgx.Tx, p identity.Principal, in MutationInput, digest string) (any, error) {
|
|
if e := validateMutation(in); e != nil {
|
|
return nil, e
|
|
}
|
|
a, e := ResolveAccess(ctx, tx, p, in.ElderID, true)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if _, e = tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,12))`, in.ID); e != nil {
|
|
return nil, e
|
|
}
|
|
var oldHash, oldSubject string
|
|
e = tx.QueryRow(ctx, `SELECT request_digest,subject_id::text FROM memory_mutation WHERE id=$1::uuid`, in.ID).Scan(&oldHash, &oldSubject)
|
|
if e == nil {
|
|
if oldSubject != a.Subject || oldHash != digest {
|
|
return nil, fault(409, "MUTATION_CONFLICT")
|
|
}
|
|
return s.mutationStatus(ctx, tx, in.ID)
|
|
}
|
|
if !errors.Is(e, pgx.ErrNoRows) {
|
|
return nil, e
|
|
}
|
|
if in.Scope == "" {
|
|
in.Scope = "MEMORY_AND_DERIVED"
|
|
}
|
|
state := "RECEIVED"
|
|
if in.Action != "ADD" && len(in.Targets) == 0 {
|
|
state = "NEEDS_CONFIRMATION"
|
|
}
|
|
_, e = tx.Exec(ctx, `INSERT INTO memory_mutation(id,tenant_id,subject_id,actor_id,action,request_digest,state,targets,expected_versions,patch,scope) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$5,$6,$7,$8::jsonb,$9::jsonb,$10::jsonb,$11)`, in.ID, a.Tenant, a.Subject, a.User, in.Action, digest, state, jsonBytes(in.Targets), jsonBytes(in.Expected), jsonBytes(in.Patch), in.Scope)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if state == "RECEIVED" {
|
|
if e = s.applyMutation(ctx, tx, a, in); e != nil {
|
|
return nil, e
|
|
}
|
|
}
|
|
return s.mutationStatus(ctx, tx, in.ID)
|
|
}
|
|
func loadMutation(ctx context.Context, tx pgx.Tx, p identity.Principal, id string, write bool) (mutation, error) {
|
|
var m mutation
|
|
if !validUUID(id) {
|
|
return m, fault(400, "INVALID_MUTATION_ID")
|
|
}
|
|
var targets, versions, patch, refs []byte
|
|
e := tx.QueryRow(ctx, `SELECT mm.id::text,sp.public_id::text,mm.subject_id::text,COALESCE(mm.actor_id::text,''),mm.action,mm.state,mm.scope,COALESCE(mm.proposal_digest,''),COALESCE(mm.confirmation_hash,''),mm.revision,mm.targets,mm.expected_versions,mm.patch,mm.confirmation_expires_at,mm.source_refs FROM memory_mutation mm JOIN subject_profile sp ON sp.id=mm.subject_id WHERE mm.id=$1::uuid FOR UPDATE OF mm`, id).Scan(&m.ID, &m.ElderID, &m.Subject, &m.Actor, &m.Action, &m.State, &m.Scope, &m.Digest, &m.TokenHash, &m.Revision, &targets, &versions, &patch, &m.Expires, &refs)
|
|
if errors.Is(e, pgx.ErrNoRows) {
|
|
return m, fault(404, "MUTATION_NOT_FOUND")
|
|
}
|
|
if e != nil {
|
|
return m, e
|
|
}
|
|
a, e := ResolveAccess(ctx, tx, p, m.ElderID, write)
|
|
if e != nil {
|
|
return m, e
|
|
}
|
|
if e = json.Unmarshal(targets, &m.Targets); e != nil {
|
|
return m, e
|
|
}
|
|
if e = json.Unmarshal(versions, &m.Expected); e != nil {
|
|
return m, e
|
|
}
|
|
if e = json.Unmarshal(refs, &m.SourceRefs); e != nil {
|
|
return m, e
|
|
}
|
|
e = json.Unmarshal(patch, &m.Patch)
|
|
if e != nil {
|
|
return m, e
|
|
}
|
|
if !a.Sensitive {
|
|
var sensitive bool
|
|
e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM curated_memory WHERE subject_id=$1::uuid AND public_id IN(SELECT value::uuid FROM jsonb_array_elements_text($2::jsonb)) AND sensitivity<>'normal')`, m.Subject, targets).Scan(&sensitive)
|
|
if e != nil {
|
|
return m, e
|
|
}
|
|
if sensitive || (m.Patch.Sensitivity != nil && *m.Patch.Sensitivity != "normal") {
|
|
return m, fault(403, "SENSITIVE_MEMORY_ACCESS_DENIED")
|
|
}
|
|
}
|
|
return m, e
|
|
}
|
|
func (s *Service) resolve(ctx context.Context, tx pgx.Tx, p identity.Principal, in ResolveInput) (any, error) {
|
|
m, e := loadMutation(ctx, tx, p, in.ID, true)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if m.State != "NEEDS_CONFIRMATION" || m.Revision != in.Revision {
|
|
return nil, fault(409, "MUTATION_VERSION_CONFLICT")
|
|
}
|
|
mi := MutationInput{ID: m.ID, ElderID: m.ElderID, Action: m.Action, Targets: in.Targets, Expected: in.Expected, Patch: in.Patch, Scope: m.Scope}
|
|
if m.Action != "ADD" && len(in.Targets) == 0 {
|
|
return nil, fault(400, "TARGET_REQUIRED")
|
|
}
|
|
if e = validateMutation(mi); e != nil {
|
|
return nil, e
|
|
}
|
|
a, e := ResolveAccess(ctx, tx, p, m.ElderID, true)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if e = checkTargets(ctx, tx, a, mi); e != nil {
|
|
return nil, e
|
|
}
|
|
tokenBytes := make([]byte, 32)
|
|
if _, e = rand.Read(tokenBytes); e != nil {
|
|
return nil, e
|
|
}
|
|
token := base64.RawURLEncoding.EncodeToString(tokenBytes)
|
|
digest := hash(jsonBytes(mi))
|
|
expires := s.now().Add(10 * time.Minute)
|
|
_, e = tx.Exec(ctx, `UPDATE memory_mutation SET targets=$2::jsonb,expected_versions=$3::jsonb,patch=$4::jsonb,proposal_digest=$5,confirmation_hash=$6,confirmation_expires_at=$7,confirmation_used_at=NULL,actor_id=$8::uuid,revision=revision+1,updated_at=now() WHERE id=$1::uuid`, m.ID, jsonBytes(in.Targets), jsonBytes(in.Expected), jsonBytes(in.Patch), digest, hash([]byte(token)), expires, a.User)
|
|
return map[string]any{"mutation_id": m.ID, "mutation_revision": m.Revision + 1, "proposal_digest": digest, "confirmation_token": token, "expires_at_ms": expires.UnixMilli(), "targets": in.Targets, "patch": in.Patch, "action": m.Action, "scope": m.Scope, "expected_record_versions": in.Expected}, e
|
|
}
|
|
func (s *Service) confirm(ctx context.Context, tx pgx.Tx, p identity.Principal, in ConfirmInput) (any, error) {
|
|
if !oneOf(in.Decision, "CONFIRM", "CANCEL") {
|
|
return nil, fault(400, "INVALID_DECISION")
|
|
}
|
|
m, e := loadMutation(ctx, tx, p, in.ID, true)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if oneOf(m.State, "PROPAGATING", "COMPLETED", "BLOCKED", "CANCELLED", "REJECTED", "EXPIRED") {
|
|
return s.mutationStatus(ctx, tx, m.ID)
|
|
}
|
|
a, e := ResolveAccess(ctx, tx, p, m.ElderID, true)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if m.Revision != in.Revision || m.State != "NEEDS_CONFIRMATION" || m.Actor != a.User || m.Digest == "" || m.Digest != in.Digest || !hmac.Equal([]byte(m.TokenHash), []byte(hash([]byte(in.Token)))) {
|
|
return nil, fault(409, "CONFIRMATION_CONFLICT")
|
|
}
|
|
if m.Expires == nil || !m.Expires.After(s.now()) {
|
|
_, e = tx.Exec(ctx, `UPDATE memory_mutation SET state='EXPIRED',confirmation_hash=NULL,revision=revision+1 WHERE id=$1::uuid`, m.ID)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
return s.mutationStatus(ctx, tx, m.ID)
|
|
}
|
|
if in.Decision == "CANCEL" {
|
|
_, e = tx.Exec(ctx, `UPDATE memory_mutation SET state='CANCELLED',confirmation_hash=NULL,confirmation_used_at=now(),revision=revision+1 WHERE id=$1::uuid`, m.ID)
|
|
} else {
|
|
mi := MutationInput{ID: m.ID, ElderID: m.ElderID, Action: m.Action, Targets: m.Targets, Expected: m.Expected, Patch: m.Patch, Scope: m.Scope, SourceRefs: m.SourceRefs}
|
|
e = s.applyMutation(ctx, tx, a, mi)
|
|
}
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
return s.mutationStatus(ctx, tx, m.ID)
|
|
}
|
|
func checkTargets(ctx context.Context, tx pgx.Tx, a Access, in MutationInput) error {
|
|
for _, id := range in.Targets {
|
|
var version int64
|
|
var sensitivity, state string
|
|
e := tx.QueryRow(ctx, `SELECT version,sensitivity,status FROM curated_memory WHERE public_id=$1::uuid AND subject_id=$2::uuid AND tenant_id=$3::uuid FOR UPDATE`, id, a.Subject, a.Tenant).Scan(&version, &sensitivity, &state)
|
|
if errors.Is(e, pgx.ErrNoRows) {
|
|
return fault(403, "MEMORY_ACCESS_DENIED")
|
|
}
|
|
if e != nil {
|
|
return e
|
|
}
|
|
if version != in.Expected[id] || oneOf(state, "DELETED", "PENDING_DELETE") {
|
|
return fault(409, "RECORD_VERSION_CONFLICT")
|
|
}
|
|
if sensitivity != "normal" && !a.Sensitive {
|
|
return fault(403, "SENSITIVE_MEMORY_ACCESS_DENIED")
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
func (s *Service) applyMutation(ctx context.Context, tx pgx.Tx, a Access, in MutationInput) error {
|
|
if e := validateMutation(in); e != nil {
|
|
return e
|
|
}
|
|
if in.Action != "DELETE" {
|
|
var consent bool
|
|
e := tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM consent_record WHERE subject_id=$1::uuid AND consent_type='personalized_memory' AND withdrawn_at IS NULL AND (expires_at IS NULL OR expires_at>now()))`, a.Subject).Scan(&consent)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
if !consent {
|
|
return fault(403, "MEMORY_CONSENT_REQUIRED")
|
|
}
|
|
}
|
|
if _, e := tx.Exec(ctx, `INSERT INTO memory_user_state(subject_id,tenant_id) VALUES($1::uuid,$2::uuid) ON CONFLICT DO NOTHING`, a.Subject, a.Tenant); e != nil {
|
|
return e
|
|
}
|
|
if _, e := tx.Exec(ctx, `SELECT subject_id FROM memory_user_state WHERE subject_id=$1::uuid FOR UPDATE`, a.Subject); e != nil {
|
|
return e
|
|
}
|
|
if e := checkTargets(ctx, tx, a, in); e != nil {
|
|
return e
|
|
}
|
|
ids := in.Targets
|
|
requestedSources := map[string]bool{}
|
|
if in.Action == "DELETE" && in.Scope == "SOURCE_AND_DERIVED" {
|
|
for _, id := range in.Targets {
|
|
requestedSources[id] = true
|
|
}
|
|
var err error
|
|
ids, err = expandSourceDeletionTargets(ctx, tx, a, in.Targets)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
_, e := tx.Exec(ctx, `UPDATE memory_mutation SET state='APPLYING',confirmation_hash=NULL,confirmation_used_at=CASE WHEN proposal_digest IS NOT NULL THEN now() END WHERE id=$1::uuid`, in.ID)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
if in.Action == "ADD" {
|
|
ids = []string{uuid.NewString()}
|
|
}
|
|
for _, publicID := range ids {
|
|
var id, content, kind, policy, sensitivity, entity, status string
|
|
var item *string
|
|
var planned, completed *time.Time
|
|
var basis []byte
|
|
var version int64
|
|
var refs []byte
|
|
if in.Action == "ADD" {
|
|
id = uuid.NewString()
|
|
version = 0
|
|
kind = "EVENT"
|
|
policy = "ALLOWED"
|
|
sensitivity = "normal"
|
|
status = "ACTIVE"
|
|
refs = jsonBytes([]map[string]any{{"type": "user_edit", "mutation_id": in.ID, "actor_id": a.User}})
|
|
if len(in.SourceRefs) > 0 {
|
|
refs = jsonBytes(in.SourceRefs)
|
|
}
|
|
} else {
|
|
e = tx.QueryRow(ctx, `SELECT cm.id::text,mv.content,cm.memory_type,cm.proactive_policy,cm.sensitivity,cm.entity_key,cm.status,cm.item_status,cm.planned_at,cm.completed_at,cm.completion_basis,cm.version,mv.source_refs FROM curated_memory cm JOIN memory_version mv ON mv.id=cm.current_version_id WHERE cm.public_id=$1::uuid AND cm.subject_id=$2::uuid FOR UPDATE OF cm`, publicID, a.Subject).Scan(&id, &content, &kind, &policy, &sensitivity, &entity, &status, &item, &planned, &completed, &basis, &version, &refs)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
// A sibling deletion already in flight still belongs to this
|
|
// mutation's projection and backup completion checks.
|
|
if in.Action == "DELETE" && status == "PENDING_DELETE" {
|
|
continue
|
|
}
|
|
}
|
|
p := in.Patch
|
|
if in.Action == "CORRECT" && len(in.SourceRefs) > 0 {
|
|
refs = jsonBytes(in.SourceRefs)
|
|
}
|
|
if p.Content != nil {
|
|
content = strings.TrimSpace(*p.Content)
|
|
}
|
|
if p.Type != nil {
|
|
kind = *p.Type
|
|
if kind != "OPEN_ITEM" && p.ItemStatus == nil {
|
|
item, planned, completed, basis = nil, nil, nil, nil
|
|
}
|
|
}
|
|
if p.Policy != nil {
|
|
policy = *p.Policy
|
|
}
|
|
if p.Sensitivity != nil {
|
|
sensitivity = *p.Sensitivity
|
|
}
|
|
if p.EntityKey != nil {
|
|
entity = *p.EntityKey
|
|
}
|
|
if p.Status != nil {
|
|
status = *p.Status
|
|
}
|
|
if p.ItemStatus != nil {
|
|
item = p.ItemStatus
|
|
if *item != "COMPLETED" {
|
|
completed, basis = nil, nil
|
|
}
|
|
}
|
|
if p.PlannedAt != nil {
|
|
planned = p.PlannedAt
|
|
}
|
|
if p.CompletedAt != nil {
|
|
completed = p.CompletedAt
|
|
}
|
|
if len(p.CompletionBasis) > 0 {
|
|
basis = p.CompletionBasis
|
|
}
|
|
if sensitivity != "normal" && !a.Sensitive {
|
|
return fault(403, "SENSITIVE_MEMORY_ACCESS_DENIED")
|
|
}
|
|
if kind == "OPEN_ITEM" && item == nil {
|
|
x := "OPEN"
|
|
item = &x
|
|
}
|
|
if kind != "OPEN_ITEM" && item != nil {
|
|
return fault(400, "ITEM_TYPE_CONFLICT")
|
|
}
|
|
if item != nil && *item == "COMPLETED" && (completed == nil || len(basis) == 0) {
|
|
return fault(400, "COMPLETION_EVIDENCE_REQUIRED")
|
|
}
|
|
if in.Action == "DELETE" {
|
|
status = "PENDING_DELETE"
|
|
policy = "NEVER"
|
|
}
|
|
if in.Action == "ADD" {
|
|
_, e = tx.Exec(ctx, `INSERT INTO curated_memory(id,public_id,tenant_id,subject_id,status,memory_type,proactive_policy,sensitivity,entity_key,item_status,planned_at,completed_at,completion_basis,source_time) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,$5,$6,$7,$8,$9,$10,$11,$12,$13::jsonb,now())`, id, publicID, a.Tenant, a.Subject, status, kind, policy, sensitivity, entity, item, planned, completed, nullableJSON(basis))
|
|
if e != nil {
|
|
return e
|
|
}
|
|
}
|
|
versionID := uuid.NewString()
|
|
_, e = tx.Exec(ctx, `INSERT INTO memory_version(id,memory_id,version_no,content,source_refs,confirmed_by,change_reason,content_digest) SELECT $1::uuid,$2::uuid,COALESCE(max(version_no),0)+1,$3,$4::jsonb,NULLIF($5,'')::uuid,$6,$7 FROM memory_version WHERE memory_id=$2::uuid`, versionID, id, content, refs, a.User, "mutation:"+in.ID, hash([]byte(content)))
|
|
if e != nil {
|
|
return e
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE curated_memory SET current_version_id=$2::uuid,version=$3,status=$4,memory_type=$5,proactive_policy=$6,sensitivity=$7,entity_key=$8,item_status=$9,planned_at=$10,completed_at=$11,completion_basis=$12::jsonb,source_time=now(),updated_at=now() WHERE id=$1::uuid`, id, versionID, version+1, status, kind, policy, sensitivity, entity, item, planned, completed, nullableJSON(basis))
|
|
if e != nil {
|
|
return e
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE memory_index_entry SET embedding=NULL,status='STALE',updated_at=now() WHERE memory_id=$1::uuid AND status IN ('ACTIVE','PENDING')`, id)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
if status == "ACTIVE" && policy != "NEVER" && sensitivity == "normal" {
|
|
_, e = tx.Exec(ctx, `INSERT INTO memory_index_entry(tenant_id,subject_id,memory_id,memory_version_id,index_type,status) VALUES($1::uuid,$2::uuid,$3::uuid,$4::uuid,'pgvector','PENDING') ON CONFLICT DO NOTHING`, a.Tenant, a.Subject, id, versionID)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
dedupe := "memory-index-v1:" + versionID
|
|
if _, e = outbox.NewStore(s.pool).Enqueue(ctx, tx, outbox.NewEvent{TenantID: &a.Tenant, AggregateType: "curated_memory", AggregateID: id, EventType: "memory.index_requested", DeduplicationKey: &dedupe, Payload: jsonBytes(map[string]any{"memory_version_id": versionID})}); e != nil {
|
|
return e
|
|
}
|
|
}
|
|
if in.Action == "DELETE" {
|
|
deletionScope := in.Scope
|
|
if deletionScope == "SOURCE_AND_DERIVED" && !requestedSources[publicID] {
|
|
deletionScope = "MEMORY_AND_DERIVED"
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE memory_deletion_ledger SET scope=$2 WHERE memory_id=$1::uuid AND backup_cleanup_verified_at IS NULL`, id, deletionScope)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE memory_source_suppression SET scope=$2 WHERE memory_id=$1::uuid`, id, deletionScope)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
}
|
|
}
|
|
stages := map[string]string{"main_store": "DONE", "serving_projection": "PENDING", "derived_context": "NOT_REQUIRED", "active_context": "NOT_REQUIRED", "source_cleanup": "NOT_REQUIRED", "backup_cleanup": "NOT_REQUIRED"}
|
|
required := []string{"main_store", "serving_projection"}
|
|
if in.Action != "ADD" {
|
|
stages["derived_context"], stages["active_context"] = "DONE", "PENDING"
|
|
required = append(required, "derived_context", "active_context")
|
|
}
|
|
if in.Action == "DELETE" {
|
|
stages["source_cleanup"], stages["backup_cleanup"] = "PENDING", "PENDING"
|
|
required = append(required, "source_cleanup", "backup_cleanup")
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE memory_mutation mm SET state='PROPAGATING',targets=$2::jsonb,revision=revision+1,memory_revision=us.memory_revision,privacy_epoch=us.privacy_epoch,stage_states=$3::jsonb,required_stages=$4::jsonb,updated_at=now() FROM memory_user_state us WHERE mm.id=$1::uuid AND us.subject_id=mm.subject_id`, in.ID, jsonBytes(ids), jsonBytes(stages), jsonBytes(required))
|
|
return e
|
|
}
|
|
func nullableJSON(b []byte) any {
|
|
if len(b) == 0 {
|
|
return nil
|
|
}
|
|
return b
|
|
}
|
|
func (s *Service) mutationStatus(ctx context.Context, tx pgx.Tx, id string) (any, error) {
|
|
var out []byte
|
|
e := tx.QueryRow(ctx, `SELECT jsonb_build_object('mutation_id',id,'mutation_revision',revision,'mutation_state',state,'action',action,'targets',targets,'patch',CASE WHEN state='NEEDS_CONFIRMATION' THEN patch ELSE '{}'::jsonb END,'expected_record_versions',expected_versions,'memory_revision',memory_revision,'privacy_epoch',privacy_epoch,'stage_states',stage_states,'required_stages',required_stages,'scope_summary',scope,'feedback_state',feedback_state,'error_code',error_code) FROM memory_mutation WHERE id=$1::uuid`, id).Scan(&out)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
var result any
|
|
e = json.Unmarshal(out, &result)
|
|
return result, e
|
|
}
|
|
|
|
type memoryQueryCursor struct {
|
|
CreatedAt time.Time `json:"created_at"`
|
|
ID string `json:"id"`
|
|
}
|
|
|
|
func (s *Service) query(ctx context.Context, tx pgx.Tx, p identity.Principal, in QueryInput) (any, error) {
|
|
a, e := ResolveAccess(ctx, tx, p, in.ElderID, false)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if in.Limit == 0 {
|
|
in.Limit = 50
|
|
}
|
|
if in.Limit < 1 || in.Limit > 100 || len(in.Cursor) > 1024 {
|
|
return nil, fault(400, "INVALID_CURSOR")
|
|
}
|
|
cursor := memoryQueryCursor{}
|
|
if in.Cursor != "" {
|
|
if validUUID(in.Cursor) {
|
|
// Old clients sent only the record UUID. Resolve its timestamp inside
|
|
// exactly the same authorized and filtered list, never another elder's.
|
|
e = tx.QueryRow(ctx, `SELECT cm.created_at,cm.public_id::text FROM curated_memory cm JOIN memory_version mv ON mv.id=cm.current_version_id WHERE cm.subject_id=$1::uuid AND ($2 OR cm.sensitivity='normal') AND ($3='' OR cm.memory_type=$3) AND (($4='' AND cm.status<>'DELETED') OR cm.status=$4) AND cm.public_id=$5::uuid`, a.Subject, a.Sensitive, in.Type, in.Status, in.Cursor).Scan(&cursor.CreatedAt, &cursor.ID)
|
|
if errors.Is(e, pgx.ErrNoRows) {
|
|
return nil, fault(400, "INVALID_CURSOR")
|
|
}
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
} else {
|
|
if !strings.HasPrefix(in.Cursor, "v1.") {
|
|
return nil, fault(400, "INVALID_CURSOR")
|
|
}
|
|
b, err := base64.RawURLEncoding.DecodeString(strings.TrimPrefix(in.Cursor, "v1."))
|
|
if err != nil || StrictJSON(b, &cursor) != nil || cursor.CreatedAt.IsZero() || !validUUID(cursor.ID) {
|
|
return nil, fault(400, "INVALID_CURSOR")
|
|
}
|
|
}
|
|
}
|
|
rows, e := tx.Query(ctx, `SELECT cm.public_id::text,cm.created_at,jsonb_build_object('memory_id',cm.public_id,'record_version',cm.version,'content',mv.content,'memory_type',cm.memory_type,'status',cm.status,'proactive_policy',cm.proactive_policy,'sensitivity',cm.sensitivity,'entity_key',cm.entity_key,'item_status',cm.item_status,'planned_at',cm.planned_at,'completed_at',cm.completed_at,'completion_basis',cm.completion_basis,'source_refs',mv.source_refs,'created_at',cm.created_at,'updated_at',cm.updated_at) FROM curated_memory cm JOIN memory_version mv ON mv.id=cm.current_version_id WHERE cm.subject_id=$1::uuid AND ($2 OR cm.sensitivity='normal') AND ($3='' OR cm.memory_type=$3) AND (($4='' AND cm.status<>'DELETED') OR cm.status=$4) AND ($5='' OR (cm.created_at,cm.public_id)<($6,NULLIF($5,'')::uuid)) ORDER BY cm.created_at DESC,cm.public_id DESC LIMIT $7`, a.Subject, a.Sensitive, in.Type, in.Status, cursor.ID, cursor.CreatedAt, in.Limit+1)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
defer rows.Close()
|
|
items := []json.RawMessage{}
|
|
cursors := []memoryQueryCursor{}
|
|
for rows.Next() {
|
|
var itemCursor memoryQueryCursor
|
|
var b []byte
|
|
if e = rows.Scan(&itemCursor.ID, &itemCursor.CreatedAt, &b); e != nil {
|
|
return nil, e
|
|
}
|
|
cursors = append(cursors, itemCursor)
|
|
items = append(items, b)
|
|
}
|
|
if e = rows.Err(); e != nil {
|
|
return nil, e
|
|
}
|
|
next := ""
|
|
if len(items) > in.Limit {
|
|
items = items[:in.Limit]
|
|
next = "v1." + base64.RawURLEncoding.EncodeToString(jsonBytes(cursors[in.Limit-1]))
|
|
}
|
|
rows.Close()
|
|
// Pending voice proposals must be discoverable; never expose sensitive targets
|
|
// to a caller whose separate memory grant excludes sensitive records.
|
|
var pending []byte
|
|
e = tx.QueryRow(ctx, `SELECT COALESCE(jsonb_agg(jsonb_build_object('mutation_id',m.id,'mutation_revision',m.revision,'mutation_state',m.state,'action',m.action,'targets',m.targets,'patch',CASE WHEN m.state='NEEDS_CONFIRMATION' THEN m.patch ELSE '{}'::jsonb END,'expected_record_versions',m.expected_versions,'stage_states',m.stage_states,'scope_summary',m.scope,'feedback_state',m.feedback_state,'error_code',m.error_code) ORDER BY m.created_at DESC),'[]'::jsonb) FROM (SELECT * FROM memory_mutation WHERE subject_id=$1::uuid AND ($2 OR (COALESCE(patch->>'sensitivity','normal')='normal' AND NOT EXISTS(SELECT 1 FROM curated_memory cm WHERE cm.subject_id=$1::uuid AND cm.sensitivity<>'normal' AND targets @> jsonb_build_array(cm.public_id::text)))) ORDER BY created_at DESC LIMIT 50) m`, a.Subject, a.Sensitive).Scan(&pending)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
return map[string]any{"items": items, "next_cursor": next, "mutations": json.RawMessage(pending), "can_write": a.Write, "can_manage_access": a.Manage}, nil
|
|
}
|
|
func (s *Service) syncStatus(ctx context.Context, tx pgx.Tx, subject string) (any, error) {
|
|
var rev, visible, epoch int64
|
|
var legacy, critical bool
|
|
var pending int
|
|
e := tx.QueryRow(ctx, `SELECT COALESCE(us.memory_revision,0),COALESCE(us.viking_visible_revision,0),COALESCE(us.privacy_epoch,0),COALESCE(us.legacy_projection_pending,false),(SELECT count(*) FROM memory_projection_job j WHERE j.subject_id=sp.id AND state<>'VISIBLE'),EXISTS(SELECT 1 FROM memory_projection_job j WHERE j.subject_id=sp.id AND j.critical AND j.state<>'VISIBLE') FROM subject_profile sp LEFT JOIN memory_user_state us ON us.subject_id=sp.id WHERE sp.id=$1::uuid`, subject).Scan(&rev, &visible, &epoch, &legacy, &pending, &critical)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
state := "IN_SYNC"
|
|
if rev == 0 {
|
|
state = "EMPTY"
|
|
}
|
|
if pending > 0 {
|
|
state = "LAGGING_NORMAL"
|
|
}
|
|
if critical || legacy {
|
|
state = "LAGGING_CRITICAL"
|
|
}
|
|
var blocked int
|
|
var reason string
|
|
e = tx.QueryRow(ctx, `SELECT count(*) FILTER(WHERE state='BLOCKED'),COALESCE((array_agg(last_error ORDER BY memory_revision) FILTER(WHERE last_error IS NOT NULL))[1],'') FROM memory_projection_job WHERE subject_id=$1::uuid AND state<>'VISIBLE'`, subject).Scan(&blocked, &reason)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if pending > 0 && s.projection == nil {
|
|
reason = "PROJECTION_NOT_CONFIGURED"
|
|
}
|
|
var uploadPending, uploaded, mirrored int
|
|
var uploadError string
|
|
if s.managedSync != nil {
|
|
e = tx.QueryRow(ctx, `SELECT count(*) FILTER(WHERE j.state='PENDING'),count(*) FILTER(WHERE j.state='UPLOADED'),count(*) FILTER(WHERE j.state='SYNCED'),COALESCE((array_agg(j.last_error ORDER BY j.created_at DESC) FILTER(WHERE j.last_error IS NOT NULL))[1],'') FROM call_memory_sync j JOIN call_session c ON c.id=j.session_id WHERE c.subject_id=$1::uuid AND j.transcript_revision=c.transcript_revision AND j.state<>'SKIPPED'`, subject).Scan(&uploadPending, &uploaded, &mirrored, &uploadError)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
if uploadError != "" {
|
|
reason = uploadError
|
|
}
|
|
if !critical && !legacy && (uploadPending > 0 || uploaded > 0 || uploadError != "") {
|
|
state = "LAGGING_NORMAL"
|
|
}
|
|
}
|
|
// A skipped job can also mean deletion, a superseded transcript, or revoked
|
|
// consent. Count missing dialogue from current authorized source calls instead.
|
|
var insufficient int
|
|
e = tx.QueryRow(ctx, `SELECT count(*) FROM call_session c
|
|
WHERE c.subject_id=$1::uuid AND c.privacy_epoch=$2 AND c.ended_at IS NOT NULL
|
|
AND c.transcript_state IN ('PARTIAL','COMPLETE','EMPTY')
|
|
AND c.budget_snapshot->>'protocol'='DIRECT-CHAT-1' AND c.memory_mode<>'OFF'
|
|
AND managed_call_memory_authorized(c.subject_id)
|
|
AND NOT EXISTS(SELECT 1 FROM memory_source_suppression s CROSS JOIN LATERAL jsonb_array_elements(s.source_refs) r WHERE s.subject_id=c.subject_id AND r->>'session_id'=c.id::text)
|
|
AND (NOT EXISTS(SELECT 1 FROM call_utterance u WHERE u.session_id=c.id AND u.speaker='USER' AND regexp_replace(u.text_content,'^[[:space:]]+|[[:space:]]+$','','g') NOT IN ('','[deleted]'))
|
|
OR NOT EXISTS(SELECT 1 FROM call_utterance u WHERE u.session_id=c.id AND u.speaker='ASSISTANT' AND regexp_replace(u.text_content,'^[[:space:]]+|[[:space:]]+$','','g') NOT IN ('','[deleted]')))`, subject, epoch).Scan(&insufficient)
|
|
if e != nil {
|
|
return nil, e
|
|
}
|
|
return map[string]any{"memory_revision": rev, "viking_visible_revision": visible, "privacy_epoch": epoch, "sync_state": state, "pending_count": pending, "critical_backlog": critical || legacy, "blocked_count": blocked, "error_code": reason, "upload_pending_count": uploadPending, "extracting_count": uploaded, "synced_call_count": mirrored, "insufficient_transcript_count": insufficient}, nil
|
|
}
|