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
|
||
|
|
}
|