Maven/internal/ipc/ipc_test.go

package ipc

import (
	"bytes"
	"context"
	"encoding/binary"
	"encoding/json"
	"errors"
	"io"
	"net"
	"os"
	"path/filepath"
	"testing"
	"time"

	"github.com/kami/maven/internal/store"
)

// tmpSocket — a socket path under a 0700 temp dir, unique per test.
func tmpSocket(t *testing.T) string {
	t.Helper()
	dir := t.TempDir()
	return filepath.Join(dir, "maven.sock")
}

// newServerWithStore spins a real store + Server + Client so the boundary
// is exercised exactly as the daemon wires it. Returns the api (for direct
// in-process expectations) and a client going through the socket.
func newServerWithStore(t *testing.T) (CoreAPI, *Server, *Client, *store.Store) {
	t.Helper()
	dir := t.TempDir()
	s, err := store.Open(context.Background(), filepath.Join(dir, "maven.db"))
	if err != nil {
		t.Fatalf("open store: %v", err)
	}
	t.Cleanup(func() { _ = s.Close() })

	api := NewStoreAPI(s)
	srv, err := Listen(tmpSocket(t), api)
	if err != nil {
		t.Fatalf("listen: %v", err)
	}
	done := make(chan struct{})
	go func() {
		_ = srv.Serve()
		close(done)
	}()
	t.Cleanup(func() {
		_ = srv.Close()
		<-done
	})
	cli, err := Dial(srv.Path())
	if err != nil {
		t.Fatalf("dial: %v", err)
	}
	t.Cleanup(func() { _ = cli.Close() })
	return api, srv, cli, s
}

// TestDialWait_WaitsForLateServer — a module may start before core's socket
// exists (core loads models first). DialWait must keep retrying until the
// socket appears rather than fail on the first attempt.
func TestDialWait_WaitsForLateServer(t *testing.T) {
	dir := t.TempDir()
	sock := filepath.Join(dir, "maven.sock")
	srvCh := make(chan *Server, 1)
	errCh := make(chan error, 1)

	// bring the server up only after DialWait is already retrying
	go func() {
		time.Sleep(300 * time.Millisecond)
		s, err := store.Open(context.Background(), filepath.Join(dir, "maven.db"))
		if err != nil {
			errCh <- err
			return
		}
		t.Cleanup(func() { _ = s.Close() })
		srv, err := Listen(sock, NewStoreAPI(s))
		if err != nil {
			errCh <- err
			return
		}
		go func() { _ = srv.Serve() }()
		srvCh <- srv
	}()

	cli, err := DialWait(sock, 5*time.Second)
	if err != nil {
		t.Fatalf("DialWait should connect once the server appears: %v", err)
	}
	select {
	case err := <-errCh:
		t.Fatalf("server setup failed: %v", err)
	case srv := <-srvCh:
		_ = cli.Close() // close client first so srv.Close's handler wait returns
		_ = srv.Close()
	}
}

// TestClient_ReconnectsAfterServerRestart — a long-lived module (e.g. mavweb)
// must survive a core restart. The first server is closed and a new one is
// brought up on the SAME socket path (as a daemon restart does); the client's
// cached conn is now dead. The next call must transparently re-dial and succeed
// instead of failing forever with "broken pipe".
func TestClient_ReconnectsAfterServerRestart(t *testing.T) {
	dir := t.TempDir()
	sock := filepath.Join(dir, "maven.sock")

	serve := func() *Server {
		s, err := store.Open(context.Background(), filepath.Join(dir, "maven.db"))
		if err != nil {
			t.Fatalf("open store: %v", err)
		}
		t.Cleanup(func() { _ = s.Close() })
		srv, err := Listen(sock, NewStoreAPI(s))
		if err != nil {
			t.Fatalf("listen: %v", err)
		}
		go func() { _ = srv.Serve() }()
		return srv
	}

	srv1 := serve()
	cli, err := Dial(sock)
	if err != nil {
		t.Fatalf("dial: %v", err)
	}
	t.Cleanup(func() { _ = cli.Close() })

	// works against the first server
	if _, err := cli.Presence(context.Background()); err != nil {
		t.Fatalf("call before restart: %v", err)
	}

	// Simulate a core restart: the client's conn dies (as it would when the
	// daemon process exits), then a fresh server binds the SAME path. Closing
	// the client side first also lets srv1's handler goroutine see EOF and
	// exit, so srv1.Close()'s wg.Wait() returns instead of blocking on a
	// parked reader.
	cli.conn.Close()
	_ = srv1.Close()
	srv2 := serve()

	// The cached conn is dead — the call must transparently re-dial and succeed.
	if _, err := cli.Presence(context.Background()); err != nil {
		t.Fatalf("call after restart should have re-dialed, got: %v", err)
	}

	// Teardown order matters: Server.Close waits for its handler goroutine,
	// which is parked reading the (now live, re-dialed) client conn. Close the
	// client first so the handler sees EOF and Close returns instead of hanging.
	_ = cli.Close()
	_ = srv2.Close()
}

