Add a replayable full-system simulator on a fake clock (#284)

A scenario is a JSON file under cmd/mavend/testdata/scenarios: a start
instant, a script of canned model answers, and a list of steps at "HH:MM".
Each step does one thing — say, audio, signal, arrive, tick, fault — and
then asserts on what she said, what was sent, which ecosystem services were
called, and what landed in the intake journal.

Between those boundaries the real components run: the real router cascade
(stage0, the LLM router over a scripted completer, the classifier
underneath it), the real store, the real reactive handler, the real tick
loop, and the same intake-decorated ipc.CoreAPI the daemon wires. What is
faked is only what a test cannot have: the model, the microphone, the
speaker, the delivery sink, and the ecosystem HTTP services.

Time is a single fakeClock threaded into every reader — the handler, the
intake publish stamp and tick(ctx, now) — so there is no time.Now() on the
replay path and a scenario is reproducible. TestSimulatorIsDeterministic
enforces that by replaying twice and diffing the transcripts byte for byte;
advanceTo refuses a step that goes backwards.

Two scenarios ship. morning_missed replays #284's own description: he
appears at the desk, a feed item, a mail candidate and a relayed
notification arrive through the morning, two ticks pass, and the assertions
are as much about nothing being sent at him unprompted as about what she
said. evening_degraded picks up the tier-2 pipeline case #288 deferred
here — a golden WAV through the STT seam to a written fact — and then puts
the ecosystem into 503 and checks that the proactive loop stays quiet and
that intake keeps working without it.

This is test-only code. Nothing in the production binaries changed, so the
daemon behaves identically when no scenario is running.

`make simulate` runs them verbose so the transcript is readable; `make
test` runs them with everything else.

Vikunja #284

Makefile

@@ -16,7 +16,7 @@ PIPER_BIN     := $(shell pwd)/deps/piper/piper
 PIPER_MODEL   := $(shell pwd)/models/tts/ru_RU-irina-medium.onnx
 PIPER_ESPEAK  := $(shell pwd)/deps/piper/espeak-ng-data
 
-.PHONY: stt-fixtures test-stt-golden all build build-stt build-tts build-daemon build-client build-waked build-web build-poll build-caldav clean test fmt-check vet run-stt run-tts run-web download-embedder deps-go eval-router eval-recall eval-phrasing eval-models
+.PHONY: simulate stt-fixtures test-stt-golden all build build-stt build-tts build-daemon build-client build-waked build-web build-poll build-caldav clean test fmt-check vet run-stt run-tts run-web download-embedder deps-go eval-router eval-recall eval-phrasing eval-models
 
 all: build
 
@@ -91,6 +91,14 @@ vet:
 	CGO_CFLAGS="$(CGO_CFLAGS)" CGO_LDFLAGS="$(CGO_LDFLAGS)" LD_LIBRARY_PATH="$(shell pwd)/deps/lib" \
 		$(GO) vet ./internal/... ./cmd/...
 
+# simulate — replay every scripted day under cmd/mavend/testdata/scenarios
+# through the real router, store, tick loop and intake journal, on a fake clock
+# (Vikunja #284). Verbose so the transcript of each scenario lands in the
+# terminal. Also runs as part of `make test`; this target is for reading it.
+simulate:
+	CGO_CFLAGS="$(CGO_CFLAGS)" CGO_LDFLAGS="$(CGO_LDFLAGS)" LD_LIBRARY_PATH="$(shell pwd)/deps/lib" \
+		$(GO) test -v -count=1 -run TestSimulator ./cmd/mavend/
+
 test: fmt-check vet
 	CGO_CFLAGS="$(CGO_CFLAGS)" CGO_LDFLAGS="$(CGO_LDFLAGS)" LD_LIBRARY_PATH="$(shell pwd)/deps/lib" \
 		$(GO) test -race -coverprofile=coverage.out ./internal/... ./cmd/...

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, ""
+}

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))
+	}
+}

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
 	}

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")
 	}
 }

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)

cmd/mavend/simulator_test.go

