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.
This commit is contained in:
Andy Dixon
2026-09-27 22:19:21 +01:00
parent 2ecdbc39b5
commit 212eef6b8c
5 changed files with 23 additions and 6 deletions
+7 -1
View File
@@ -31,13 +31,16 @@ func (s *Server) beliefRoutes() {
// Emitter returns a beliefs.Emitter that appends through the event store and // Emitter returns a beliefs.Emitter that appends through the event store and
// publishes after the surrounding transaction (or immediately on the pool). // 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 { func (s *Server) Emitter() beliefs.Emitter {
return func(ctx context.Context, q storage.Querier, n events.New) (events.Event, error) { return func(ctx context.Context, q storage.Querier, n events.New) (events.Event, error) {
out, err := s.deps.Events.AppendIn(ctx, q, n) out, err := s.deps.Events.AppendIn(ctx, q, n)
if err != nil { if err != nil {
return events.Event{}, err return events.Event{}, err
} }
s.deps.Events.Publish(out...)
return out[0], nil return out[0], nil
} }
} }
@@ -117,6 +120,7 @@ func (s *Server) beliefsCreate(w http.ResponseWriter, r *http.Request) {
fail(w, r, beliefErr(err)) fail(w, r, beliefErr(err))
return 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}) 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) writeJSON(w, http.StatusCreated, res)
} }
@@ -263,6 +267,7 @@ func (s *Server) beliefEvidence(w http.ResponseWriter, r *http.Request) {
fail(w, r, beliefErr(err)) fail(w, r, beliefErr(err))
return 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}) 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) writeJSON(w, http.StatusOK, res)
} }
@@ -319,6 +324,7 @@ func (s *Server) beliefRetract(w http.ResponseWriter, r *http.Request) {
fail(w, r, beliefErr(err)) fail(w, r, beliefErr(err))
return 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}) 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) writeJSON(w, http.StatusOK, res)
} }
+2 -1
View File
@@ -86,7 +86,7 @@ func (s *Server) mindImport(w http.ResponseWriter, r *http.Request) {
if weight <= 0 { if weight <= 0 {
weight = 0.5 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)}}}) 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 err != nil {
if errors.Is(err, beliefs.ErrInvalid) || errors.Is(err, beliefs.ErrBlocked) { 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) fail(w, r, err)
return return
} }
s.deps.Events.Publish(res.Events...)
beliefCount++ beliefCount++
} }
s.deps.Brain.Reconcile(r.Context()) //nolint:errcheck s.deps.Brain.Reconcile(r.Context()) //nolint:errcheck
+5 -2
View File
@@ -283,8 +283,11 @@ func (s *Server) propagateMemoryRetraction(r *http.Request, mind, memoryID ids.I
return return
} }
for _, bid := range affected { 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) logger(r).Warn("re-evaluate failed", "belief", bid, "err", err)
} continue
}
s.deps.Events.Publish(res.Events...)
} }
} }
+7 -2
View File
@@ -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") u.flash(w, r, "error", err.Error(), "/minds/"+m.Slug+"/beliefs")
return 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.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()) 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 r.ParseForm() //nolint:errcheck
weight, _ := strconv.ParseFloat(r.PostFormValue("weight"), 64) weight, _ := strconv.ParseFloat(r.PostFormValue("weight"), 64)
note := r.PostFormValue("note") 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()) u.flash(w, r, "error", err.Error(), "/minds/"+m.Slug+"/beliefs/"+id.String())
return 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.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()) 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") reason := r.PostFormValue("reason")
dni := r.PostFormValue("do_not_infer") == "1" 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()) u.flash(w, r, "error", err.Error(), "/minds/"+m.Slug+"/beliefs/"+id.String())
return 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.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()) u.flash(w, r, "ok", "Belief marked wrong. Its history is preserved.", "/minds/"+m.Slug+"/beliefs/"+id.String())
} }
+3 -1
View File
@@ -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}) 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 { if affected, err := u.deps.Beliefs.RetractEvidence(r.Context(), u.deps.DB, m.ID, "memory", string(id)); err == nil {
for _, bid := range affected { 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()) u.flash(w, r, "ok", "Memory marked inaccurate. It remains in history.", "/minds/"+m.Slug+"/memory/"+id.String())