Normalize every intake path into one event envelope (#283)
Things arrive at Maven from eight directions — a relayed Android notification on POST /api/ambient, mail candidates from mavmaild, RSS items, changed pages from the crawler, zenmoney and wg reads from mavpoll, CalDAV events, presence probes, meeting transcripts and image descriptions. Each grew its own shape and its own log line, and nothing could answer "what came in today, from where". internal/event is that answer: a flat source-agnostic envelope (Source, Kind, EntityIDs, Title, Body, Priority, OccurredAt, Payload) plus a bounded in-memory journal. Both are pure — Publish and Normalize take `now` as a parameter, so no clock read sits on a path a replay would drive. Adopting it did not touch eight callers, because every intake path already converges on three ipc.CoreAPI methods: WriteFact, WriteNote and CaptureTask. cmd/mavend/intake.go decorates that ONE interface, so mavweb, mavcaldav, mavpoll, mavmaild and the in-core feed/crawl/capture/ vision workers publish envelopes without knowing events exist. The lone exception is cmd/mavend/mail.go, which captures through the store directly and now publishes explicitly. Nothing dispatches on an event. It is a report that something arrived, never an instruction to speak — "a feed item appeared" becoming a notification is the nag this repo refuses. Digestion may read the journal later; it will still go through internal/loop's rules and the severity/presence routing table. Read surface: ipc.MethodRecentEvents (AuthRead, daemon-cached like TickTrace — a bare store cannot serve a ring) and a read-only /events page in mavweb. Production is unchanged when nobody is watching: a nil *event.Bus makes Publish a no-op and newIntakeAPI returns the wrapped API untouched, so config.intake_journal < 0 leaves no decorator on the call path at all. The default is 512 entries; the "off unless configured" rule is for capabilities that reach out, and a bounded in-memory log of writes core already performed reaches nowhere. Verified: make build, make test (go test -race) both clean. New tests cover the envelope and ring (internal/event, 95.7%), the decorator's invariants — a failed write publishes nothing, a deduped capture publishes nothing, OccurredAt is the fact's Ts and not notice time — and the /events page including escaping of feed-supplied titles.
This commit is contained in:
kami
@@ -135,7 +135,13 @@ func Requirement(m ipc.Method) Authority {
|
||||
ipc.MethodListSpeakers,
|
||||
// The read side of the model swap: which model is resident, which ones are
|
||||
// allowlisted. It loads nothing and changes nothing.
|
||||
ipc.MethodModelStatus:
|
||||
ipc.MethodModelStatus,
|
||||
// The unified intake journal (Vikunja #283). AuthRead, and listed
|
||||
// explicitly rather than inherited so the reasoning is on the record: it
|
||||
// reports what already arrived — sources, keys, note headlines — which is
|
||||
// the same material RecentFacts and RecentNotes already return at this
|
||||
// rung. It writes nothing, and it holds nothing a fact read does not.
|
||||
ipc.MethodRecentEvents:
|
||||
return AuthRead
|
||||
}
|
||||
// Unknown method ⇒ AuthRead, but ipc.dispatch returns ErrUnknownMethod
|
||||
|
||||
@@ -169,6 +169,20 @@ type Config struct {
|
||||
// live in the reader (cmd/mavmaild), never here.
|
||||
Email *EmailConfig `json:"email,omitempty"`
|
||||
|
||||
// IntakeJournal — how many entries the unified intake journal keeps
|
||||
// (Vikunja #283): one envelope per thing that arrived, whatever direction it
|
||||
// came from. Absent ⇒ DefaultIntakeJournal. A NEGATIVE value turns the
|
||||
// journal off entirely, and then there is no decorator on the intake path at
|
||||
// all.
|
||||
//
|
||||
// Not gated behind an "off unless configured" block like feeds or telegram,
|
||||
// and the distinction is the one CLAUDE.md draws: that rule exists for
|
||||
// capabilities that reach OUT — a fetch, a send, a third party. This reaches
|
||||
// nowhere. It is a bounded in-memory log of writes core already performed,
|
||||
// it is read only by /events and the simulator, and nothing Maven says
|
||||
// depends on it.
|
||||
IntakeJournal int `json:"intake_journal,omitempty"`
|
||||
|
||||
// Feeds — RSS/Atom feed reading (Vikunja #258). nil / absent ⇒ no feed is
|
||||
// ever fetched: reading the outside world is off unless configured, like
|
||||
// the weather and telegram. See FeedsConfig.
|
||||
@@ -968,7 +982,11 @@ const (
|
||||
DefaultRepeatInterval = 5 * time.Minute
|
||||
DefaultAutotuneInterval = 10 * time.Minute
|
||||
DefaultRouterThreshold = 0.55
|
||||
DefaultQueryMinScore = 0.55
|
||||
// DefaultIntakeJournal — entries kept in the unified intake journal
|
||||
// (Vikunja #283). A busy day is a few hundred intake writes, so this is
|
||||
// roughly "today and yesterday" at a few hundred KB of memory.
|
||||
DefaultIntakeJournal = 512
|
||||
DefaultQueryMinScore = 0.55
|
||||
// Read off the margin sweep in internal/memory/recalleval on the e5
|
||||
// embedder: 0.008 answers 68% of real questions (down from 72%) and cuts
|
||||
// false recall from 5/5 to 1/5. Every larger delta costs real recall
|
||||
@@ -1018,6 +1036,9 @@ func Load(path string) (*Config, error) {
|
||||
}
|
||||
|
||||
func (c *Config) applyDefaults() {
|
||||
if c.IntakeJournal == 0 {
|
||||
c.IntakeJournal = DefaultIntakeJournal
|
||||
}
|
||||
if c.TickInterval == 0 {
|
||||
c.TickInterval = Duration(DefaultTickInterval)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,129 @@
|
||||
package event
|
||||
|
||||
import (
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Bus — the in-memory intake journal: a bounded ring of recent Events plus
|
||||
// zero or more subscribers.
|
||||
//
|
||||
// Two properties are load-bearing, both about not changing production
|
||||
// behaviour when nobody is watching:
|
||||
//
|
||||
// - A nil *Bus is a working no-op. Publish on nil returns immediately, so
|
||||
// an intake path can call b.Publish(...) unconditionally and a daemon that
|
||||
// never built a bus behaves exactly as it did before. This is what let
|
||||
// eight callers adopt the envelope without a config flag each.
|
||||
// - Publish never blocks on a subscriber and never propagates a panic from
|
||||
// one. Intake is on the request path of POST /api/ambient and of every
|
||||
// fact write; a slow or broken observer must not be able to stall or kill
|
||||
// a write that already succeeded.
|
||||
//
|
||||
// The ring is bounded because it is memory that nothing prunes otherwise. Its
|
||||
// contents are a window, not a record: the durable consequence of an event is
|
||||
// the fact, note or task the intake path wrote.
|
||||
type Bus struct {
|
||||
mu sync.Mutex
|
||||
ring []Event // len == cap once full; oldest at (next % cap)
|
||||
next int
|
||||
n int
|
||||
subs []func(Event)
|
||||
}
|
||||
|
||||
// DefaultCapacity — how many recent events a bus keeps. A busy day is a few
|
||||
// hundred intake events (a feed poll is one per new item), so this is roughly
|
||||
// "today and yesterday" at a few hundred KB.
|
||||
const DefaultCapacity = 512
|
||||
|
||||
// NewBus returns a bus keeping the last capacity events. capacity <= 0 uses
|
||||
// DefaultCapacity.
|
||||
func NewBus(capacity int) *Bus {
|
||||
if capacity <= 0 {
|
||||
capacity = DefaultCapacity
|
||||
}
|
||||
return &Bus{ring: make([]Event, capacity)}
|
||||
}
|
||||
|
||||
// Publish normalizes e, drops it if it is not Valid, appends it to the ring and
|
||||
// hands it to every subscriber. Safe on a nil receiver and safe from any
|
||||
// goroutine.
|
||||
//
|
||||
// now is passed in rather than read from the clock: the whole point of #284's
|
||||
// replay is that no time.Now() sits inside a path a scenario drives.
|
||||
func (b *Bus) Publish(e Event, now time.Time) {
|
||||
if b == nil {
|
||||
return
|
||||
}
|
||||
e = e.Normalize(now)
|
||||
if !e.Valid() {
|
||||
return
|
||||
}
|
||||
|
||||
b.mu.Lock()
|
||||
b.ring[b.next] = e
|
||||
b.next = (b.next + 1) % len(b.ring)
|
||||
if b.n < len(b.ring) {
|
||||
b.n++
|
||||
}
|
||||
subs := make([]func(Event), len(b.subs))
|
||||
copy(subs, b.subs)
|
||||
b.mu.Unlock()
|
||||
|
||||
for _, fn := range subs {
|
||||
notify(fn, e)
|
||||
}
|
||||
}
|
||||
|
||||
// notify calls one subscriber, swallowing a panic. A test double or a page
|
||||
// renderer must not be able to take down a daemon from the intake path.
|
||||
func notify(fn func(Event), e Event) {
|
||||
defer func() { _ = recover() }()
|
||||
fn(e)
|
||||
}
|
||||
|
||||
// Subscribe registers fn to be called for every subsequent event, in publish
|
||||
// order. There is no unsubscribe: subscribers are wired at startup and live as
|
||||
// long as the daemon. Safe on a nil receiver (the subscription is dropped,
|
||||
// which is the honest outcome when there is no bus to subscribe to).
|
||||
func (b *Bus) Subscribe(fn func(Event)) {
|
||||
if b == nil || fn == nil {
|
||||
return
|
||||
}
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
b.subs = append(b.subs, fn)
|
||||
}
|
||||
|
||||
// Recent returns up to limit events, newest first. limit <= 0 returns
|
||||
// everything held. Safe on a nil receiver (returns nil).
|
||||
func (b *Bus) Recent(limit int) []Event {
|
||||
if b == nil {
|
||||
return nil
|
||||
}
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
if b.n == 0 {
|
||||
return nil
|
||||
}
|
||||
if limit <= 0 || limit > b.n {
|
||||
limit = b.n
|
||||
}
|
||||
out := make([]Event, 0, limit)
|
||||
// next points one past the newest; walk backwards.
|
||||
for i := 0; i < limit; i++ {
|
||||
idx := (b.next - 1 - i + len(b.ring)*2) % len(b.ring)
|
||||
out = append(out, b.ring[idx])
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Len reports how many events the ring currently holds. Safe on nil.
|
||||
func (b *Bus) Len() int {
|
||||
if b == nil {
|
||||
return 0
|
||||
}
|
||||
b.mu.Lock()
|
||||
defer b.mu.Unlock()
|
||||
return b.n
|
||||
}
|
||||
@@ -0,0 +1,198 @@
|
||||
// Package event is the unified intake envelope (Vikunja #283,
|
||||
// 20-07-2026-BACKLOG.md item 1).
|
||||
//
|
||||
// # The problem it solves
|
||||
//
|
||||
// Things arrive at Maven from a lot of directions: a relayed Android
|
||||
// notification (POST /api/ambient), a mail the reader extracted candidates
|
||||
// from (ingest_mail), an RSS item, a changed page the crawler noticed, a
|
||||
// zenmoney spend, a CalDAV event, a wg handshake that means he is home, a
|
||||
// photo he sent, a meeting she was asked to record. Each of those grew its own
|
||||
// shape, its own storage decision and its own log line. Nothing could answer
|
||||
// "what came in today, from where" without reading eight packages.
|
||||
//
|
||||
// An Event is that answer: one flat, source-agnostic description of "something
|
||||
// arrived". It is deliberately NOT a new storage layer and NOT a new transport.
|
||||
// Every intake path keeps writing exactly what it wrote before — a fact, a
|
||||
// note, a candidate task — and additionally describes what it did as an Event.
|
||||
// The envelope is a VIEW over intake, not a replacement for it, which is why
|
||||
// adopting it did not require touching eight callers.
|
||||
//
|
||||
// # What it is not
|
||||
//
|
||||
// - Not a command. An Event is a report of something that happened; nothing
|
||||
// in Maven executes one. Digestion may read them; it may not be driven by
|
||||
// an event alone, because "a thing arrived" is not "a thing must be said".
|
||||
// - Not durable. The bus is a bounded in-memory ring. An event's durable
|
||||
// consequence is the fact/note/task the intake path already wrote; the
|
||||
// envelope is the recent-history window on top. A restart losing the ring
|
||||
// loses nothing that mattered.
|
||||
// - Not a secret store. Body carries what the intake path was already willing
|
||||
// to log or store. Nothing puts a mail body, an IMAP password or a
|
||||
// voiceprint in here, and callers must keep it that way.
|
||||
package event
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Event — one thing that arrived, normalized.
|
||||
//
|
||||
// The field set is the one recorded in the backlog, and it is intentionally
|
||||
// small: anything source-specific goes in Payload, so adding a source never
|
||||
// widens the struct and never breaks a reader.
|
||||
type Event struct {
|
||||
// Source — provenance, in the facts vocabulary already used across the
|
||||
// repo: "ambient:notif", "caldav:personal", "poll:zenmoney", "rss:<feed>",
|
||||
// "crawl:<watch>", "email:<mailbox>", "infer:wg", "tap:voice". Same string
|
||||
// the fact or note was written under, so an event and its row can be
|
||||
// matched up by eye.
|
||||
Source string `json:"source"`
|
||||
|
||||
// Kind — what sort of thing arrived, from the closed set below. This is the
|
||||
// field digestion switches on; Source is for provenance and display.
|
||||
Kind string `json:"kind"`
|
||||
|
||||
// EntityIDs — Nexus entity ids this event is about, when the intake path
|
||||
// knew any. Usually empty: most intake happens before enrichment resolves a
|
||||
// subject to an entity.
|
||||
EntityIDs []string `json:"entity_ids,omitempty"`
|
||||
|
||||
// Title — one short line, safe to show on a page. For a fact it is the key,
|
||||
// for a note the first line, for a task the task text.
|
||||
Title string `json:"title"`
|
||||
|
||||
// Body — optional detail, already truncated by the caller.
|
||||
Body string `json:"body,omitempty"`
|
||||
|
||||
// Priority — one of PriorityLow / PriorityNormal / PriorityHigh. It is a
|
||||
// hint about attention, not a delivery instruction: nothing here decides
|
||||
// whether Maven speaks. That stays with internal/loop and internal/delivery,
|
||||
// where the severity/presence routing table lives.
|
||||
Priority string `json:"priority"`
|
||||
|
||||
// OccurredAt — when the thing happened, NOT when Maven noticed it. A wg
|
||||
// handshake carries the handshake instant; an RSS item carries its publish
|
||||
// time. Intake paths already make this distinction when writing facts, and
|
||||
// the envelope must not flatten it.
|
||||
OccurredAt time.Time `json:"occurred_at"`
|
||||
|
||||
// Payload — source-specific extra, opaque here. Optional.
|
||||
Payload json.RawMessage `json:"payload,omitempty"`
|
||||
}
|
||||
|
||||
// Kinds. Closed set: a reader may switch on these exhaustively. A new intake
|
||||
// path picks the closest existing kind before it adds one — the point of the
|
||||
// envelope is that digestion has a small stable input.
|
||||
const (
|
||||
// KindFact — something was written to the facts table: a calendar read, a
|
||||
// zenmoney window, a presence probe, a crawler watermark.
|
||||
KindFact = "fact"
|
||||
|
||||
// KindNote — something was written to the notes table: an RSS item, a
|
||||
// changed page, a meeting transcript, an image description.
|
||||
KindNote = "note"
|
||||
|
||||
// KindTask — a candidate task was captured: the mail reader, the web form,
|
||||
// the voice path.
|
||||
KindTask = "task"
|
||||
|
||||
// KindMessage — an inbound message on a reach channel. Nothing produces
|
||||
// this yet (telegram is send-only today); the kind exists so the bridge,
|
||||
// when it lands, is a constructor and not a schema change.
|
||||
KindMessage = "message"
|
||||
|
||||
// KindHealth — a service or probe reported its own state.
|
||||
KindHealth = "health"
|
||||
)
|
||||
|
||||
// Priorities.
|
||||
const (
|
||||
PriorityLow = "low"
|
||||
PriorityNormal = "normal"
|
||||
PriorityHigh = "high"
|
||||
)
|
||||
|
||||
// TitleMaxRunes / BodyMaxRunes bound what an envelope carries. The ring is
|
||||
// in memory and served to a web page; a 40 KB crawled article has no business
|
||||
// in either. Cut on a rune boundary — most of this text is Russian and half a
|
||||
// cyrillic letter is a broken line.
|
||||
const (
|
||||
TitleMaxRunes = 120
|
||||
BodyMaxRunes = 400
|
||||
)
|
||||
|
||||
// Normalize returns e with its fields put in range: whitespace collapsed out
|
||||
// of Title, Title and Body truncated, an unknown or empty Priority forced to
|
||||
// PriorityNormal, and a zero OccurredAt filled from now.
|
||||
//
|
||||
// It takes now as a parameter rather than reading the clock, so the whole
|
||||
// package stays pure and the simulator (Vikunja #284) can replay intake against
|
||||
// a scripted clock.
|
||||
func (e Event) Normalize(now time.Time) Event {
|
||||
e.Title = truncateRunes(strings.Join(strings.Fields(e.Title), " "), TitleMaxRunes)
|
||||
e.Body = truncateRunes(strings.TrimSpace(e.Body), BodyMaxRunes)
|
||||
if !validPriority(e.Priority) {
|
||||
e.Priority = PriorityNormal
|
||||
}
|
||||
if e.Kind == "" {
|
||||
e.Kind = KindFact
|
||||
}
|
||||
if e.OccurredAt.IsZero() {
|
||||
e.OccurredAt = now
|
||||
}
|
||||
return e
|
||||
}
|
||||
|
||||
// Valid reports whether e carries the minimum a reader can rely on: a source,
|
||||
// a known kind, a title and a time. The bus drops anything that fails — an
|
||||
// envelope with no provenance is worse than no envelope, because it looks like
|
||||
// evidence.
|
||||
func (e Event) Valid() bool {
|
||||
return e.Source != "" && validKind(e.Kind) && e.Title != "" && !e.OccurredAt.IsZero()
|
||||
}
|
||||
|
||||
func validKind(k string) bool {
|
||||
switch k {
|
||||
case KindFact, KindNote, KindTask, KindMessage, KindHealth:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func validPriority(p string) bool {
|
||||
switch p {
|
||||
case PriorityLow, PriorityNormal, PriorityHigh:
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// truncateRunes cuts s to n runes, marking the cut.
|
||||
func truncateRunes(s string, n int) string {
|
||||
r := []rune(s)
|
||||
if len(r) <= n {
|
||||
return s
|
||||
}
|
||||
return string(r[:n]) + "…"
|
||||
}
|
||||
|
||||
// SourceKind guesses the Kind for a source string when the caller has not said
|
||||
// otherwise. It exists so the one intake decorator in cmd/mavend does not need
|
||||
// a switch per writer: the source prefix already tells you what arrived.
|
||||
//
|
||||
// Unknown prefixes get fallback, which is what the caller was going to write
|
||||
// anyway (a WriteFact call knows it is a fact).
|
||||
func SourceKind(source, fallback string) string {
|
||||
switch {
|
||||
case strings.HasPrefix(source, "rss:"), strings.HasPrefix(source, "crawl:"):
|
||||
return KindNote
|
||||
case strings.HasPrefix(source, "email:"):
|
||||
return KindTask
|
||||
case strings.HasPrefix(source, "probe:"), strings.HasPrefix(source, "health:"):
|
||||
return KindHealth
|
||||
}
|
||||
return fallback
|
||||
}
|
||||
@@ -0,0 +1,185 @@
|
||||
package event
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
var testNow = time.Date(2026, 8, 1, 9, 30, 0, 0, time.UTC)
|
||||
|
||||
func TestNormalizeFillsDefaults(t *testing.T) {
|
||||
got := Event{Source: "poll:zenmoney", Title: " spent today "}.Normalize(testNow)
|
||||
if got.Title != "spent today" {
|
||||
t.Errorf("title = %q, want collapsed whitespace", got.Title)
|
||||
}
|
||||
if got.Priority != PriorityNormal {
|
||||
t.Errorf("priority = %q, want %q", got.Priority, PriorityNormal)
|
||||
}
|
||||
if got.Kind != KindFact {
|
||||
t.Errorf("kind = %q, want %q", got.Kind, KindFact)
|
||||
}
|
||||
if !got.OccurredAt.Equal(testNow) {
|
||||
t.Errorf("occurred_at = %v, want %v", got.OccurredAt, testNow)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeKeepsRealOccurredAt(t *testing.T) {
|
||||
// A wg handshake carries the handshake instant, not "now". Flattening that
|
||||
// would make every intake look like it happened at notice time.
|
||||
real := testNow.Add(-3 * time.Hour)
|
||||
got := Event{Source: "infer:wg", Title: "wg_handshake", OccurredAt: real}.Normalize(testNow)
|
||||
if !got.OccurredAt.Equal(real) {
|
||||
t.Errorf("occurred_at = %v, want the supplied %v", got.OccurredAt, real)
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeTruncatesOnRuneBoundary(t *testing.T) {
|
||||
long := strings.Repeat("я", TitleMaxRunes+50)
|
||||
got := Event{Source: "rss:x", Title: long}.Normalize(testNow)
|
||||
r := []rune(got.Title)
|
||||
if len(r) != TitleMaxRunes+1 { // +1 for the ellipsis marker
|
||||
t.Fatalf("title runes = %d, want %d", len(r), TitleMaxRunes+1)
|
||||
}
|
||||
if r[len(r)-1] != '…' {
|
||||
t.Errorf("truncated title does not mark the cut: %q", string(r[len(r)-3:]))
|
||||
}
|
||||
for _, c := range r[:TitleMaxRunes] {
|
||||
if c != 'я' {
|
||||
t.Fatalf("truncation broke a rune: got %q", c)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestNormalizeRejectsUnknownPriority(t *testing.T) {
|
||||
got := Event{Source: "s", Title: "t", Priority: "URGENT!!"}.Normalize(testNow)
|
||||
if got.Priority != PriorityNormal {
|
||||
t.Errorf("priority = %q, want %q", got.Priority, PriorityNormal)
|
||||
}
|
||||
}
|
||||
|
||||
func TestValid(t *testing.T) {
|
||||
base := Event{Source: "rss:tech", Kind: KindNote, Title: "заголовок", OccurredAt: testNow}
|
||||
if !base.Valid() {
|
||||
t.Fatal("well-formed event reported invalid")
|
||||
}
|
||||
for name, mut := range map[string]func(Event) Event{
|
||||
"no source": func(e Event) Event { e.Source = ""; return e },
|
||||
"no title": func(e Event) Event { e.Title = ""; return e },
|
||||
"no time": func(e Event) Event { e.OccurredAt = time.Time{}; return e },
|
||||
"bad kind": func(e Event) Event { e.Kind = "whatever"; return e },
|
||||
} {
|
||||
if mut(base).Valid() {
|
||||
t.Errorf("%s: reported valid", name)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestSourceKind(t *testing.T) {
|
||||
cases := map[string]string{
|
||||
"rss:tech": KindNote,
|
||||
"crawl:kernel": KindNote,
|
||||
"email:inbox": KindTask,
|
||||
"probe:netdata": KindHealth,
|
||||
"ambient:notif": KindFact,
|
||||
"tap:voice": KindFact,
|
||||
}
|
||||
for src, want := range cases {
|
||||
if got := SourceKind(src, KindFact); got != want {
|
||||
t.Errorf("SourceKind(%q) = %q, want %q", src, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusNilIsANoOp(t *testing.T) {
|
||||
// The whole adoption story depends on this: an intake path calls Publish
|
||||
// unconditionally, and a daemon with no bus behaves as it did before.
|
||||
var b *Bus
|
||||
b.Publish(Event{Source: "s", Kind: KindFact, Title: "t"}, testNow)
|
||||
b.Subscribe(func(Event) { t.Error("nil bus delivered to a subscriber") })
|
||||
if got := b.Recent(10); got != nil {
|
||||
t.Errorf("Recent on nil bus = %v, want nil", got)
|
||||
}
|
||||
if got := b.Len(); got != 0 {
|
||||
t.Errorf("Len on nil bus = %d, want 0", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusRecentIsNewestFirst(t *testing.T) {
|
||||
b := NewBus(8)
|
||||
for _, title := range []string{"one", "two", "three"} {
|
||||
b.Publish(Event{Source: "rss:t", Kind: KindNote, Title: title}, testNow)
|
||||
}
|
||||
got := b.Recent(0)
|
||||
if len(got) != 3 {
|
||||
t.Fatalf("len = %d, want 3", len(got))
|
||||
}
|
||||
want := []string{"three", "two", "one"}
|
||||
for i, w := range want {
|
||||
if got[i].Title != w {
|
||||
t.Errorf("Recent()[%d] = %q, want %q", i, got[i].Title, w)
|
||||
}
|
||||
}
|
||||
if lim := b.Recent(2); len(lim) != 2 || lim[0].Title != "three" {
|
||||
t.Errorf("Recent(2) = %v, want the two newest", lim)
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusRingEvicts(t *testing.T) {
|
||||
b := NewBus(3)
|
||||
for _, title := range []string{"a", "b", "c", "d", "e"} {
|
||||
b.Publish(Event{Source: "s", Kind: KindFact, Title: title}, testNow)
|
||||
}
|
||||
if b.Len() != 3 {
|
||||
t.Fatalf("Len = %d, want the capacity 3", b.Len())
|
||||
}
|
||||
got := b.Recent(0)
|
||||
want := []string{"e", "d", "c"}
|
||||
for i, w := range want {
|
||||
if got[i].Title != w {
|
||||
t.Errorf("Recent()[%d] = %q, want %q", i, got[i].Title, w)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusDropsInvalid(t *testing.T) {
|
||||
b := NewBus(4)
|
||||
b.Publish(Event{Kind: KindFact, Title: "no source"}, testNow)
|
||||
b.Publish(Event{Source: "s", Kind: KindFact}, testNow)
|
||||
if b.Len() != 0 {
|
||||
t.Errorf("Len = %d, want 0 — an envelope with no provenance must not be kept", b.Len())
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusSubscriberPanicDoesNotBreakIntake(t *testing.T) {
|
||||
b := NewBus(4)
|
||||
var seen int
|
||||
b.Subscribe(func(Event) { panic("observer is broken") })
|
||||
b.Subscribe(func(Event) { seen++ })
|
||||
b.Publish(Event{Source: "s", Kind: KindFact, Title: "t"}, testNow)
|
||||
if seen != 1 {
|
||||
t.Errorf("healthy subscriber called %d times, want 1", seen)
|
||||
}
|
||||
if b.Len() != 1 {
|
||||
t.Errorf("event not recorded despite a panicking subscriber")
|
||||
}
|
||||
}
|
||||
|
||||
func TestBusConcurrentPublish(t *testing.T) {
|
||||
b := NewBus(256)
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < 16; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for j := 0; j < 10; j++ {
|
||||
b.Publish(Event{Source: "s", Kind: KindFact, Title: "t"}, testNow)
|
||||
}
|
||||
}()
|
||||
}
|
||||
wg.Wait()
|
||||
if b.Len() != 160 {
|
||||
t.Errorf("Len = %d, want 160", b.Len())
|
||||
}
|
||||
}
|
||||
@@ -660,6 +660,31 @@ type CoreAPI interface {
|
||||
// (router → dialogue → action → replier) and returns the reply text.
|
||||
// No audio or stt/tts — for text channels (mavweb, telegram).
|
||||
Chat(ctx context.Context, text string) (string, error)
|
||||
|
||||
// RecentEvents returns the daemon's unified intake journal, newest first
|
||||
// (Vikunja #283) — one envelope per thing that arrived, whatever direction
|
||||
// it came from: a relayed notification, a mail candidate, a feed item, a
|
||||
// changed page, a spend, a presence probe.
|
||||
//
|
||||
// Read-only and daemon-cached, the same shape as TickTrace and DayPlan:
|
||||
// the store adapter returns an error, because the journal is a bounded
|
||||
// in-memory ring and not a table. Its contents are a window over intake,
|
||||
// never the durable record — that is still the fact, note or task the
|
||||
// intake path wrote.
|
||||
RecentEvents(ctx context.Context, n int) ([]IntakeEvent, error)
|
||||
}
|
||||
|
||||
// IntakeEvent — one entry of the unified intake journal on the wire. Mirrors
|
||||
// event.Event field for field; the ipc package does not import internal/event
|
||||
// so the wire shape stays independent of the in-process type.
|
||||
type IntakeEvent struct {
|
||||
Source string `json:"source"`
|
||||
Kind string `json:"kind"`
|
||||
EntityIDs []string `json:"entity_ids,omitempty"`
|
||||
Title string `json:"title"`
|
||||
Body string `json:"body,omitempty"`
|
||||
Priority string `json:"priority"`
|
||||
OccurredAt time.Time `json:"occurred_at"`
|
||||
}
|
||||
|
||||
// --- Rule trace / explanation DTOs ---
|
||||
|
||||
@@ -74,6 +74,7 @@ var readOnlyMethods = map[Method]bool{
|
||||
MethodMorningStatus: true,
|
||||
MethodMCPServers: true,
|
||||
MethodDayPlan: true,
|
||||
MethodRecentEvents: true,
|
||||
}
|
||||
|
||||
// Dial connects to a core socket at path and returns a Client. The module
|
||||
@@ -593,6 +594,14 @@ func (c *Client) TickTrace(ctx context.Context) (TickTrace, error) {
|
||||
return t, nil
|
||||
}
|
||||
|
||||
func (c *Client) RecentEvents(ctx context.Context, n int) ([]IntakeEvent, error) {
|
||||
var e []IntakeEvent
|
||||
if err := c.call(ctx, MethodRecentEvents, nReq{N: n}, &e); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return e, nil
|
||||
}
|
||||
|
||||
func (c *Client) MCPServers(ctx context.Context) ([]MCPServerStatus, error) {
|
||||
var s []MCPServerStatus
|
||||
if err := c.call(ctx, MethodMCPServers, nil, &s); err != nil {
|
||||
|
||||
@@ -211,6 +211,12 @@ func (a *storeAPI) MorningStatus(ctx context.Context) ([]MorningRoutineStatus, e
|
||||
return nil, errors.New("store: morning status not available via direct store API")
|
||||
}
|
||||
|
||||
// RecentEvents — same shape as TickTrace: the intake journal is a bounded ring
|
||||
// in the daemon's memory, not a table, so a bare store cannot serve it.
|
||||
func (a *storeAPI) RecentEvents(ctx context.Context, n int) ([]IntakeEvent, error) {
|
||||
return nil, errors.New("store: intake events not available via direct store API")
|
||||
}
|
||||
|
||||
func (a *storeAPI) MCPServers(ctx context.Context) ([]MCPServerStatus, error) {
|
||||
return nil, nil // no manager behind a bare store: nothing configured
|
||||
}
|
||||
@@ -884,6 +890,16 @@ var methodTable = map[Method]handlerFunc{
|
||||
MethodMorningStatus: withoutParams(func(ctx context.Context, api CoreAPI) ([]MorningRoutineStatus, error) {
|
||||
return api.MorningStatus(ctx)
|
||||
}),
|
||||
MethodRecentEvents: withParams(func(ctx context.Context, api CoreAPI, p nReq) ([]IntakeEvent, error) {
|
||||
out, err := api.RecentEvents(ctx, p.N)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if out == nil {
|
||||
out = []IntakeEvent{}
|
||||
}
|
||||
return out, nil
|
||||
}),
|
||||
MethodMCPServers: withoutParams(func(ctx context.Context, api CoreAPI) ([]MCPServerStatus, error) {
|
||||
out, err := api.MCPServers(ctx)
|
||||
if err != nil {
|
||||
|
||||
@@ -122,6 +122,9 @@ func (UnimplementedCoreAPI) TickTrace(ctx context.Context) (TickTrace, error) {
|
||||
func (UnimplementedCoreAPI) MorningStatus(ctx context.Context) ([]MorningRoutineStatus, error) {
|
||||
return nil, ErrNotImplemented
|
||||
}
|
||||
func (UnimplementedCoreAPI) RecentEvents(ctx context.Context, n int) ([]IntakeEvent, error) {
|
||||
return nil, ErrNotImplemented
|
||||
}
|
||||
func (UnimplementedCoreAPI) MCPServers(ctx context.Context) ([]MCPServerStatus, error) {
|
||||
return nil, ErrNotImplemented
|
||||
}
|
||||
|
||||
@@ -62,6 +62,7 @@ const (
|
||||
MethodEnrollSpeaker Method = "enroll_speaker"
|
||||
MethodListSpeakers Method = "list_speakers"
|
||||
MethodForgetSpeaker Method = "forget_speaker"
|
||||
MethodRecentEvents Method = "recent_events"
|
||||
)
|
||||
|
||||
// Request — one frame from module to core. Params is the JSON-encoded argument
|
||||
|
||||
Reference in New Issue
Block a user