Merge branch 'fix/g10' into fix/integrated

2026-08-01 14:24:05 +04:00
parent b436be69c3 d0e98a9419
commit 8846b7e43c
26 changed files with 1411 additions and 160 deletions
@@ -55,8 +55,10 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"net/http"
 	"os"
 	"path/filepath"
+	"regexp"
 	"strings"
 	"sync"
 	"testing"
@@ -102,9 +104,22 @@ type scenario struct {
 	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.
@@ -125,6 +140,16 @@ type scriptEntry struct {
 // 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"`
@@ -176,6 +201,13 @@ type signalStep struct {
 	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 {
@@ -225,6 +257,28 @@ type simWorld struct {
 	// 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
@@ -324,6 +378,25 @@ func newSimWorld(t *testing.T, sc scenario) *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
@@ -345,7 +418,10 @@ func newSimWorld(t *testing.T, sc scenario) *simWorld {
 	// 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)
+	// 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{
@@ -355,6 +431,7 @@ func newSimWorld(t *testing.T, sc scenario) *simWorld {
 		embedder:           emb,
 		api:                api,
 		matcher:            matcher,
+		tools:              tool.NewExecutor(api, 5*time.Second),
 		phraser:            phraser.NewStub(),
 		replier:            newLLMReplier(scripted, nil),
 		now:                clock.Now,
@@ -416,8 +493,9 @@ func (w *simWorld) advanceTo(at string) {
 	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",
+		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))
 }
@@ -446,8 +524,8 @@ func (w *simWorld) run(sc scenario) {
 			w.logf("# %s", s.Note)
 		}
 		sendsBefore := w.sink.count()
-		callsBefore := w.callCount()
-		eventsBefore := w.bus.Len()
+		callsBefore := w.callMark()
+		eventsBefore := w.publishCount()

 		w.stimulate(ctx, s)
 		w.assert(i, s, sendsBefore, callsBefore, eventsBefore)
@@ -503,10 +581,14 @@ func (w *simWorld) write(ctx context.Context, sig signalStep, ts time.Time) {
 	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: 1.0,
+		Ts: ts, Kind: kind, Key: sig.Key, Value: sig.Value, Source: sig.Source, Confidence: conf,
 	}); err != nil {
-		w.t.Fatalf("write fact %s: %v", sig.Key, err)
+		w.fatalf("write fact %s: %v", sig.Key, err)
 	}
 }

@@ -552,29 +634,39 @@ func (w *simWorld) arrive(ctx context.Context, a arriveStep) {
 // 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")
-	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
+	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
 		}
 	}
-	w.t.Fatalf("audio step %q: no such case in %s", ref, manifest)
+	if text, ok := w.audio[ref]; ok && ref != "" {
+		return text
+	}
+	w.fatalf("audio step %q: no such case in %s", ref, manifest)
 	return ""
 }

@@ -582,35 +674,62 @@ 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} {
+// 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 {
-			n += len(fs.Requests())
+			mark[i] = len(fs.Requests())
 		}
 	}
-	return n
+	return mark
 }

-func (w *simWorld) callPaths() []string {
+// 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 _, fs := range []*fakeServer{w.praxis, w.nexus, w.hexis} {
+	for i, fs := range w.fakes() {
 		if fs == nil {
 			continue
 		}
-		for _, r := range fs.Requests() {
+		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, callsBefore, eventsBefore int) {
+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 != "" {
@@ -654,9 +773,10 @@ func (w *simWorld) assert(i int, s step, sendsBefore, callsBefore, eventsBefore
 			fail("no ecosystem call matches %q; calls so far: %v", want, paths)
 		}
 	}
+	since := w.callPathsSince(callsBefore)
 	for _, unwanted := range s.ExpectNotCalled {
-		if anyContains(paths[callsBefore:], unwanted) {
-			fail("an ecosystem call matched %q and must not have: %v", unwanted, paths[callsBefore:])
+		if anyContains(since, unwanted) {
+			fail("an ecosystem call matched %q and must not have: %v", unwanted, since)
 		}
 	}

@@ -666,8 +786,12 @@ func (w *simWorld) assert(i int, s step, sendsBefore, callsBefore, eventsBefore
 			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())
+	// 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)
 	}
 }

@@ -682,7 +806,10 @@ func sendableTexts(sends []delivery.Sendable) []string {
 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))
+		// 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
 }
@@ -773,26 +900,143 @@ func TestSimulatorIsDeterministic(t *testing.T) {
 	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()")
+	// 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) {
-	// 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)
+	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)
 	}
-	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)
+	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)
 	}
 }