Maven/cmd/mavend/intake_test.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"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))
	}
}

// Maven's own bookkeeping is not intake. The feed watermark, the crawl hash,
// the praxis trace of an act she performed and the quiet-hours toggle he
// pressed all landed on a page headed "everything that arrived", and on a cold
// start a handful of feeds could evict real intake behind their marks.
func TestIntakeSkipsHerOwnBookkeeping(t *testing.T) {
	api, bus := newIntakeTestAPI(t)
	ctx := context.Background()
	for _, req := range []ipc.WriteFactReq{
		{Ts: intakeNow, Kind: "config", Key: "rss:latest:tech", Value: "2026-08-01T09:00:00Z", Source: "poll:rss", Confidence: 1.0},
		{Ts: intakeNow, Kind: "config", Key: "crawl:hash:kernel", Value: "deadbeef", Source: "poll:crawl", Confidence: 1.0},
		{Ts: intakeNow, Kind: "config", Key: "quiet_hours", Value: "true", Source: "tap:voice", Confidence: 1.0},
		{Ts: intakeNow, Kind: "env", Key: "praxis:list_attention", Value: "ok", Source: "praxis:trace", Confidence: 1.0},
	} {
		if _, err := api.WriteFact(ctx, req); err != nil {
			t.Fatalf("WriteFact(%s): %v", req.Key, err)
		}
	}
	if n := bus.Len(); n != 0 {
		t.Fatalf("journalled %d bookkeeping writes, want 0: %+v", n, bus.Recent(0))
	}
	// A real arrival under the same decorator still lands.
	if _, err := api.WriteFact(ctx, ipc.WriteFactReq{
		Ts: intakeNow, Kind: "env", Key: "spend_today", Value: "1200",
		Source: "poll:zenmoney", Confidence: 1.0,
	}); err != nil {
		t.Fatal(err)
	}
	if bus.Len() != 1 {
		t.Fatalf("a real intake write was dropped: %+v", bus.Recent(0))
	}
}

// Confidence is the distinction between an inference and a credentialled read,
// and the three-value priority bucket cannot carry it: unset and 1.0 land in
// the same bucket, and 0.6 is gone entirely once mapped. Payload keeps it.
func TestIntakeCarriesConfidenceAndFactKind(t *testing.T) {
	api, bus := newIntakeTestAPI(t)
	ctx := context.Background()
	if _, err := api.WriteFact(ctx, ipc.WriteFactReq{
		Ts: intakeNow, Kind: "env", Key: "calendar_event_x", Value: "18:00 планёрка",
		Source: "ambient:notif", Confidence: 0.6,
	}); err != nil {
		t.Fatal(err)
	}
	if _, err := api.WriteFact(ctx, ipc.WriteFactReq{
		Ts: intakeNow, Kind: "self", Key: "mood", Value: "ok", Source: "tap:web", Confidence: 1.0,
	}); err != nil {
		t.Fatal(err)
	}
	got := bus.Recent(0)
	if len(got) != 2 {
		t.Fatalf("journal has %d entries, want 2", len(got))
	}
	var relayed, stated factDetail
	if err := json.Unmarshal(got[1].Payload, &relayed); err != nil {
		t.Fatalf("payload: %v", err)
	}
	if relayed.Confidence == nil || *relayed.Confidence != 0.6 {
		t.Errorf("confidence = %v, want 0.6 recoverable from the payload", relayed.Confidence)
	}
	if relayed.FactKind != "env" {
		t.Errorf("fact_kind = %q, want env", relayed.FactKind)
	}
	// Both writes land in PriorityNormal or PriorityLow buckets that cannot be
	// told apart from the outside; the payload is where the two numbers stay
	// distinguishable.
	if err := json.Unmarshal(got[0].Payload, &stated); err != nil {
		t.Fatalf("payload: %v", err)
	}
	if stated.Confidence == nil || *stated.Confidence != 1.0 || stated.FactKind != "self" {
		t.Errorf("payload = %+v, want confidence 1.0 and fact_kind self", stated)
	}
}

// A retraction is not an observation. It used to publish an envelope
// indistinguishable from a fresh reading of the same key.
func TestIntakeMarksARetraction(t *testing.T) {
	api, bus := newIntakeTestAPI(t)
	ctx := context.Background()
	id, err := api.WriteFact(ctx, ipc.WriteFactReq{
		Ts: intakeNow, Kind: "env", Key: "weight", Value: "82", Source: "tap:web", Confidence: 1.0,
	})
	if err != nil {
		t.Fatal(err)
	}
	if _, err := api.WriteFact(ctx, ipc.WriteFactReq{
		Ts: intakeNow, Kind: "env", Key: "weight", Value: "81", Source: "tap:web",
		Confidence: 1.0, VoidsID: &id,
	}); err != nil {
		t.Fatal(err)
	}
	e := bus.Recent(1)[0]
	if e.Priority != event.PriorityLow {
		t.Errorf("priority = %q, want low for a correction", e.Priority)
	}
	if !strings.HasPrefix(e.Title, "отмена:") {
		t.Errorf("title = %q, want it marked as a retraction", e.Title)
	}
	var d factDetail
	if err := json.Unmarshal(e.Payload, &d); err != nil {
		t.Fatal(err)
	}
	if d.VoidsID == nil || *d.VoidsID != id {
		t.Errorf("voids_id = %v, want %d", d.VoidsID, id)
	}
}