// TestFrame_Roundtrip — JSON over a length prefix survives the loop, and the
// prefix itself encodes the length exactly. The framing is the only thing
// keeping a module's request paired with core's reply; it's worth a direct test.
func TestFrame_Roundtrip(t *testing.T) {
	var buf bytes.Buffer
	type payload struct {
		Msg string `json:"m"`
		N   int    `json:"n"`
	}
	want := payload{Msg: "hello", N: 42}
	if err := writeFrame(&buf, want); err != nil {
		t.Fatalf("writeFrame: %v", err)
	}
	// header length must equal the JSON body length that follows.
	var hdr [4]byte
	if _, err := io.ReadFull(&buf, hdr[:]); err != nil {
		t.Fatalf("read hdr: %v", err)
	}
	bodyLen := binary.BigEndian.Uint32(hdr[:])
	if int(bodyLen) != buf.Len() {
		t.Fatalf("prefix length %d != body %d", bodyLen, buf.Len())
	}
	// readFrame consumes header+body together; recombine so it sees a whole frame.
	full := append(hdr[:], buf.Bytes()...)
	var got payload
	if err := readFrame(bytes.NewReader(full), &got); err != nil {
		t.Fatalf("readFrame: %v", err)
	}
	if got != want {
		t.Fatalf("roundtrip mismatch: got %+v want %+v", got, want)
	}
}

func TestFrame_TooLarge(t *testing.T) {
	// Encode-side guard refuses to ship anything bigger than maxFrame; the
	// socket never sees it. Defense against a confused peer, not a real path.
	big := make([]byte, maxFrame+1)
	if err := writeFrame(io.Discard, big); !errors.Is(err, ErrFrameTooLarge) {
		t.Fatalf("writeFrame: got %v, want ErrFrameTooLarge", err)
	}
	// Decode-side guard refuses a header claiming a too-large body; the conn
	// is now desynced (length read, body not), but readFrame doesn't have to
	// recover — the caller closes it.
	var hdr [4]byte
	binary.BigEndian.PutUint32(hdr[:], maxFrame+1)
	if err := readFrame(bytes.NewReader(hdr[:]), nil); !errors.Is(err, ErrFrameTooLarge) {
		t.Fatalf("readFrame: got %v, want ErrFrameTooLarge", err)
	}
}

// TestSocket_Perms — the auth floor. 0600 ⇒ only the same unix user can
// connect. If this regresses to world-readable, every user on the box is a
// module; that's the entire auth model today, so assert it.
func TestSocket_Perms(t *testing.T) {
	_, srv, _, _ := newServerWithStore(t)
	fi, err := os.Stat(srv.Path())
	if err != nil {
		t.Fatalf("stat socket: %v", err)
	}
	mode := fi.Mode().Perm()
	if mode != 0o600 {
		t.Fatalf("socket perm = %#o, want 0600", mode)
	}
}