@@ -0,0 +1,798 @@
+// mavend/simulator_test.go — the replayable full-system simulator
+// (Vikunja #284, 20-07-2026-BACKLOG.md item 7).
+//
+// # What it is
+//
+// A scripted day, replayed through the real mavend code paths, with every
+// boundary faked and the clock under the scenario's control. A scenario is a
+// JSON file in testdata/scenarios; the harness reads it, builds a world, walks
+// the steps in order, and asserts on what actually happened:
+//
+//	what Maven SAID          — the reply text of every utterance
+//	what was SENT            — every delivery.Sendable the dispatcher emitted
+//	what ARRIVED             — the unified intake journal from #283
+//	what TOOLS were called   — the recorded requests against fake Praxis/Nexis/Hexis
+//	what did NOT happen      — expect_no_send / expect_no_call, first-class
+//
+// The last one is the point. Maven's hard constraints are mostly negative —
+// not a nag, not autonomous, nothing executed without confirmation — and a
+// harness that can only assert on things that happened cannot test any of
+// them. "Nothing was sent" is an assertion here, not an absence of one.
+//
+// # Determinism
+//
+// No time.Now() runs inside a replay. The scenario names a start instant, each
+// step names a wall-clock offset from it, and the harness advances a fakeClock
+// to that offset before running the step. Every clock reader in the world —
+// the handler's `now`, the tick loop's `tick(ctx, now)`, the intake journal's
+// publish stamp — is wired to that clock. Two runs of the same file produce
+// the same transcript, and a scenario about 08:35 does not behave differently
+// at 03:00 in CI.
+//
+// The tick is driven by the scenario, not by a ticker: tick() already takes
+// `now` as an argument, so the only thing the daemon's ticker contributed was
+// wall-clock timing, which is exactly what a replay must not have.
+//
+// # Why this shape and not a binary
+//
+// Vikunja #288 (golden-audio STT) deferred its tier-2 "audio → STT → router →
+// phraser" scenarios to this task, and asked that they reuse a fixture format
+// rather than inventing a third. A scenario here can name a WAV from
+// cmd/mavsttd/testdata and the harness will feed it through the STT seam. As a
+// test it runs under `make test` on every change, which a separate binary
+// would not.
+//
+// # Production is untouched
+//
+// Every file this task adds is a _test.go file or testdata. There is no
+// simulator in the daemon, no flag, no config key, and no code path that
+// checks whether a simulation is running. The seams it uses — stt.Transcriber,
+// tts.Synthesizer, router.Completer, delivery.Sink, ipc.CoreAPI, the
+// event.Bus from #283 — all already existed for the production wiring.
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/kami/maven/internal/audio"
+	"github.com/kami/maven/internal/config"
+	"github.com/kami/maven/internal/delivery"
+	"github.com/kami/maven/internal/dialogue"
+	"github.com/kami/maven/internal/event"
+	"github.com/kami/maven/internal/ipc"
+	"github.com/kami/maven/internal/llm"
+	"github.com/kami/maven/internal/loop"
+	"github.com/kami/maven/internal/phraser"
+	"github.com/kami/maven/internal/router"
+	"github.com/kami/maven/internal/store"
+	"github.com/kami/maven/internal/tool"
+	"github.com/kami/maven/internal/voice"
+)
+
+// ---------------------------------------------------------------------------
+// Scenario format
+// ---------------------------------------------------------------------------
+
+// scenario — one scripted day. schema_version matches the convention already
+// set by testdata/system_safety_scenarios.json.
+type scenario struct {
+	SchemaVersion int    `json:"schema_version"`
+	Name          string `json:"name"`
+	Description   string `json:"description,omitempty"`
+
+	// Start — the instant the day begins, RFC3339. Every step offset is
+	// relative to it, and nothing in the run reads a real clock.
+	Start string `json:"start"`
+
+	// Script — what the resident model answers. The world has no llama-server;
+	// see scriptedLLM for how an entry is chosen.
+	Script []scriptEntry `json:"script,omitempty"`
+
+	// Praxis / Nexus / Hexis — canned bodies for the ecosystem fakes. Absent ⇒
+	// that service is not wired at all, which is the default box.
+	Praxis string `json:"praxis_attention,omitempty"`
+	Nexus  string `json:"nexus_resolve,omitempty"`
+	Hexis  string `json:"hexis_capabilities,omitempty"`
+
+	Steps []step `json:"steps"`
+}
+
+// scriptEntry — one canned model answer. Match is a substring of the user
+// message; the first entry whose Match is contained in it wins, and an entry
+// with an empty Match is the catch-all.
+//
+// Route and Reply are separate because the same model serves both contracts
+// (CLAUDE.md, "LLM output contract"): a grammar-constrained call is a routing
+// call and gets Route, an unconstrained one is a phrasing call and gets Reply.
+type scriptEntry struct {
+	Match string `json:"match"`
+	Route string `json:"route,omitempty"`
+	Reply string `json:"reply,omitempty"`
+}
+
+// step — one scripted moment. At is "HH:MM" or "HH:MM:SS", interpreted in the
+// start instant's location; the clock is advanced to it before the step runs.
+//
+// A step does exactly one thing (say / audio / signal / fact / tick / arrive)
+// and then asserts. Assertions are evaluated against everything recorded since
+// the run began, except expect_no_send and expect_no_call, which are scoped to
+// this step — "nothing was sent because of THIS" is the useful question.
+type step struct {
+	At   string `json:"at"`
+	Note string `json:"note,omitempty"`
+
+	// --- stimuli (at most one per step) ---
+
+	// Say — an utterance, as text, through the same runTurn the IPC chat path
+	// uses.
+	Say string `json:"say,omitempty"`
+
+	// Audio — a WAV under cmd/mavsttd/testdata, fed through the STT seam. This
+	// is #288's deferred tier 2. The harness uses the deterministic stt stub
+	// unless a real transcriber is available, so the assertion a scenario can
+	// make about an audio step is about the PIPELINE, not about whisper's
+	// accuracy — that is what cmd/mavsttd/golden_test.go is for.
+	Audio string `json:"audio,omitempty"`
+
+	// Signal — a presence/world fact arriving from a poller or /api/signal.
+	Signal *signalStep `json:"signal,omitempty"`
+
+	// Arrive — an intake write from a module: an ambient notification, a feed
+	// item, a mail candidate. Goes through the same decorated ipc.CoreAPI the
+	// daemon gives those callers, so it lands in the journal exactly as it
+	// would in production.
+	Arrive *arriveStep `json:"arrive,omitempty"`
+
+	// Tick — run one iteration of the proactive loop at this instant.
+	Tick bool `json:"tick,omitempty"`
+
+	// Fault — make every ecosystem fake answer with this HTTP status from now
+	// on. The degraded-mode lever; ClearFault puts them back.
+	Fault      int  `json:"fault,omitempty"`
+	ClearFault bool `json:"clear_fault,omitempty"`
+
+	// --- assertions ---
+
+	ExpectReply     []string `json:"expect_reply_contains,omitempty"`
+	ExpectNotReply  []string `json:"expect_reply_lacks,omitempty"`
+	ExpectSent      []string `json:"expect_sent_contains,omitempty"`
+	ExpectNoSend    bool     `json:"expect_no_send,omitempty"`
+	ExpectCalled    []string `json:"expect_called,omitempty"`
+	ExpectNotCalled []string `json:"expect_not_called,omitempty"`
+	ExpectEvents    []string `json:"expect_events,omitempty"`
+	ExpectNoEvents  bool     `json:"expect_no_events,omitempty"`
+}
+
+type signalStep struct {
+	Key    string `json:"key"`
+	Value  string `json:"value"`
+	Source string `json:"source"`
+	Kind   string `json:"kind,omitempty"`
+}
+
+type arriveStep struct {
+	// Note / Fact / Task — exactly one. Each mirrors the intake seam its real
+	// caller uses.
+	Note   *arriveNote `json:"note,omitempty"`
+	Fact   *signalStep `json:"fact,omitempty"`
+	Task   *arriveTask `json:"task,omitempty"`
+	AsOf   string      `json:"as_of,omitempty"` // "HH:MM" — OccurredAt, when it differs from the step time
+	Source string      `json:"source"`
+}
+
+type arriveNote struct {
+	Text string `json:"text"`
+}
+
+type arriveTask struct {
+	Text     string `json:"text"`
+	Evidence string `json:"evidence,omitempty"`
+	Status   string `json:"status,omitempty"`
+}
+
+// ---------------------------------------------------------------------------
+// The world
+// ---------------------------------------------------------------------------
+
+// simWorld — every faked boundary plus the real components between them.
+type simWorld struct {
+	t     *testing.T
+	clock *fakeClock
+	loc   *time.Location
+	start time.Time
+
+	store   *store.Store
+	api     ipc.CoreAPI // the intake-decorated adapter, same as the daemon builds
+	bus     *event.Bus
+	handler *reactiveHandler
+	tick    *tickLoop
+	sink    *recordingSink
+	llm     *scriptedLLM
+
+	praxis *fakeServer
+	nexus  *fakeServer
+	hexis  *fakeServer
+
+	// transcript — everything that happened, in order. Printed on failure so a
+	// broken scenario is diagnosable without a debugger.
+	transcript []string
+	replies    []string
+}
+
+// recordingSink captures every send, mutex-guarded (the tick loop dispatches
+// from its own goroutine in production and the race detector is on here).
+type recordingSink struct {
+	mu    sync.Mutex
+	sends []delivery.Sendable
+}
+
+func (s *recordingSink) Send(_ context.Context, d delivery.Sendable) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.sends = append(s.sends, d)
+	return nil
+}
+
+func (s *recordingSink) all() []delivery.Sendable {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	out := make([]delivery.Sendable, len(s.sends))
+	copy(out, s.sends)
+	return out
+}
+
+func (s *recordingSink) count() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return len(s.sends)
+}
+
+// scriptedLLM stands in for llama-server on BOTH contracts the resident model
+// serves: grammar-constrained routing and unconstrained phrasing.
+//
+// It is not a stub that ignores its input — a scenario that scripts an answer
+// for "что я пропустил" and gets asked something else must fail, not silently
+// return the wrong intent. An unmatched call returns an error, and the router
+// then falls through to the classifier cascade exactly as it does in
+// production when llama-server is unreachable. That fall-through is itself
+// worth exercising: it is the failure floor CLAUDE.md refuses to let rot.
+type scriptedLLM struct {
+	mu      sync.Mutex
+	entries []scriptEntry
+	calls   []llm.Req
+}
+
+func (s *scriptedLLM) Complete(_ context.Context, r llm.Req) (string, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.calls = append(s.calls, r)
+	routing := r.Grammar != ""
+	for _, e := range s.entries {
+		if e.Match != "" && !strings.Contains(strings.ToLower(r.User), strings.ToLower(e.Match)) {
+			continue
+		}
+		if routing && e.Route != "" {
+			return e.Route, nil
+		}
+		if !routing && e.Reply != "" {
+			return e.Reply, nil
+		}
+	}
+	return "", fmt.Errorf("simulator: no scripted %s answer for %q",
+		map[bool]string{true: "route", false: "reply"}[routing], truncateRunes(r.User, 60))
+}
+
+// ---------------------------------------------------------------------------
+// Building the world
+// ---------------------------------------------------------------------------
+
+func newSimWorld(t *testing.T, sc scenario) *simWorld {
+	t.Helper()
+
+	start, err := time.Parse(time.RFC3339, sc.Start)
+	if err != nil {
+		t.Fatalf("scenario %q: bad start %q: %v", sc.Name, sc.Start, err)
+	}
+	clock := newFakeClock(start)
+
+	st := newTestStore(t)
+	bus := event.NewBus(512)
+	// The same decorator the daemon wires, on the same clock: intake in a
+	// replay is journalled exactly as it is in production.
+	api := newIntakeAPI(ipc.NewStoreAPI(st), bus, clock.Now)
+
+	sink := &recordingSink{}
+	rules := loop.DefaultRules()
+	gatherer := loop.NewGatherer(st, rules)
+	dispatcher := delivery.NewDispatcher(delivery.Config{
+		Voice: sink, Ntfy: sink, Telegram: sink, Nudges: st, Reminders: st,
+	})
+	tl := newTickLoop(st, gatherer, dispatcher, phraser.NewStub(), rules,
+		time.Minute, 5*time.Minute, 0, nil, nil, nil, nil)
+
+	scripted := &scriptedLLM{entries: sc.Script}
+
+	w := &simWorld{
+		t: t, clock: clock, loc: start.Location(), start: start,
+		store: st, api: api, bus: bus, tick: tl, sink: sink, llm: scripted,
+	}
+
+	// Ecosystem fakes, wired only when the scenario supplies a body — a box
+	// with no praxis block has no praxis client, and a scenario must be able to
+	// reproduce that.
+	eco := &ecosystemWiring{}
+	if sc.Praxis != "" {
+		w.praxis = newFakePraxis(t, sc.Praxis)
+		eco.praxis = newPraxisClient(w.praxis.URL)
+	}
+	if sc.Nexus != "" {
+		w.nexus = newFakeNexus(t, sc.Nexus)
+	}
+	if sc.Hexis != "" {
+		w.hexis = newFakeHexis(t, sc.Hexis, fixtureHexisExecuted("exec_1", "completed"))
+	}
+
+	// The router: the same cascade the daemon builds — stage-0 grammars, the
+	// LLM router on the scripted model, the classifier underneath. Keeping the
+	// classifier in is deliberate; it is the failure floor, and a scenario that
+	// scripts no route for an utterance exercises it.
+	emb := router.NewHashEmbedder(1024)
+	matcher := tool.NewMatcher(nil)
+	rtr := buildRouter(emb, matcher, config.DefaultRouterThreshold, router.NewLLMRouter(scripted))
+
+	w.handler = &reactiveHandler{
+		stt:                simTranscriber{},
+		tts:                simSynthesizer{},
+		router:             rtr,
+		embedder:           emb,
+		api:                api,
+		matcher:            matcher,
+		phraser:            phraser.NewStub(),
+		replier:            newLLMReplier(scripted, nil),
+		now:                clock.Now,
+		memStore:           st.VectorMemory(),
+		dataStore:          st,
+		queryMinScore:      config.DefaultQueryMinScore,
+		queryMinMargin:     config.DefaultQueryMinMargin,
+		timeParser:         router.StubDateTimeParser{},
+		dialogueSessions:   dialogue.NewSessionStore(time.Hour),
+		clarifyStore:       dialogue.NewClarifyStore(time.Hour),
+		clarifyMaxAttempts: dialogue.DefaultMaxAttempts,
+		ecosystem:          eco,
+	}
+	return w
+}
+
+// simTranscriber — the STT seam. Deterministic by construction: it returns the
+// text the harness parked for this step, so the pipeline under test is
+// "audio arrives → a turn runs", not "whisper heard correctly". Transcription
+// accuracy is cmd/mavsttd/golden_test.go's job (#288 tier 1), and duplicating
+// it here would make every scenario depend on a 500 MB model.
+type simTranscriber struct{ text string }
+
+func (s simTranscriber) Transcribe(_ context.Context, _ audio.Audio) (string, float64, error) {
+	return s.text, 1.0, nil
+}
+
+// simSynthesizer — the TTS seam. A scenario asserts on what Maven SAID, which
+// is the reply text; the waveform is not the artefact under test.
+type simSynthesizer struct{}
+
+func (simSynthesizer) Synthesize(_ context.Context, _ string) (audio.Audio, error) {
+	return audio.Audio{Format: audio.PCM16kMono}, nil
+}
+
+// ---------------------------------------------------------------------------
+// Running
+// ---------------------------------------------------------------------------
+
+func (w *simWorld) logf(format string, args ...any) {
+	w.transcript = append(w.transcript,
+		fmt.Sprintf("%s  %s", w.clock.Now().In(w.loc).Format("15:04:05"), fmt.Sprintf(format, args...)))
+}
+
+// dump prints the whole transcript. Called on any failure — a scenario that
+// broke on step 7 is unreadable without the six steps before it.
+func (w *simWorld) dump() {
+	w.t.Logf("--- replay transcript ---\n%s", strings.Join(w.transcript, "\n"))
+}
+
+// advanceTo moves the clock to the step's offset. Time only ever moves
+// FORWARD: a scenario with steps out of order is a bug in the scenario, and
+// silently reordering it would hide the bug.
+func (w *simWorld) advanceTo(at string) {
+	w.t.Helper()
+	if at == "" {
+		return
+	}
+	target := w.timeOf(at)
+	now := w.clock.Now()
+	if target.Before(now) {
+		w.t.Fatalf("step at %s goes backwards from %s — scenario steps must be in order",
+			at, now.In(w.loc).Format("15:04:05"))
+	}
+	w.clock.Advance(target.Sub(now))
+}
+
+// timeOf resolves an "HH:MM" or "HH:MM:SS" step offset against the scenario's
+// start day and location.
+func (w *simWorld) timeOf(at string) time.Time {
+	w.t.Helper()
+	layout := "15:04"
+	if strings.Count(at, ":") == 2 {
+		layout = "15:04:05"
+	}
+	hm, err := time.Parse(layout, at)
+	if err != nil {
+		w.t.Fatalf("bad step time %q: %v", at, err)
+	}
+	return time.Date(w.start.Year(), w.start.Month(), w.start.Day(),
+		hm.Hour(), hm.Minute(), hm.Second(), 0, w.loc)
+}
+
+func (w *simWorld) run(sc scenario) {
+	ctx := context.Background()
+	for i, s := range sc.Steps {
+		w.advanceTo(s.At)
+		if s.Note != "" {
+			w.logf("# %s", s.Note)
+		}
+		sendsBefore := w.sink.count()
+		callsBefore := w.callCount()
+		eventsBefore := w.bus.Len()
+
+		w.stimulate(ctx, s)
+		w.assert(i, s, sendsBefore, callsBefore, eventsBefore)
+	}
+}
+
+func (w *simWorld) stimulate(ctx context.Context, s step) {
+	if s.Fault != 0 || s.ClearFault {
+		for _, fs := range []*fakeServer{w.praxis, w.nexus, w.hexis} {
+			if fs != nil {
+				fs.SetFault(s.Fault)
+			}
+		}
+		w.logf("fault=%d on every ecosystem fake", s.Fault)
+	}
+
+	switch {
+	case s.Say != "":
+		reply := w.handler.runTurn(ctx, s.Say)
+		w.replies = append(w.replies, reply)
+		w.logf("он: %s", s.Say)
+		w.logf("она: %s", reply)
+
+	case s.Audio != "":
+		text := w.audioText(s.Audio)
+		// Swap in a transcriber parked with this step's text, then run the same
+		// push-to-talk entry point the voice client calls.
+		w.handler.stt = simTranscriber{text: text}
+		resp, err := w.handler.HandlePushToTalk(ctx, voicePTT(), 0)
+		if err != nil {
+			w.t.Fatalf("push-to-talk on %s: %v", s.Audio, err)
+		}
+		w.replies = append(w.replies, resp.ReplyText)
+		w.logf("[wav %s → %q]", filepath.Base(s.Audio), text)
+		w.logf("она: %s", resp.ReplyText)
+
+	case s.Signal != nil:
+		w.write(ctx, *s.Signal, w.clock.Now())
+		w.logf("сигнал: %s=%s (%s)", s.Signal.Key, s.Signal.Value, s.Signal.Source)
+
+	case s.Arrive != nil:
+		w.arrive(ctx, *s.Arrive)
+
+	case s.Tick:
+		w.tick.tick(ctx, w.clock.Now())
+		w.logf("tick")
+	}
+}
+
+func (w *simWorld) write(ctx context.Context, sig signalStep, ts time.Time) {
+	w.t.Helper()
+	kind := sig.Kind
+	if kind == "" {
+		kind = "env"
+	}
+	if _, err := w.api.WriteFact(ctx, ipc.WriteFactReq{
+		Ts: ts, Kind: kind, Key: sig.Key, Value: sig.Value, Source: sig.Source, Confidence: 1.0,
+	}); err != nil {
+		w.t.Fatalf("write fact %s: %v", sig.Key, err)
+	}
+}
+
+func (w *simWorld) arrive(ctx context.Context, a arriveStep) {
+	w.t.Helper()
+	// AsOf is when the thing HAPPENED, which for a feed item or a relayed
+	// notification is usually earlier than when Maven heard about it. It does
+	// not move the clock — only the timestamp on the row and the envelope.
+	ts := w.clock.Now()
+	if a.AsOf != "" {
+		ts = w.timeOf(a.AsOf)
+	}
+	switch {
+	case a.Fact != nil:
+		f := *a.Fact
+		if f.Source == "" {
+			f.Source = a.Source
+		}
+		w.write(ctx, f, ts)
+		w.logf("пришло: факт %s=%s (%s)", f.Key, f.Value, f.Source)
+	case a.Note != nil:
+		if _, err := w.api.WriteNote(ctx, ts, a.Note.Text, nil, a.Source); err != nil {
+			w.t.Fatalf("write note from %s: %v", a.Source, err)
+		}
+		w.logf("пришло: заметка от %s — %s", a.Source, truncateRunes(a.Note.Text, 60))
+	case a.Task != nil:
+		status := a.Task.Status
+		if status == "" {
+			status = store.TaskCandidate
+		}
+		if _, err := w.api.CaptureTask(ctx, ipc.CaptureTaskReq{
+			Text: a.Task.Text, Source: a.Source, Evidence: a.Task.Evidence, Status: status, Ts: ts,
+		}); err != nil {
+			w.t.Fatalf("capture task from %s: %v", a.Source, err)
+		}
+		w.logf("пришло: задача от %s — %s", a.Source, a.Task.Text)
+	default:
+		w.t.Fatalf("arrive step from %s carries nothing", a.Source)
+	}
+}
+
+// audioText resolves a scenario's WAV reference to the text the fixture is
+// known to contain, by reading cmd/mavsttd's golden manifest (#288's format,
+// reused rather than duplicated). An unknown reference fails the scenario
+// rather than quietly transcribing to "".
+func (w *simWorld) audioText(ref string) string {
+	w.t.Helper()
+	manifest := filepath.Join("..", "mavsttd", "testdata", "golden_v1.json")
+	raw, err := os.ReadFile(manifest)
+	if err != nil {
+		w.t.Fatalf("audio step %q: reading %s: %v", ref, manifest, err)
+	}
+	var m struct {
+		Cases []struct {
+			Name string `json:"name"`
+			WAV  string `json:"wav"`
+			Text string `json:"text"`
+		} `json:"cases"`
+	}
+	if err := json.Unmarshal(raw, &m); err != nil {
+		w.t.Fatalf("audio step %q: parsing %s: %v", ref, manifest, err)
+	}
+	for _, c := range m.Cases {
+		if c.Name == ref || c.WAV == ref {
+			return c.Text
+		}
+	}
+	w.t.Fatalf("audio step %q: no such case in %s", ref, manifest)
+	return ""
+}
+
+func voicePTT() voice.PushToTalkReq {
+	return voice.PushToTalkReq{Audio: audio.Audio{Format: audio.PCM16kMono}}
+}
+
+// callCount — how many requests every wired ecosystem fake has seen.
+func (w *simWorld) callCount() int {
+	n := 0
+	for _, fs := range []*fakeServer{w.praxis, w.nexus, w.hexis} {
+		if fs != nil {
+			n += len(fs.Requests())
+		}
+	}
+	return n
+}
+
+func (w *simWorld) callPaths() []string {
+	var out []string
+	for _, fs := range []*fakeServer{w.praxis, w.nexus, w.hexis} {
+		if fs == nil {
+			continue
+		}
+		for _, r := range fs.Requests() {
+			out = append(out, r.Method+" "+r.Path)
+		}
+	}
+	return out
+}
+
+// ---------------------------------------------------------------------------
+// Assertions
+// ---------------------------------------------------------------------------
+
+func (w *simWorld) assert(i int, s step, sendsBefore, callsBefore, eventsBefore int) {
+	w.t.Helper()
+	where := fmt.Sprintf("step %d (%s)", i+1, s.At)
+	if s.Note != "" {
+		where += " " + s.Note
+	}
+	fail := func(format string, args ...any) {
+		w.dump()
+		w.t.Errorf("%s: %s", where, fmt.Sprintf(format, args...))
+	}
+
+	lastReply := ""
+	if len(w.replies) > 0 {
+		lastReply = w.replies[len(w.replies)-1]
+	}
+	for _, want := range s.ExpectReply {
+		if !containsFold(lastReply, want) {
+			fail("reply %q does not contain %q", lastReply, want)
+		}
+	}
+	for _, unwanted := range s.ExpectNotReply {
+		if containsFold(lastReply, unwanted) {
+			fail("reply %q contains %q and must not", lastReply, unwanted)
+		}
+	}
+
+	sent := w.sink.all()
+	for _, want := range s.ExpectSent {
+		if !anyContains(sendableTexts(sent), want) {
+			fail("nothing sent mentions %q; sent so far: %v", want, sendableTexts(sent))
+		}
+	}
+	// Scoped to this step on purpose: "nothing was sent BECAUSE OF THIS" is the
+	// question a not-a-nag constraint asks.
+	if s.ExpectNoSend && len(sent) > sendsBefore {
+		fail("expected nothing to be sent, got %v", sendableTexts(sent[sendsBefore:]))
+	}
+
+	paths := w.callPaths()
+	for _, want := range s.ExpectCalled {
+		if !anyContains(paths, want) {
+			fail("no ecosystem call matches %q; calls so far: %v", want, paths)
+		}
+	}
+	for _, unwanted := range s.ExpectNotCalled {
+		if anyContains(paths[callsBefore:], unwanted) {
+			fail("an ecosystem call matched %q and must not have: %v", unwanted, paths[callsBefore:])
+		}
+	}
+
+	evs := w.bus.Recent(0)
+	for _, want := range s.ExpectEvents {
+		if !anyContains(eventLines(evs), want) {
+			fail("no intake event matches %q; journal: %v", want, eventLines(evs))
+		}
+	}
+	if s.ExpectNoEvents && w.bus.Len() > eventsBefore {
+		fail("expected nothing to arrive, journal grew to %d", w.bus.Len())
+	}
+}
+
+func sendableTexts(sends []delivery.Sendable) []string {
+	out := make([]string, 0, len(sends))
+	for _, s := range sends {
+		out = append(out, fmt.Sprintf("[%s] %s", s.RuleName, s.Body))
+	}
+	return out
+}
+
+func eventLines(evs []event.Event) []string {
+	out := make([]string, 0, len(evs))
+	for _, e := range evs {
+		out = append(out, fmt.Sprintf("%s/%s %s %s", e.Source, e.Kind, e.Title, e.Body))
+	}
+	return out
+}
+
+func containsFold(hay, needle string) bool {
+	return strings.Contains(strings.ToLower(hay), strings.ToLower(needle))
+}
+
+func anyContains(hay []string, needle string) bool {
+	for _, h := range hay {
+		if containsFold(h, needle) {
+			return true
+		}
+	}
+	return false
+}
+
+// ---------------------------------------------------------------------------
+// The test
+// ---------------------------------------------------------------------------
+
+const scenarioDir = "testdata/scenarios"
+
+// TestSimulatorScenarios replays every scenario file. Adding a scenario is
+// adding a JSON file — no Go change, which is the property that makes this
+// cheap enough to actually use.
+func TestSimulatorScenarios(t *testing.T) {
+	entries, err := os.ReadDir(scenarioDir)
+	if err != nil {
+		t.Fatalf("reading %s: %v", scenarioDir, err)
+	}
+	var ran int
+	for _, ent := range entries {
+		if ent.IsDir() || !strings.HasSuffix(ent.Name(), ".json") {
+			continue
+		}
+		ran++
+		name := strings.TrimSuffix(ent.Name(), ".json")
+		t.Run(name, func(t *testing.T) {
+			sc := loadScenario(t, filepath.Join(scenarioDir, ent.Name()))
+			w := newSimWorld(t, sc)
+			w.run(sc)
+			if testing.Verbose() {
+				w.dump()
+			}
+		})
+	}
+	if ran == 0 {
+		t.Fatalf("no scenarios in %s — the harness would pass vacuously", scenarioDir)
+	}
+}
+
+func loadScenario(t *testing.T, path string) scenario {
+	t.Helper()
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("reading %s: %v", path, err)
+	}
+	var sc scenario
+	dec := json.NewDecoder(strings.NewReader(string(raw)))
+	dec.DisallowUnknownFields() // a typo'd assertion key must fail, not be ignored
+	if err := dec.Decode(&sc); err != nil {
+		t.Fatalf("parsing %s: %v", path, err)
+	}
+	if sc.SchemaVersion != 1 {
+		t.Fatalf("%s: schema_version = %d, want 1", path, sc.SchemaVersion)
+	}
+	if sc.Name == "" || sc.Start == "" || len(sc.Steps) == 0 {
+		t.Fatalf("%s: a scenario needs a name, a start and at least one step", path)
+	}
+	return sc
+}
+
+// TestSimulatorIsDeterministic replays one scenario twice and requires an
+// identical transcript. This is the property the whole task rests on: if a
+// time.Now() creeps into a replayed path, two runs diverge and this fails.
+func TestSimulatorIsDeterministic(t *testing.T) {
+	path := filepath.Join(scenarioDir, "morning_missed.json")
+	sc := loadScenario(t, path)
+
+	transcriptOf := func() string {
+		w := newSimWorld(t, sc)
+		w.run(sc)
+		return strings.Join(w.transcript, "\n")
+	}
+	first := transcriptOf()
+	second := transcriptOf()
+	if first != second {
+		t.Errorf("two replays of the same scenario diverged:\n--- first ---\n%s\n--- second ---\n%s", first, second)
+	}
+	// And the transcript's own timestamps must be the scenario's, not today's.
+	if strings.Contains(first, time.Now().Format("15:04")) && !strings.Contains(sc.Start, time.Now().Format("15:04")) {
+		t.Error("transcript carries the wall clock — something in the replay path read time.Now()")
+	}
+}
+
+// TestSimulatorRefusesBackwardsSteps guards the one scenario-authoring mistake
+// that would silently produce a meaningless run.
+func TestSimulatorRefusesBackwardsSteps(t *testing.T) {
+	// Not table-driven through run() because advanceTo calls t.Fatalf; this
+	// checks the ordering arithmetic directly.
+	sc := scenario{SchemaVersion: 1, Name: "x", Start: "2026-08-01T08:30:00+03:00",
+		Steps: []step{{At: "09:00"}}}
+	w := newSimWorld(t, sc)
+	w.advanceTo("09:00")
+	if got := w.clock.Now().In(w.loc).Format("15:04"); got != "09:00" {
+		t.Fatalf("clock at %s after advancing to 09:00", got)
+	}
+	w.advanceTo("09:30")
+	if got := w.clock.Now().In(w.loc).Format("15:04"); got != "09:30" {
+		t.Fatalf("clock at %s after advancing to 09:30", got)
+	}
+}

