From 03fa52dfd481053d71c9ee657dbfeb661e96af31 Mon Sep 17 00:00:00 2001 From: kami Date: Mon, 20 Jul 2026 12:00:37 +0400 Subject: [PATCH] Add entity-aware fact resolution against Nexus (Vikunja #279) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit facts gain a Subject/EntityID/ResolutionState triple and an async enrichment worker that resolves free-text subjects to canonical Nexus entity_ids, mirroring Praxis's enrichment-worker pattern. Ambiguous or unreachable Nexus never guesses an entity_id — the fact stays pending or terminal-ambiguous instead. Co-Authored-By: Claude Sonnet 5 Claude-Session: https://claude.ai/code/session_018ghELqYhZNLub2TXGMazqA --- cmd/mavend/factenrichment.go | 82 +++++++++++++++++ cmd/mavend/factenrichment_test.go | 100 +++++++++++++++++++++ cmd/mavend/main.go | 13 +++ internal/config/config.go | 11 +++ internal/store/entityfacts.go | 136 +++++++++++++++++++++++++++++ internal/store/entityfacts_test.go | 112 ++++++++++++++++++++++++ internal/store/migrations.go | 7 ++ internal/store/store.go | 24 +++++ 8 files changed, 485 insertions(+) create mode 100644 cmd/mavend/factenrichment.go create mode 100644 cmd/mavend/factenrichment_test.go create mode 100644 internal/store/entityfacts.go create mode 100644 internal/store/entityfacts_test.go diff --git a/cmd/mavend/factenrichment.go b/cmd/mavend/factenrichment.go new file mode 100644 index 0000000..e228215 --- /dev/null +++ b/cmd/mavend/factenrichment.go @@ -0,0 +1,82 @@ +// factenrichment.go — resolves the Subject of entity-aware facts against +// Nexus, turning free-text ("the espresso machine", "Kate") into a canonical +// entity_id (Vikunja #279). Mirrors Praxis's async enrichment worker +// (internal/enrichment/worker.go there): a plain poll-queue-update loop, no +// shared state with the tick loop beyond the store and ecosystemWiring both +// already hold. +package main + +import ( + "context" + "log" + "time" + + "github.com/kami/maven/internal/store" +) + +// factEnrichmentWorker polls store.PendingFactResolutions and resolves each +// one's Subject through ecosystemWiring.resolveEntityReference. A nil nexus +// client (ecosystem.nexus unconfigured) makes every tick a no-op rather than +// erroring — entity-aware memory degrades to "facts just aren't tagged", +// never to a crash loop. +type factEnrichmentWorker struct { + store *store.Store + eco *ecosystemWiring + interval time.Duration + batch int // facts resolved per tick; keeps a single slow tick bounded +} + +func newFactEnrichmentWorker(st *store.Store, eco *ecosystemWiring, interval time.Duration) *factEnrichmentWorker { + return &factEnrichmentWorker{store: st, eco: eco, interval: interval, batch: 20} +} + +func (w *factEnrichmentWorker) run(ctx context.Context) { + if w.eco == nil || w.eco.nexus == nil { + log.Printf("factenrichment: nexus not configured, worker idle") + <-ctx.Done() + return + } + + ticker := time.NewTicker(w.interval) + defer ticker.Stop() + + w.tick(ctx) + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + w.tick(ctx) + } + } +} + +func (w *factEnrichmentWorker) tick(ctx context.Context) { + pending, err := w.store.PendingFactResolutions(ctx, w.batch) + if err != nil { + log.Printf("factenrichment: list pending: %v", err) + return + } + for _, f := range pending { + w.resolveOne(ctx, f) + } +} + +func (w *factEnrichmentWorker) resolveOne(ctx context.Context, f store.Fact) { + entityID, _, ambiguous, err := w.eco.resolveEntityReference(ctx, f.Subject, nil) + if err != nil { + // Transient (Nexus unreachable) — leave pending, retry next tick. + log.Printf("factenrichment: resolve fact %d subject %q: %v", f.ID, f.Subject, err) + return + } + state := store.ResolutionNotFound + switch { + case entityID != "": + state = store.ResolutionResolved + case len(ambiguous) > 0: + state = store.ResolutionAmbiguous + } + if err := w.store.ResolveFactEntity(ctx, f.ID, entityID, state); err != nil { + log.Printf("factenrichment: record resolution for fact %d: %v", f.ID, err) + } +} diff --git a/cmd/mavend/factenrichment_test.go b/cmd/mavend/factenrichment_test.go new file mode 100644 index 0000000..9dfdfa3 --- /dev/null +++ b/cmd/mavend/factenrichment_test.go @@ -0,0 +1,100 @@ +package main + +import ( + "context" + "database/sql" + "testing" + "time" + + "github.com/kami/maven/internal/store" +) + +func TestFactEnrichmentWorker_ResolvesPendingSubject(t *testing.T) { + ctx := context.Background() + nexus := newFakeNexus(t, fixtureNexusResolved("ent_espresso", "the espresso machine", "device")) + st := newTestStore(t) + + id, err := st.WriteFactAboutSubject(ctx, time.Now(), store.KindEnv, "likes", "the espresso machine", `"true"`, "infer:pref", 0.8, sql.NullInt64{}) + if err != nil { + t.Fatalf("WriteFactAboutSubject: %v", err) + } + + w := newFactEnrichmentWorker(st, stubEcosystem(nexus.URL, ""), time.Hour) + w.tick(ctx) + + facts, err := st.FactsByEntity(ctx, "ent_espresso", 10) + if err != nil { + t.Fatalf("FactsByEntity: %v", err) + } + if len(facts) != 1 || facts[0].ID != id { + t.Fatalf("expected fact %d resolved to ent_espresso, got %+v", id, facts) + } + + pending, err := st.PendingFactResolutions(ctx, 10) + if err != nil { + t.Fatalf("PendingFactResolutions: %v", err) + } + if len(pending) != 0 { + t.Fatalf("expected no facts left pending, got %+v", pending) + } +} + +// TestFactEnrichmentWorker_NexusDownLeavesFactPending covers the fail-closed +// contract: a Nexus outage must not mark a fact resolved/not_found — it +// should stay pending so a later successful tick can still resolve it. +func TestFactEnrichmentWorker_NexusDownLeavesFactPending(t *testing.T) { + ctx := context.Background() + nexus := newFakeNexus(t, fixtureNexusResolved("ent_espresso", "the espresso machine", "device")) + nexus.SetFault(503) + st := newTestStore(t) + + id, err := st.WriteFactAboutSubject(ctx, time.Now(), store.KindEnv, "likes", "the espresso machine", `"true"`, "infer:pref", 0.8, sql.NullInt64{}) + if err != nil { + t.Fatalf("WriteFactAboutSubject: %v", err) + } + + w := newFactEnrichmentWorker(st, stubEcosystem(nexus.URL, ""), time.Hour) + w.tick(ctx) + + pending, err := st.PendingFactResolutions(ctx, 10) + if err != nil { + t.Fatalf("PendingFactResolutions: %v", err) + } + if len(pending) != 1 || pending[0].ID != id { + t.Fatalf("expected fact %d to remain pending during nexus outage, got %+v", id, pending) + } +} + +func TestFactEnrichmentWorker_AmbiguousLeavesEntityIDUnset(t *testing.T) { + ctx := context.Background() + nexus := newFakeNexus(t, fixtureNexusAmbiguous( + map[string]string{"entity_id": "ent_kate_1", "display_name": "Kate Smith"}, + map[string]string{"entity_id": "ent_kate_2", "display_name": "Kate Jones"}, + )) + st := newTestStore(t) + + id, err := st.WriteFactAboutSubject(ctx, time.Now(), store.KindEnv, "likes", "kate", `"true"`, "infer:pref", 0.8, sql.NullInt64{}) + if err != nil { + t.Fatalf("WriteFactAboutSubject: %v", err) + } + + w := newFactEnrichmentWorker(st, stubEcosystem(nexus.URL, ""), time.Hour) + w.tick(ctx) + + pending, err := st.PendingFactResolutions(ctx, 10) + if err != nil { + t.Fatalf("PendingFactResolutions: %v", err) + } + if len(pending) != 0 { + t.Fatalf("ambiguous resolution should leave the pending queue, got %+v", pending) + } + + facts, err := st.FactsByEntity(ctx, "ent_kate_1", 10) + if err != nil { + t.Fatalf("FactsByEntity: %v", err) + } + if len(facts) != 0 { + t.Fatalf("ambiguous resolution must not guess an entity_id, got %+v", facts) + } + _ = id +} diff --git a/cmd/mavend/main.go b/cmd/mavend/main.go index d6c6ea2..4c251da 100644 --- a/cmd/mavend/main.go +++ b/cmd/mavend/main.go @@ -249,6 +249,7 @@ func run(args []string) error { tl *tickLoop coreAPI ipc.CoreAPI eco *ecosystemWiring + factWorker *factEnrichmentWorker ) if !locked { @@ -341,6 +342,7 @@ func run(args []string) error { repeatInterval := time.Duration(cfg.RepeatInterval) autotuneInterval := time.Duration(cfg.AutotuneInterval) tl = newTickLoop(st, gatherer, dispatcher, phr, rules, tickInterval, repeatInterval, autotuneInterval, cfg.Digest, routinesFromConfig(cfg.Routines)) + factWorker = newFactEnrichmentWorker(st, eco, time.Duration(cfg.FactEnrichmentInterval)) coreAPI = &daemonAPI{ CoreAPI: ipc.NewStoreAPI(st), @@ -500,6 +502,7 @@ func run(args []string) error { repeatInterval := time.Duration(cfg.RepeatInterval) autotuneInterval := time.Duration(cfg.AutotuneInterval) tl = newTickLoop(st, gatherer, dispatcher, phr, rules, tickInterval, repeatInterval, autotuneInterval, cfg.Digest, routinesFromConfig(cfg.Routines)) + factWorker = newFactEnrichmentWorker(st, eco, time.Duration(cfg.FactEnrichmentInterval)) // Swap the CoreAPI from lockedAPI to the real store adapter. newAPI := &daemonAPI{ @@ -530,6 +533,11 @@ func run(args []string) error { tl.run(ctx) }() + // Start fact-entity enrichment worker. + go func() { + factWorker.run(ctx) + }() + dl.unlock() log.Printf("mavend: unlocked via passkey assertion") return nil @@ -563,6 +571,11 @@ func run(args []string) error { defer wg.Done() tl.run(ctx) }() + wg.Add(1) + go func() { + defer wg.Done() + factWorker.run(ctx) + }() } <-ctx.Done() diff --git a/internal/config/config.go b/internal/config/config.go index a1fc583..34733cd 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -84,6 +84,12 @@ type Config struct { // falls back to the rule's static Base, matching pre-autotune behavior). AutotuneInterval Duration `json:"autotune_interval,omitempty"` + // FactEnrichmentInterval — how often the fact-entity enrichment worker + // polls for facts with resolution_state='pending' and resolves their + // subject against Nexus. Default 30s. Only runs when Nexus is configured; + // no-ops (harmlessly) otherwise. + FactEnrichmentInterval Duration `json:"fact_enrichment_interval,omitempty"` + // Ntfy — the ntfy push sink config. nil ⇒ ntfy channel not wired. // sev3 (ops soft) away + sev4 (ops hard) present + reminders away all // route here; not wiring ntfy means those routes drop silently. @@ -357,6 +363,8 @@ const ( DefaultRouterThreshold = 0.55 DefaultQueryMinScore = 0.55 DefaultToolTimeout = 30 * time.Second + + DefaultFactEnrichmentInterval = 30 * time.Second ) // Load reads the JSON config at path and applies defaults. A missing file is @@ -394,6 +402,9 @@ func (c *Config) applyDefaults() { if c.AutotuneInterval == 0 { c.AutotuneInterval = Duration(DefaultAutotuneInterval) } + if c.FactEnrichmentInterval == 0 { + c.FactEnrichmentInterval = Duration(DefaultFactEnrichmentInterval) + } // StateDir — when set, use it as the base for both db and socket if their // paths are still relative (empty). If StateDir is empty, fall back to the // XDG-style defaults (data dir for db, runtime dir for socket). diff --git a/internal/store/entityfacts.go b/internal/store/entityfacts.go new file mode 100644 index 0000000..7681194 --- /dev/null +++ b/internal/store/entityfacts.go @@ -0,0 +1,136 @@ +package store + +import ( + "context" + "database/sql" + "errors" + "fmt" + "time" +) + +// WriteFactAboutSubject is WriteFact plus a free-text subject the fact is +// about (e.g. "the espresso machine", "Kate"). If subject is non-empty the +// row starts life at ResolutionPending and the fact-enrichment worker +// (cmd/mavend/factenrichment.go) later resolves it against Nexus into +// EntityID. Facts with no subject are ordinary — ResolutionState stays +// ResolutionNone and no enrichment work is queued for them. +func (s *Store) WriteFactAboutSubject(ctx context.Context, ts time.Time, kind FactKind, key, subject, value, source string, confidence float64, voidsID sql.NullInt64) (int64, error) { + if confidence <= 0.0 || confidence > 1.0 { + return 0, fmt.Errorf("%w: %f", ErrConfidence, confidence) + } + if voidsID.Valid { + var ok int64 + err := s.db.QueryRowContext(ctx, "SELECT 1 FROM facts WHERE id = ?", voidsID.Int64).Scan(&ok) + if errors.Is(err, sql.ErrNoRows) { + return 0, fmt.Errorf("%w: id=%d", ErrVoidsMissing, voidsID.Int64) + } + if err != nil { + return 0, fmt.Errorf("voids lookup: %w", err) + } + } + state := ResolutionNone + if subject != "" { + state = ResolutionPending + } + res, err := s.db.ExecContext(ctx, + `INSERT INTO facts (ts, kind, key, value, source, confidence, voids_id, subject, resolution_state) + VALUES (?,?,?,?,?,?,?,?,?)`, + ts.UnixMilli(), string(kind), key, value, source, confidence, voidsID, subject, string(state)) + if err != nil { + return 0, fmt.Errorf("write fact about subject: %w", err) + } + id, err := res.LastInsertId() + if err != nil { + return 0, fmt.Errorf("last insert id: %w", err) + } + return id, nil +} + +// PendingFactResolutions returns up to limit facts awaiting entity +// resolution (ResolutionPending), oldest first — the enrichment worker's +// work queue. Voided facts are included: a corrected fact's subject still +// deserves resolution so history stays queryable by entity. +func (s *Store) PendingFactResolutions(ctx context.Context, limit int) ([]Fact, error) { + rows, err := s.db.QueryContext(ctx, ` + SELECT id, ts, kind, key, value, source, confidence, voids_id, subject, entity_id, resolution_state + FROM facts + WHERE resolution_state = ? + ORDER BY id ASC + LIMIT ?`, string(ResolutionPending), limit) + if err != nil { + return nil, fmt.Errorf("pending fact resolutions: %w", err) + } + defer rows.Close() + var out []Fact + for rows.Next() { + f, err := scanFactWithEntity(rows) + if err != nil { + return nil, err + } + out = append(out, f) + } + return out, rows.Err() +} + +// ResolveFactEntity records the outcome of resolving a pending fact's +// subject against Nexus. entityID is ignored (left NULL) unless state is +// ResolutionResolved — an ambiguous or not-found result must not leave a +// stale/guessed entity_id behind. +func (s *Store) ResolveFactEntity(ctx context.Context, factID int64, entityID string, state FactResolutionState) error { + var idArg sql.NullString + if state == ResolutionResolved { + idArg = sql.NullString{String: entityID, Valid: entityID != ""} + } + _, err := s.db.ExecContext(ctx, + `UPDATE facts SET entity_id = ?, resolution_state = ? WHERE id = ?`, + idArg, string(state), factID) + if err != nil { + return fmt.Errorf("resolve fact entity: %w", err) + } + return nil +} + +// FactsByEntity returns facts resolved to the given Nexus entity_id, newest +// first, up to limit. Only ResolutionResolved rows carry an entity_id so +// this naturally excludes pending/ambiguous/not_found rows. +func (s *Store) FactsByEntity(ctx context.Context, entityID string, limit int) ([]Fact, error) { + rows, err := s.db.QueryContext(ctx, ` + SELECT id, ts, kind, key, value, source, confidence, voids_id, subject, entity_id, resolution_state + FROM facts + WHERE entity_id = ? + ORDER BY ts DESC, id DESC + LIMIT ?`, entityID, limit) + if err != nil { + return nil, fmt.Errorf("facts by entity: %w", err) + } + defer rows.Close() + var out []Fact + for rows.Next() { + f, err := scanFactWithEntity(rows) + if err != nil { + return nil, err + } + out = append(out, f) + } + return out, rows.Err() +} + +func scanFactWithEntity(r rowScanner) (Fact, error) { + var f Fact + var tsMilli int64 + var kind string + var voids sql.NullInt64 + var state string + if err := r.Scan(&f.ID, &tsMilli, &kind, &f.Key, &f.Value, &f.Source, &f.Confidence, &voids, + &f.Subject, &f.EntityID, &state); err != nil { + if errors.Is(err, sql.ErrNoRows) { + return Fact{}, ErrNoFact + } + return Fact{}, err + } + f.Ts = time.UnixMilli(tsMilli).UTC() + f.Kind = FactKind(kind) + f.VoidsID = voids + f.ResolutionState = FactResolutionState(state) + return f, nil +} diff --git a/internal/store/entityfacts_test.go b/internal/store/entityfacts_test.go new file mode 100644 index 0000000..7cbb5ce --- /dev/null +++ b/internal/store/entityfacts_test.go @@ -0,0 +1,112 @@ +package store + +import ( + "context" + "database/sql" + "testing" + "time" +) + +func TestWriteFactAboutSubject_NoSubjectStaysNone(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + id, err := s.WriteFactAboutSubject(ctx, time.Now(), KindSelf, "mood", "", `"content"`, "tap:mood", 1.0, sql.NullInt64{}) + if err != nil { + t.Fatalf("WriteFactAboutSubject: %v", err) + } + + pending, err := s.PendingFactResolutions(ctx, 10) + if err != nil { + t.Fatalf("PendingFactResolutions: %v", err) + } + for _, f := range pending { + if f.ID == id { + t.Fatalf("fact %d written with no subject should not be queued for resolution", id) + } + } +} + +func TestWriteFactAboutSubject_QueuesPendingResolution(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + id, err := s.WriteFactAboutSubject(ctx, time.Now(), KindEnv, "likes", "the espresso machine", `"true"`, "infer:pref", 0.8, sql.NullInt64{}) + if err != nil { + t.Fatalf("WriteFactAboutSubject: %v", err) + } + + pending, err := s.PendingFactResolutions(ctx, 10) + if err != nil { + t.Fatalf("PendingFactResolutions: %v", err) + } + if len(pending) != 1 || pending[0].ID != id { + t.Fatalf("expected fact %d in pending queue, got %+v", id, pending) + } + if pending[0].Subject != "the espresso machine" { + t.Fatalf("expected subject preserved, got %q", pending[0].Subject) + } + if pending[0].ResolutionState != ResolutionPending { + t.Fatalf("expected ResolutionPending, got %q", pending[0].ResolutionState) + } +} + +func TestResolveFactEntity_ResolvedMakesItFindableByEntity(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + id, err := s.WriteFactAboutSubject(ctx, time.Now(), KindEnv, "likes", "the espresso machine", `"true"`, "infer:pref", 0.8, sql.NullInt64{}) + if err != nil { + t.Fatalf("WriteFactAboutSubject: %v", err) + } + + if err := s.ResolveFactEntity(ctx, id, "ent_espresso", ResolutionResolved); err != nil { + t.Fatalf("ResolveFactEntity: %v", err) + } + + pending, err := s.PendingFactResolutions(ctx, 10) + if err != nil { + t.Fatalf("PendingFactResolutions: %v", err) + } + if len(pending) != 0 { + t.Fatalf("expected resolved fact to leave the pending queue, got %+v", pending) + } + + facts, err := s.FactsByEntity(ctx, "ent_espresso", 10) + if err != nil { + t.Fatalf("FactsByEntity: %v", err) + } + if len(facts) != 1 || facts[0].ID != id { + t.Fatalf("expected fact %d under ent_espresso, got %+v", id, facts) + } + if !facts[0].EntityID.Valid || facts[0].EntityID.String != "ent_espresso" { + t.Fatalf("expected EntityID set, got %+v", facts[0].EntityID) + } +} + +// TestResolveFactEntity_AmbiguousDoesNotStoreAnEntityID covers the +// ecosystem-wide invariant that ambiguity blocks mutation: an ambiguous +// resolve must never guess an entity_id, even transiently. +func TestResolveFactEntity_AmbiguousDoesNotStoreAnEntityID(t *testing.T) { + s := newTestStore(t) + ctx := context.Background() + + id, err := s.WriteFactAboutSubject(ctx, time.Now(), KindEnv, "likes", "kate", `"true"`, "infer:pref", 0.8, sql.NullInt64{}) + if err != nil { + t.Fatalf("WriteFactAboutSubject: %v", err) + } + + // Simulate an ambiguous Nexus response: candidates found, but no single + // entity_id — the enrichment worker passes "" for entityID in this case. + if err := s.ResolveFactEntity(ctx, id, "", ResolutionAmbiguous); err != nil { + t.Fatalf("ResolveFactEntity: %v", err) + } + + pending, err := s.PendingFactResolutions(ctx, 10) + if err != nil { + t.Fatalf("PendingFactResolutions: %v", err) + } + if len(pending) != 0 { + t.Fatalf("ambiguous fact should leave the pending queue (it's terminal), got %+v", pending) + } +} diff --git a/internal/store/migrations.go b/internal/store/migrations.go index 8221303..c19d52f 100644 --- a/internal/store/migrations.go +++ b/internal/store/migrations.go @@ -63,6 +63,13 @@ ALTER TABLE reminders ADD COLUMN next_fire_ts INTEGER;`, // #2 completed_ts INTEGER ); CREATE INDEX IF NOT EXISTS idx_delivery_attempts_status ON delivery_attempts (status);`, // #6 — durable delivery outbox + + `ALTER TABLE facts ADD COLUMN subject TEXT NOT NULL DEFAULT ''; + ALTER TABLE facts ADD COLUMN entity_id TEXT; + ALTER TABLE facts ADD COLUMN resolution_state TEXT NOT NULL DEFAULT 'none' + CHECK (resolution_state IN ('none','pending','resolved','ambiguous','not_found')); + CREATE INDEX IF NOT EXISTS idx_facts_entity_id ON facts (entity_id) WHERE entity_id IS NOT NULL; + CREATE INDEX IF NOT EXISTS idx_facts_resolution_pending ON facts (resolution_state) WHERE resolution_state = 'pending';`, // #7 — entity-aware memory (Vikunja #279): facts about a subject get resolved to a Nexus entity_id async } // migrate applies every migration with a number greater than the DB's current diff --git a/internal/store/store.go b/internal/store/store.go index 2131245..757c42e 100644 --- a/internal/store/store.go +++ b/internal/store/store.go @@ -40,8 +40,32 @@ type Fact struct { Source string // tap:* | infer:* | poll:* | ambient | promote | feedback Confidence float64 VoidsID sql.NullInt64 + + // Subject, EntityID, ResolutionState — entity-aware memory (Vikunja #279). + // Subject is the free-text "who/what this fact is about" supplied at write + // time by WriteFactAboutSubject; empty means the fact isn't about a + // resolvable entity (ResolutionState stays "none"). EntityID is the + // canonical Nexus entity_id once the enrichment worker resolves Subject. + Subject string + EntityID sql.NullString + ResolutionState FactResolutionState } +// FactResolutionState — where a fact's Subject stands in Nexus entity +// resolution. "none" = no subject given (most facts). "pending" = subject +// given, not yet resolved. Terminal states: "resolved", "ambiguous" (Nexus +// returned candidates, not stored — matches the ecosystem's ambiguity-blocks +// invariant), "not_found". +type FactResolutionState string + +const ( + ResolutionNone FactResolutionState = "none" + ResolutionPending FactResolutionState = "pending" + ResolutionResolved FactResolutionState = "resolved" + ResolutionAmbiguous FactResolutionState = "ambiguous" + ResolutionNotFound FactResolutionState = "not_found" +) + // Bucket — presence hysteresis state. type Bucket string