// TestStoreAPI_Direct — the in-process adapter path (no socket) maps store
// sentinels to ipc sentinels. The boundary's contract is that error identity
// is the same on both sides; this pins it for the daemon-embedded modules
// (router, delivery today) that never go over the wire.
func TestStoreAPI_Direct(t *testing.T) {
	api, _, _, _ := newServerWithStore(t)
	ctx := context.Background()

	// missing key ⇒ ErrNoFact
	if _, err := api.LatestFact(ctx, "nope"); !errors.Is(err, ErrNoFact) {
		t.Fatalf("LatestFact missing: got %v, want ErrNoFact", err)
	}
	// bad confidence ⇒ ErrConfidence
	if _, err := api.WriteFact(ctx, WriteFactReq{
		Ts: time.Now(), Kind: "self", Key: "water", Value: "1", Source: "tap:water", Confidence: 0,
	}); !errors.Is(err, ErrConfidence) {
		t.Fatalf("WriteFact conf=0: got %v, want ErrConfidence", err)
	}
	// since on missing key ⇒ ErrNoFact
	if _, err := api.Since(ctx, "nope", time.Now()); !errors.Is(err, ErrNoFact) {
		t.Fatalf("Since missing: got %v, want ErrNoFact", err)
	}
	// reminder idempotency: invalid status ⇒ ErrReminderState
	if err := api.MarkReminder(ctx, 99999, "weird"); !errors.Is(err, ErrReminderState) {
		t.Fatalf("MarkReminder weird: got %v, want ErrReminderState", err)
	}
	// resolve nonexistent nudge ⇒ ErrNudgeNotFound
	if err := api.ResolveNudge(ctx, 99999, "acted", time.Now()); !errors.Is(err, ErrNudgeNotFound) {
		t.Fatalf("ResolveNudge none: got %v, want ErrNudgeNotFound", err)
	}
}

// TestClient_E2E — full socket round trip against a real store. Drives every
// method end-to-end and asserts sentinel identity survives the wire. This is
// the test that catches the boundary bugs: param shape mismatch, sentinel
// code drift, dto mapping, framing interleaving.
func TestClient_E2E(t *testing.T) {
	_, _, cli, _ := newServerWithStore(t)
	ctx := context.Background()
	now := time.Now().UTC().Truncate(time.Millisecond)

	// write a tap (self, confidence 1.0) and read it back.
	id, err := cli.WriteFact(ctx, WriteFactReq{
		Ts: now, Kind: "self", Key: "water", Value: "1", Source: "tap:water", Confidence: 1.0,
	})
	if err != nil {
		t.Fatalf("WriteFact: %v", err)
	}
	if id <= 0 {
		t.Fatalf("WriteFact returned id %d", id)
	}
	f, err := cli.LatestFact(ctx, "water")
	if err != nil {
		t.Fatalf("LatestFact: %v", err)
	}
	if f.Key != "water" || f.Value != "1" || f.Source != "tap:water" || f.Confidence != 1.0 {
		t.Fatalf("LatestFact mismatch: %+v", f)
	}
	if !f.Ts.Equal(now) {
		t.Fatalf("Ts roundtrip: got %v want %v", f.Ts, now)
	}

	// provenance scope: a foreign source doesn't see the tap value.
	if _, err := cli.LatestFactBySource(ctx, "water", "poll:evil"); !errors.Is(err, ErrNoFact) {
		t.Fatalf("LatestFactBySource foreign: got %v, want ErrNoFact", err)
	}
	if _, err := cli.LatestFactBySource(ctx, "water", "tap:water"); err != nil {
		t.Fatalf("LatestFactBySource own: %v", err)
	}

	// since: ~0 elapsed since "now".
	d, err := cli.Since(ctx, "water", now.Add(time.Second))
	if err != nil {
		t.Fatalf("Since: %v", err)
	}
	if d != time.Second {
		t.Fatalf("Since dur = %v, want 1s", d)
	}
	// since missing ⇒ ErrNoFact over the wire.
	if _, err := cli.Since(ctx, "nope", now); !errors.Is(err, ErrNoFact) {
		t.Fatalf("Since missing: got %v, want ErrNoFact", err)
	}

	// presence cold-start ⇒ away, score 0.
	pres, err := cli.Presence(ctx)
	if err != nil {
		t.Fatalf("Presence: %v", err)
	}
	if pres.Bucket != Away || pres.Score != 0 {
		t.Fatalf("Presence cold-start = %+v, want away/0", pres)
	}

	// reminder lifecycle: create → mark fired → re-mark ⇒ ErrReminderState.
	rid, err := cli.CreateReminder(ctx, now.Add(time.Hour), `{"text":"wake me 7"}`, "")
	if err != nil {
		t.Fatalf("CreateReminder: %v", err)
	}
	if err := cli.MarkReminder(ctx, rid, "fired"); err != nil {
		t.Fatalf("MarkReminder fired: %v", err)
	}
	if err := cli.MarkReminder(ctx, rid, "fired"); !errors.Is(err, ErrReminderState) {
		t.Fatalf("MarkReminder twice: got %v, want ErrReminderState", err)
	}

	// nudge lifecycle: record → resolve acted → resolve again ⇒ ErrNudgeOutcome.
	nid, err := cli.RecordNudge(ctx, "water", "voice", "drink", now)
	if err != nil {
		t.Fatalf("RecordNudge: %v", err)
	}
	if err := cli.ResolveNudge(ctx, nid, "acted", now); err != nil {
		t.Fatalf("ResolveNudge acted: %v", err)
	}
	if err := cli.ResolveNudge(ctx, nid, "ignored", now); !errors.Is(err, ErrNudgeOutcome) {
		t.Fatalf("ResolveNudge twice: got %v, want ErrNudgeOutcome", err)
	}

	// feedback loop read: RecentOutcomes returns the resolved outcome.
	out, err := cli.RecentOutcomes(ctx, "water", 5)
	if err != nil {
		t.Fatalf("RecentOutcomes: %v", err)
	}
	if len(out) != 1 || out[0] != "acted" {
		t.Fatalf("RecentOutcomes = %v, want [acted]", out)
	}
	// empty result over the wire is a stable [] not null (server coerces).
	if got, err := cli.RecentOutcomes(ctx, "never_fired_rule", 5); err != nil || len(got) != 0 {
		t.Fatalf("RecentOutcomes empty = %v err=%v, want []", got, err)
	}
}

