diff --git a/cmd/mavend/intake.go b/cmd/mavend/intake.go
new file mode 100644
index 0000000..0e7542d
--- /dev/null
+++ b/cmd/mavend/intake.go
@@ -0,0 +1,235 @@
+// mavend/intake.go — the unified event intake envelope, wired (Vikunja #283).
+//
+// internal/event defines the envelope and the bounded in-memory journal. This
+// file is the one place that FILLS it, and the reason it is one place is worth
+// stating, because the alternative was eight patches:
+//
+// Every intake path in Maven already converges on three writes, and all three
+// are ipc.CoreAPI methods —
+//
+// WriteFact ← POST /api/ambient, mavcaldav, mavpoll's zenmoney + wg reads,
+// /api/signal presence probes, the RSS/crawl watermarks
+// WriteNote ← the RSS poller, the page crawler, meeting transcripts,
+// image descriptions
+// CaptureTask ← the voice path, the web form, and the mail reader
+//
+// — so decorating that ONE interface with a publish covers the lot without a
+// caller knowing about events at all. cmd/mavmaild, cmd/mavcaldav, cmd/mavpoll,
+// cmd/mavweb and the in-core feed/crawl/capture/vision workers are unchanged:
+// they call the same interface they always called, and it now also narrates.
+//
+// The exception is cmd/mavend/mail.go, which reaches past the interface to
+// st.CaptureTask directly. It publishes explicitly; see mailIntake.ingest.
+//
+// # Production behaviour when nobody is watching
+//
+// A nil *event.Bus makes Publish a no-op, and newIntakeAPI with a nil bus
+// returns the wrapped API unchanged, so there is not even a decorator on the
+// call path. The journal is memory-only and is never consulted by the tick
+// loop, the router, or delivery — nothing Maven says depends on it. It is a
+// read surface (`/events`, `recent_events`) and an observation seam for the
+// simulator.
+//
+// # What is deliberately NOT here
+//
+// No dispatch. An event 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 one day read the journal; it will still
+// go through internal/loop's rules and the severity/presence routing table.
+package main
+
+import (
+ "context"
+ "log"
+ "strings"
+ "time"
+
+ "github.com/kami/maven/internal/config"
+ "github.com/kami/maven/internal/event"
+ "github.com/kami/maven/internal/ipc"
+ "github.com/kami/maven/internal/store"
+)
+
+// newEventBus builds the journal, or returns nil when the operator turned it
+// off (a negative config.intake_journal). nil is the "behave exactly as before"
+// value all the way down: no decorator, no ring, no /events rows.
+func newEventBus(cfg *config.Config) *event.Bus {
+ if cfg == nil || cfg.IntakeJournal < 0 {
+ log.Printf("intake journal: off (intake_journal < 0)")
+ return nil
+ }
+ n := cfg.IntakeJournal
+ if n == 0 {
+ n = config.DefaultIntakeJournal
+ }
+ log.Printf("intake journal: keeping the last %d intake events in memory", n)
+ return event.NewBus(n)
+}
+
+// intakeEventsFn is the daemonAPI.getEvents closure: the bus's ring rendered as
+// the wire type. Returns nil for a nil bus, which the daemonAPI reports as an
+// empty journal rather than an error.
+func intakeEventsFn(bus *event.Bus) func(n int) []ipc.IntakeEvent {
+ if bus == nil {
+ return nil
+ }
+ return func(n int) []ipc.IntakeEvent {
+ evs := bus.Recent(n)
+ out := make([]ipc.IntakeEvent, 0, len(evs))
+ for _, e := range evs {
+ out = append(out, ipc.IntakeEvent{
+ Source: e.Source,
+ Kind: e.Kind,
+ EntityIDs: e.EntityIDs,
+ Title: e.Title,
+ Body: e.Body,
+ Priority: e.Priority,
+ OccurredAt: e.OccurredAt,
+ })
+ }
+ return out
+ }
+}
+
+// intakeAPI decorates a CoreAPI, publishing one envelope per successful
+// intake write. Embedding the interface means every other method passes
+// through untouched, and a new CoreAPI method is inherited rather than
+// silently dropped.
+type intakeAPI struct {
+ ipc.CoreAPI
+ bus *event.Bus
+ now func() time.Time
+}
+
+// newIntakeAPI wraps api so its intake writes are journalled. A nil bus
+// returns api itself — no decorator, no allocation, no behaviour change.
+func newIntakeAPI(api ipc.CoreAPI, bus *event.Bus, now func() time.Time) ipc.CoreAPI {
+ if bus == nil || api == nil {
+ return api
+ }
+ if now == nil {
+ now = time.Now
+ }
+ return &intakeAPI{CoreAPI: api, bus: bus, now: now}
+}
+
+// WriteFact journals the fact after it lands. Order matters: an event is a
+// report of something that HAPPENED, so a failed write publishes nothing.
+func (a *intakeAPI) WriteFact(ctx context.Context, req ipc.WriteFactReq) (int64, error) {
+ id, err := a.CoreAPI.WriteFact(ctx, req)
+ if err != nil {
+ return id, err
+ }
+ // OccurredAt is req.Ts, not now: mavpoll's wg read carries the handshake
+ // instant and the ambient path carries the meeting's start. Flattening
+ // those to notice-time would make the journal lie about when things
+ // happened, which is the one thing it is for.
+ a.bus.Publish(event.Event{
+ Source: req.Source,
+ Kind: event.SourceKind(req.Source, event.KindFact),
+ Title: req.Key,
+ Body: req.Value,
+ Priority: factPriority(req),
+ OccurredAt: req.Ts,
+ EntityIDs: entityIDs(req.Subject),
+ }, a.now())
+ return id, nil
+}
+
+// WriteNote journals a note. This is the RSS and crawler path, and also the
+// meeting transcript and image description paths, which write their derived
+// text as ordinary notes.
+func (a *intakeAPI) WriteNote(ctx context.Context, ts time.Time, text string, embedding []float32, source string) (int64, error) {
+ id, err := a.CoreAPI.WriteNote(ctx, ts, text, embedding, source)
+ if err != nil {
+ return id, err
+ }
+ title, body := splitFirstLine(text)
+ a.bus.Publish(event.Event{
+ Source: source,
+ Kind: event.SourceKind(source, event.KindNote),
+ Title: title,
+ Body: body,
+ Priority: event.PriorityLow,
+ OccurredAt: ts,
+ }, a.now())
+ return id, nil
+}
+
+// CaptureTask journals a captured task, but only when a row was actually
+// created. CaptureTask dedupes on normalised text among live rows, so a
+// mailbox re-read after a restart must not refill the journal with tasks that
+// were already there.
+func (a *intakeAPI) CaptureTask(ctx context.Context, req ipc.CaptureTaskReq) (ipc.CaptureTaskResp, error) {
+ resp, err := a.CoreAPI.CaptureTask(ctx, req)
+ if err != nil || !resp.Created {
+ return resp, err
+ }
+ a.bus.Publish(publishableTask(store.Task{
+ CreatedTs: req.Ts,
+ Text: req.Text,
+ Source: req.Source,
+ Evidence: req.Evidence,
+ Status: req.Status,
+ Due: req.Due,
+ }, a.now()), a.now())
+ return resp, nil
+}
+
+// publishableTask is the task→envelope shape, shared with mail.go, which
+// captures through the store directly rather than through the interface.
+//
+// Priority is high for a candidate with a due date and normal otherwise. That
+// is the only place this file makes a judgement, and it is a display hint on a
+// review page — nothing routes on it.
+func publishableTask(t store.Task, now time.Time) event.Event {
+ occurred := t.CreatedTs
+ if occurred.IsZero() {
+ occurred = now
+ }
+ prio := event.PriorityNormal
+ if t.Due != nil {
+ prio = event.PriorityHigh
+ }
+ return event.Event{
+ Source: t.Source,
+ Kind: event.KindTask,
+ Title: t.Text,
+ Body: t.Evidence,
+ Priority: prio,
+ OccurredAt: occurred,
+ }
+}
+
+// factPriority is the attention hint for a fact write. Deliberately crude:
+// a low-confidence inference (the ambient notification path writes below 1.0)
+// is worth less attention than a read he or a credentialled poller made, and
+// nothing else is distinguishable from here.
+func factPriority(req ipc.WriteFactReq) string {
+ if req.Confidence > 0 && req.Confidence < 1.0 {
+ return event.PriorityLow
+ }
+ return event.PriorityNormal
+}
+
+// entityIDs turns a fact's free-text Subject into the EntityIDs slot when it
+// already looks resolved. Intake runs BEFORE the fact enrichment worker
+// resolves a subject against Nexus, so this is almost always empty — the slot
+// exists for the paths that do know (the ecosystem acts), not for guessing.
+func entityIDs(subject string) []string {
+ subject = strings.TrimSpace(subject)
+ if subject == "" || !strings.HasPrefix(subject, "entity:") {
+ return nil
+ }
+ return []string{strings.TrimPrefix(subject, "entity:")}
+}
+
+// splitFirstLine renders a note as title + body. Feed and crawl notes are
+// written "headline\nsummary\nlink", so the first line is already the title.
+func splitFirstLine(text string) (title, body string) {
+ text = strings.TrimSpace(text)
+ if i := strings.IndexByte(text, '\n'); i >= 0 {
+ return strings.TrimSpace(text[:i]), strings.TrimSpace(text[i+1:])
+ }
+ return text, ""
+}
diff --git a/cmd/mavend/intake_test.go b/cmd/mavend/intake_test.go
new file mode 100644
index 0000000..1105628
--- /dev/null
+++ b/cmd/mavend/intake_test.go
@@ -0,0 +1,178 @@
+package main
+
+import (
+ "context"
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/kami/maven/internal/config"
+ "github.com/kami/maven/internal/event"
+ "github.com/kami/maven/internal/ipc"
+)
+
+var intakeNow = time.Date(2026, 8, 1, 10, 0, 0, 0, time.UTC)
+
+func intakeClock() time.Time { return intakeNow }
+
+// failingAPI wraps the store adapter, failing the three intake writes on
+// demand, so the "a failed write publishes nothing" invariant is testable.
+type failingAPI struct {
+ ipc.CoreAPI
+ fail bool
+}
+
+func (f *failingAPI) WriteFact(ctx context.Context, req ipc.WriteFactReq) (int64, error) {
+ if f.fail {
+ return 0, errors.New("injected")
+ }
+ return f.CoreAPI.WriteFact(ctx, req)
+}
+
+func newIntakeTestAPI(t *testing.T) (ipc.CoreAPI, *event.Bus) {
+ t.Helper()
+ st := newTestStore(t)
+ bus := event.NewBus(32)
+ return newIntakeAPI(ipc.NewStoreAPI(st), bus, intakeClock), bus
+}
+
+func TestIntakeAPIWithoutBusIsTheBareAPI(t *testing.T) {
+ // The adoption invariant: with the journal off there is not even a
+ // decorator on the intake path, so production behaves exactly as before.
+ st := newTestStore(t)
+ bare := ipc.NewStoreAPI(st)
+ if got := newIntakeAPI(bare, nil, intakeClock); got != ipc.CoreAPI(bare) {
+ t.Errorf("newIntakeAPI with a nil bus returned a wrapper, want the bare API")
+ }
+}
+
+func TestNewEventBusOffWhenNegative(t *testing.T) {
+ if b := newEventBus(&config.Config{IntakeJournal: -1}); b != nil {
+ t.Error("intake_journal = -1 still built a bus")
+ }
+ if b := newEventBus(&config.Config{IntakeJournal: 4}); b == nil {
+ t.Error("intake_journal = 4 built no bus")
+ }
+}
+
+func TestIntakeJournalsAFactWrite(t *testing.T) {
+ api, bus := newIntakeTestAPI(t)
+ ctx := context.Background()
+ // The ambient path's shape: an env fact below full confidence, timestamped
+ // at the meeting's start rather than at notice time.
+ start := intakeNow.Add(2 * time.Hour)
+ if _, err := api.WriteFact(ctx, ipc.WriteFactReq{
+ Ts: start, Kind: "env", Key: "calendar_event_20260801_планёрка",
+ Value: "10:00-11:00 планёрка", Source: "ambient:notif", Confidence: 0.6,
+ }); err != nil {
+ t.Fatalf("WriteFact: %v", err)
+ }
+ got := bus.Recent(0)
+ if len(got) != 1 {
+ t.Fatalf("journal has %d entries, want 1", len(got))
+ }
+ e := got[0]
+ if e.Source != "ambient:notif" || e.Kind != event.KindFact {
+ t.Errorf("source/kind = %q/%q", e.Source, e.Kind)
+ }
+ if e.Title != "calendar_event_20260801_планёрка" {
+ t.Errorf("title = %q, want the fact key", e.Title)
+ }
+ if !e.OccurredAt.Equal(start) {
+ t.Errorf("occurred_at = %v, want the fact's Ts %v — the journal must not flatten intake to notice time", e.OccurredAt, start)
+ }
+ if e.Priority != event.PriorityLow {
+ t.Errorf("priority = %q, want %q for a sub-1.0 confidence read", e.Priority, event.PriorityLow)
+ }
+}
+
+func TestIntakeDoesNotJournalAFailedWrite(t *testing.T) {
+ st := newTestStore(t)
+ bus := event.NewBus(8)
+ api := newIntakeAPI(&failingAPI{CoreAPI: ipc.NewStoreAPI(st), fail: true}, bus, intakeClock)
+ if _, err := api.WriteFact(context.Background(), ipc.WriteFactReq{
+ Ts: intakeNow, Kind: "env", Key: "k", Value: "v", Source: "poll:zenmoney", Confidence: 1,
+ }); err == nil {
+ t.Fatal("expected the injected error")
+ }
+ if bus.Len() != 0 {
+ t.Errorf("journal has %d entries after a failed write, want 0 — an event reports something that happened", bus.Len())
+ }
+}
+
+func TestIntakeJournalsANoteAsTitlePlusBody(t *testing.T) {
+ api, bus := newIntakeTestAPI(t)
+ // The RSS shape: "headline\nsummary\nlink".
+ if _, err := api.WriteNote(context.Background(), intakeNow,
+ "Вышло ядро 6.19\nкраткое содержание\nhttps://example.org/a", nil, "rss:tech"); err != nil {
+ t.Fatalf("WriteNote: %v", err)
+ }
+ got := bus.Recent(1)
+ if len(got) != 1 {
+ t.Fatalf("journal has %d entries, want 1", len(got))
+ }
+ if got[0].Title != "Вышло ядро 6.19" {
+ t.Errorf("title = %q, want the headline", got[0].Title)
+ }
+ if got[0].Kind != event.KindNote {
+ t.Errorf("kind = %q, want %q", got[0].Kind, event.KindNote)
+ }
+ if got[0].Body == "" {
+ t.Error("body is empty, want the rest of the note")
+ }
+}
+
+func TestIntakeJournalsOnlyCreatedTasks(t *testing.T) {
+ api, bus := newIntakeTestAPI(t)
+ ctx := context.Background()
+ req := ipc.CaptureTaskReq{Text: "оплатить интернет", Source: "email:inbox", Status: "candidate", Ts: intakeNow}
+ if _, err := api.CaptureTask(ctx, req); err != nil {
+ t.Fatalf("CaptureTask: %v", err)
+ }
+ // Same text again: CaptureTask dedupes among live rows, and a re-read of a
+ // mailbox must not refill the journal.
+ resp, err := api.CaptureTask(ctx, req)
+ if err != nil {
+ t.Fatalf("CaptureTask (repeat): %v", err)
+ }
+ if resp.Created {
+ t.Fatal("store did not dedupe; the test cannot check what it means to")
+ }
+ if bus.Len() != 1 {
+ t.Errorf("journal has %d entries, want 1 — a deduped capture must not publish", bus.Len())
+ }
+ if got := bus.Recent(1)[0]; got.Kind != event.KindTask || got.Title != "оплатить интернет" {
+ t.Errorf("entry = %+v, want the captured task", got)
+ }
+}
+
+func TestIntakeEventsFnRendersNewestFirst(t *testing.T) {
+ api, bus := newIntakeTestAPI(t)
+ ctx := context.Background()
+ for _, key := range []string{"a", "b", "c"} {
+ if _, err := api.WriteFact(ctx, ipc.WriteFactReq{
+ Ts: intakeNow, Kind: "env", Key: key, Value: "1", Source: "poll:zenmoney", Confidence: 1,
+ }); err != nil {
+ t.Fatalf("WriteFact %s: %v", key, err)
+ }
+ }
+ fn := intakeEventsFn(bus)
+ got := fn(2)
+ if len(got) != 2 || got[0].Title != "c" || got[1].Title != "b" {
+ t.Errorf("intakeEventsFn(2) = %+v, want the two newest, newest first", got)
+ }
+ if intakeEventsFn(nil) != nil {
+ t.Error("intakeEventsFn(nil) returned a closure, want nil so daemonAPI reports an empty journal")
+ }
+}
+
+func TestDaemonAPIRecentEventsEmptyWithoutABus(t *testing.T) {
+ d := &daemonAPI{CoreAPI: ipc.UnimplementedCoreAPI{}}
+ got, err := d.RecentEvents(context.Background(), 10)
+ if err != nil {
+ t.Fatalf("RecentEvents with no journal errored: %v", err)
+ }
+ if len(got) != 0 {
+ t.Errorf("got %d events, want none", len(got))
+ }
+}
diff --git a/cmd/mavend/mail.go b/cmd/mavend/mail.go
index ebde48a..5167e71 100644
--- a/cmd/mavend/mail.go
+++ b/cmd/mavend/mail.go
@@ -26,6 +26,7 @@ import (
"github.com/kami/maven/internal/config"
"github.com/kami/maven/internal/email"
+ "github.com/kami/maven/internal/event"
"github.com/kami/maven/internal/ipc"
"github.com/kami/maven/internal/phraser"
"github.com/kami/maven/internal/store"
@@ -42,6 +43,11 @@ type mailIntake struct {
ex *email.Extractor
timeout time.Duration
now func() time.Time
+ // bus — the unified intake journal (Vikunja #283). This path captures
+ // through the store directly rather than through ipc.CoreAPI, so the
+ // decorator in intake.go does not see it and the publish is explicit here.
+ // nil is a working no-op.
+ bus *event.Bus
}
// newMailIntake returns nil when mail ingestion must not be available, which is
@@ -52,7 +58,7 @@ type mailIntake struct {
// no keyword fallback: "the subject line became a task" is not extraction,
// it is a mailbox rendered as a to-do list, and it would fill the review
// page faster than he could clear it.
-func newMailIntake(st *store.Store, phr phraser.Phraser, cfg *config.Config) *mailIntake {
+func newMailIntake(st *store.Store, phr phraser.Phraser, cfg *config.Config, bus *event.Bus) *mailIntake {
if cfg.Email == nil {
return nil
}
@@ -67,7 +73,7 @@ func newMailIntake(st *store.Store, phr phraser.Phraser, cfg *config.Config) *ma
}
ex := email.NewExtractor(llmClientFor(lp, timeout), cfg.Email.MaxTasks, contextBlockFn(cfg, time.Now))
log.Printf("mail intake: enabled (max %d candidates per message, timeout %s)", cfg.Email.MaxTasks, timeout)
- return &mailIntake{st: st, ex: ex, timeout: timeout, now: time.Now}
+ return &mailIntake{st: st, ex: ex, timeout: timeout, now: time.Now, bus: bus}
}
// ingest handles one ipc.MethodIngestMail call.
@@ -128,6 +134,10 @@ func (m *mailIntake) ingest(ctx context.Context, req ipc.IngestMailReq) (ipc.Ing
resp.TaskIDs = append(resp.TaskIDs, id)
if created {
resp.Created++
+ // Only a row that was actually created. CaptureTask dedupes on
+ // normalised text among live rows, so a mailbox re-read after a
+ // restart must not refill the journal with tasks already in it.
+ m.bus.Publish(publishableTask(t, now), now)
}
}
// Counts only: the log line names the mailbox and the UID, never the subject,
@@ -139,8 +149,8 @@ func (m *mailIntake) ingest(ctx context.Context, req ipc.IngestMailReq) (ipc.Ing
// wireMailIntake installs the IPC hook, or leaves it nil so the method reports
// ErrUnknownMethod. Called on both startup paths (unlocked boot and passkey
// unlock) so mail behaves the same either way.
-func wireMailIntake(srv *ipc.Server, st *store.Store, phr phraser.Phraser, cfg *config.Config) {
- mi := newMailIntake(st, phr, cfg)
+func wireMailIntake(srv *ipc.Server, st *store.Store, phr phraser.Phraser, cfg *config.Config, bus *event.Bus) {
+ mi := newMailIntake(st, phr, cfg, bus)
if mi == nil {
return
}
diff --git a/cmd/mavend/mail_test.go b/cmd/mavend/mail_test.go
index d06bb70..05a5112 100644
--- a/cmd/mavend/mail_test.go
+++ b/cmd/mavend/mail_test.go
@@ -171,12 +171,12 @@ func TestIngestTruncatesEvidence(t *testing.T) {
// exist at all.
func TestNewMailIntakeOffWithoutConfig(t *testing.T) {
st := newTestStore(t)
- if mi := newMailIntake(st, nil, &config.Config{}); mi != nil {
+ if mi := newMailIntake(st, nil, &config.Config{}, nil); mi != nil {
t.Error("no email block must mean no mail intake")
}
// Configured but with a non-LLM phraser: still off — there is no fallback
// extraction, by design.
- if mi := newMailIntake(st, nil, &config.Config{Email: &config.EmailConfig{}}); mi != nil {
+ if mi := newMailIntake(st, nil, &config.Config{Email: &config.EmailConfig{}}, nil); mi != nil {
t.Error("without a llama-server phraser there is nothing to extract with")
}
}
diff --git a/cmd/mavend/main.go b/cmd/mavend/main.go
index 0702ac0..4913d91 100644
--- a/cmd/mavend/main.go
+++ b/cmd/mavend/main.go
@@ -204,6 +204,16 @@ func run(args []string) error {
crawlWkr *crawlWorker // nil ⇒ no page is watched (the default)
)
+ // The unified intake journal (Vikunja #283). Built before anything else
+ // that holds a CoreAPI, because intakeAPI wraps that one interface and
+ // every intake path in the daemon reaches its sink through it. nil (the
+ // operator set intake_journal negative) means no decorator at all.
+ evBus := newEventBus(cfg)
+ // coreFor is what every in-process holder of a CoreAPI now takes, instead
+ // of a bare ipc.NewStoreAPI(st). Identical behaviour plus one published
+ // envelope per successful intake write.
+ coreFor := func() ipc.CoreAPI { return newIntakeAPI(ipc.NewStoreAPI(st), evBus, time.Now) }
+
if !locked {
rules = loop.DefaultRules()
gatherer = loop.NewGatherer(st, rules)
@@ -247,7 +257,7 @@ func run(args []string) error {
eco = wireEcosystem(cfg)
// voice
- voiceW, err = wireVoice(cfg, ipc.NewStoreAPI(st), phr, st.VectorMemory(), st, eco)
+ voiceW, err = wireVoice(cfg, coreFor(), phr, st.VectorMemory(), st, eco)
if err != nil {
return fmt.Errorf("wire voice: %w", err)
}
@@ -297,14 +307,15 @@ func run(args []string) error {
tl = newTickLoop(st, gatherer, dispatcher, phr, rules, tickInterval, repeatInterval, autotuneInterval, cfg.Digest, routinesFromConfig(cfg.Routines), config.MorningRoutinesFromConfig(cfg.MorningRoutines), cfg.PatternProposals)
factWorker = newFactEnrichmentWorker(st, eco, time.Duration(cfg.FactEnrichmentInterval))
evalWorker = newMemoryEvalWorker(st, phr, cfg)
- feedWkr = newFeedWorker(ipc.NewStoreAPI(st), embedderOf(voiceW), cfg)
- crawlWkr = newCrawlWorker(newCrawler(cfg), ipc.NewStoreAPI(st), embedderOf(voiceW), cfg)
+ feedWkr = newFeedWorker(coreFor(), embedderOf(voiceW), cfg)
+ crawlWkr = newCrawlWorker(newCrawler(cfg), coreFor(), embedderOf(voiceW), cfg)
coreAPI = &daemonAPI{
- CoreAPI: ipc.NewStoreAPI(st),
+ CoreAPI: coreFor(),
getTrace: tl.trace,
getMorningStatus: func(ctx context.Context) []ipc.MorningRoutineStatus { return tl.morningStatus(ctx, time.Now()) },
getDayPlan: func(ctx context.Context) ipc.DayPlan { return tl.dayPlan(ctx, time.Now()) },
+ getEvents: intakeEventsFn(evBus),
}
if voiceW != nil && voiceW.handler != nil {
api := coreAPI.(*daemonAPI)
@@ -361,7 +372,7 @@ func run(args []string) error {
// configured and there is a llama-server to extract with, in which case
// ipc.MethodIngestMail reports ErrUnknownMethod.
if !locked {
- wireMailIntake(srv, st, phr, cfg)
+ wireMailIntake(srv, st, phr, cfg, evBus)
wireModelSwap(srv, phr, cfg)
// Vision + the media blob store (Vikunja #252). Both stay dark without a
// media block; MethodDescribeImage answers ErrUnknownMethod then.
@@ -482,7 +493,7 @@ func run(args []string) error {
eco = wireEcosystem(cfg)
- voiceW, err = wireVoice(cfg, ipc.NewStoreAPI(st), phr, st.VectorMemory(), st, eco)
+ voiceW, err = wireVoice(cfg, coreFor(), phr, st.VectorMemory(), st, eco)
if err != nil {
return fmt.Errorf("wire voice: %w", err)
}
@@ -526,22 +537,23 @@ func run(args []string) error {
tl = newTickLoop(st, gatherer, dispatcher, phr, rules, tickInterval, repeatInterval, autotuneInterval, cfg.Digest, routinesFromConfig(cfg.Routines), config.MorningRoutinesFromConfig(cfg.MorningRoutines), cfg.PatternProposals)
factWorker = newFactEnrichmentWorker(st, eco, time.Duration(cfg.FactEnrichmentInterval))
evalWorker = newMemoryEvalWorker(st, phr, cfg)
- feedWkr = newFeedWorker(ipc.NewStoreAPI(st), embedderOf(voiceW), cfg)
- crawlWkr = newCrawlWorker(newCrawler(cfg), ipc.NewStoreAPI(st), embedderOf(voiceW), cfg)
+ feedWkr = newFeedWorker(coreFor(), embedderOf(voiceW), cfg)
+ crawlWkr = newCrawlWorker(newCrawler(cfg), coreFor(), embedderOf(voiceW), cfg)
// Swap the CoreAPI from the locked placeholder to the real store adapter.
newAPI := &daemonAPI{
- CoreAPI: ipc.NewStoreAPI(st),
+ CoreAPI: coreFor(),
getTrace: tl.trace,
getMorningStatus: func(ctx context.Context) []ipc.MorningRoutineStatus { return tl.morningStatus(ctx, time.Now()) },
getDayPlan: func(ctx context.Context) ipc.DayPlan { return tl.dayPlan(ctx, time.Now()) },
+ getEvents: intakeEventsFn(evBus),
}
if voiceW != nil && voiceW.handler != nil {
newAPI.chatFn = voiceW.handler.handleText
}
srv.SetAPI(newAPI)
srv.Check = (&auth.Gate{Enrollment: auth.NewFloorEnrollment(), Session: passkeySess}).Check
- wireMailIntake(srv, st, phr, cfg)
+ wireMailIntake(srv, st, phr, cfg, evBus)
wireModelSwap(srv, phr, cfg)
keeper := wireVision(ctx, srv, st, embedderOf(voiceW), cfg)
wireCapture(srv, keeper, st, voiceW, phr, cfg)
diff --git a/cmd/mavend/tick.go b/cmd/mavend/tick.go
index 746cbe7..b3b7c7d 100644
--- a/cmd/mavend/tick.go
+++ b/cmd/mavend/tick.go
@@ -942,6 +942,18 @@ type daemonAPI struct {
getDayPlan func(ctx context.Context) ipc.DayPlan
chatFn func(ctx context.Context, text string) string
getMCPServers func() []ipc.MCPServerStatus
+ getEvents func(n int) []ipc.IntakeEvent
+}
+
+// RecentEvents — the unified intake journal (Vikunja #283). Empty, not an
+// error, when no bus was wired: "nothing has arrived" and "the journal is off"
+// look the same to a reader on purpose, because neither is a fault and the
+// page renders both as an empty table.
+func (d *daemonAPI) RecentEvents(ctx context.Context, n int) ([]ipc.IntakeEvent, error) {
+ if d.getEvents == nil {
+ return nil, nil
+ }
+ return d.getEvents(n), nil
}
func (d *daemonAPI) Chat(ctx context.Context, text string) (string, error) {
diff --git a/cmd/mavweb/events.html b/cmd/mavweb/events.html
new file mode 100644
index 0000000..df655e9
--- /dev/null
+++ b/cmd/mavweb/events.html
@@ -0,0 +1,24 @@
+{{template "shellTop" "events"}}
+Intake
+Everything that arrived, newest first — a relayed notification, a mail candidate, a feed
+item, a changed page, a spend, a presence probe. One envelope per write; the durable row is still the
+fact, note or task itself. Held in memory only, so a restart empties this.
+{{if .Err}}journal unavailable: {{.Err}}
{{end}}
+{{if and (not .Events) (not .Err)}}
+nothing has arrived yet
+{{end}}
+{{if .Events}}
+
+{{end}}
+{{template "shellBottom"}}
+