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.
335 lines
16 KiB
Go
335 lines
16 KiB
Go
package callmemory
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"github.com/jackc/pgx/v5"
|
|
"time"
|
|
)
|
|
|
|
type ProjectionRecord struct {
|
|
Tenant, Subject, ID, Kind, Content, Policy, Status, ItemStatus string
|
|
Version int64
|
|
}
|
|
type ProjectionResult struct {
|
|
Kind, ID string
|
|
Visible bool
|
|
}
|
|
|
|
// ReconcileOnly forbids another create after an uncertain acknowledgement.
|
|
type ProjectionProvider interface {
|
|
Project(context.Context, ProjectionRecord, string, string, bool) (ProjectionResult, error)
|
|
}
|
|
type SourceTurn struct {
|
|
ID string `json:"utterance_id"`
|
|
Revision int64 `json:"source_revision"`
|
|
Text string `json:"text"`
|
|
Time time.Time `json:"source_at"`
|
|
}
|
|
type Intent struct {
|
|
Utterance string `json:"utterance_id"`
|
|
Revision int64 `json:"source_revision"`
|
|
Action string `json:"action"`
|
|
Targets []string `json:"targets"`
|
|
Patch MemoryPatch `json:"patch"`
|
|
Confidence float64 `json:"confidence"`
|
|
Quote string `json:"source_quote"`
|
|
Start int `json:"source_start"`
|
|
End int `json:"source_end"`
|
|
}
|
|
type IntentExtractor interface {
|
|
ExtractMemoryIntents(context.Context, string, []SourceTurn, []map[string]any) ([]Intent, error)
|
|
}
|
|
|
|
func (s *Service) ConfigureProjection(p ProjectionProvider) { s.projection = p }
|
|
func (s *Service) ConfigureExtractor(e IntentExtractor) { s.extractor = e }
|
|
func (s *Service) ConfigureManagedSync(sync func(context.Context) error) { s.managedSync = sync }
|
|
func (s *Service) Run(ctx context.Context) {
|
|
ticker := time.NewTicker(time.Second)
|
|
defer ticker.Stop()
|
|
for {
|
|
if e := s.Maintain(ctx); e != nil && ctx.Err() == nil {
|
|
s.logger.Error("call memory maintenance failed")
|
|
}
|
|
if s.managedSync != nil {
|
|
if e := s.managedSync(ctx); e != nil && ctx.Err() == nil {
|
|
s.logger.Warn("managed memory synchronization pending", "code", e.Error())
|
|
}
|
|
}
|
|
if e := s.ProjectNext(ctx); e != nil && ctx.Err() == nil {
|
|
s.logger.Warn("memory projection pending")
|
|
}
|
|
if e := s.ExtractNext(ctx); e != nil && ctx.Err() == nil {
|
|
s.logger.Warn("memory extraction pending")
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}
|
|
func (s *Service) Maintain(ctx context.Context) error {
|
|
tx, e := s.pool.Begin(ctx)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
defer tx.Rollback(context.Background())
|
|
// Direct chat has a bounded local lease. Missing end is an interrupted call,
|
|
// never evidence that the supplier finished or that every subtitle arrived.
|
|
_, e = tx.Exec(ctx, `UPDATE call_session SET call_state='FAILED',manifest_status='DISCARDED',manifest_b64=NULL,ended_at=now(),drain_started_at=now(),transcript_state='DRAINING',text_gap=true WHERE budget_snapshot->>'protocol'='DIRECT-CHAT-1' AND ended_at IS NULL AND (lease_expires_at<=now() OR expires_at<=now())`)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
// Clock-based expiry must revoke projections even without a consent UPDATE.
|
|
_, e = tx.Exec(ctx, `UPDATE curated_memory SET status='DISABLED',version=version+1,updated_at=now() WHERE status='ACTIVE' AND NOT call_memory_authorized(subject_id)`)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
_, e = tx.Exec(ctx, `WITH revoked AS (
|
|
UPDATE call_session SET manifest_status='REVOKED',call_state=CASE WHEN call_state IN ('PREPARING','STARTING','TALKING') THEN 'STOPPING' ELSE call_state END
|
|
WHERE manifest_status='ISSUED' AND lease_expires_at<=now() RETURNING id,privacy_epoch)
|
|
INSERT INTO call_command(session_id,privacy_epoch,type,reason,expires_at) SELECT id,privacy_epoch,'STOP_CALL','LEASE_EXPIRED',now()+interval '7 days' FROM revoked ON CONFLICT DO NOTHING`)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE call_session c SET transcript_state=CASE
|
|
WHEN text_gap OR EXISTS(SELECT 1 FROM call_text_fragment f WHERE f.session_id=c.id GROUP BY utterance_id,source_revision HAVING count(*)<max(part_count)) THEN 'PARTIAL'
|
|
WHEN EXISTS(SELECT 1 FROM call_utterance u WHERE u.session_id=c.id) THEN 'COMPLETE' ELSE 'EMPTY' END
|
|
WHERE transcript_state='DRAINING' AND (drain_started_at<=now()-interval '60 seconds' OR GREATEST(COALESCE(last_text_at,drain_started_at),drain_started_at)<=now()-interval '15 seconds')`)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
_, e = tx.Exec(ctx, `INSERT INTO call_postprocess_job(session_id,transcript_revision,extractor_version) SELECT id,transcript_revision,'memory-1' FROM call_session WHERE transcript_state IN ('COMPLETE','PARTIAL') AND transcript_revision>0 AND (NOT $1::boolean OR COALESCE(budget_snapshot->>'protocol','')<>'DIRECT-CHAT-1') ON CONFLICT DO NOTHING`, s.managedSync != nil)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
if s.managedSync != nil {
|
|
if _, e = tx.Exec(ctx, `UPDATE call_postprocess_job j SET state='SUPERSEDED',last_error=NULL FROM call_session c WHERE c.id=j.session_id AND c.budget_snapshot->>'protocol'='DIRECT-CHAT-1' AND j.state='PENDING'`); e != nil {
|
|
return e
|
|
}
|
|
}
|
|
if e = s.enqueueDirectStories(ctx, tx); e != nil {
|
|
return e
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE memory_mutation SET state='EXPIRED',confirmation_hash=NULL,revision=revision+1,updated_at=now() WHERE state='NEEDS_CONFIRMATION' AND confirmation_expires_at<=now()`)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
if e = s.advanceMutations(ctx, tx); e != nil {
|
|
return e
|
|
}
|
|
// Receipts can expire; deletion ledgers and suppressions must survive cleanup.
|
|
_, e = tx.Exec(ctx, `DELETE FROM call_request_result WHERE expires_at<now()`)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
return tx.Commit(ctx)
|
|
}
|
|
func (s *Service) advanceMutations(ctx context.Context, tx pgx.Tx) error {
|
|
rows, e := tx.Query(ctx, `SELECT id::text,subject_id::text,action,scope,COALESCE(memory_revision,0),COALESCE(privacy_epoch,0),targets FROM memory_mutation WHERE state IN ('PROPAGATING','BLOCKED') ORDER BY created_at FOR UPDATE SKIP LOCKED LIMIT 50`)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
type work struct {
|
|
id, subject, action, scope string
|
|
rev, epoch int64
|
|
targets []byte
|
|
}
|
|
jobs := []work{}
|
|
for rows.Next() {
|
|
var w work
|
|
if e = rows.Scan(&w.id, &w.subject, &w.action, &w.scope, &w.rev, &w.epoch, &w.targets); e != nil {
|
|
rows.Close()
|
|
return e
|
|
}
|
|
jobs = append(jobs, w)
|
|
}
|
|
e = rows.Err()
|
|
rows.Close()
|
|
if e != nil {
|
|
return e
|
|
}
|
|
for _, w := range jobs {
|
|
var visible int64
|
|
var legacy, active, projectionBlocked bool
|
|
e = tx.QueryRow(ctx, `SELECT viking_visible_revision,legacy_projection_pending,EXISTS(SELECT 1 FROM call_session WHERE subject_id=$1::uuid AND privacy_epoch<$2 AND manifest_status IN ('ISSUED','REVOKED') AND NOT(local_audio_off AND sdk_stopped AND remote_state IN ('NOT_STARTED','STOPPED_OBSERVED'))) OR call_legacy_cleanup_pending($1::uuid),EXISTS(SELECT 1 FROM memory_projection_job WHERE subject_id=$1::uuid AND memory_revision<=$3 AND state='BLOCKED') FROM memory_user_state WHERE subject_id=$1::uuid`, w.subject, w.epoch, w.rev).Scan(&visible, &legacy, &active, &projectionBlocked)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
stage := map[string]string{"main_store": "DONE", "serving_projection": "PENDING", "derived_context": "DONE", "active_context": "PENDING", "source_cleanup": "DONE", "backup_cleanup": "DONE"}
|
|
if visible >= w.rev && !legacy {
|
|
stage["serving_projection"] = "DONE"
|
|
}
|
|
if !active {
|
|
stage["active_context"] = "DONE"
|
|
}
|
|
if w.action == "ADD" {
|
|
stage["derived_context"], stage["active_context"] = "NOT_REQUIRED", "NOT_REQUIRED"
|
|
}
|
|
if w.action != "DELETE" {
|
|
stage["source_cleanup"], stage["backup_cleanup"] = "NOT_REQUIRED", "NOT_REQUIRED"
|
|
}
|
|
if w.action == "DELETE" {
|
|
if e = s.cleanSources(ctx, tx, w.subject, w.targets, w.scope); e != nil {
|
|
return e
|
|
}
|
|
var pendingBackup bool
|
|
e = tx.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM memory_deletion_ledger l JOIN curated_memory cm ON cm.id=l.memory_id WHERE cm.subject_id=$1::uuid AND cm.public_id IN (SELECT value::uuid FROM jsonb_array_elements_text($2::jsonb)) AND l.backup_cleanup_verified_at IS NULL)`, w.subject, w.targets).Scan(&pendingBackup)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
if pendingBackup {
|
|
stage["backup_cleanup"] = "PENDING"
|
|
}
|
|
}
|
|
state := "PROPAGATING"
|
|
complete := true
|
|
for _, v := range stage {
|
|
if v != "DONE" && v != "NOT_REQUIRED" {
|
|
complete = false
|
|
}
|
|
}
|
|
if complete {
|
|
state = "COMPLETED"
|
|
if w.action == "DELETE" {
|
|
if _, e = tx.Exec(ctx, `UPDATE curated_memory SET status='DELETED',updated_at=now() WHERE subject_id=$1::uuid AND public_id IN(SELECT value::uuid FROM jsonb_array_elements_text($2::jsonb)) AND status='PENDING_DELETE'`, w.subject, w.targets); e != nil {
|
|
return e
|
|
}
|
|
}
|
|
}
|
|
if projectionBlocked {
|
|
state = "BLOCKED"
|
|
stage["serving_projection"] = "BLOCKED"
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE memory_mutation SET state=$2,stage_states=$3::jsonb,revision=revision+CASE WHEN state<>$2 OR stage_states<>$3::jsonb THEN 1 ELSE 0 END,updated_at=now() WHERE id=$1::uuid`, w.id, state, jsonBytes(stage))
|
|
if e != nil {
|
|
return e
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
func (s *Service) cleanSources(ctx context.Context, tx pgx.Tx, subject string, targets []byte, scope string) error {
|
|
// Preserve only tombstones. Source-suppression carries IDs and provenance, not plaintext.
|
|
_, e := tx.Exec(ctx, `UPDATE memory_version SET content='[deleted]',change_reason=NULL,content_digest=$3 WHERE memory_id IN (SELECT id FROM curated_memory WHERE subject_id=$1::uuid AND public_id IN(SELECT value::uuid FROM jsonb_array_elements_text($2::jsonb))) AND content<>'[deleted]'`, subject, targets, hash([]byte("[deleted]")))
|
|
if e != nil {
|
|
return e
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE curated_memory SET entity_key='',completion_basis=CASE WHEN completion_basis IS NOT NULL THEN '{"redacted":true}'::jsonb END WHERE subject_id=$1::uuid AND public_id IN(SELECT value::uuid FROM jsonb_array_elements_text($2::jsonb))`, subject, targets)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE memory_index_entry SET embedding=NULL,status='DELETED',updated_at=now() WHERE memory_id IN(SELECT id FROM curated_memory WHERE subject_id=$1::uuid AND public_id IN(SELECT value::uuid FROM jsonb_array_elements_text($2::jsonb)))`, subject, targets)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE memory_mutation m SET patch='{}'::jsonb,confirmation_hash=NULL,proposal_digest=NULL,confirmation_expires_at=NULL,state=CASE WHEN state IN ('RECEIVED','NEEDS_CONFIRMATION') THEN 'CANCELLED' ELSE state END WHERE subject_id=$1::uuid AND EXISTS(SELECT 1 FROM jsonb_array_elements_text(m.targets) a JOIN jsonb_array_elements_text($2::jsonb) b ON a.value=b.value)`, subject, targets)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
if scope == "SOURCE_AND_DERIVED" {
|
|
if e = s.cleanStorySources(ctx, tx, subject, targets); e != nil {
|
|
return e
|
|
}
|
|
}
|
|
// Old derived summaries cannot continue carrying a removed fact.
|
|
_, e = tx.Exec(ctx, `UPDATE conversation_summary SET content='[deleted]',status='SUPERSEDED' WHERE subject_id=$1::uuid AND status='SUPERSEDED' AND content<>'[deleted]'`, subject)
|
|
return e
|
|
}
|
|
func (s *Service) ProjectNext(ctx context.Context) error {
|
|
if s.projection == nil {
|
|
return nil
|
|
}
|
|
conn, e := s.pool.Acquire(ctx)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
defer conn.Release()
|
|
var subject string
|
|
e = conn.QueryRow(ctx, `SELECT j.subject_id::text FROM memory_projection_job j WHERE j.state NOT IN ('VISIBLE','BLOCKED') AND j.next_attempt_at<=now() AND j.memory_revision=(SELECT min(x.memory_revision) FROM memory_projection_job x WHERE x.subject_id=j.subject_id AND x.state<>'VISIBLE') ORDER BY j.next_attempt_at,j.created_at LIMIT 1`).Scan(&subject)
|
|
if errors.Is(e, pgx.ErrNoRows) {
|
|
return nil
|
|
}
|
|
if e != nil {
|
|
return e
|
|
}
|
|
var locked bool
|
|
e = conn.QueryRow(ctx, `SELECT pg_try_advisory_lock(hashtextextended($1,13))`, subject).Scan(&locked)
|
|
if e != nil || !locked {
|
|
return e
|
|
}
|
|
defer conn.Exec(context.Background(), `SELECT pg_advisory_unlock(hashtextextended($1,13))`, subject)
|
|
var jobID, state, priorID, priorKind string
|
|
var r ProjectionRecord
|
|
var jobVersion, revision int64
|
|
e = conn.QueryRow(ctx, `SELECT j.id::text,j.state,j.record_version,j.memory_revision,cm.tenant_id::text,cm.subject_id::text,cm.public_id::text,cm.memory_type,mv.content,cm.proactive_policy,CASE WHEN cm.sensitivity='normal' AND call_memory_authorized(cm.subject_id) THEN cm.status ELSE 'DISABLED' END,COALESCE(cm.item_status,''),cm.version,COALESCE(mp.provider_id,''),COALESCE(mp.provider_kind,'') FROM memory_projection_job j JOIN curated_memory cm ON cm.id=j.memory_id JOIN memory_version mv ON mv.id=cm.current_version_id LEFT JOIN memory_projection mp ON mp.memory_id=cm.id WHERE j.subject_id=$1::uuid AND j.state<>'VISIBLE' ORDER BY j.memory_revision LIMIT 1`, subject).Scan(&jobID, &state, &jobVersion, &revision, &r.Tenant, &r.Subject, &r.ID, &r.Kind, &r.Content, &r.Policy, &r.Status, &r.ItemStatus, &r.Version, &priorID, &priorKind)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
if state == "BLOCKED" {
|
|
return nil
|
|
}
|
|
// Project current authoritative state even if this job predates a correction.
|
|
// A critical job is never skipped without verifying the resulting projection.
|
|
reconcile := state == "WRITING" || state == "RECONCILING" || state == "WRITE_ACKED"
|
|
if _, e = conn.Exec(ctx, `UPDATE memory_projection_job SET state='WRITING',attempts=attempts+1 WHERE id=$1::uuid`, jobID); e != nil {
|
|
return e
|
|
}
|
|
result, writeErr := s.projection.Project(ctx, r, priorKind, priorID, reconcile)
|
|
nextState := "WRITE_ACKED"
|
|
code := ""
|
|
if writeErr != nil {
|
|
nextState = "RECONCILING"
|
|
code = "PROVIDER_RESULT_UNKNOWN"
|
|
var f *Error
|
|
if errors.As(writeErr, &f) && f.Status < 500 {
|
|
nextState = "BLOCKED"
|
|
code = f.Code
|
|
}
|
|
} else if result.Visible {
|
|
nextState = "VISIBLE"
|
|
}
|
|
tx, e := conn.Begin(ctx)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
defer tx.Rollback(context.Background())
|
|
if _, e = tx.Exec(ctx, `SELECT subject_id FROM memory_user_state WHERE subject_id=$1::uuid FOR UPDATE`, subject); e != nil {
|
|
return e
|
|
}
|
|
_, e = tx.Exec(ctx, `UPDATE memory_projection_job SET state=$2,provider_kind=NULLIF($3,''),provider_id=NULLIF($4,''),last_error=NULLIF($5,''),next_attempt_at=now()+interval '5 seconds' WHERE id=$1::uuid`, jobID, nextState, result.Kind, result.ID, code)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
if writeErr == nil && result.ID != "" {
|
|
_, e = tx.Exec(ctx, `INSERT INTO memory_projection(memory_id,subject_id,provider_kind,provider_id,record_version,visible_version,operation_key) SELECT id,subject_id,$2,$3,$4,$5,$6 FROM curated_memory WHERE public_id=$1::uuid ON CONFLICT(memory_id) DO UPDATE SET provider_kind=excluded.provider_kind,provider_id=excluded.provider_id,record_version=excluded.record_version,visible_version=excluded.visible_version,operation_key=excluded.operation_key,updated_at=now()`, r.ID, result.Kind, result.ID, r.Version, map[bool]int64{true: r.Version, false: 0}[result.Visible], fmt.Sprintf("%s:%d", r.ID, r.Version))
|
|
if e != nil {
|
|
return e
|
|
}
|
|
}
|
|
// A concurrent correction can commit while the external write is in flight.
|
|
// Reconcile that version before acknowledging visibility of this job.
|
|
var currentVersion int64
|
|
var currentStatus string
|
|
if e = tx.QueryRow(ctx, `SELECT version,CASE WHEN sensitivity='normal' AND call_memory_authorized(subject_id) THEN status ELSE 'DISABLED' END FROM curated_memory WHERE public_id=$1::uuid FOR UPDATE`, r.ID).Scan(¤tVersion, ¤tStatus); e != nil {
|
|
return e
|
|
}
|
|
if currentVersion != r.Version || currentStatus != r.Status {
|
|
nextState = "RECONCILING"
|
|
if _, e = tx.Exec(ctx, `UPDATE memory_projection_job SET state='RECONCILING',next_attempt_at=now() WHERE id=$1::uuid`, jobID); e != nil {
|
|
return e
|
|
}
|
|
}
|
|
if nextState == "VISIBLE" {
|
|
_, e = tx.Exec(ctx, `UPDATE memory_user_state us SET viking_visible_revision=COALESCE((SELECT min(memory_revision)-1 FROM memory_projection_job j WHERE j.subject_id=us.subject_id AND j.state<>'VISIBLE'),us.memory_revision),updated_at=now() WHERE subject_id=$1::uuid`, subject)
|
|
if e != nil {
|
|
return e
|
|
}
|
|
}
|
|
if e = tx.Commit(ctx); e != nil {
|
|
return e
|
|
}
|
|
return writeErr
|
|
}
|