// TestCaller_Peercred — when the client dials, core sees a Caller with the
// test process's own uid via SO_PEERCRED. This is the seam auth scopes on;
// asserting it's populated today means the future auth layer has its input.
func TestCaller_Peercred(t *testing.T) {
	_, srv, cli, _ := newServerWithStore(t)
	ctx := context.Background()

	// round-trip any call; the server annotates ctx with a Caller on accept.
	if _, err := cli.LatestFact(ctx, "nope"); err != nil && !errors.Is(err, ErrNoFact) {
		t.Fatalf("LatestFact: %v", err)
	}

	// introspect the server's view: re-accept a conn manually and read creds.
	uc, err := dialRaw(srv.Path())
	if err != nil {
		t.Fatalf("dialRaw: %v", err)
	}
	defer uc.Close()
	c, ok := peerCaller(uc)
	if !ok {
		t.Skip("SO_PEERCRED unavailable on this platform; skipping")
	}
	if c.Uid != int32(os.Getuid()) {
		t.Fatalf("peercred uid = %d, want %d", c.Uid, os.Getuid())
	}
}

// TestChatViaClient — Chat round-trips over the wire. Uses a custom CoreAPI
// that implements Chat (storeAPI returns an error for it).
func TestChatViaClient(t *testing.T) {
	dir := t.TempDir()
	sock := filepath.Join(dir, "maven.sock")
	chatAPI := &chatTestAPI{}
	srv, err := Listen(sock, chatAPI)
	if err != nil {
		t.Fatalf("listen: %v", err)
	}
	done := make(chan struct{})
	go func() {
		_ = srv.Serve()
		close(done)
	}()
	t.Cleanup(func() {
		_ = srv.Close()
		<-done
	})
	cli, err := Dial(srv.Path())
	if err != nil {
		t.Fatalf("dial: %v", err)
	}
	t.Cleanup(func() { _ = cli.Close() })

	reply, err := cli.Chat(context.Background(), "привет")
	if err != nil {
		t.Fatalf("Chat: %v", err)
	}
	if reply != "и тебе привет!" {
		t.Fatalf("Chat = %q, want %q", reply, "и тебе привет!")
	}
}

// chatTestAPI — a minimal CoreAPI that only implements Chat for testing.
type chatTestAPI struct{}