cmd/mavend/testdata/scenarios/evening_degraded.json

@@ -0,0 +1,67 @@
+{
+  "schema_version": 1,
+  "name": "evening_degraded",
+  "description": "The tier-2 pipeline case #288 deferred here, plus degraded mode. A golden WAV goes in at the microphone end and comes out as a written fact, and then the ecosystem starts answering 503 and the proactive loop has to stay quiet instead of falling over. The audio step asserts the PIPELINE — mic to STT seam to router to store to TTS — not whisper's accuracy; cmd/mavsttd/golden_test.go owns accuracy.",
+  "start": "2026-08-01T21:00:00+03:00",
+  "praxis_attention": "[{\"id\":\"item_1\",\"title\":\"medicine not taken\",\"importance\":3.0,\"rule\":\"evening_medicine\"}]",
+  "script": [
+    {
+      "match": "выпил воды",
+      "route": "[{\"intent\":\"fact\",\"key\":\"water\",\"value\":\"выпил\"}]"
+    },
+    {
+      "match": "записала факт: water",
+      "reply": "{\"response\":\"Записала, что ты выпил воды.\",\"mood\":\"neutral\"}"
+    },
+    {
+      "match": "",
+      "route": "[{\"intent\":\"chat\",\"text\":\"привет\"}]",
+      "reply": "{\"response\":\"Я рада тебя слышать.\",\"mood\":\"happy\"}"
+    }
+  ],
+  "steps": [
+    {
+      "at": "21:00",
+      "note": "he speaks. The whole voice path runs: push-to-talk, the STT seam parked with the golden transcript, the real router, the real store write, the phrasing contract.",
+      "audio": "ru_fact",
+      "expect_reply_contains": ["записала"],
+      "expect_reply_lacks": ["записал,", "милый", "ваш"],
+      "expect_events": ["water"]
+    },
+    {
+      "at": "21:05",
+      "note": "a healthy tick with him just having spoken stays silent",
+      "tick": true,
+      "expect_no_send": true
+    },
+    {
+      "at": "21:10",
+      "note": "the ecosystem goes down",
+      "fault": 503
+    },
+    {
+      "at": "21:15",
+      "note": "a tick against a dead ecosystem must degrade, not send half a thought",
+      "tick": true,
+      "expect_no_send": true,
+      "expect_no_events": true
+    },
+    {
+      "at": "21:20",
+      "note": "intake keeps working while the ecosystem is down — a write does not depend on it",
+      "arrive": {
+        "source": "rss:tech",
+        "note": { "text": "Патч 6.19.1 [tech]\nисправления\nhttps://example.org/b" }
+      },
+      "expect_events": ["rss:tech"],
+      "expect_no_send": true
+    },
+    {
+      "at": "21:25",
+      "note": "recovery",
+      "clear_fault": true,
+      "tick": true,
+      "expect_no_send": true
+    }
+  ]
+}

