package stdio import ( "bufio" "encoding/json" "io" "testing" "time" ) // emitRuntime drives a bundle over pipes as the runtime does, reading every line it says. type emitRuntime struct { t *testing.T lines *bufio.Scanner in io.Writer } func (r *emitRuntime) next() map[string]any { done := make(chan map[string]any, 1) go func() { if r.lines.Scan() { var m map[string]any _ = json.Unmarshal(r.lines.Bytes(), &m) done <- m } }() select { case m := <-done: return m case <-time.After(3 * time.Second): r.t.Fatal("the bundle said nothing") return nil } } func serveEmit(t *testing.T, tools []Tool) *emitRuntime { t.Helper() t.Setenv(ServedAs, "chime") inR, inW := io.Pipe() outR, outW := io.Pipe() go func() { _ = ServeOn("", tools, inR, outW) }() t.Cleanup(func() { _ = inW.Close() }) return &emitRuntime{t: t, lines: bufio.NewScanner(outR), in: inW} } func (r *emitRuntime) call(id int, tool string) { b, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": id, "method": "tools/call", "params": map[string]any{"name": tool, "arguments": map[string]any{}}}) _, _ = r.in.Write(append(b, '\n')) } func (r *emitRuntime) answer(id any) { b, _ := json.Marshal(map[string]any{"jsonrpc": "2.0", "id": id, "result": map[string]any{}}) _, _ = r.in.Write(append(b, '\n')) } // eventIDOf is a "mesh/publish" ask's x-event-id header. func eventIDOf(asked map[string]any) string { return asked["params"].(map[string]any)["headers"].(map[string]any)["x-event-id"].(string) } // 1187/AC1: given EmitOptions with a message id, when a bundle emits, then the event's x-event-id header is // that id; and a random one when none is given, as on main. func TestEmitSendsTheChosenMessageIDAsXEventID(t *testing.T) { rt := serveEmit(t, []Tool{ {Name: "chosen", Run: func(map[string]any) (any, error) { return nil, Emit("sent", map[string]any{}, EmitOptions{MessageID: "send-42"}) }}, {Name: "unchosen", Run: func(map[string]any) (any, error) { return nil, Emit("sent", map[string]any{}) }}, }) rt.call(1, "chosen") asked := rt.next() if got := eventIDOf(asked); got != "send-42" { t.Fatalf("x-event-id: %q, want the chosen id %q", got, "send-42") } rt.answer(asked["id"]) rt.next() // the tool's own answer rt.call(2, "unchosen") asked = rt.next() if got := eventIDOf(asked); len(got) != 36 { t.Fatalf("x-event-id with no message id chosen: %q, want a random UUID as on main", got) } rt.answer(asked["id"]) rt.next() } // 1187/AC2: given two emits with the same chosen id, when sent through the runtime's protocol stand-in, // then both carry the same x-event-id, which the tool runner gives the bus as Nats-Msg-Id (mesh-tools // node-tools/internal/bus/bus.go ~748-773, :770): a repeated publish is kept to one by the bus's dedup. func TestEmitRepeatedWithTheSameChosenIDCarriesTheSameXEventIDTwice(t *testing.T) { rt := serveEmit(t, []Tool{ {Name: "chosen", Run: func(map[string]any) (any, error) { return nil, Emit("sent", map[string]any{}, EmitOptions{MessageID: "send-77"}) }}, }) var ids []string for i := 1; i <= 2; i++ { rt.call(i, "chosen") asked := rt.next() ids = append(ids, eventIDOf(asked)) rt.answer(asked["id"]) rt.next() } if ids[0] != "send-77" || ids[1] != "send-77" { t.Fatalf("the two emits' x-event-id: %v, want %q both times", ids, "send-77") } }