wip, rework bridge

.gitignore

@@ -1 +1,2 @@
 app.log
+.direnv/

internal/bridge/client.go

@@ -1,15 +1,15 @@
 package bridge
 
 import (
-	"bufio"
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
 	"io"
-	"log/slog"
+	"os"
 	"os/exec"
-	"slices"
 	"sync"
+	"sync/atomic"
 )
 
 var ErrNotRunning = errors.New("bridge not running")
@@ -21,349 +21,480 @@ type ExecutionPoint struct {
 }
 
 type Bridge struct {
-	stdin  io.Writer
+	path string
-	stdout *bufio.Reader
+
+	lifecycleMu sync.Mutex
+	started     bool
+	finished    bool
+	closing     bool
+	failure     error
+	exitErr     error
 
 	cmd      *exec.Cmd
+	stdin    io.WriteCloser
+	protocol io.ReadCloser
+	encoder  *json.Encoder
+	decoder  *json.Decoder
 
-	input chan string
+	writeMu sync.Mutex
 
-	outputLock *sync.RWMutex
+	nextID    atomic.Uint64
-	output     []chan string
+	pendingMu sync.Mutex
+	pending   map[uint64]chan Message
 
-	executionStopLock *sync.RWMutex
+	events chan Message
-	executionStop     []chan ExecutionPoint
+	output chan Output
+	done   chan struct{}
 
-	registry     map[string]chan string
+	finishOnce sync.Once
-	registryLock *sync.RWMutex
+}
 
-	breakpoints map[string][]int
+type OutputStream string
 
-	callbacksLock *sync.RWMutex
+const (
-	callbacks     map[string]map[int]func()
+	StdoutStream OutputStream = "stdout"
+	StderrStream OutputStream = "stderr"
+)
 
-	running bool
+type Output struct {
-
+	Stream OutputStream
-	path string
+	Text   string
 }
 
 func NewBridge(path string) *Bridge {
 	return &Bridge{
 		path:    path,
-		input:        make(chan string),
+		pending: make(map[uint64]chan Message),
-		registry:     make(map[string]chan string),
+		events:  make(chan Message, 32),
-		registryLock: &sync.RWMutex{},
+		output:  make(chan Output, 128),
-		breakpoints:  make(map[string][]int),
+		done:    make(chan struct{}),
-
-		callbacksLock: &sync.RWMutex{},
-		callbacks:     make(map[string]map[int]func()),
-
-		output:     make([]chan string, 0),
-		outputLock: &sync.RWMutex{},
-
-		executionStop:     make([]chan ExecutionPoint, 0),
-		executionStopLock: &sync.RWMutex{},
-
-		running: false,
 	}
 }
 