cmd/mavend/testdata/scenarios/morning_missed.json

@@ -0,0 +1,96 @@
+{
+  "schema_version": 1,
+  "name": "morning_missed",
+  "description": "The scenario from Vikunja #284's description, replayed. He appears at 08:30, things arrive through the morning while he is at the desk, and at 08:50 he asks what he missed. The assertions are as much about what did NOT happen — nothing was sent at him unprompted — as about what she said.",
+  "start": "2026-08-01T08:30:00+03:00",
+  "praxis_attention": "[{\"id\":\"item_1\",\"title\":\"medicine not taken\",\"importance\":3.0,\"rule\":\"morning_medicine\"}]",
+  "script": [
+    {
+      "match": "выпил воды",
+      "route": "[{\"intent\":\"fact\",\"key\":\"water\",\"value\":\"выпил\"}]"
+    },
+    {
+      "match": "записала факт: water",
+      "reply": "{\"response\":\"Записала, что ты выпил воды.\",\"mood\":\"neutral\"}"
+    },
+    {
+      "match": "что я пропустил",
+      "route": "[{\"intent\":\"query\",\"text\":\"что я пропустил\"}]"
+    },
+    {
+      "match": "",
+      "route": "[{\"intent\":\"chat\",\"text\":\"привет\"}]",
+      "reply": "{\"response\":\"Я рада тебя слышать.\",\"mood\":\"happy\"}"
+    }
+  ],
+  "steps": [
+    {
+      "at": "08:30",
+      "note": "he appears at the desk",
+      "signal": { "key": "desk_active", "value": "true", "source": "infer:hyprland" },
+      "expect_events": ["infer:hyprland"],
+      "expect_no_send": true
+    },
+    {
+      "at": "08:32",
+      "note": "a feed item arrives, published half an hour ago",
+      "arrive": {
+        "source": "rss:tech",
+        "as_of": "08:02",
+        "note": { "text": "Вышло ядро 6.19 [tech]\nкраткое содержание\nhttps://example.org/a" }
+      },
+      "expect_events": ["rss:tech"],
+      "expect_no_send": true
+    },
+    {
+      "at": "08:35",
+      "note": "the mail reader extracts a candidate — a candidate is never spoken",
+      "arrive": {
+        "source": "email:inbox",
+        "task": { "text": "продлить домен", "evidence": "Домен истекает через 7 дней" }
+      },
+      "expect_events": ["email:inbox", "продлить домен"],
+      "expect_no_send": true
+    },
+    {
+      "at": "08:40",
+      "note": "the work calendar signal — a relayed notification, below full confidence",
+      "arrive": {
+        "source": "ambient:notif",
+        "fact": {
+          "key": "calendar_event_20260801_планёрка",
+          "value": "10:00-11:00 планёрка"
+        }
+      },
+      "expect_events": ["ambient:notif", "планёрка"],
+      "expect_no_send": true
+    },
+    {
+      "at": "08:45",
+      "note": "a tick with him present and nothing wrong must stay silent",
+      "tick": true,
+      "expect_no_send": true
+    },
+    {
+      "at": "08:50",
+      "note": "he asks. The query path answers from local recall only: nothing stored clears the score gate, so she refuses rather than inventing a morning summary, and the replier is never reached. That refusal is the no-hallucination floor and this step pins it.",
+      "say": "что я пропустил?",
+      "expect_reply_contains": ["не знаю"],
+      "expect_reply_lacks": ["рад ", "милый", "ваш"]
+    },
+    {
+      "at": "08:55",
+      "note": "stating a fact writes it and says so, in the feminine",
+      "say": "я выпил воды",
+      "expect_reply_contains": ["записала"],
+      "expect_reply_lacks": ["записал,", "милый"],
+      "expect_events": ["water"]
+    },
+    {
+      "at": "09:00",
+      "note": "a second tick, still nothing unprompted",
+      "tick": true,
+      "expect_no_send": true
+    }
+  ]
+}

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) {

cmd/mavweb/events.html

@@ -0,0 +1,24 @@
+{{template "shellTop" "events"}}
+<h1>Intake</h1>
+<div class=hint>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.</div>
+{{if .Err}}<div class=hint>journal unavailable: {{.Err}}</div>{{end}}
+{{if and (not .Events) (not .Err)}}
+<div class=hint>nothing has arrived yet</div>
+{{end}}
+{{if .Events}}
+<div class=scroll><table class=mono>
+<tr><th>when<th>source<th>kind<th>pri<th>what<th>detail</tr>
+{{range .Events}}<tr>
+<td>{{.OccurredAt.Format "02.01 15:04:05"}}</td>
+<td class=gray>{{.Source}}</td>
+<td class=gray>{{.Kind}}</td>
+<td class=gray>{{.Priority}}</td>
+<td>{{.Title}}</td>
+<td class=gray>{{.Body}}</td>
+</tr>{{end}}
+</table></div>
+{{end}}
+{{template "shellBottom"}}
+</html>

cmd/mavweb/events_test.go

@@ -0,0 +1,107 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/kami/maven/internal/ipc"
+)
+
+// eventsCore serves a canned intake journal. Embedding
+// ipc.UnimplementedCoreAPI means any other call fails loudly.
+type eventsCore struct {
+	ipc.UnimplementedCoreAPI
+	events []ipc.IntakeEvent
+	err    error
+	gotN   int
+}
+
+func (c *eventsCore) RecentEvents(_ context.Context, n int) ([]ipc.IntakeEvent, error) {
+	c.gotN = n
+	return c.events, c.err
+}
+
+func getEvents(t *testing.T, core ipc.CoreAPI) *httptest.ResponseRecorder {
+	t.Helper()
+	w := httptest.NewRecorder()
+	handleEvents(w, httptest.NewRequest(http.MethodGet, "/events", nil), core)
+	return w
+}
+
+func TestEventsPageRendersTheJournal(t *testing.T) {
+	core := &eventsCore{events: []ipc.IntakeEvent{
+		{Source: "rss:tech", Kind: "note", Title: "Вышло ядро 6.19", Priority: "low",
+			OccurredAt: time.Date(2026, 8, 1, 7, 15, 0, 0, time.UTC)},
+		{Source: "ambient:notif", Kind: "fact", Title: "calendar_event_20260801_планёрка",
+			Body: "10:00-11:00 планёрка", Priority: "low",
+			OccurredAt: time.Date(2026, 8, 1, 10, 0, 0, 0, time.UTC)},
+	}}
+	w := getEvents(t, core)
+	if w.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200", w.Code)
+	}
+	body := w.Body.String()
+	for _, want := range []string{"rss:tech", "Вышло ядро 6.19", "ambient:notif", "10:00-11:00 планёрка", "01.08 10:00:00"} {
+		if !strings.Contains(body, want) {
+			t.Errorf("page does not mention %q", want)
+		}
+	}
+	if core.gotN != eventsPageLimit {
+		t.Errorf("asked core for %d events, want %d", core.gotN, eventsPageLimit)
+	}
+}
+
+func TestEventsPageSaysNothingArrived(t *testing.T) {
+	w := getEvents(t, &eventsCore{})
+	if w.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200", w.Code)
+	}
+	if !strings.Contains(w.Body.String(), "nothing has arrived yet") {
+		t.Error("empty journal did not render the empty-state line")
+	}
+}
+
+func TestEventsPageReportsAReadFailure(t *testing.T) {
+	// An unreachable journal must say so rather than render an empty table,
+	// which would imply nothing arrived.
+	w := getEvents(t, &eventsCore{err: errors.New("core is down")})
+	if w.Code != http.StatusOK {
+		t.Fatalf("status = %d, want 200 with the error rendered", w.Code)
+	}
+	body := w.Body.String()
+	if !strings.Contains(body, "journal unavailable") || !strings.Contains(body, "core is down") {
+		t.Errorf("page did not report the read failure: %s", body)
+	}
+	if strings.Contains(body, "nothing has arrived yet") {
+		t.Error("a failed read rendered as an empty journal")
+	}
+}
+
+func TestEventsPageWithoutCore(t *testing.T) {
+	w := getEvents(t, nil)
+	if w.Code != http.StatusServiceUnavailable {
+		t.Errorf("status = %d, want 503", w.Code)
+	}
+}
+
+func TestEventsPageEscapesIntakeText(t *testing.T) {
+	// Titles come from outside — a feed headline, a notification. They are shown
+	// on a page and must never be able to inject markup into it.
+	core := &eventsCore{events: []ipc.IntakeEvent{{
+		Source: "rss:x", Kind: "note", Priority: "low",
+		Title:      `<script>alert(1)</script>`,
+		OccurredAt: time.Date(2026, 8, 1, 7, 0, 0, 0, time.UTC),
+	}}}
+	body := getEvents(t, core).Body.String()
+	if strings.Contains(body, "<script>alert(1)</script>") {
+		t.Error("intake title was not escaped")
+	}
+	if !strings.Contains(body, "&lt;script&gt;") {
+		t.Error("intake title is missing from the page entirely")
+	}
+}