func (a *chatTestAPI) WriteFact(ctx context.Context, req WriteFactReq) (int64, error) {
	return 0, ErrUnknownMethod
}
func (a *chatTestAPI) LatestFact(ctx context.Context, key string) (Fact, error) {
	return Fact{}, ErrUnknownMethod
}
func (a *chatTestAPI) LatestFactBySource(ctx context.Context, key, source string) (Fact, error) {
	return Fact{}, ErrUnknownMethod
}
func (a *chatTestAPI) Since(ctx context.Context, key string, now time.Time) (time.Duration, error) {
	return 0, ErrUnknownMethod
}
func (a *chatTestAPI) Presence(ctx context.Context) (Presence, error) {
	return Presence{}, ErrUnknownMethod
}
func (a *chatTestAPI) CreateReminder(ctx context.Context, fire time.Time, payload, cron string) (int64, error) {
	return 0, ErrUnknownMethod
}
func (a *chatTestAPI) MarkReminder(ctx context.Context, id int64, status string) error {
	return ErrUnknownMethod
}
func (a *chatTestAPI) ListReminders(ctx context.Context, n int) ([]Reminder, error) {
	return nil, ErrUnknownMethod
}
func (a *chatTestAPI) RecordNudge(ctx context.Context, rule, channel, message string, ts time.Time) (int64, error) {
	return 0, ErrUnknownMethod
}
func (a *chatTestAPI) ResolveNudge(ctx context.Context, id int64, outcome string, ts time.Time) error {
	return ErrUnknownMethod
}
func (a *chatTestAPI) RecentOutcomes(ctx context.Context, rule string, n int) ([]string, error) {
	return nil, ErrUnknownMethod
}
func (a *chatTestAPI) RecentFacts(ctx context.Context, n int) ([]Fact, error) {
	return nil, ErrUnknownMethod
}
func (a *chatTestAPI) CalendarEvents(ctx context.Context, from, to time.Time) ([]Fact, error) {
	return nil, ErrUnknownMethod
}
func (a *chatTestAPI) RecentNudges(ctx context.Context, n int) ([]Nudge, error) {
	return nil, ErrUnknownMethod
}
func (a *chatTestAPI) WriteNote(ctx context.Context, ts time.Time, text string, embedding []float32, source string) (int64, error) {
	return 0, ErrUnknownMethod
}
func (a *chatTestAPI) QueryNotes(ctx context.Context, embedding []float32, k int) ([]Note, error) {
	return nil, ErrUnknownMethod
}
func (a *chatTestAPI) RecentNotes(ctx context.Context, n int) ([]Note, error) {
	return nil, ErrUnknownMethod
}
func (a *chatTestAPI) ProposeTool(ctx context.Context, name, utterance, scope string, ts time.Time) (bool, error) {
	return false, ErrUnknownMethod
}
func (a *chatTestAPI) EnableTool(ctx context.Context, name string, cmd []string, destructive bool, scope string, ts time.Time) error {
	return ErrUnknownMethod
}
func (a *chatTestAPI) DisableTool(ctx context.Context, name string) error {
	return ErrUnknownMethod
}
func (a *chatTestAPI) DeleteTool(ctx context.Context, name string) error {
	return ErrUnknownMethod
}
func (a *chatTestAPI) ListProposedRoutines(ctx context.Context) ([]ProposedRoutine, error) {
	return nil, ErrUnknownMethod
}
func (a *chatTestAPI) DismissProposedRoutine(ctx context.Context, id int64) error {
	return ErrUnknownMethod
}
func (a *chatTestAPI) LookupTool(ctx context.Context, name string) (Tool, error) {
	return Tool{}, ErrUnknownMethod
}
func (a *chatTestAPI) ListTools(ctx context.Context, status string) ([]Tool, error) {
	return nil, ErrUnknownMethod
}
func (a *chatTestAPI) RevertFact(ctx context.Context, key string) (int64, error) {
	return 0, ErrUnknownMethod
}
func (a *chatTestAPI) TickTrace(ctx context.Context) (TickTrace, error) {
	return TickTrace{}, ErrUnknownMethod
}
func (a *chatTestAPI) MorningStatus(ctx context.Context) ([]MorningRoutineStatus, error) {
	return nil, ErrUnknownMethod
}
func (a *chatTestAPI) Chat(ctx context.Context, text string) (string, error) {
	if text == "привет" {
		return "и тебе привет!", nil
	}
	return "поговорили.", nil
}

