Add Claude Code stream-json client, report gate, and terminal driver

2026-10-13 22:01:12 -07:00
commit 555a55efc8
8 changed files with 1820 additions and 0 deletions
@@ -0,0 +1,525 @@
+package claude
+
+import (
+	"bufio"
+	"context"
+	"crypto/rand"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log/slog"
+	"os"
+	"os/exec"
+	"sync"
+	"time"
+
+	"g.fc.run/misc/byre/internal/mcp"
+)
+
+type HookFunc func(ctx context.Context, input json.RawMessage) (map[string]any, error)
+
+type HookMatcher struct {
+	Matcher string
+	Func    HookFunc
+}
+
+type Options struct {
+	Cwd         string
+	Model       string
+	Resume      string
+	ForkSession bool
+	GitName     string
+	GitEmail    string
+	Hooks       map[string][]HookMatcher
+	MCPServers  []*mcp.Server
+	Transcript  io.Writer
+}
+
+type Event struct {
+	Type    string
+	Subtype string
+	Raw     json.RawMessage
+}
+
+type Block struct {
+	Type    string          `json:"type"`
+	Text    string          `json:"text"`
+	Name    string          `json:"name"`
+	Input   json.RawMessage `json:"input"`
+	Content json.RawMessage `json:"content"`
+	IsError bool            `json:"is_error"`
+}
+
+type Result struct {
+	Subtype      string  `json:"subtype"`
+	IsError      bool    `json:"is_error"`
+	Result       string  `json:"result"`
+	DurationMS   int     `json:"duration_ms"`
+	NumTurns     int     `json:"num_turns"`
+	TotalCostUSD float64 `json:"total_cost_usd"`
+}
+
+func (e Event) Blocks() ([]Block, error) {
+	msg := struct {
+		Message struct {
+			Content json.RawMessage `json:"content"`
+		} `json:"message"`
+	}{}
+	err := json.Unmarshal(e.Raw, &msg)
+	if err != nil {
+		return nil, err
+	}
+
+	if len(msg.Message.Content) > 0 && msg.Message.Content[0] == '"' {
+		text := ""
+		err = json.Unmarshal(msg.Message.Content, &text)
+		if err != nil {
+			return nil, err
+		}
+		return []Block{{Type: "text", Text: text}}, nil
+	}
+
+	blocks := []Block{}
+	err = json.Unmarshal(msg.Message.Content, &blocks)
+	if err != nil {
+		return nil, err
+	}
+	return blocks, nil
+}
+
+func (e Event) Result() (Result, error) {
+	res := Result{}
+	err := json.Unmarshal(e.Raw, &res)
+	return res, err
+}
+
+type Session struct {
+	cmd    *exec.Cmd
+	stdin  io.WriteCloser
+	events chan Event
+	done   chan struct{}
+
+	writeMu    sync.Mutex
+	transcript io.Writer
+
+	pendingMu sync.Mutex
+	pending   map[string]chan controlResult
+
+	hooks   map[string]HookFunc
+	servers map[string]*mcp.Server
+
+	idMu sync.Mutex
+	id   string
+}
+
+type controlResult struct {
+	response json.RawMessage
+	err      error
+}
+
+type envelope struct {
+	Type      string          `json:"type"`
+	Subtype   string          `json:"subtype"`
+	SessionID string          `json:"session_id"`
+	RequestID string          `json:"request_id"`
+	Request   json.RawMessage `json:"request"`
+	Response  json.RawMessage `json:"response"`
+}
+
+func Start(ctx context.Context, opts Options) (*Session, error) {
+	s := &Session{
+		events:     make(chan Event, 256),
+		done:       make(chan struct{}),
+		transcript: opts.Transcript,
+		pending:    map[string]chan controlResult{},
+		hooks:      map[string]HookFunc{},
+		servers:    map[string]*mcp.Server{},
+	}
+
+	args, err := s.buildArgs(opts)
+	if err != nil {
+		return nil, err
+	}
+
+	s.cmd = exec.Command("claude", args...)
+	s.cmd.Dir = opts.Cwd
+	s.cmd.Stderr = os.Stderr
+	s.cmd.Env = buildEnv(opts)
+
+	s.stdin, err = s.cmd.StdinPipe()
+	if err != nil {
+		return nil, err
+	}
+
+	stdout, err := s.cmd.StdoutPipe()
+	if err != nil {
+		return nil, err
+	}
+
+	err = s.cmd.Start()
+	if err != nil {
+		return nil, fmt.Errorf("start claude: %w", err)
+	}
+
+	go s.readLoop(stdout)
+
+	_, err = s.Control(ctx, map[string]any{
+		"subtype": "initialize",
+		"hooks":   s.hookConfig(opts.Hooks),
+	})
+	if err != nil {
+		s.Close()
+		return nil, fmt.Errorf("initialize: %w", err)
+	}
+
+	return s, nil
+}
+
+func (s *Session) buildArgs(opts Options) ([]string, error) {
+	settings, err := json.Marshal(map[string]any{
+		"attribution": map[string]any{"commit": "", "pr": ""},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	servers := map[string]any{}
+	for _, srv := range opts.MCPServers {
+		s.servers[srv.Name] = srv
+		servers[srv.Name] = map[string]any{"type": "sdk", "name": srv.Name}
+	}
+
+	mcpConfig, err := json.Marshal(map[string]any{"mcpServers": servers})
+	if err != nil {
+		return nil, err
+	}
+
+	args := []string{
+		"--output-format", "stream-json",
+		"--input-format", "stream-json",
+		"--verbose",
+		"--permission-mode", "bypassPermissions",
+		"--allow-dangerously-skip-permissions",
+		"--settings", string(settings),
+		"--mcp-config", string(mcpConfig),
+	}
+
+	if opts.Model != "" {
+		args = append(args, "--model", opts.Model)
+	}
+
+	if opts.Resume != "" {
+		args = append(args, "--resume="+opts.Resume)
+	}
+
+	if opts.ForkSession {
+		args = append(args, "--fork-session")
+	}
+
+	return args, nil
+}
+
+func buildEnv(opts Options) []string {
+	env := []string{}
+	for _, kv := range os.Environ() {
+		if len(kv) >= 11 && kv[:11] == "CLAUDECODE=" {
+			continue
+		}
+		env = append(env, kv)
+	}
+
+	return append(env,
+		"CLAUDE_CODE_ENTRYPOINT=sdk-go",
+		"CLAUDE_CODE_DISABLE_AUTOUPDATER=1",
+		"GIT_AUTHOR_NAME="+opts.GitName,
+		"GIT_AUTHOR_EMAIL="+opts.GitEmail,
+		"GIT_COMMITTER_NAME="+opts.GitName,
+		"GIT_COMMITTER_EMAIL="+opts.GitEmail,
+	)
+}
+
+func (s *Session) hookConfig(hooks map[string][]HookMatcher) map[string]any {
+	if len(hooks) == 0 {
+		return nil
+	}
+
+	cfg := map[string]any{}
+	n := 0
+	for event, matchers := range hooks {
+		list := []map[string]any{}
+		for _, m := range matchers {
+			id := fmt.Sprintf("hook_%d", n)
+			n++
+			s.hooks[id] = m.Func
+			list = append(list, map[string]any{
+				"matcher":         m.Matcher,
+				"hookCallbackIds": []string{id},
+			})
+		}
+		cfg[event] = list
+	}
+
+	return cfg
+}
+
+func (s *Session) Events() <-chan Event {
+	return s.events
+}
+
+func (s *Session) Done() <-chan struct{} {
+	return s.done
+}
+
+func (s *Session) ID() string {
+	s.idMu.Lock()
+	defer s.idMu.Unlock()
+	return s.id
+}
+
+func (s *Session) Send(text string) error {
+	return s.write(map[string]any{
+		"type":               "user",
+		"message":            map[string]any{"role": "user", "content": text},
+		"parent_tool_use_id": nil,
+		"session_id":         s.ID(),
+	})
+}
+
+func (s *Session) Interrupt(ctx context.Context) error {
+	_, err := s.Control(ctx, map[string]any{"subtype": "interrupt"})
+	return err
+}
+
+func (s *Session) Control(ctx context.Context, req map[string]any) (json.RawMessage, error) {
+	id := newRequestID()
+	ch := make(chan controlResult, 1)
+
+	s.pendingMu.Lock()
+	s.pending[id] = ch
+	s.pendingMu.Unlock()
+
+	defer func() {
+		s.pendingMu.Lock()
+		delete(s.pending, id)
+		s.pendingMu.Unlock()
+	}()
+
+	err := s.write(map[string]any{
+		"type":       "control_request",
+		"request_id": id,
+		"request":    req,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	select {
+	case res := <-ch:
+		return res.response, res.err
+	case <-s.done:
+		return nil, errors.New("claude exited")
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+}
+
+func (s *Session) Close() error {
+	s.writeMu.Lock()
+	s.stdin.Close()
+	s.writeMu.Unlock()
+
+	select {
+	case <-s.done:
+	case <-time.After(5 * time.Second):
+		s.cmd.Process.Kill()
+		<-s.done
+	}
+
+	return s.cmd.Wait()
+}
+
+func (s *Session) write(v any) error {
+	b, err := json.Marshal(v)
+	if err != nil {
+		return err
+	}
+
+	s.writeMu.Lock()
+	defer s.writeMu.Unlock()
+
+	s.record("send", b)
+
+	_, err = s.stdin.Write(append(b, '\n'))
+	return err
+}
+
+func (s *Session) record(dir string, line []byte) {
+	if s.transcript == nil {
+		return
+	}
+
+	b, err := json.Marshal(map[string]any{"dir": dir, "msg": json.RawMessage(line)})
+	if err != nil {
+		slog.Error("record transcript", "err", err)
+		return
+	}
+
+	s.transcript.Write(append(b, '\n'))
+}
+
+func (s *Session) readLoop(stdout io.Reader) {
+	defer close(s.done)
+	defer close(s.events)
+
+	r := bufio.NewReader(stdout)
+	for {
+		line, err := r.ReadBytes('\n')
+		if len(line) > 1 && line[0] == '{' {
+			s.handleLine(line[:len(line)-1])
+		}
+		if err != nil {
+			if err != io.EOF {
+				slog.Error("read claude stdout", "err", err)
+			}
+			return
+		}
+	}
+}
+
+func (s *Session) handleLine(line []byte) {
+	s.writeMu.Lock()
+	s.record("recv", line)
+	s.writeMu.Unlock()
+
+	env := envelope{}
+	err := json.Unmarshal(line, &env)
+	if err != nil {
+		slog.Error("parse claude message", "err", err, "line", string(line))
+		return
+	}
+
+	switch env.Type {
+	case "control_response":
+		s.handleControlResponse(env.Response)
+		return
+
+	case "control_request":
+		go s.handleControlRequest(env.RequestID, env.Request)
+		return
+
+	case "control_cancel_request", "keep_alive":
+		return
+	}
+
+	if env.Type == "system" && env.Subtype == "init" {
+		s.idMu.Lock()
+		s.id = env.SessionID
+		s.idMu.Unlock()
+	}
+
+	s.events <- Event{Type: env.Type, Subtype: env.Subtype, Raw: append(json.RawMessage{}, line...)}
+}
+
+func (s *Session) handleControlResponse(raw json.RawMessage) {
+	resp := struct {
+		Subtype   string          `json:"subtype"`
+		RequestID string          `json:"request_id"`
+		Response  json.RawMessage `json:"response"`
+		Error     string          `json:"error"`
+	}{}
+	err := json.Unmarshal(raw, &resp)
+	if err != nil {
+		slog.Error("parse control response", "err", err)
+		return
+	}
+
+	s.pendingMu.Lock()
+	ch := s.pending[resp.RequestID]
+	s.pendingMu.Unlock()
+
+	if ch == nil {
+		slog.Warn("control response for unknown request", "request_id", resp.RequestID)
+		return
+	}
+
+	if resp.Subtype == "error" {
+		ch <- controlResult{err: errors.New(resp.Error)}
+		return
+	}
+
+	ch <- controlResult{response: resp.Response}
+}
+
+func (s *Session) handleControlRequest(id string, raw json.RawMessage) {
+	result, err := s.dispatchControl(raw)
+	if err != nil {
+		slog.Error("control request failed", "request_id", id, "err", err)
+		s.write(map[string]any{
+			"type": "control_response",
+			"response": map[string]any{
+				"subtype":    "error",
+				"request_id": id,
+				"error":      err.Error(),
+			},
+		})
+		return
+	}
+
+	s.write(map[string]any{
+		"type": "control_response",
+		"response": map[string]any{
+			"subtype":    "success",
+			"request_id": id,
+			"response":   result,
+		},
+	})
+}
+
+func (s *Session) dispatchControl(raw json.RawMessage) (any, error) {
+	req := struct {
+		Subtype    string          `json:"subtype"`
+		CallbackID string          `json:"callback_id"`
+		Input      json.RawMessage `json:"input"`
+		ServerName string          `json:"server_name"`
+		Message    json.RawMessage `json:"message"`
+	}{}
+	err := json.Unmarshal(raw, &req)
+	if err != nil {
+		return nil, err
+	}
+
+	ctx := context.Background()
+
+	switch req.Subtype {
+	case "hook_callback":
+		fn := s.hooks[req.CallbackID]
+		if fn == nil {
+			return nil, fmt.Errorf("no hook callback %s", req.CallbackID)
+		}
+		return fn(ctx, req.Input)
+
+	case "mcp_message":
+		srv := s.servers[req.ServerName]
+		if srv == nil {
+			return nil, fmt.Errorf("no mcp server %s", req.ServerName)
+		}
+		resp, err := srv.Handle(ctx, req.Message)
+		if err != nil {
+			return nil, err
+		}
+		return map[string]any{"mcp_response": resp}, nil
+
+	case "can_use_tool":
+		return map[string]any{"behavior": "allow", "updatedInput": req.Input}, nil
+	}
+
+	return nil, fmt.Errorf("unsupported control request %s", req.Subtype)
+}
+
+func newRequestID() string {
+	b := make([]byte, 8)
+	rand.Read(b)
+	return "req_" + hex.EncodeToString(b)
+}