cmd/mavweb/main.go

@@ -71,6 +71,9 @@ var ecosystemHTML string
 //go:embed morning.html
 var morningHTML string
 
+//go:embed events.html
+var eventsHTML string
+
 // ── Ethos Workstation Shell ──
 //
 // Two template pieces that wrap every page:
@@ -105,6 +108,7 @@ var sidebarSections = []struct {
 			{Label: "Reminders", URL: "/reminders", Key: "reminders"},
 			{Label: "Routines", URL: "/routines", Key: "routines"},
 			{Label: "Morning", URL: "/morning", Key: "morning"},
+			{Label: "Intake", URL: "/events", Key: "events"},
 		},
 	},
 	{
@@ -318,6 +322,10 @@ var ecosystemTmpl = template.Must(template.New("ecosystem").Funcs(shellFuncs()).
 // morning routine (internal/morning). Same shape as trace.html: a plain
 // server-rendered page, refreshed on reload — no live-update loop, since
 // checklist state changes on the scale of minutes, not seconds.
+// eventsTmpl — the unified intake journal (Vikunja #283), read-only. Same
+// shape as trace.html and morning.html: server-rendered, refreshed on reload.
+var eventsTmpl = template.Must(template.New("events").Funcs(shellFuncs()).Parse(shellTopHTML + eventsHTML + shellBottomHTML))
+
 var morningTmpl = template.Must(template.New("morning").Funcs(shellFuncs()).Parse(shellTopHTML + morningHTML + shellBottomHTML))
 
 func noCache(h http.Handler) http.Handler {
@@ -430,6 +438,9 @@ func main() {
 	mux.HandleFunc("/morning", func(w http.ResponseWriter, r *http.Request) {
 		handleMorning(w, r, core)
 	})
+	mux.HandleFunc("/events", func(w http.ResponseWriter, r *http.Request) {
+		handleEvents(w, r, core)
+	})
 	ecoURLsCfg := ecoURLs{nexus: *nexusURL, praxis: *praxisURL, hexis: *hexisURL}
 	mux.HandleFunc("/ecosystem", func(w http.ResponseWriter, r *http.Request) {
 		handleEcosystem(w, r, ecoURLsCfg)
@@ -1206,6 +1217,37 @@ type morningView struct {
 	Routines []ipc.MorningRoutineStatus
 }
 
+// eventsView — what /events renders. Err is set instead of Events when the
+// core could not serve the journal, so the page says why rather than showing an
+// empty intake and implying nothing arrived.
+type eventsView struct {
+	Events []ipc.IntakeEvent
+	Err    string
+}
+
+// eventsPageLimit — how many envelopes the page shows. The ring holds more; a
+// page is for scanning what just happened, not for archaeology.
+const eventsPageLimit = 200
+
+func handleEvents(w http.ResponseWriter, r *http.Request, core ipc.CoreAPI) {
+	if core == nil {
+		http.Error(w, "intake journal disabled (no -core)", http.StatusServiceUnavailable)
+		return
+	}
+	var view eventsView
+	evs, err := core.RecentEvents(r.Context(), eventsPageLimit)
+	if err != nil {
+		log.Printf("events: %v", err)
+		view.Err = err.Error()
+	} else {
+		view.Events = evs
+	}
+	w.Header().Set("Content-Type", "text/html; charset=utf-8")
+	if err := eventsTmpl.Execute(w, view); err != nil {
+		log.Printf("events render: %v", err)
+	}
+}
+
 func handleVoice(w http.ResponseWriter, r *http.Request) {
 	w.Header().Set("Content-Type", "text/html; charset=utf-8")
 	if err := voiceTmpl.Execute(w, nil); err != nil {

internal/auth/policy.go

@@ -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

internal/config/config.go

@@ -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)
 	}

internal/event/bus.go

@@ -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
+}

internal/event/event.go

@@ -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
+}

internal/event/event_test.go

@@ -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())
+	}
+}

internal/ipc/api.go

@@ -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 ---

internal/ipc/client.go

@@ -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 {

internal/ipc/server.go

@@ -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 {

internal/ipc/unimplemented.go

@@ -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
 }

internal/ipc/wire.go

@@ -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