9f51596e2f
The seed path was relative to the working directory, which is cmd/mavend under `go test`. Every open failed, and the three scenarios replayed a whole scripted day against a classifier holding zero examples. They passed. A green simulator was proving something other than the routing the box runs, and a regression in the seed set could not have surfaced there. seedPath walks up to five levels to find models/seeds, so the daemon started from the repo root behaves exactly as before and a test started anywhere inside the tree finds the same files. All three scenarios still pass with 339 seeds loaded, so the outcome was not resting on the empty classifier. The new test asserts the count rather than logging it. A silent zero is the failure that hid here.
1069 lines
37 KiB
Go
1069 lines
37 KiB
Go
// mavend/simulator_test.go — the replayable full-system simulator
|
|
// (Vikunja #284).
|
|
//
|
|
// # 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"
|
|
"net/http"
|
|
"os"
|
|
"path/filepath"
|
|
"regexp"
|
|
"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"`
|
|
|
|
// Tools — allowlist rows to enable before the first step. An act is only
|
|
// dispatched when its verb is on the enabled allowlist, so a scenario that
|
|
// wants to exercise one has to say which rows Kami had enabled.
|
|
Tools []toolRow `json:"tools,omitempty"`
|
|
|
|
Steps []step `json:"steps"`
|
|
}
|
|
|
|
// toolRow — one enabled allowlist row. Cmd is empty for an ecosystem verb,
|
|
// which is intercepted before the process executor is ever reached.
|
|
type toolRow struct {
|
|
Name string `json:"name"`
|
|
Cmd []string `json:"cmd,omitempty"`
|
|
Destructive bool `json:"destructive,omitempty"`
|
|
}
|
|
|
|
// 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.
|
|
//
|
|
// The asymmetry is worth stating plainly, because it changes what a scenario
|
|
// author is writing. expect_sent_contains, expect_called and expect_events are
|
|
// RUN-scoped: they pass if the thing ever happened, at any earlier step. So
|
|
// repeating expect_events: ["rss:tech"] on a later step asserts nothing new,
|
|
// it just re-checks the earlier arrival. The negatives — expect_no_send,
|
|
// expect_not_called, expect_no_events — are STEP-scoped, and are the ones that
|
|
// say something about this moment. expect_reply_contains and
|
|
// expect_reply_lacks read the most recent reply only, so a step with no
|
|
// utterance re-checks the previous one.
|
|
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"`
|
|
|
|
// Confidence — 0 ⇒ 1.0, an observation Maven made herself. A relayed
|
|
// notification is not that: the ambient path writes
|
|
// calendar.AmbientConfidence, 0.6, and factPriority in intake.go branches
|
|
// on exactly that difference. The field exists so a replay can reach the
|
|
// low branch, which it could not while write() hardcoded 1.0.
|
|
Confidence float64 `json:"confidence,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
|
|
|
|
// fatalf — the abort seam. Defaults to t.Fatalf. It exists so a test can
|
|
// reach the harness's own refusals (a backwards step, a missing WAV) and
|
|
// assert on them instead of dying with the scenario.
|
|
fatalf func(format string, args ...any)
|
|
|
|
// published — how many events the bus accepted, counted through a
|
|
// subscriber. bus.Len() saturates at the ring capacity and cannot answer
|
|
// "did anything arrive during this step" once a long scenario has filled
|
|
// it.
|
|
mu sync.Mutex
|
|
published int
|
|
|
|
// audio — golden_v1.json, parsed once. A scenario with twenty audio steps
|
|
// used to read and parse the manifest twenty times.
|
|
audio map[string]string
|
|
}
|
|
|
|
func (w *simWorld) publishCount() int {
|
|
w.mu.Lock()
|
|
defer w.mu.Unlock()
|
|
return w.published
|
|
}
|
|
|
|
// 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)
|
|
// A grammar no longer separates the two contracts — the replier carries one
|
|
// too since phraser.ResponseGrammar was attached to it. Only the router's
|
|
// grammar names the intent enum, so that is what tells them apart.
|
|
routing := strings.Contains(r.Grammar, "intent")
|
|
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,
|
|
}
|
|
w.fatalf = t.Fatalf
|
|
bus.Subscribe(func(event.Event) {
|
|
w.mu.Lock()
|
|
w.published++
|
|
w.mu.Unlock()
|
|
})
|
|
|
|
// Allowlist rows the scenario asked for, enabled before the first step. An
|
|
// act only reaches a dispatch if a verb is on the enabled list, so without
|
|
// this a scenario cannot script one at all.
|
|
for _, tr := range sc.Tools {
|
|
cmd := tr.Cmd
|
|
if len(cmd) == 0 {
|
|
cmd = []string{"true"}
|
|
}
|
|
if err := st.EnableTool(context.Background(), tr.Name, cmd, tr.Destructive, "sim", start); err != nil {
|
|
t.Fatalf("scenario %q: enabling tool %q: %v", sc.Name, tr.Name, err)
|
|
}
|
|
}
|
|
|
|
// 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)
|
|
// The matcher reads the live enabled allowlist, same as the daemon's. It
|
|
// used to be built on a nil API, which meant any scenario that produced an
|
|
// act panicked the moment the matcher was consulted.
|
|
matcher := tool.NewMatcher(api)
|
|
rtr := buildRouter(emb, matcher, config.DefaultRouterThreshold, router.NewLLMRouter(scripted))
|
|
|
|
w.handler = &reactiveHandler{
|
|
stt: simTranscriber{},
|
|
tts: simSynthesizer{},
|
|
router: rtr,
|
|
embedder: emb,
|
|
api: api,
|
|
matcher: matcher,
|
|
tools: tool.NewExecutor(api, 5*time.Second),
|
|
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.fatalf("step at %s goes backwards from %s — scenario steps must be in order",
|
|
at, now.In(w.loc).Format("15:04:05"))
|
|
return
|
|
}
|
|
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.callMark()
|
|
eventsBefore := w.publishCount()
|
|
|
|
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, sourceText)
|
|
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"
|
|
}
|
|
conf := sig.Confidence
|
|
if conf == 0 {
|
|
conf = 1.0
|
|
}
|
|
if _, err := w.api.WriteFact(ctx, ipc.WriteFactReq{
|
|
Ts: ts, Kind: kind, Key: sig.Key, Value: sig.Value, Source: sig.Source, Confidence: conf,
|
|
}); err != nil {
|
|
w.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 "".
|
|
//
|
|
// The manifest is read and parsed once per world, not once per step: a
|
|
// scenario with twenty audio steps should cost one file read.
|
|
func (w *simWorld) audioText(ref string) string {
|
|
w.t.Helper()
|
|
manifest := filepath.Join("..", "mavsttd", "testdata", "golden_v1.json")
|
|
if w.audio == nil {
|
|
raw, err := os.ReadFile(manifest)
|
|
if err != nil {
|
|
w.fatalf("audio step %q: reading %s: %v", ref, manifest, err)
|
|
return ""
|
|
}
|
|
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.fatalf("audio step %q: parsing %s: %v", ref, manifest, err)
|
|
return ""
|
|
}
|
|
w.audio = make(map[string]string, len(m.Cases)*2)
|
|
for _, c := range m.Cases {
|
|
w.audio[c.Name] = c.Text
|
|
w.audio[c.WAV] = c.Text
|
|
}
|
|
}
|
|
if text, ok := w.audio[ref]; ok && ref != "" {
|
|
return text
|
|
}
|
|
w.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}}
|
|
}
|
|
|
|
// fakes — every ecosystem fake, in a fixed order. Both the mark and the paths
|
|
// walk this same order, which is the whole point: they have to agree.
|
|
func (w *simWorld) fakes() []*fakeServer { return []*fakeServer{w.praxis, w.nexus, w.hexis} }
|
|
|
|
// callMark takes a PER-SERVER snapshot of how many requests each fake has
|
|
// seen. It is not a total.
|
|
//
|
|
// A total cannot be used to slice the concatenated path list, and the harness
|
|
// used to do exactly that. callPaths concatenates praxis, then nexus, then
|
|
// hexis; a total counts arrivals across all three. With praxis on 3 requests
|
|
// and nexus on 1, the total is 4 and the list is [p1 p2 p3 n1]. A step that
|
|
// calls praxis once makes the list [p1 p2 p3 p4 n1], and paths[4:] is [n1].
|
|
// The new praxis call sits at index 3 and is never looked at, so
|
|
// expect_not_called on praxis passed on a step that called praxis. The same
|
|
// slice reported the stale nexus call as new, so expect_not_called on
|
|
// "/resolve" failed on a step that resolved nothing.
|
|
func (w *simWorld) callMark() []int {
|
|
mark := make([]int, len(w.fakes()))
|
|
for i, fs := range w.fakes() {
|
|
if fs != nil {
|
|
mark[i] = len(fs.Requests())
|
|
}
|
|
}
|
|
return mark
|
|
}
|
|
|
|
// callPathsSince returns the calls each fake took after its own mark. A nil
|
|
// mark means "everything, from the beginning of the run".
|
|
func (w *simWorld) callPathsSince(mark []int) []string {
|
|
var out []string
|
|
for i, fs := range w.fakes() {
|
|
if fs == nil {
|
|
continue
|
|
}
|
|
reqs := fs.Requests()
|
|
from := 0
|
|
if mark != nil && i < len(mark) {
|
|
from = mark[i]
|
|
}
|
|
if from > len(reqs) {
|
|
from = len(reqs)
|
|
}
|
|
for _, r := range reqs[from:] {
|
|
out = append(out, r.Method+" "+r.Path)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (w *simWorld) callPaths() []string { return w.callPathsSince(nil) }
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Assertions
|
|
// ---------------------------------------------------------------------------
|
|
|
|
func (w *simWorld) assert(i int, s step, sendsBefore int, callsBefore []int, 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)
|
|
}
|
|
}
|
|
since := w.callPathsSince(callsBefore)
|
|
for _, unwanted := range s.ExpectNotCalled {
|
|
if anyContains(since, unwanted) {
|
|
fail("an ecosystem call matched %q and must not have: %v", unwanted, since)
|
|
}
|
|
}
|
|
|
|
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))
|
|
}
|
|
}
|
|
// Counted publishes, not bus.Len(): the ring saturates at its capacity, so
|
|
// a long scenario that filled it made every later expect_no_events pass
|
|
// unconditionally.
|
|
if s.ExpectNoEvents && w.publishCount() > eventsBefore {
|
|
fail("expected nothing to arrive, %d event(s) were published",
|
|
w.publishCount()-eventsBefore)
|
|
}
|
|
}
|
|
|
|
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 {
|
|
// Priority is in the line so a scenario can assert on it. It is the one
|
|
// field factPriority derives from confidence, and without it a replay
|
|
// could set a confidence but never see what the journal did with it.
|
|
out = append(out, fmt.Sprintf("%s/%s pri=%s %s %s", e.Source, e.Kind, e.Priority, 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 every transcript timestamp must lie inside the scenario's own span.
|
|
//
|
|
// This used to compare the transcript against time.Now().Format("15:04"),
|
|
// which failed whenever the suite happened to run during the half hour the
|
|
// scenario covers: morning_missed logs 08:30 through 09:00, sc.Start
|
|
// contains only 08:30, so a run at 08:35 reported a wall-clock read that
|
|
// had not happened. A determinism test that depends on the time of day is
|
|
// the bug it is looking for.
|
|
start, err := time.Parse(time.RFC3339, sc.Start)
|
|
if err != nil {
|
|
t.Fatalf("bad start: %v", err)
|
|
}
|
|
last := start
|
|
for _, s := range sc.Steps {
|
|
if at := stepInstant(t, start, s.At); at.After(last) {
|
|
last = at
|
|
}
|
|
}
|
|
// Only the lines logf stamped. A note or a feed item can carry its own
|
|
// newlines, and those continuation lines have no timestamp.
|
|
stamp := regexp.MustCompile(`^(\d\d:\d\d:\d\d) `)
|
|
for _, line := range strings.Split(first, "\n") {
|
|
m := stamp.FindStringSubmatch(line)
|
|
if m == nil {
|
|
continue
|
|
}
|
|
at := stepInstant(t, start, m[1])
|
|
if at.Before(start) || at.After(last) {
|
|
t.Errorf("transcript line %q is stamped outside the scenario span %s..%s — "+
|
|
"something in the replay path read time.Now()",
|
|
line, start.Format("15:04:05"), last.Format("15:04:05"))
|
|
}
|
|
}
|
|
}
|
|
|
|
// TestCallsSinceAreScopedPerServer pins the ordering bug that made
|
|
// expect_not_called unsound. The mark is per server; a total cannot slice a
|
|
// list that is concatenated per server.
|
|
func TestCallsSinceAreScopedPerServer(t *testing.T) {
|
|
sc := scenario{SchemaVersion: 1, Name: "x", Start: "2026-08-01T08:30:00+03:00",
|
|
Praxis: fixturePraxisAttentionItems(), Nexus: fixtureNexusResolved("ent_1", "thing", "device"),
|
|
Steps: []step{{At: "08:30"}}}
|
|
w := newSimWorld(t, sc)
|
|
|
|
hit := func(fs *fakeServer, path string) {
|
|
t.Helper()
|
|
resp, err := http.Get(fs.URL + path)
|
|
if err != nil {
|
|
t.Fatalf("hitting %s: %v", path, err)
|
|
}
|
|
resp.Body.Close()
|
|
}
|
|
// Praxis runs ahead of nexus, so the concatenated list already has a nexus
|
|
// call sitting after three praxis ones.
|
|
hit(w.praxis, "/api/v1/tools/attention")
|
|
hit(w.praxis, "/api/v1/tools/attention")
|
|
hit(w.praxis, "/api/v1/tools/attention")
|
|
hit(w.nexus, "/api/v1/resolve")
|
|
|
|
mark := w.callMark()
|
|
hit(w.praxis, "/api/v1/tools/attention")
|
|
|
|
since := w.callPathsSince(mark)
|
|
if !anyContains(since, "attention") {
|
|
t.Errorf("the praxis call made after the mark is missing from %v", since)
|
|
}
|
|
if anyContains(since, "resolve") {
|
|
t.Errorf("a nexus call from before the mark was reported as new: %v", since)
|
|
}
|
|
}
|
|
|
|
// TestPublishCountDoesNotSaturateWithTheRing pins expect_no_events on a bus
|
|
// that has already wrapped. bus.Len() stops at the capacity, so it can no
|
|
// longer answer "did anything arrive".
|
|
func TestPublishCountDoesNotSaturateWithTheRing(t *testing.T) {
|
|
sc := scenario{SchemaVersion: 1, Name: "x", Start: "2026-08-01T08:30:00+03:00",
|
|
Steps: []step{{At: "08:30"}}}
|
|
w := newSimWorld(t, sc)
|
|
|
|
for i := 0; i < event.DefaultCapacity+5; i++ {
|
|
w.bus.Publish(event.Event{
|
|
Source: "sim:test", Kind: event.KindFact, Title: fmt.Sprintf("f%d", i),
|
|
}, w.clock.Now())
|
|
}
|
|
if got := w.bus.Len(); got != event.DefaultCapacity {
|
|
t.Fatalf("ring holds %d, expected it to be saturated at %d", got, event.DefaultCapacity)
|
|
}
|
|
before := w.publishCount()
|
|
w.bus.Publish(event.Event{Source: "sim:test", Kind: event.KindFact, Title: "one more"}, w.clock.Now())
|
|
if w.publishCount() != before+1 {
|
|
t.Errorf("publish count went %d → %d on a full ring, expected it to keep counting",
|
|
before, w.publishCount())
|
|
}
|
|
}
|
|
|
|
// stepInstant resolves "HH:MM" or "HH:MM:SS" against the scenario's start day.
|
|
func stepInstant(t *testing.T, start time.Time, at string) time.Time {
|
|
t.Helper()
|
|
layout := "15:04"
|
|
if strings.Count(at, ":") == 2 {
|
|
layout = "15:04:05"
|
|
}
|
|
hm, err := time.Parse(layout, at)
|
|
if err != nil {
|
|
t.Fatalf("bad step time %q: %v", at, err)
|
|
}
|
|
return time.Date(start.Year(), start.Month(), start.Day(),
|
|
hm.Hour(), hm.Minute(), hm.Second(), 0, start.Location())
|
|
}
|
|
|
|
// TestSimulatorRefusesBackwardsSteps guards the one scenario-authoring mistake
|
|
// that would silently produce a meaningless run.
|
|
//
|
|
// It used to advance to 09:00 and then to 09:30 and assert the clock had
|
|
// moved, which is the forwards case: the backwards branch it is named after
|
|
// was never reached, because reaching it ended the test. The fatalf seam is
|
|
// what makes it testable.
|
|
func TestSimulatorRefusesBackwardsSteps(t *testing.T) {
|
|
sc := scenario{SchemaVersion: 1, Name: "x", Start: "2026-08-01T08:30:00+03:00",
|
|
Steps: []step{{At: "09:00"}}}
|
|
w := newSimWorld(t, sc)
|
|
var refusal string
|
|
w.fatalf = func(format string, args ...any) { refusal = fmt.Sprintf(format, args...) }
|
|
|
|
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)
|
|
}
|
|
if refusal != "" {
|
|
t.Fatalf("a forwards step was refused: %s", refusal)
|
|
}
|
|
|
|
w.advanceTo("08:45")
|
|
if refusal == "" {
|
|
t.Fatal("a step going backwards to 08:45 was accepted")
|
|
}
|
|
if got := w.clock.Now().In(w.loc).Format("15:04"); got != "09:00" {
|
|
t.Errorf("the clock moved to %s on a refused step, it must stay at 09:00", got)
|
|
}
|
|
}
|
|
|
|
// TestSimulatorRoutesWithTheDeployedSeeds — the scenarios must replay against
|
|
// the classifier the deploy runs, not an empty one.
|
|
//
|
|
// They did not. The seed path was relative to the working directory, which is
|
|
// cmd/mavend under `go test`, so every file failed to open and the whole
|
|
// simulator scored three green scenarios with zero examples loaded (Vikunja
|
|
// #465). The count is asserted rather than logged, because a silent zero is
|
|
// exactly the failure that hid here for as long as it did.
|
|
func TestSimulatorRoutesWithTheDeployedSeeds(t *testing.T) {
|
|
cls := router.NewClassifier(router.NewHashEmbedder(1024))
|
|
seedClassifier(cls)
|
|
total := 0
|
|
for _, intent := range cls.Intents() {
|
|
total += len(cls.Examples(intent))
|
|
}
|
|
if total == 0 {
|
|
t.Fatalf("no seed examples loaded from %s — the simulator would route on nothing", seedPath())
|
|
}
|
|
if len(cls.Intents()) != 7 {
|
|
t.Fatalf("seeded %d intents, want all 7", len(cls.Intents()))
|
|
}
|
|
}
|