claude_session/playground/web/server.go

// Command server is a browser chat over a single hot claude_session daemon. Unlike
// the claude_pipe/claude_wire chats (which spawn a fresh one-shot process per
// message), this server launches ONE claude_session at boot and keeps it alive, so
// the conversation has memory across turns and each message answers in ~2-3s.
//
// Per message it writes {"cmd":"send","prompt":...} to the daemon's stdin and
// relays the daemon's NDJSON (text_delta + result) to the browser as SSE. A
// /restart endpoint sends {"cmd":"restart"} to start a fresh conversation.
//
// The daemon's stdout is read only while holding a mutex, so requests are
// serialized (a chat is sequential anyway).
//
// Run:
//
//	cd apps/claude_session && go build -o claude_session .
//	cd playground && go run ./web      # http://localhost:8101
package main

import (
	"bufio"
	_ "embed"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"sync"
	"time"
)

//go:embed index.html
var indexHTML []byte

type daemonEvent struct {
	Type    string `json:"type"`
	Text    string `json:"text"`
	Result  string `json:"result"`
	Message string `json:"message"`
}

var (
	mu      sync.Mutex
	stdin   io.Writer
	stdout  *bufio.Scanner
	session *exec.Cmd
)

func main() {
	port := flag.String("port", "8101", "port to listen on")
	bin := flag.String("session", "/home/enmanuel/fn_registry/apps/claude_session/claude_session", "claude_session binary")
	cwd := flag.String("cwd", "/home/enmanuel/fn_registry", "cwd for the claude session")
	flag.Parse()

	abs, err := filepath.Abs(*bin)
	if err != nil || !fileExists(abs) {
		log.Fatalf("claude_session binary not found at %s — build it first (cd .. && go build -o claude_session .)", abs)
	}

	if err := startDaemon(abs, *cwd); err != nil {
		log.Fatalf("start daemon: %v", err)
	}

	http.HandleFunc("/", handleIndex)
	http.HandleFunc("/chat", handleChat)
	http.HandleFunc("/restart", handleRestart)

	log.Printf("claude_session web chat → http://localhost:%s  (daemon=%s cwd=%s)", *port, abs, *cwd)
	log.Fatal(http.ListenAndServe(":"+*port, nil))
}

// startDaemon launches claude_session and blocks until its initial {"type":"ready"}.
func startDaemon(bin, cwd string) error {
	cmd := exec.Command(bin, "--cwd", cwd)
	cmd.Stderr = os.Stderr
	in, err := cmd.StdinPipe()
	if err != nil {
		return err
	}
	out, err := cmd.StdoutPipe()
	if err != nil {
		return err
	}
	if err := cmd.Start(); err != nil {
		return err
	}
	sc := bufio.NewScanner(out)
	sc.Buffer(make([]byte, 1024*1024), 1024*1024)
	stdin, stdout, session = in, sc, cmd

	log.Printf("waiting for claude session to warm up...")
	for sc.Scan() {
		var ev daemonEvent
		if json.Unmarshal(sc.Bytes(), &ev) == nil && ev.Type == "ready" {
			log.Printf("claude session ready")
			return nil
		}
	}
	return fmt.Errorf("daemon exited before ready")
}

func handleIndex(w http.ResponseWriter, r *http.Request) {
	if r.URL.Path != "/" {
		http.NotFound(w, r)
		return
	}
	w.Header().Set("Content-Type", "text/html; charset=utf-8")
	_, _ = w.Write(indexHTML)
}

// handleChat sends one prompt to the live daemon and relays its events as SSE.
func handleChat(w http.ResponseWriter, r *http.Request) {
	prompt := r.URL.Query().Get("prompt")
	if prompt == "" {
		http.Error(w, "missing prompt", http.StatusBadRequest)
		return
	}
	flusher, ok := w.(http.Flusher)
	if !ok {
		http.Error(w, "streaming unsupported", http.StatusInternalServerError)
		return
	}
	w.Header().Set("Content-Type", "text/event-stream")
	w.Header().Set("Cache-Control", "no-cache")
	w.Header().Set("X-Accel-Buffering", "no")

	mu.Lock()
	defer mu.Unlock()

	cmd, _ := json.Marshal(map[string]string{"cmd": "send", "prompt": prompt})
	if _, err := fmt.Fprintln(stdin, string(cmd)); err != nil {
		sse(w, flusher, "error", fmt.Sprintf(`{"message":%q}`, err.Error()))
		return
	}

	// Relay daemon events until result, then consume the trailing ready.
	gotResult := false
	for stdout.Scan() {
		line := stdout.Bytes()
		var ev daemonEvent
		if json.Unmarshal(line, &ev) != nil {
			continue
		}
		switch ev.Type {
		case "text_delta", "result", "error":
			sse(w, flusher, "", string(line))
			if ev.Type == "result" || ev.Type == "error" {
				gotResult = true
			}
		case "ready":
			if gotResult {
				sse(w, flusher, "done", "{}")
				return
			}
		}
	}
	sse(w, flusher, "error", `{"message":"daemon stream ended"}`)
}

// handleRestart tells the daemon to start a fresh conversation.
func handleRestart(w http.ResponseWriter, r *http.Request) {
	mu.Lock()
	defer mu.Unlock()

	if _, err := fmt.Fprintln(stdin, `{"cmd":"restart"}`); err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}
	deadline := time.Now().Add(30 * time.Second)
	for stdout.Scan() {
		var ev daemonEvent
		if json.Unmarshal(stdout.Bytes(), &ev) == nil && ev.Type == "ready" {
			w.WriteHeader(http.StatusOK)
			_, _ = w.Write([]byte(`{"ok":true}`))
			return
		}
		if time.Now().After(deadline) {
			break
		}
	}
	http.Error(w, "restart timeout", http.StatusGatewayTimeout)
}

func sse(w http.ResponseWriter, f http.Flusher, event, data string) {
	if event != "" {
		fmt.Fprintf(w, "event: %s\n", event)
	}
	fmt.Fprintf(w, "data: %s\n\n", data)
	f.Flush()
}

func fileExists(p string) bool {
	_, err := os.Stat(p)
	return err == nil
}