From 212eef6b8c6fb377b31a95e4307d041660efa3f8 Mon Sep 17 00:00:00 2001 From: Andy Dixon Date: Sun, 27 Sep 2026 22:19:21 +0100 Subject: [PATCH] fix(api): belief events were published before their transaction committed The API's belief emitter published every event as soon as it was appended. Belief retraction runs inside its own transaction, so subscribers on the event bus saw belief_retracted while the belief row still held its old status, breaking the rule that subscribers never see uncommitted events; a plan reacting to a retracted assumption read the belief before the change and did nothing. The emitter now only appends, and every API and UI caller publishes the operation's events after it succeeds. --- internal/api/beliefs.go | 8 +++++++- internal/api/import.go | 3 ++- internal/api/memory.go | 5 ++++- internal/ui/beliefs.go | 9 +++++++-- internal/ui/memory.go | 4 +++- 5 files changed, 23 insertions(+), 6 deletions(-) diff --git a/internal/api/beliefs.go b/internal/api/beliefs.go index 3db791e..7ce1f19 100644 --- a/internal/api/beliefs.go +++ b/internal/api/beliefs.go @@ -31,13 +31,16 @@ func (s *Server) beliefRoutes() { // Emitter returns a beliefs.Emitter that appends through the event store and // publishes after the surrounding transaction (or immediately on the pool). +// +// It does not publish: belief operations may run inside their own +// transaction, and subscribers must never see uncommitted events. Callers +// publish the returned Result's events after the operation succeeds. func (s *Server) Emitter() beliefs.Emitter { return func(ctx context.Context, q storage.Querier, n events.New) (events.Event, error) { out, err := s.deps.Events.AppendIn(ctx, q, n) if err != nil { return events.Event{}, err } - s.deps.Events.Publish(out...) return out[0], nil } } @@ -117,6 +120,7 @@ func (s *Server) beliefsCreate(w http.ResponseWriter, r *http.Request) { fail(w, r, beliefErr(err)) return } + s.deps.Events.Publish(res.Events...) s.deps.Audit.Log(r.Context(), "belief.create", "belief", res.Belief.ID.String(), map[string]any{"mind": m.Slug, "proposition": res.Belief.Proposition}) writeJSON(w, http.StatusCreated, res) } @@ -263,6 +267,7 @@ func (s *Server) beliefEvidence(w http.ResponseWriter, r *http.Request) { fail(w, r, beliefErr(err)) return } + s.deps.Events.Publish(res.Events...) s.deps.Audit.Log(r.Context(), "belief.evidence", "belief", id.String(), map[string]any{"direction": in.Direction, "note": in.Note}) writeJSON(w, http.StatusOK, res) } @@ -319,6 +324,7 @@ func (s *Server) beliefRetract(w http.ResponseWriter, r *http.Request) { fail(w, r, beliefErr(err)) return } + s.deps.Events.Publish(res.Events...) s.deps.Audit.Log(r.Context(), "belief.retract", "belief", id.String(), map[string]any{"reason": in.Reason, "do_not_infer": in.DoNotInfer}) writeJSON(w, http.StatusOK, res) } diff --git a/internal/api/import.go b/internal/api/import.go index 3e26321..24dc3f6 100644 --- a/internal/api/import.go +++ b/internal/api/import.go @@ -86,7 +86,7 @@ func (s *Server) mindImport(w http.ResponseWriter, r *http.Request) { if weight <= 0 { weight = 0.5 } - _, err := s.deps.Beliefs.Create(r.Context(), s.deps.DB, s.Emitter(), beliefs.New{MindID: m.ID, Proposition: sb.Proposition, Subject: sb.Subject, Epistemic: sb.Epistemic, Source: "seed", + res, err := s.deps.Beliefs.Create(r.Context(), s.deps.DB, s.Emitter(), beliefs.New{MindID: m.ID, Proposition: sb.Proposition, Subject: sb.Subject, Epistemic: sb.Epistemic, Source: "seed", Evidence: []beliefs.EvidenceInput{{Direction: "for", Kind: "operator", Source: "seed:" + actor(r), Weight: weight, Credibility: events.TrustOperator.Credibility(), Note: firstNonEmpty(sb.Note, "seeded by "+actor(r)), Derivation: beliefs.DerivationOperator, Family: "seed:" + actor(r)}}}) if err != nil { if errors.Is(err, beliefs.ErrInvalid) || errors.Is(err, beliefs.ErrBlocked) { @@ -96,6 +96,7 @@ func (s *Server) mindImport(w http.ResponseWriter, r *http.Request) { fail(w, r, err) return } + s.deps.Events.Publish(res.Events...) beliefCount++ } s.deps.Brain.Reconcile(r.Context()) //nolint:errcheck diff --git a/internal/api/memory.go b/internal/api/memory.go index 57aedbd..6ada995 100644 --- a/internal/api/memory.go +++ b/internal/api/memory.go @@ -283,8 +283,11 @@ func (s *Server) propagateMemoryRetraction(r *http.Request, mind, memoryID ids.I return } for _, bid := range affected { - if _, err := s.deps.Beliefs.Reevaluate(r.Context(), s.deps.DB, s.Emitter(), mind, bid, "supporting memory retracted by operator"); err != nil { + res, err := s.deps.Beliefs.Reevaluate(r.Context(), s.deps.DB, s.Emitter(), mind, bid, "supporting memory retracted by operator") + if err != nil { logger(r).Warn("re-evaluate failed", "belief", bid, "err", err) + continue } + s.deps.Events.Publish(res.Events...) } } diff --git a/internal/ui/beliefs.go b/internal/ui/beliefs.go index f48676a..ec7153c 100644 --- a/internal/ui/beliefs.go +++ b/internal/ui/beliefs.go @@ -58,6 +58,7 @@ func (u *UI) beliefAssert(w http.ResponseWriter, r *http.Request) { u.flash(w, r, "error", err.Error(), "/minds/"+m.Slug+"/beliefs") return } + u.deps.Events.Publish(res.Events...) u.deps.Audit.Log(r.Context(), "belief.create", "belief", res.Belief.ID.String(), map[string]any{"mind": m.Slug, "proposition": res.Belief.Proposition}) u.flash(w, r, "ok", "Belief recorded with confidence "+strconv.FormatFloat(res.Belief.Confidence, 'f', 2, 64)+".", "/minds/"+m.Slug+"/beliefs/"+res.Belief.ID.String()) } @@ -93,10 +94,12 @@ func (u *UI) beliefEvidence(w http.ResponseWriter, r *http.Request) { r.ParseForm() //nolint:errcheck weight, _ := strconv.ParseFloat(r.PostFormValue("weight"), 64) note := r.PostFormValue("note") - if _, err := u.deps.Beliefs.AddEvidence(r.Context(), u.deps.DB, u.emitF(), m.ID, id, beliefs.EvidenceInput{Direction: r.PostFormValue("direction"), Kind: "operator", Source: "operator:" + username(r), Weight: weight, Credibility: events.TrustOperator.Credibility(), Note: note}, "operator evidence: "+note, ids.Nil, ids.Nil); err != nil { + res, err := u.deps.Beliefs.AddEvidence(r.Context(), u.deps.DB, u.emitF(), m.ID, id, beliefs.EvidenceInput{Direction: r.PostFormValue("direction"), Kind: "operator", Source: "operator:" + username(r), Weight: weight, Credibility: events.TrustOperator.Credibility(), Note: note}, "operator evidence: "+note, ids.Nil, ids.Nil) + if err != nil { u.flash(w, r, "error", err.Error(), "/minds/"+m.Slug+"/beliefs/"+id.String()) return } + u.deps.Events.Publish(res.Events...) u.deps.Audit.Log(r.Context(), "belief.evidence", "belief", id.String(), map[string]any{"direction": r.PostFormValue("direction"), "note": note}) u.flash(w, r, "ok", "Evidence added and confidence re-evaluated.", "/minds/"+m.Slug+"/beliefs/"+id.String()) } @@ -130,10 +133,12 @@ func (u *UI) beliefRetract(w http.ResponseWriter, r *http.Request) { } reason := r.PostFormValue("reason") dni := r.PostFormValue("do_not_infer") == "1" - if _, err := u.deps.Beliefs.Retract(r.Context(), u.emitF(), m.ID, id, reason, username(r), dni); err != nil { + res, err := u.deps.Beliefs.Retract(r.Context(), u.emitF(), m.ID, id, reason, username(r), dni) + if err != nil { u.flash(w, r, "error", err.Error(), "/minds/"+m.Slug+"/beliefs/"+id.String()) return } + u.deps.Events.Publish(res.Events...) u.deps.Audit.Log(r.Context(), "belief.retract", "belief", id.String(), map[string]any{"reason": reason, "do_not_infer": dni}) u.flash(w, r, "ok", "Belief marked wrong. Its history is preserved.", "/minds/"+m.Slug+"/beliefs/"+id.String()) } diff --git a/internal/ui/memory.go b/internal/ui/memory.go index 987f690..6440343 100644 --- a/internal/ui/memory.go +++ b/internal/ui/memory.go @@ -122,7 +122,9 @@ func (u *UI) memoryRetract(w http.ResponseWriter, r *http.Request) { u.deps.Audit.Log(r.Context(), "memory.retract", "memory", id.String(), map[string]any{"reason": reason}) if affected, err := u.deps.Beliefs.RetractEvidence(r.Context(), u.deps.DB, m.ID, "memory", string(id)); err == nil { for _, bid := range affected { - u.deps.Beliefs.Reevaluate(r.Context(), u.deps.DB, u.emitF(), m.ID, bid, "supporting memory retracted by operator") //nolint:errcheck + if res, err := u.deps.Beliefs.Reevaluate(r.Context(), u.deps.DB, u.emitF(), m.ID, bid, "supporting memory retracted by operator"); err == nil { + u.deps.Events.Publish(res.Events...) + } } } u.flash(w, r, "ok", "Memory marked inaccurate. It remains in history.", "/minds/"+m.Slug+"/memory/"+id.String())