1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160

package acp_test

import (
	"bufio"
	"encoding/json"
	"fmt"
	"net"
	"sync"
	"testing"
	"time"

	"codeberg.org/turbo-editors/turbo-core/acp"
)

// fakeAgent is an agent living in this process, at the far end of an in-memory
// pipe.
//
// Driving the client against it exercises the real framing, the real
// concurrency and the real decoding, with no subprocess, no model and no
// timing to get lucky with. It is deliberately not built on the client's own
// types for the wire: a peer that shared them could not catch the client
// encoding a field wrongly.
type fakeAgent struct {
	t      *testing.T
	stream net.Conn
	reader *bufio.Reader

	mu       sync.Mutex
	received []string
}

// newFakeAgent returns an agent and the stream a client talks to it over.
func newFakeAgent(t *testing.T) (*fakeAgent, net.Conn) {
	t.Helper()

	theirs, ours := net.Pipe()
	agent := &fakeAgent{t: t, stream: theirs, reader: bufio.NewReader(theirs)}
	t.Cleanup(func() { _ = theirs.Close() })
	return agent, ours
}

// read returns the next message the client sent.
func (a *fakeAgent) read() map[string]any {
	a.t.Helper()

	_ = a.stream.SetReadDeadline(time.Now().Add(5 * time.Second))
	line, err := a.reader.ReadBytes('\n')
	if err != nil {
		a.t.Fatalf("reading from the client: %v", err)
	}

	var msg map[string]any
	if err := json.Unmarshal(line, &msg); err != nil {
		a.t.Fatalf("the client wrote %q, which is not JSON: %v", line, err)
	}

	a.mu.Lock()
	a.received = append(a.received, fmt.Sprint(msg["method"]))
	a.mu.Unlock()
	return msg
}

// send writes one raw line to the client.
func (a *fakeAgent) send(line string) {
	a.t.Helper()

	_ = a.stream.SetWriteDeadline(time.Now().Add(5 * time.Second))
	if _, err := fmt.Fprintf(a.stream, "%s\n", line); err != nil {
		a.t.Fatalf("writing to the client: %v", err)
	}
}

// reply answers a request with a result.
func (a *fakeAgent) reply(id any, result string) {
	a.send(fmt.Sprintf(`{"jsonrpc":"2.0","id":%v,"result":%s}`, id, result))
}

// update sends one session/update.
func (a *fakeAgent) update(sessionID, update string) {
	a.send(fmt.Sprintf(`{"jsonrpc":"2.0","method":"session/update","params":{"sessionId":%q,"update":%s}}`, sessionID, update))
}

// methods returns what the client has sent, in order.
func (a *fakeAgent) methods() []string {
	a.mu.Lock()
	defer a.mu.Unlock()

	out := make([]string, len(a.received))
	copy(out, a.received)
	return out
}

// handshake plays the opening exchange the way the real docker agent does,
// and returns the session id it handed out.
//
// The shapes here are copied from a recorded conversation with docker agent
// v1.139.0 against a local llama.cpp, not invented: the agentCapabilities
// object really does carry sessionCapabilities, and authMethods really is an
// empty array rather than absent.
func (a *fakeAgent) handshake() string {
	return a.handshakeWith(`{"auth":{},"mcpCapabilities":{},"promptCapabilities":{"embeddedContext":true,"image":true},"sessionCapabilities":{"close":{},"list":{},"resume":{}}}`)
}

// handshakeWith is handshake with the agent declaring the given capabilities
// object, for an agent that takes less than docker agent does.
func (a *fakeAgent) handshakeWith(capabilities string) string {
	initialize := a.read()
	if initialize["method"] != acp.MethodInitialize {
		a.t.Fatalf("the client opened with %v, want initialize", initialize["method"])
	}
	a.reply(initialize["id"], `{"agentCapabilities":`+capabilities+`,"agentInfo":{"name":"docker agent","title":"docker agent","version":"v1.139.0"},"authMethods":[],"protocolVersion":1}`)

	newSession := a.read()
	if newSession["method"] != acp.MethodNewSession {
		a.t.Fatalf("the client asked %v, want session/new", newSession["method"])
	}
	const id = "eefb58ee-4f23-4064-9ea4-dba74b957d3a"
	a.reply(newSession["id"], fmt.Sprintf(`{"sessionId":%q}`, id))
	return id
}

// waitFor polls until want is true, or fails the test.
//
// The client's work happens on goroutines of its own, so a test that looked
// once would pass or fail by how fast the machine is.
func waitFor(t *testing.T, what string, want func() bool) {
	t.Helper()

	deadline := time.Now().Add(5 * time.Second)
	for time.Now().Before(deadline) {
		if want() {
			return
		}
		time.Sleep(time.Millisecond)
	}
	t.Fatalf("timed out waiting for %s", what)
}

// textOf joins an entry list into something a test can assert on.
func textOf(entries []acp.Entry, kind acp.EntryKind) string {
	var out []string
	for _, entry := range entries {
		if entry.Kind == kind {
			out = append(out, entry.Text)
		}
	}
	return joinWith(out, "\n")
}

// joinWith joins pieces, so the tests need no import for one call.
func joinWith(pieces []string, sep string) string {
	result := ""
	for i, piece := range pieces {
		if i > 0 {
			result += sep
		}
		result += piece
	}
	return result
}