+func (b *Bridge) Events() <-chan Message {
+	return b.events
+}
+
+func (b *Bridge) Output() <-chan Output {
+	return b.output
+}
+
+func (b *Bridge) Done() <-chan struct{} {
+	return b.done
+}
+
 func (b *Bridge) Start() error {
-	if b.running {
+	b.lifecycleMu.Lock()
+	defer b.lifecycleMu.Unlock()
+
+	if b.started {
 		return ErrAlreadyStarted
 	}
 
-	b.cmd = exec.Command(
+	protocolReader, protocolWriter, err := os.Pipe()
+	if err != nil {
+		return fmt.Errorf("create protocol pipe: %w", err)
+	}
+
+	cmd := exec.Command(
 		"python",
 		"-u",
 		"python/pybug_runtime.py",
 		b.path,
 	)
 
-	var err error
+	cmd.ExtraFiles = []*os.File{protocolWriter}
-	b.stdin, err = b.cmd.StdinPipe()
+	cmd.Env = append(os.Environ(), "PYBUG_PROTOCOL_FD=3")
+
+	stdin, err := cmd.StdinPipe()
 	if err != nil {
-		return err
+
+		protocolWriter.Close()
+		protocolReader.Close()
+		return fmt.Errorf("create command pipe: %w", err)
 	}
 
-	reader, err := b.cmd.StdoutPipe()
+	stdout, err := cmd.StdoutPipe()
 	if err != nil {
-		return err
+		stdin.Close()
-	}
-	b.stdout = bufio.NewReader(reader)
 
-	err = b.cmd.Start()
+		protocolReader.Close()
-	if err != nil {
+		protocolWriter.Close()
-		return err
+		return fmt.Errorf("create stdout pipe: %w", err)
 	}
 
-	b.running = true
+	stderr, err := cmd.StderrPipe()
-	go b.readLoop()
+	if err != nil {
-	go b.writeLoop()
+		stdout.Close()
+		protocolReader.Close()
+		protocolWriter.Close()
+		return fmt.Errorf("create stderr pipe: %w", err)
+	}
+
+	if err := cmd.Start(); err != nil {
+		stdin.Close()
+		stdout.Close()
+		stderr.Close()
+		protocolWriter.Close()
+		protocolReader.Close()
+		return fmt.Errorf("start debugger runtime: %w", err)
+	}
+
+	if err := protocolWriter.Close(); err != nil {
+		cmd.Process.Kill()
+		stdin.Close()
+		stdout.Close()
+		stderr.Close()
+		protocolWriter.Close()
+		protocolReader.Close()
+		return fmt.Errorf("close parent protocol writer: %w", err)
+	}
+
+	b.cmd = cmd
+	b.stdin = stdin
+	b.protocol = protocolReader
+	b.encoder = json.NewEncoder(stdin)
+	b.decoder = json.NewDecoder(protocolReader)
+	b.started = true
+
+	go b.readProtocol()
+	go b.readOutput(stdout, StdoutStream)
+	go b.readOutput(stderr, StderrStream)
+	go b.waitForProcess()
 
 	return nil
 }
 
-func (b *Bridge) Subscribe() chan string {
+func (b *Bridge) ensureRunning() error {
-	b.outputLock.Lock()
+	b.lifecycleMu.Lock()
-	defer b.outputLock.Unlock()
+	defer b.lifecycleMu.Unlock()
 
-	c := make(chan string)
+	if !b.started || b.finished {
-	b.output = append(b.output, c)
-
-	return c
-}
-
-func (b *Bridge) SubscribeStopped() chan ExecutionPoint {
-	b.executionStopLock.Lock()
-	defer b.executionStopLock.Unlock()
-
-	c := make(chan ExecutionPoint)
-	b.executionStop = append(b.executionStop, c)
-
-	return c
-}
-
-func (b *Bridge) Locals() (map[string]any, error) {
-	if !b.running {
-		return nil, ErrNotRunning
-	}
-
-	requestId, cmd := makeCommand(LocalsCommand, map[string]any{})
-
-	c := b.sendCommand(requestId, cmd)
-
-	obj := <-c
-
-	var m map[string]any
-	err := json.Unmarshal([]byte(obj), &m)
-	if err != nil {
-		return nil, err
-	}
-
-	vars, ok := m["vars"].(map[string]any)
-	if !ok {
-		return nil, errors.New("could not extract vars from response")
-	}
-
-	return vars, nil
-}
-
-func (b *Bridge) Step() error {
-	if !b.running {
 		return ErrNotRunning
 	}
 
-	_, cmd := makeCommand(StepCommand, nil)
+	return nil
-	b.sendCommandNoResponse(cmd)
+}
+
+func (b *Bridge) writeMessage(msg Message) error {
+	b.writeMu.Lock()
+	defer b.writeMu.Unlock()
+
+	if err := b.encoder.Encode(msg); err != nil {
+		return fmt.Errorf("write protocol message: %w", err)
+	}
 
 	return nil
 }
 
-func (b *Bridge) Breakpoint(file string, line int) (set bool, err error) {
+func (b *Bridge) request(
-	if !b.running {
+	ctx context.Context,
-		return false, ErrNotRunning
+	name string,
+	body any,
+) (Message, error) {
+	if err := b.ensureRunning(); err != nil {
+		return Message{}, err
 	}
 
-	var command CommandType
+	id := b.nextID.Add(1)
-	if _, ok := b.breakpoints[file]; ok && slices.Contains(b.breakpoints[file], line) {
+	response := make(chan Message, 1)
-		command = UnbreakCommand
+
-	} else {
+	b.pendingMu.Lock()
-		command = BreakCommand
+	b.pending[id] = response
+	b.pendingMu.Unlock()
+
+	defer func() {
+		b.pendingMu.Lock()
+		delete(b.pending, id)
+		b.pendingMu.Unlock()
+	}()
+
+	msg, err := newMessage(Request, &id, name, body)
+	if err != nil {
+		return Message{}, fmt.Errorf("create %q request: %w", name, err)
 	}
 
-	requestId, cmd := makeCommand(command, map[string]any{
+	if err := msg.Validate(); err != nil {
-		"file": file,
+		return Message{}, fmt.Errorf("validate %q request: %w", name, err)
-		"line": line,
+	}
+
+	if err := b.writeMessage(msg); err != nil {
+		return Message{}, err
+	}
+
+	select {
+	case response := <-response:
+		if response.Name != name {
+			return Message{}, fmt.Errorf("response name mismatch: requested %q, received %q", name, response.Name)
+		}
+
+		if response.Error != nil {
+			return Message{}, response.Error
+		}
+		return response, nil
+	case <-ctx.Done():
+		return Message{}, ctx.Err()
+	case <-b.done:
+		return Message{}, ErrNotRunning
+	}
+}
+
+func (b *Bridge) deliverResponse(msg Message) {
+	id := *msg.RequestId
+
+	b.pendingMu.Lock()
+	response, ok := b.pending[id]
+	b.pendingMu.Unlock()
+
+	if !ok {
+		return
+	}
+
+	select {
+	case response <- msg:
+	case <-b.done:
+	default:
+	}
+}
+
+func (b *Bridge) readProtocol() {
+	for {
+		var msg Message
+		if err := b.decoder.Decode(&msg); err != nil {
+			if errors.Is(err, io.EOF) {
+				return
+			}
+
+			select {
+			case <-b.done:
+				return
+			default:
+			}
+
+			b.abort(fmt.Errorf("decode protocol message: %w", err))
+			return
+		}
+
+		if err := msg.Validate(); err != nil {
+			b.abort(fmt.Errorf("invalid protocol message: %w", err))
+			return
+		}
+
+		switch msg.Type {
+		case Response:
+			b.deliverResponse(msg)
+		case Event:
+			select {
+			case b.events <- msg:
+			case <-b.done:
+				return
+			}
+		default:
+			b.abort(fmt.Errorf("unexpected message type from runtime: %q", msg.Type))
+			return
+		}
+	}
+}
+
+func (b *Bridge) readOutput(reader io.ReadCloser, stream OutputStream) {
+	defer reader.Close()
+
+	buffer := make([]byte, 4096)
+
+	for {
+		n, err := reader.Read(buffer)
+
+		if n > 0 {
+			output := Output{
+				Stream: stream,
+				Text:   string(buffer[:n]),
+			}
+
+			select {
+			case b.output <- output:
+			case <-b.done:
+				return
+			}
+		}
+
+		if err != nil {
+			if !errors.Is(err, io.EOF) {
+				b.abort(fmt.Errorf("read target %s: %w", stream, err))
+			}
+			return
+		}
+	}
+}
+
+func (b *Bridge) waitForProcess() {
+	err := b.cmd.Wait()
+	b.finish(err)
+}
+
+func (b *Bridge) abort(err error) {
+	b.lifecycleMu.Lock()
+	if b.failure == nil {
+		b.failure = err
+	}
+
+	var process *os.Process
+	if b.cmd != nil {
+		process = b.cmd.Process
+	}
+	finished := b.finished
+
+	b.lifecycleMu.Unlock()
+
+	if !finished && process != nil {
+		_ = process.Kill()
+	}
+}
+
+func (b *Bridge) finish(waitErr error) {
+	b.finishOnce.Do(func() {
+		b.lifecycleMu.Lock()
+
+		b.finished = true
+		if b.failure != nil {
+			b.exitErr = b.failure
+		} else if !b.closing {
+			b.exitErr = waitErr
+		}
+
+		stdin := b.stdin
+		protocol := b.protocol
+
+		b.lifecycleMu.Unlock()
+
+		if stdin != nil {
+			_ = stdin.Close()
+		}
+
+		if protocol != nil {
+			_ = protocol.Close()
+		}
+
+		close(b.done)
 	})
-
-	c := b.sendCommand(requestId, cmd)
-
-	obj := <-c
-
-	var m map[string]any
-	err = json.Unmarshal([]byte(obj), &m)
-	if err != nil {
-		return false, err
 }
 
-	if m["status"] != "ok" {
+func (b *Bridge) Wait(ctx context.Context) error {
-		return false, fmt.Errorf("error occured on break, err: %s", m["error"])
+	b.lifecycleMu.Lock()
-	}
+	started := b.started
+	b.lifecycleMu.Unlock()
 
-	if command == BreakCommand {
+	if !started {
-		b.breakpoints[file] = append(b.breakpoints[file], line)
-		return true, nil
-	} else {
-		breakpointsLen := len(b.breakpoints[file])
-		index := slices.Index(b.breakpoints[file], line)
-		b.breakpoints[file][index] = b.breakpoints[file][breakpointsLen-1]
-
-		b.breakpoints[file] = b.breakpoints[file][0 : breakpointsLen-1]
-		return false, nil
-	}
-}
-
-func (b *Bridge) Continue() error {
-	if !b.running {
 		return ErrNotRunning
 	}
 
-	_, cmd := makeCommand(ContinueCommand, map[string]any{})
+	select {
+	case <-b.done:
+		return b.processError()
 
-	b.sendCommandNoResponse(cmd)
+	case <-ctx.Done():
-	return nil
+		return ctx.Err()
+	}
 }
 
-func (b *Bridge) sendCommandNoResponse(command string) {
+func (b *Bridge) Close(ctx context.Context) error {
-	b.input <- command
+	b.lifecycleMu.Lock()
+
+	if !b.started {
+		b.lifecycleMu.Unlock()
+		return ErrNotRunning
 	}
 
-func (b *Bridge) sendCommand(requestId string, command string) chan string {
+	if b.finished {
-	b.registryLock.Lock()
+		b.lifecycleMu.Unlock()
-	defer b.registryLock.Unlock()
+		return b.processError()
-
-	channel := make(chan string)
-
-	b.registry[requestId] = channel
-	b.input <- command
-
-	return channel
 	}
 
-func (b *Bridge) writeLoop() {
+	b.closing = true
-	slog.Debug("started writeLoop")
+	process := b.cmd.Process
-	defer slog.Debug("writeLoop exited")
+	b.lifecycleMu.Unlock()
-	for {
-		cmd := <-b.input
 
-		if !b.running {
+	if process != nil {
-			return
+		err := process.Kill()
+		if err != nil && !errors.Is(err, os.ErrProcessDone) {
+			return fmt.Errorf("kill debugger runtime: %w", err)
+		}
 	}
 
-		slog.Info("Received command", "cmd", cmd)
+	return b.Wait(ctx)
+}
 
-		_, err := b.stdin.Write([]byte(cmd + "\n"))
+func (b *Bridge) processError() error {
+	b.lifecycleMu.Lock()
+	defer b.lifecycleMu.Unlock()
+
+	return b.exitErr
+}
+
+func (b *Bridge) Locals(ctx context.Context) (LocalsResponse, error) {
+	msg, err := b.request(ctx, "locals/get", nil)
 	if err != nil {
-			slog.Error("Error occured while writing to stdin", "error", err)
+		return LocalsResponse{}, err
 	}
 
-		slog.Debug("Command written", "cmd", cmd)
+	var result LocalsResponse
-	}
+	if err := json.Unmarshal(msg.Body, &result); err != nil {
+		return LocalsResponse{}, fmt.Errorf("decode locals response: %w", err)
 	}
 
-func (b *Bridge) readLoop() {
+	return result, nil
-	slog.Debug("started readLoop")
-	defer slog.Debug("readLoop exited")
-	for {
-		slog.Debug("reading string from stdout waiting for newline")
-		line, err := b.stdout.ReadString('\n')
-		if err != nil && !errors.Is(err, io.EOF) {
-			slog.Error("Error occured while reading from stdout", "error", err)
-			continue
-		} else if errors.Is(err, io.EOF) {
-			b.running = false
-			return
 }
 
-		var msg map[string]any
+func (b *Bridge) Continue(ctx context.Context) error {
-		err = json.Unmarshal([]byte(line), &msg)
+	_, err := b.request(ctx, "execution/continue", nil)
-		if err != nil {
+	return err
-			slog.Debug("read line from stdout", "line", line)
-
-			b.outputLock.RLock()
-			for _, c := range b.output {
-				c <- line
-			}
-			b.outputLock.RUnlock()
-
-		} else if requestId, ok := msg["request_id"].(string); ok {
-			b.registryLock.RLock()
-			c, ok := b.registry[requestId]
-			b.registryLock.RUnlock()
-
-			if !ok {
-				slog.Error("Could not find requestId in registry", "requestId", requestId)
-				continue
 }
 
-			c <- line
+func (b *Bridge) StepInto(ctx context.Context) error {
-		} else {
+	_, err := b.request(ctx, "execution/step-into", nil)
-			b.handleStopped(msg)
+	return err
-		}
-	}
 }
 
-func (b *Bridge) handleStopped(msg map[string]any) error {
+func (b *Bridge) StepOver(ctx context.Context) error {
-	// TODO: set to stopped
+	_, err := b.request(ctx, "execution/step-over", nil)
-	if event, ok := msg["event"]; !ok || event != "stopped" {
+	return err
-		slog.Warn("received unkown event", "msg", msg)
-		return errors.New("unknown event encountered")
 }
 
-	file := msg["file"].(string)
+func (b *Bridge) SetBreakpoint(ctx context.Context, file string, line int) (SetBreakpointResponse, error) {
-	line, ok := toInt(msg["line"])
+	msg, err := b.request(
-	if !ok {
+		ctx,
-		slog.Error("could not convert line to int", "line", msg["line"])
+		"breakpoint/set",
-		return errors.New("could not convert line to int")
+		SetBreakpointRequest{
-	}
-
-	slog.Info("received stopped event")
-	b.callbacksLock.RLock()
-	defer b.callbacksLock.RUnlock()
-
-	if callback, ok := b.callbacks[file][line]; ok {
-		slog.Info("found callback, now running", "file", file, "line", line)
-		go callback()
-	}
-
-	b.executionStopLock.RLock()
-	defer b.executionStopLock.RUnlock()
-
-	for _, c := range b.executionStop {
-		c <- ExecutionPoint{
 			File: file,
 			Line: line,
-		}
+		},
+	)
+	if err != nil {
+		return SetBreakpointResponse{}, err
 	}
 
-	return nil
+	var response SetBreakpointResponse
+	if err := json.Unmarshal(msg.Body, &response); err != nil {
+		return SetBreakpointResponse{}, fmt.Errorf(
+			"decode breakpoint response: %w", err,
+		)
 	}
 
-func (b *Bridge) OnBreakpoint(file string, line int, callback func()) {
+	return response, nil
-	b.callbacksLock.Lock()
-	defer b.callbacksLock.Unlock()
-
-	if f, ok := b.callbacks[file]; ok {
-		f[line] = callback
-	} else {
-		b.callbacks[file] = map[int]func(){
-			line: callback,
-		}
-	}
 }
 
-func (b *Bridge) Wait() error {
+func (b *Bridge) RemoveBreakpoint(
-	return b.cmd.Wait()
+	ctx context.Context,
-}
+	file string,
+	line int,
+) error {
+	_, err := b.request(
+		ctx,
+		"breakpoint/remove",
+		RemoveBreakpointRequest{
+			File: file,
+			Line: line,
+		},
+	)
 
-func toInt(v any) (int, bool) {
+	return err
-	switch x := v.(type) {
-	case int:
-		return x, true
-	case int8:
-		return int(x), true
-	case int16:
-		return int(x), true
-	case int32:
-		return int(x), true
-	case int64:
-		return int(x), true
-	case float32:
-		return int(x), true
-	case float64:
-		return int(x), true
-	default:
-		return 0, false
-	}
 }

internal/bridge/client_test.go

@@ -0,0 +1,354 @@
+package bridge
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"os"
+	"os/exec"
+	"strings"
+	"testing"
+	"time"
+)
+
+type bridgeHarness struct {
+	bridge          *Bridge
+	requestDecoder  *json.Decoder
+	responseEncoder *json.Encoder
+	requestReader   *io.PipeReader
+	responseWriter  *io.PipeWriter
+}
+
+func newBridgeHarness(t *testing.T) *bridgeHarness {
+	t.Helper()
+
+	requestReader, requestWriter := io.Pipe()
+	responseReader, responseWriter := io.Pipe()
+
+	b := NewBridge("unused.py")
+	b.started = true
+	b.stdin = requestWriter
+	b.protocol = responseReader
+	b.encoder = json.NewEncoder(requestWriter)
+	b.decoder = json.NewDecoder(responseReader)
+
+	h := &bridgeHarness{
+		bridge:          b,
+		requestDecoder:  json.NewDecoder(requestReader),
+		responseEncoder: json.NewEncoder(responseWriter),
+		requestReader:   requestReader,
+		responseWriter:  responseWriter,
+	}
+
+	go b.readProtocol()
+
+	t.Cleanup(func() {
+		responseWriter.Close()
+		requestReader.Close()
+		b.finish(nil)
+	})
+
+	return h
+}
+
+func testContext(t *testing.T) context.Context {
+	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	t.Cleanup(cancel)
+	return ctx
+}
+
+func (h *bridgeHarness) receiveRequest(t *testing.T) Message {
+	t.Helper()
+
+	var request Message
+	if err := h.requestDecoder.Decode(&request); err != nil {
+		t.Fatalf("decode request: %v", err)
+	}
+	return request
+}
+
+func (h *bridgeHarness) sendMessage(t *testing.T, msg Message) {
+	t.Helper()
+
+	if err := h.responseEncoder.Encode(msg); err != nil {
+		t.Fatalf("encode runtime message: %v", err)
+	}
+}
+
+func TestRequestCorrelatesResponse(t *testing.T) {
+	h := newBridgeHarness(t)
+
+	type requestResult struct {
+		message Message
+		err     error
+	}
+	resultChannel := make(chan requestResult, 1)
+	ctx := testContext(t)
+
+	go func() {
+		msg, err := h.bridge.request(ctx, "locals/get", nil)
+		resultChannel <- requestResult{message: msg, err: err}
+	}()
+
+	request := h.receiveRequest(t)
+	if request.Type != Request || request.Name != "locals/get" || request.RequestId == nil {
+		t.Fatalf("unexpected request: %+v", request)
+	}
+
+	response, err := newMessage(Response, request.RequestId, request.Name, LocalsResponse{
+		Variables: map[string]string{"answer": "42"},
+	})
+	if err != nil {
+		t.Fatalf("newMessage() error = %v", err)
+	}
+	h.sendMessage(t, response)
+
+	result := <-resultChannel
+	if result.err != nil {
+		t.Fatalf("request() error = %v", result.err)
+	}
+
+	var body LocalsResponse
+	if err := json.Unmarshal(result.message.Body, &body); err != nil {
+		t.Fatalf("decode response body: %v", err)
+	}
+	if body.Variables["answer"] != "42" {
+		t.Fatalf("response variables = %#v", body.Variables)
+	}
+}
+
+func TestConcurrentRequestsAreCorrelatedByID(t *testing.T) {
+	h := newBridgeHarness(t)
+	ctx := testContext(t)
+
+	type requestResult struct {
+		name string
+		body string
+		err  error
+	}
+
+	const requestCount = 8
+	results := make(chan requestResult, requestCount)
+
+	for i := range requestCount {
+		name := fmt.Sprintf("test/request-%d", i)
+		go func() {
+			msg, err := h.bridge.request(ctx, name, nil)
+			if err != nil {
+				results <- requestResult{name: name, err: err}
+				return
+			}
+
+			var body struct {
+				Name string `json:"name"`
+			}
+			err = json.Unmarshal(msg.Body, &body)
+			results <- requestResult{name: name, body: body.Name, err: err}
+		}()
+	}
+
+	requests := make([]Message, 0, requestCount)
+	ids := make(map[uint64]struct{}, requestCount)
+	for range requestCount {
+		request := h.receiveRequest(t)
+		if request.RequestId == nil {
+			t.Fatalf("request has no ID: %+v", request)
+		}
+		if _, exists := ids[*request.RequestId]; exists {
+			t.Fatalf("duplicate request ID %d", *request.RequestId)
+		}
+		ids[*request.RequestId] = struct{}{}
+		requests = append(requests, request)
+	}
+
+	for i := len(requests) - 1; i >= 0; i-- {
+		request := requests[i]
+		response, err := newMessage(Response, request.RequestId, request.Name, struct {
+			Name string `json:"name"`
+		}{Name: request.Name})
+		if err != nil {
+			t.Fatalf("newMessage() error = %v", err)
+		}
+		h.sendMessage(t, response)
+	}
+
+	for range requestCount {
+		result := <-results
+		if result.err != nil {
+			t.Fatalf("request %q error = %v", result.name, result.err)
+		}
+		if result.body != result.name {
+			t.Fatalf("request %q received body for %q", result.name, result.body)
+		}
+	}
+}
+
+func TestRequestReturnsProtocolError(t *testing.T) {
+	h := newBridgeHarness(t)
+	errChannel := make(chan error, 1)
+	ctx := testContext(t)
+
+	go func() {
+		_, err := h.bridge.request(ctx, "execution/continue", nil)
+		errChannel <- err
+	}()
+
+	request := h.receiveRequest(t)
+	h.sendMessage(t, newErrorResponse(
+		*request.RequestId,
+		request.Name,
+		"not_paused",
+		"execution is not paused",
+	))
+
+	err := <-errChannel
+	var protocolErr *ProtocolError
+	if !errors.As(err, &protocolErr) {
+		t.Fatalf("request() error = %v, want ProtocolError", err)
+	}
+	if protocolErr.Code != "not_paused" {
+		t.Fatalf("protocol error code = %q", protocolErr.Code)
+	}
+}
+
+func TestRequestRejectsMismatchedResponseName(t *testing.T) {
+	h := newBridgeHarness(t)
+	errChannel := make(chan error, 1)
+	ctx := testContext(t)
+
+	go func() {
+		_, err := h.bridge.request(ctx, "locals/get", nil)
+		errChannel <- err
+	}()
+
+	request := h.receiveRequest(t)
+	response, err := newMessage(Response, request.RequestId, "execution/continue", nil)
+	if err != nil {
+		t.Fatalf("newMessage() error = %v", err)
+	}
+	h.sendMessage(t, response)
+
+	err = <-errChannel
+	if err == nil || !strings.Contains(err.Error(), "response name mismatch") {
+		t.Fatalf("request() error = %v, want response name mismatch", err)
+	}
+}
+
+func TestCancelledRequestIsRemovedFromRegistry(t *testing.T) {
+	h := newBridgeHarness(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	errChannel := make(chan error, 1)
+
+	go func() {
+		_, err := h.bridge.request(ctx, "locals/get", nil)
+		errChannel <- err
+	}()
+
+	request := h.receiveRequest(t)
+	cancel()
+
+	if err := <-errChannel; !errors.Is(err, context.Canceled) {
+		t.Fatalf("request() error = %v, want context.Canceled", err)
+	}
+
+	h.bridge.pendingMu.Lock()
+	_, exists := h.bridge.pending[*request.RequestId]
+	h.bridge.pendingMu.Unlock()
+	if exists {
+		t.Fatalf("request ID %d remained in pending registry", *request.RequestId)
+	}
+}
+
+func TestReadProtocolPublishesEvent(t *testing.T) {
+	h := newBridgeHarness(t)
+	event, err := newMessage(Event, nil, "execution/stopped", ExecutionStoppedEvent{
+		Reason: "step",
+		File:   "test.py",
+		Line:   12,
+	})
+	if err != nil {
+		t.Fatalf("newMessage() error = %v", err)
+	}
+	h.sendMessage(t, event)
+
+	select {
+	case received := <-h.bridge.Events():
+		if received.Name != "execution/stopped" {
+			t.Fatalf("event name = %q", received.Name)
+		}
+
+		var body ExecutionStoppedEvent
+		if err := json.Unmarshal(received.Body, &body); err != nil {
+			t.Fatalf("decode event body: %v", err)
+		}
+		if body.Reason != "step" || body.File != "test.py" || body.Line != 12 {
+			t.Fatalf("event body = %+v", body)
+		}
+
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for event")
+	}
+}
+
+func TestReadOutputPublishesStreamAndText(t *testing.T) {
+	b := NewBridge("unused.py")
+	reader := io.NopCloser(strings.NewReader("hello from target\n"))
+
+	go b.readOutput(reader, StdoutStream)
+
+	select {
+	case output := <-b.Output():
+		if output.Stream != StdoutStream {
+			t.Fatalf("output stream = %q", output.Stream)
+		}
+		if output.Text != "hello from target\n" {
+			t.Fatalf("output text = %q", output.Text)
+		}
+
+	case <-time.After(time.Second):
+		t.Fatal("timed out waiting for output")
+	}
+}
+
+func TestRequestBeforeStartReturnsErrNotRunning(t *testing.T) {
+	b := NewBridge("unused.py")
+
+	_, err := b.request(testContext(t), "locals/get", nil)
+	if !errors.Is(err, ErrNotRunning) {
+		t.Fatalf("request() error = %v, want ErrNotRunning", err)
+	}
+}
+
+func TestCloseTerminatesRunningProcess(t *testing.T) {
+	if os.Getenv("PYBUG_BRIDGE_HELPER_PROCESS") == "1" {
+		time.Sleep(time.Minute)
+		return
+	}
+
+	cmd := exec.Command(os.Args[0], "-test.run=TestCloseTerminatesRunningProcess")
+	cmd.Env = append(os.Environ(), "PYBUG_BRIDGE_HELPER_PROCESS=1")
+	if err := cmd.Start(); err != nil {
+		t.Fatalf("start helper process: %v", err)
+	}
+
+	b := NewBridge("unused.py")
+	b.cmd = cmd
+	b.started = true
+	go b.waitForProcess()
+
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	if err := b.Close(ctx); err != nil {
+		t.Fatalf("Close() error = %v", err)
+	}
+
+	select {
+	case <-b.Done():
+	default:
+		t.Fatal("Done channel was not closed")
+	}
+}

internal/bridge/protocol.go

@@ -2,43 +2,156 @@ package bridge
 
 import (
 	"encoding/json"
-	"maps"
+	"errors"
-	"math/rand"
+	"fmt"
 )
 
-type CommandType string
+const ProtocolVersion = 1
+
+type MessageType string
 
 const (
-	ContinueCommand CommandType = "continue"
+	Request  MessageType = "request"
-	BreakCommand    CommandType = "break"
+	Response MessageType = "response"
-	UnbreakCommand  CommandType = "unbreak"
+	Event    MessageType = "event"
-	LocalsCommand   CommandType = "locals"
-	StepCommand     CommandType = "step"
 )
 
-func makeCommand(cmd CommandType, values map[string]any) (requestId string, request string) {
+type Message struct {
-	m := make(map[string]any, len(values)+2)
+	Version   int             `json:"version"`
-	maps.Copy(m, values)
+	Type      MessageType     `json:"type"`
+	RequestId *uint64         `json:"id,omitempty"`
+	Name      string          `json:"name"`
+	Body      json.RawMessage `json:"body,omitempty"`
+	Error     *ProtocolError  `json:"error,omitempty"`
+}
 
-	requestId = randString(8)
+func (m Message) Validate() error {
+	if m.Version != ProtocolVersion {
+		return fmt.Errorf("unsupported protocol version %d", m.Version)
+	}
 
-	m["cmd"] = cmd
+	if m.Name == "" {
-	m["request_id"] = requestId
+		return errors.New("message name is required")
+	}
 
-	b, err := json.Marshal(m)
+	switch m.Type {
+	case Request:
+		if m.RequestId == nil {
+			return errors.New("request ID is required")
+		}
+		if m.Error != nil {
+			return errors.New("request cannot contain an error")
+		}
+	case Response:
+		if m.RequestId == nil {
+			return errors.New("response ID is required")
+		}
+		if m.Error != nil && len(m.Body) != 0 {
+			return errors.New("response cannot contain both body and error")
+		}
+
+	case Event:
+		if m.RequestId != nil {
+			return errors.New("event cannot contain a request ID")
+		}
+		if m.Error != nil {
+			return errors.New("event cannot contain an error")
+		}
+	default:
+		return fmt.Errorf("unknown message type %q", m.Type)
+	}
+
+	return nil
+}
+
+type ProtocolError struct {
+	Code    string          `json:"code"`
+	Message string          `json:"message"`
+	Data    json.RawMessage `json:"data,omitempty"`
+}
+
+func (e *ProtocolError) Error() string {
+	if e == nil {
+		return ""
+	}
+
+	if e.Code == "" {
+		return e.Message
+	}
+
+	return fmt.Sprintf("%s: %s", e.Code, e.Message)
+}
+
+type SetBreakpointRequest struct {
+	File string `json:"file"`
+	Line int    `json:"line"`
+}
+
+type SetBreakpointResponse struct {
+	File     string `json:"file"`
+	Line     int    `json:"line"`
+	Verified bool   `json:"verified"`
+}
+
+type RemoveBreakpointRequest struct {
+	File string `json:"file"`
+	Line int    `json:"line"`
+}
+
+type LocalsResponse struct {
+	Variables map[string]string `json:"variables"`
+}
+
+type ExecutionStoppedEvent struct {
+	Reason string `json:"reason"`
+	File   string `json:"file"`
+	Line   int    `json:"line"`
+}
+
+type ExecutionExitedEvent struct {
+	ExitCode int `json:"exit_code"`
+}
+
+func newMessage(
+	messageType MessageType,
+	id *uint64,
+	name string,
+	body any,
+) (Message, error) {
+	var raw json.RawMessage
+
+	if body != nil {
+		encoded, err := json.Marshal(body)
 		if err != nil {
-		panic("failed to marshal command " + err.Error())
+			return Message{}, err
 		}
 
-	return requestId, string(b)
+		raw = encoded
 	}
 
-const letterBytes = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
+	return Message{
+		Version:   ProtocolVersion,
+		Type:      messageType,
+		RequestId: id,
+		Name:      name,
+		Body:      raw,
+	}, nil
+}
 
-func randString(n int) string {
+func newErrorResponse(
-	b := make([]byte, n)
+	id uint64,
-	for i := range b {
+	name string,
-		b[i] = letterBytes[rand.Intn(len(letterBytes))]
+	code string,
+	message string,
+) Message {
+	return Message{
+		Version:   ProtocolVersion,
+		Type:      Response,
+		RequestId: &id,
+		Name:      name,
+		Error: &ProtocolError{
+			Code:    code,
+			Message: message,
+		},
 	}
-	return string(b)
 }

internal/bridge/protocol_test.go

@@ -0,0 +1,214 @@
+package bridge
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+)
+
+func messageID(id uint64) *uint64 {
+	return &id
+}
+
+func TestMessageValidate(t *testing.T) {
+	tests := []struct {
+		name    string
+		message Message
+		wantErr string
+	}{
+		{
+			name: "request",
+			message: Message{
+				Version:   ProtocolVersion,
+				Type:      Request,
+				RequestId: messageID(1),
+				Name:      "locals/get",
+			},
+		},
+		{
+			name: "successful response",
+			message: Message{
+				Version:   ProtocolVersion,
+				Type:      Response,
+				RequestId: messageID(1),
+				Name:      "locals/get",
+				Body:      json.RawMessage(`{"variables":{}}`),
+			},
+		},
+		{
+			name: "error response",
+			message: Message{
+				Version:   ProtocolVersion,
+				Type:      Response,
+				RequestId: messageID(1),
+				Name:      "locals/get",
+				Error: &ProtocolError{
+					Code:    "not_paused",
+					Message: "execution is not paused",
+				},
+			},
+		},
+		{
+			name: "event",
+			message: Message{
+				Version: ProtocolVersion,
+				Type:    Event,
+				Name:    "execution/stopped",
+				Body:    json.RawMessage(`{"reason":"step","file":"test.py","line":1}`),
+			},
+		},
+		{
+			name: "unsupported version",
+			message: Message{
+				Version:   ProtocolVersion + 1,
+				Type:      Request,
+				RequestId: messageID(1),
+				Name:      "locals/get",
+			},
+			wantErr: "unsupported protocol version",
+		},
+		{
+			name: "missing name",
+			message: Message{
+				Version:   ProtocolVersion,
+				Type:      Request,
+				RequestId: messageID(1),
+			},
+			wantErr: "message name is required",
+		},
+		{
+			name: "request without ID",
+			message: Message{
+				Version: ProtocolVersion,
+				Type:    Request,
+				Name:    "locals/get",
+			},
+			wantErr: "request ID is required",
+		},
+		{
+			name: "request with error",
+			message: Message{
+				Version:   ProtocolVersion,
+				Type:      Request,
+				RequestId: messageID(1),
+				Name:      "locals/get",
+				Error:     &ProtocolError{Code: "invalid"},
+			},
+			wantErr: "request cannot contain an error",
+		},
+		{
+			name: "response without ID",
+			message: Message{
+				Version: ProtocolVersion,
+				Type:    Response,
+				Name:    "locals/get",
+			},
+			wantErr: "response ID is required",
+		},
+		{
+			name: "response with body and error",
+			message: Message{
+				Version:   ProtocolVersion,
+				Type:      Response,
+				RequestId: messageID(1),
+				Name:      "locals/get",
+				Body:      json.RawMessage(`{}`),
+				Error:     &ProtocolError{Code: "invalid"},
+			},
+			wantErr: "response cannot contain both body and error",
+		},
+		{
+			name: "event with ID",
+			message: Message{
+				Version:   ProtocolVersion,
+				Type:      Event,
+				RequestId: messageID(1),
+				Name:      "execution/stopped",
+			},
+			wantErr: "event cannot contain a request ID",
+		},
+		{
+			name: "event with error",
+			message: Message{
+				Version: ProtocolVersion,
+				Type:    Event,
+				Name:    "execution/stopped",
+				Error:   &ProtocolError{Code: "invalid"},
+			},
+			wantErr: "event cannot contain an error",
+		},
+		{
+			name: "unknown type",
+			message: Message{
+				Version:   ProtocolVersion,
+				Type:      MessageType("notification"),
+				RequestId: messageID(1),
+				Name:      "locals/get",
+			},
+			wantErr: "unknown message type",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := tt.message.Validate()
+			if tt.wantErr == "" {
+				if err != nil {
+					t.Fatalf("Validate() error = %v", err)
+				}
+				return
+			}
+
+			if err == nil {
+				t.Fatalf("Validate() error = nil, want error containing %q", tt.wantErr)
+			}
+			if !strings.Contains(err.Error(), tt.wantErr) {
+				t.Fatalf("Validate() error = %q, want error containing %q", err, tt.wantErr)
+			}
+		})
+	}
+}
+
+func TestNewMessageEncodesTypedBody(t *testing.T) {
+	id := uint64(7)
+	msg, err := newMessage(Request, &id, "breakpoint/set", SetBreakpointRequest{
+		File: "example.py",
+		Line: 42,
+	})
+	if err != nil {
+		t.Fatalf("newMessage() error = %v", err)
+	}
+
+	if err := msg.Validate(); err != nil {
+		t.Fatalf("Validate() error = %v", err)
+	}
+
+	var body SetBreakpointRequest
+	if err := json.Unmarshal(msg.Body, &body); err != nil {
+		t.Fatalf("unmarshal body: %v", err)
+	}
+
+	if body.File != "example.py" || body.Line != 42 {
+		t.Fatalf("decoded body = %+v", body)
+	}
+}
+
+func TestProtocolErrorString(t *testing.T) {
+	tests := []struct {
+		name string
+		err  *ProtocolError
+		want string
+	}{
+		{name: "code and message", err: &ProtocolError{Code: "not_paused", Message: "execution is not paused"}, want: "not_paused: execution is not paused"},
+		{name: "message only", err: &ProtocolError{Message: "execution is not paused"}, want: "execution is not paused"},
+		{name: "nil", err: nil, want: ""},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.err.Error(); got != tt.want {
+				t.Fatalf("Error() = %q, want %q", got, tt.want)
+			}
+		})
+	}
+}

python/pybug_runtime.py

@@ -1,23 +1,39 @@
 import bdb
 import json
+import os
 import sys
 from types import FrameType
 
+PROTOCOL_FD = int(os.environ["PYBUG_PROTOCOL_FD"])
+
 
 class PyBugBridgeDebugger(bdb.Bdb):
     def __init__(self):
         super().__init__()
+        self.protocol = os.fdopen(
+            PROTOCOL_FD, mode="w", encoding="utf-8", buffering=1, closefd=True
+        )
         self.waiting = False
 
     def send(self, msg: dict):
-        sys.stdout.write(json.dumps(msg) + "\n")
+        json.dump(msg, self.protocol, separators=(",", ":"))
-        sys.stdout.flush()
+        self.protocol.write("\n")
+        self.protocol.flush()
 
     def recv(self) -> dict:
-        return json.loads(sys.stdin.readline())
+        line = sys.stdin.readline()
+        if line == "":
+            raise EOFError("debugger command channel closed")
+        return json.loads(line)
 
     def user_line(self, frame: FrameType):
-        print("TRACE:", frame.f_code.co_filename, frame.f_lineno)
+        print(
+            "TRACE:",
+            frame.f_code.co_filename,
+            frame.f_lineno,
+            file=sys.stderr,
+            flush=True,
+        )
         self.send(
             {
                 "event": "stopped",