// TestDispatch_UnknownMethod — an unknown method over the wire comes back as
// ErrUnknownMethod, not a panic or a dropped conn. The server must stay up
// for the next (legitimate) request on the same conn.
func TestDispatch_UnknownMethod(t *testing.T) {
	_, srv, _, _ := newServerWithStore(t)

	uc, err := dialRaw(srv.Path())
	if err != nil {
		t.Fatalf("dialRaw: %v", err)
	}
	defer uc.Close()

	// send garbage method on the raw conn, read back its error, then send a
	// real method on the SAME conn to confirm the server survived.
	if err := writeFrame(uc, Request{Method: Method("definitely_not_a_method")}); err != nil {
		t.Fatalf("writeFrame: %v", err)
	}
	var resp Response
	if err := readFrame(uc, &resp); err != nil {
		t.Fatalf("readFrame: %v", err)
	}
	if resp.Error == nil || !errors.Is(hydrate(resp.Error), ErrUnknownMethod) {
		t.Fatalf("unknown method response = %+v, want ErrUnknownMethod", resp.Error)
	}
	// same conn, legit follow-up: prove the goroutine is still alive.
	if err := writeFrame(uc, Request{Method: MethodLatestFact, Params: mustJSON(keyReq{Key: "nope"})}); err != nil {
		t.Fatalf("writeFrame follow-up: %v", err)
	}
	if err := readFrame(uc, &resp); err != nil {
		t.Fatalf("readFrame follow-up: %v", err)
	}
	if resp.Error == nil || !errors.Is(hydrate(resp.Error), ErrNoFact) {
		t.Fatalf("follow-up response = %+v, want ErrNoFact", resp.Error)
	}
}

// dialRaw — a bare unix conn for tests that want to script the wire directly
// (send an unknown method, follow up on the same conn, inspect framing).
func dialRaw(path string) (net.Conn, error) {
	return net.Dial("unix", path)
}

// crashAfterReceiveServer accepts exactly one connection, reads exactly one
// request frame (so, from the client's point of view, the request definitely
// reached the server — a real write could have already committed at this
// point), then closes the connection without ever writing a reply. This is
// the "commit-then-disconnect" scenario the audit finding is about: the
// client cannot tell success from failure from the dropped connection alone.
func crashAfterReceiveServer(t *testing.T, path string) {
	t.Helper()
	l, err := net.Listen("unix", path)
	if err != nil {
		t.Fatalf("listen: %v", err)
	}
	t.Cleanup(func() { _ = l.Close() })
	go func() {
		conn, err := l.Accept()
		if err != nil {
			return
		}
		var req Request
		_ = readFrame(conn, &req)
		_ = conn.Close() // crash: request received, no reply ever sent
	}()
}

// TestClient_MutationNotRetriedOnAmbiguousDisconnect — the audit's core
// finding (Vikunja #269): a write whose reply never arrived (server received
// the frame, then died before replying) must not be silently retried, since
// the original request may have already committed. The client must surface
// ErrAmbiguousOutcome instead of guessing either way.
func TestClient_MutationNotRetriedOnAmbiguousDisconnect(t *testing.T) {
	sock := tmpSocket(t)
	crashAfterReceiveServer(t, sock)

	cli, err := Dial(sock)
	if err != nil {
		t.Fatalf("dial: %v", err)
	}
	defer cli.Close()

	_, err = cli.WriteFact(context.Background(), WriteFactReq{Key: "k", Value: "v", Source: "test"})
	if !errors.Is(err, ErrAmbiguousOutcome) {
		t.Fatalf("expected ErrAmbiguousOutcome on commit-then-disconnect, got %v", err)
	}
}

// TestClient_ReadRetriedOnAmbiguousDisconnect — the same disconnect timing on
// a read-only method is safe to retry (replaying a read can't double-apply):
// the first connection receives the request and dies without replying, the
// second connection (the client's automatic retry redial) gets a real reply.
// No sleeps: each accepted connection is handled deterministically by index.
func TestClient_ReadRetriedOnAmbiguousDisconnect(t *testing.T) {
	sock := tmpSocket(t)

	l, err := net.Listen("unix", sock)
	if err != nil {
		t.Fatalf("listen: %v", err)
	}
	t.Cleanup(func() { _ = l.Close() })

	var n int
	go func() {
		for {
			conn, err := l.Accept()
			if err != nil {
				return
			}
			n++
			attempt := n
			go func(conn net.Conn, attempt int) {
				defer conn.Close()
				var req Request
				if err := readFrame(conn, &req); err != nil {
					return
				}
				if attempt == 1 {
					return // crash: request received, no reply ever sent
				}
				_ = writeFrame(conn, Response{Result: mustJSON(Presence{})})
			}(conn, attempt)
		}
	}()

	cli, err := Dial(sock)
	if err != nil {
		t.Fatalf("dial: %v", err)
	}
	defer cli.Close()

	if _, err := cli.Presence(context.Background()); err != nil {
		t.Fatalf("read should have retried past the ambiguous disconnect, got: %v", err)
	}
}

func mustJSON(v any) []byte {
	b, err := json.Marshal(v)
	if err != nil {
		panic(err)
	}
	return b
}