update build

shellctl-go/Makefile

@@ -0,0 +1,39 @@
+.PHONY: build clean test lint integration-up integration-test integration-down integration
+
+BIN_DIR := bin
+
+build: $(BIN_DIR)/shellctl $(BIN_DIR)/shellctl-sanitize-pty $(BIN_DIR)/shellctl-runner-exit
+
+$(BIN_DIR)/shellctl: $(shell find cmd/shellctl internal -name '*.go')
+	go build -o $@ ./cmd/shellctl
+
+$(BIN_DIR)/shellctl-sanitize-pty: $(shell find cmd/sanitize-pty internal/sanitize -name '*.go')
+	go build -o $@ ./cmd/sanitize-pty
+
+$(BIN_DIR)/shellctl-runner-exit: $(shell find cmd/runner-exit internal/runner_exit -name '*.go')
+	go build -o $@ ./cmd/runner-exit
+
+test:
+	go test ./...
+
+lint:
+	golangci-lint run ./...
+
+clean:
+	rm -rf $(BIN_DIR)
+
+# --- Integration tests (Docker) ---
+
+integration-up:
+	docker compose -f tests/docker-compose.yml up --build -d
+	@echo "Waiting for services to be healthy..."
+	@sleep 5
+
+integration-test:
+	go test -tags=integration -v -count=1 ./tests/...
+
+integration-down:
+	docker compose -f tests/docker-compose.yml down -v
+
+# Full cycle: up → test → down
+integration: integration-up integration-test integration-down

shellctl-go/README.md

@@ -0,0 +1,62 @@
+# shellctl-go
+
+Go implementation of the shellctl server and runtime utilities.
+
+This is a rewrite of the Python `shellctl` package (`dify-agent/src/shellctl/` and
+`dify-agent/src/shellctl_runtime/`). The original Python code is kept as reference.
+
+## Architecture
+
+```
+cmd/
+  shellctl/          — main server binary (shellctl serve)
+  sanitize-pty/      — tmux pipe-pane PTY sanitizer (stdin→stdout filter)
+  runner-exit/       — post-drain SQLite exit recorder
+
+internal/
+  sanitize/          — PTY ANSI stripping + CR normalization
+  runner_exit/       — SQLite CAS update for job exit
+  server/            — HTTP API, job service, tmux controller, output reader
+```
+
+## Building
+
+```bash
+make build
+```
+
+Produces three binaries in `bin/`:
+- `shellctl` — the main server (`shellctl serve --listen 0.0.0.0:5004`)
+- `shellctl-sanitize-pty` — PTY sanitizer for tmux pipe-pane
+- `shellctl-runner-exit` — exit state writer
+
+
+### Building docker image
+
+```
+docker build -f shellctl-go/docker/Dockerfile \
+  --build-context agent=./dify-agent \
+  -t shellctl-go:latest \
+  shellctl-go/
+```
+
+## Testing
+
+```bash
+make test
+```
+
+## Dependencies
+
+- Go 1.23+
+- `modernc.org/sqlite` (pure-Go SQLite driver, no CGO required)
+- tmux (runtime dependency, not a build dependency)
+
+## Migration from Python
+
+The Go binaries are drop-in replacements for the Python console scripts:
+- `shellctl-sanitize-pty` replaces the Python `shellctl-sanitize-pty` entrypoint
+- `shellctl-runner-exit` replaces the Python `shellctl-runner-exit` entrypoint
+- `shellctl serve` replaces the Python `shellctl serve` (FastAPI/uvicorn)
+
+The HTTP API contract, SQLite schema, and filesystem artifact layout are identical.

shellctl-go/cmd/runner-exit/main.go

@@ -0,0 +1,30 @@
+// shellctl-runner-exit persists a drained runner exit into the shellctl SQLite DB.
+// It runs after the tmux output pipe reaches EOF and output.log is fully flushed.
+package main
+
+import (
+	"flag"
+	"fmt"
+	"os"
+
+	runnerexit "github.com/langgenius/dify/shellctl-go/internal/runner_exit"
+)
+
+func main() {
+	stateDir := flag.String("state-dir", "", "shellctl state directory")
+	jobID := flag.String("job-id", "", "job identifier")
+	exitCode := flag.Int("exit-code", 0, "runner process exit code")
+	endedAt := flag.String("ended-at", "", "ISO-8601 ended_at timestamp")
+	busyTimeoutMs := flag.Int("sqlite-busy-timeout-ms", 5000, "SQLite busy timeout in milliseconds")
+	flag.Parse()
+
+	if *stateDir == "" || *jobID == "" || *endedAt == "" {
+		fmt.Fprintf(os.Stderr, "shellctl-runner-exit: --state-dir, --job-id, and --ended-at are required\n")
+		os.Exit(1)
+	}
+
+	if err := runnerexit.RecordRunnerExit(*stateDir, *jobID, *exitCode, *endedAt, *busyTimeoutMs); err != nil {
+		fmt.Fprintf(os.Stderr, "shellctl-runner-exit: %v\n", err)
+		os.Exit(1)
+	}
+}

shellctl-go/cmd/sanitize-pty/main.go

@@ -0,0 +1,22 @@
+// shellctl-sanitize-pty is a stdin→stdout PTY sanitizer used by tmux pipe-pane.
+// It strips ANSI escape sequences, normalizes carriage-return progress lines,
+// and performs incremental UTF-8 decoding.
+package main
+
+import (
+	"flag"
+	"fmt"
+	"os"
+
+	"github.com/langgenius/dify/shellctl-go/internal/sanitize"
+)
+
+func main() {
+	readyFile := flag.String("ready-file", "", "path to touch before reading stdin")
+	flag.Parse()
+
+	if err := sanitize.Run(*readyFile, os.Stdin, os.Stdout); err != nil {
+		fmt.Fprintf(os.Stderr, "shellctl-sanitize-pty: %v\n", err)
+		os.Exit(1)
+	}
+}

shellctl-go/cmd/shellctl/main.go

@@ -0,0 +1,87 @@
+// shellctl is the main server binary that exposes a REST API for managing
+// tmux-backed shell jobs inside a sandbox container.
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/langgenius/dify/shellctl-go/internal/server"
+)
+
+func main() {
+	if err := run(); err != nil {
+		fmt.Fprintf(os.Stderr, "shellctl: %v\n", err)
+		os.Exit(1)
+	}
+}
+
+func run() error {
+	// Parse CLI flags
+	serveCmd := flag.NewFlagSet("serve", flag.ExitOnError)
+	listen := serveCmd.String("listen", "", "address to listen on (host:port)")
+	stateDir := serveCmd.String("state-dir", "", "state directory path")
+	token := serveCmd.String("token", "", "bearer auth token")
+
+	if len(os.Args) < 2 || os.Args[1] != "serve" {
+		fmt.Fprintf(os.Stderr, "Usage: shellctl serve [flags]\n")
+		return nil
+	}
+	serveCmd.Parse(os.Args[2:])
+
+	// Build config
+	config := server.DefaultConfig()
+	if *listen != "" {
+		config.Listen = *listen
+	}
+	if *stateDir != "" {
+		config.StateDir = *stateDir
+	}
+	if *token != "" {
+		config.AuthToken = *token
+	}
+
+	// Initialize service
+	svc := server.NewService(config)
+	if err := svc.Initialize(); err != nil {
+		return fmt.Errorf("initialize: %w", err)
+	}
+	defer svc.Shutdown()
+
+	svc.StartBackgroundGC()
+	svc.StartBackgroundPipeMonitor()
+
+	// Create HTTP server
+	handler := server.Handler(svc, config)
+	srv := &http.Server{
+		Addr:         config.Listen,
+		Handler:      handler,
+		ReadTimeout:  0, // Long-poll requires no read timeout
+		WriteTimeout: 0,
+	}
+
+	// Graceful shutdown on signal
+	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer stop()
+
+	go func() {
+		<-ctx.Done()
+		log.Println("shutting down...")
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		srv.Shutdown(shutdownCtx)
+	}()
+
+	log.Printf("shellctl serving on %s", config.Listen)
+	if err := srv.ListenAndServe(); err != http.ErrServerClosed {
+		return err
+	}
+	return nil
+}

shellctl-go/docker/Dockerfile

@@ -0,0 +1,123 @@
+# Go shellctl sandbox image — runtime environment mirrors the Python
+# local-sandbox (dify-agent/docker/local-sandbox/Dockerfile) so that
+# agent-managed jobs see the same binaries, libraries, and tools.
+#
+# Build from shellctl-go root:
+#   docker build -f docker/Dockerfile -t dify-agent-local-sandbox:local .
+#
+# If you also need the dify-agent CLI tools inside the container, run a
+# two-context build (see comments in the dify-agent tools stage below).
+
+# ── Go build stage ───────────────────────────────────────────────────────────
+FROM golang:1.26 AS go-builder
+
+WORKDIR /src
+COPY go.mod go.sum ./
+RUN go mod download
+COPY . .
+RUN CGO_ENABLED=0 go build -o /bin/shellctl ./cmd/shellctl && \
+    CGO_ENABLED=0 go build -o /bin/shellctl-sanitize-pty ./cmd/sanitize-pty && \
+    CGO_ENABLED=0 go build -o /bin/shellctl-runner-exit ./cmd/runner-exit
+
+# ── dify-agent tools stage (optional, requires --build-context agent=../dify-agent) ──
+# To include dify-agent CLI: build with
+#   docker build -f docker/Dockerfile \
+#     --build-context agent=../dify-agent \
+#     -t dify-agent-local-sandbox:local .
+FROM python:3.12-slim-bookworm AS base
+
+ARG NODE_VERSION=22.22.1
+ARG PNPM_VERSION=11.9.0
+ARG UV_VERSION=0.8.9
+ARG DIFY_AGENT_TOOL_SPEC=.[grpc,shellctl-server]
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1
+
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends \
+        bash \
+        ca-certificates \
+        curl \
+        file \
+        git \
+        jq \
+        less \
+        openssh-client \
+        procps \
+        ripgrep \
+        tmux \
+        unzip \
+        xz-utils \
+        zip \
+    && node_arch="$(dpkg --print-architecture)" \
+    && case "${node_arch}" in \
+        amd64) node_arch="x64" ;; \
+        arm64) node_arch="arm64" ;; \
+        *) echo "Unsupported Node.js architecture: ${node_arch}" >&2; exit 1 ;; \
+    esac \
+    && node_dist="node-v${NODE_VERSION}-linux-${node_arch}" \
+    && curl -fsSLO "https://nodejs.org/dist/v${NODE_VERSION}/SHASUMS256.txt" \
+    && curl -fsSLO "https://nodejs.org/dist/v${NODE_VERSION}/${node_dist}.tar.xz" \
+    && grep " ${node_dist}.tar.xz\$" SHASUMS256.txt | sha256sum -c - \
+    && tar -xJf "${node_dist}.tar.xz" -C /usr/local --strip-components=1 \
+    && rm -f SHASUMS256.txt "${node_dist}.tar.xz" \
+    && npm install --global "pnpm@${PNPM_VERSION}" \
+    && npm cache clean --force \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN python -m pip install --no-cache-dir "uv==${UV_VERSION}"
+
+WORKDIR /opt/dify-agent
+
+
+FROM base AS tools
+
+ARG DIFY_AGENT_TOOL_SPEC
+
+# Copy dify-agent source from the build context named "agent" if available.
+# When building without --build-context agent=..., this stage can be skipped
+# by targeting the "production" stage directly.
+COPY --from=agent pyproject.toml uv.lock README.md ./
+COPY --from=agent src ./src
+
+RUN uv export --frozen --no-dev --all-extras --no-emit-project --no-hashes \
+        > /tmp/dify-agent-constraints.txt \
+    && UV_TOOL_DIR=/opt/dify-agent-tools/envs \
+        UV_TOOL_BIN_DIR=/opt/dify-agent-tools/bin \
+        uv tool install --force --python /usr/local/bin/python --no-python-downloads \
+        --constraints /tmp/dify-agent-constraints.txt --link-mode=copy "${DIFY_AGENT_TOOL_SPEC}" \
+    && rm -f /tmp/dify-agent-constraints.txt
+
+
+# ── Production stage ─────────────────────────────────────────────────────────
+FROM base AS production
+
+ENV PATH="/opt/dify-agent-tools/bin:${PATH}"
+
+# dify-agent CLI tools (from tools stage)
+COPY --from=tools /opt/dify-agent-tools/envs /opt/dify-agent-tools/envs
+COPY --from=tools /opt/dify-agent-tools/bin /opt/dify-agent-tools/bin
+
+# Go shellctl binaries (replace Python shellctl-server)
+COPY --from=go-builder /bin/shellctl /usr/local/bin/shellctl
+COPY --from=go-builder /bin/shellctl-sanitize-pty /usr/local/bin/shellctl-sanitize-pty
+COPY --from=go-builder /bin/shellctl-runner-exit /usr/local/bin/shellctl-runner-exit
+
+# Remove Python shellctl console scripts so the Go binaries take precedence
+RUN rm -f /opt/dify-agent-tools/bin/shellctl \
+          /opt/dify-agent-tools/bin/shellctl-sanitize-pty \
+          /opt/dify-agent-tools/bin/shellctl-runner-exit
+
+RUN useradd --create-home --shell /bin/sh dify \
+    && mkdir -p /mnt/drive \
+    && chown dify:dify /home \
+    && chown -R dify:dify /home/dify /mnt/drive
+
+USER dify
+WORKDIR /home/dify
+
+EXPOSE 5004
+
+CMD ["shellctl", "serve", "--listen", "0.0.0.0:5004"]

shellctl-go/go.mod

@@ -0,0 +1,18 @@
+module github.com/langgenius/dify/shellctl-go
+
+go 1.26
+
+require modernc.org/sqlite v1.37.1
+
+require (
+	github.com/dustin/go-humanize v1.0.1 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/ncruces/go-strftime v0.1.9 // indirect
+	github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
+	golang.org/x/exp v0.0.0-20250408133849-7e4ce0ab07d0 // indirect
+	golang.org/x/sys v0.33.0 // indirect
+	modernc.org/libc v1.65.7 // indirect
+	modernc.org/mathutil v1.7.1 // indirect
+	modernc.org/memory v1.11.0 // indirect
+)

shellctl-go/go.sum

@@ -0,0 +1,47 @@
+github.com/dustin/go-humanize v1.0.1 h1:GzkhY7T5VNhEkwH0PVJgjz+fX1rhBrR7pRT3mDkpeCY=
+github.com/dustin/go-humanize v1.0.1/go.mod h1:Mu1zIs6XwVuF/gI1OepvI0qD18qycQx+mFykh5fBlto=
+github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs=
+github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/mattn/go-isatty v0.0.20 h1:xfD0iDuEKnDkl03q4limB+vH+GxLEtL/jb4xVJSWWEY=
+github.com/mattn/go-isatty v0.0.20/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
+github.com/ncruces/go-strftime v0.1.9 h1:bY0MQC28UADQmHmaF5dgpLmImcShSi2kHU9XLdhx/f4=
+github.com/ncruces/go-strftime v0.1.9/go.mod h1:Fwc5htZGVVkseilnfgOVb9mKy6w1naJmn9CehxcKcls=
+github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE=
+github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo=
+golang.org/x/exp v0.0.0-20250408133849-7e4ce0ab07d0 h1:R84qjqJb5nVJMxqWYb3np9L5ZsaDtB+a39EqjV0JSUM=
+golang.org/x/exp v0.0.0-20250408133849-7e4ce0ab07d0/go.mod h1:S9Xr4PYopiDyqSyp5NjCrhFrqg6A5zA2E/iPHPhqnS8=
+golang.org/x/mod v0.24.0 h1:ZfthKaKaT4NrhGVZHO1/WDTwGES4De8KtWO0SIbNJMU=
+golang.org/x/mod v0.24.0/go.mod h1:IXM97Txy2VM4PJ3gI61r1YEk/gAj6zAHN3AdZt6S9Ww=
+golang.org/x/sync v0.14.0 h1:woo0S4Yywslg6hp4eUFjTVOyKt0RookbpAHG4c1HmhQ=
+golang.org/x/sync v0.14.0/go.mod h1:1dzgHSNfp02xaA81J2MS99Qcpr2w7fw1gpm99rleRqA=
+golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.33.0 h1:q3i8TbbEz+JRD9ywIRlyRAQbM0qF7hu24q3teo2hbuw=
+golang.org/x/sys v0.33.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k=
+golang.org/x/tools v0.33.0 h1:4qz2S3zmRxbGIhDIAgjxvFutSvH5EfnsYrRBj0UI0bc=
+golang.org/x/tools v0.33.0/go.mod h1:CIJMaWEY88juyUfo7UbgPqbC8rU2OqfAV1h2Qp0oMYI=
+modernc.org/cc/v4 v4.26.1 h1:+X5NtzVBn0KgsBCBe+xkDC7twLb/jNVj9FPgiwSQO3s=
+modernc.org/cc/v4 v4.26.1/go.mod h1:uVtb5OGqUKpoLWhqwNQo/8LwvoiEBLvZXIQ/SmO6mL0=
+modernc.org/ccgo/v4 v4.28.0 h1:rjznn6WWehKq7dG4JtLRKxb52Ecv8OUGah8+Z/SfpNU=
+modernc.org/ccgo/v4 v4.28.0/go.mod h1:JygV3+9AV6SmPhDasu4JgquwU81XAKLd3OKTUDNOiKE=
+modernc.org/fileutil v1.3.1 h1:8vq5fe7jdtEvoCf3Zf9Nm0Q05sH6kGx0Op2CPx1wTC8=
+modernc.org/fileutil v1.3.1/go.mod h1:HxmghZSZVAz/LXcMNwZPA/DRrQZEVP9VX0V4LQGQFOc=
+modernc.org/gc/v2 v2.6.5 h1:nyqdV8q46KvTpZlsw66kWqwXRHdjIlJOhG6kxiV/9xI=
+modernc.org/gc/v2 v2.6.5/go.mod h1:YgIahr1ypgfe7chRuJi2gD7DBQiKSLMPgBQe9oIiito=
+modernc.org/libc v1.65.7 h1:Ia9Z4yzZtWNtUIuiPuQ7Qf7kxYrxP1/jeHZzG8bFu00=
+modernc.org/libc v1.65.7/go.mod h1:011EQibzzio/VX3ygj1qGFt5kMjP0lHb0qCW5/D/pQU=
+modernc.org/mathutil v1.7.1 h1:GCZVGXdaN8gTqB1Mf/usp1Y/hSqgI2vAGGP4jZMCxOU=
+modernc.org/mathutil v1.7.1/go.mod h1:4p5IwJITfppl0G4sUEDtCr4DthTaT47/N3aT6MhfgJg=
+modernc.org/memory v1.11.0 h1:o4QC8aMQzmcwCK3t3Ux/ZHmwFPzE6hf2Y5LbkRs+hbI=
+modernc.org/memory v1.11.0/go.mod h1:/JP4VbVC+K5sU2wZi9bHoq2MAkCnrt2r98UGeSK7Mjw=
+modernc.org/opt v0.1.4 h1:2kNGMRiUjrp4LcaPuLY2PzUfqM/w9N23quVwhKt5Qm8=
+modernc.org/opt v0.1.4/go.mod h1:03fq9lsNfvkYSfxrfUhZCWPk1lm4cq4N+Bh//bEtgns=
+modernc.org/sortutil v1.2.1 h1:+xyoGf15mM3NMlPDnFqrteY07klSFxLElE2PVuWIJ7w=
+modernc.org/sortutil v1.2.1/go.mod h1:7ZI3a3REbai7gzCLcotuw9AC4VZVpYMjDzETGsSMqJE=
+modernc.org/sqlite v1.37.1 h1:EgHJK/FPoqC+q2YBXg7fUmES37pCHFc97sI7zSayBEs=
+modernc.org/sqlite v1.37.1/go.mod h1:XwdRtsE1MpiBcL54+MbKcaDvcuej+IYSMfLN6gSKV8g=
+modernc.org/strutil v1.2.1 h1:UneZBkQA+DX2Rp35KcM69cSsNES9ly8mQWD71HKlOA0=
+modernc.org/strutil v1.2.1/go.mod h1:EHkiggD70koQxjVdSBM3JKM7k6L0FbGE5eymy9i3B9A=
+modernc.org/token v1.1.0 h1:Xl7Ap9dKaEs5kLoOQeQmPWevfnk/DM5qcLcYlA8ys6Y=
+modernc.org/token v1.1.0/go.mod h1:UGzOrNV1mAFSEB63lOFHIpNRUVMvYTc6yu1SMY/XTDM=

shellctl-go/internal/runner_exit/runner_exit.go

@@ -0,0 +1,108 @@
+// Package runner_exit persists a drained shellctl job exit into SQLite.
+//
+// This runs out-of-process from the main shellctl server, after the tmux
+// output pipe reaches EOF. It uses the same CAS semantics as the Python
+// shellctl_runtime/runner_exit.py: only non-terminal rows are updated.
+package runner_exit
+
+import (
+	"database/sql"
+	"fmt"
+	"path/filepath"
+
+	_ "modernc.org/sqlite"
+)
+
+var (
+	nonterminalStatuses = []string{"created", "starting", "running"}
+	terminalStatuses    = []string{"exited", "terminated", "failed", "lost"}
+)
+
+// RecordRunnerExit persists exit_code and ended_at into the shellctl SQLite DB.
+// The update is idempotent for terminal rows.
+func RecordRunnerExit(stateDir, jobID string, exitCode int, endedAt string, busyTimeoutMs int) error {
+	dbPath := filepath.Join(stateDir, "shellctl.db")
+
+	dsn := fmt.Sprintf("file:%s?_busy_timeout=%d", dbPath, busyTimeoutMs)
+	db, err := sql.Open("sqlite", dsn)
+	if err != nil {
+		return fmt.Errorf("open database: %w", err)
+	}
+	defer db.Close()
+
+	// Check current status
+	var status string
+	err = db.QueryRow("SELECT status FROM jobs WHERE job_id = ?", jobID).Scan(&status)
+	if err == sql.ErrNoRows {
+		return fmt.Errorf("unknown job id: %s", jobID)
+	}
+	if err != nil {
+		return fmt.Errorf("query job status: %w", err)
+	}
+
+	if isTerminal(status) {
+		return nil
+	}
+
+	// CAS update: only transition non-terminal rows
+	result, err := db.Exec(`
+		UPDATE jobs
+		SET status = CASE
+				WHEN status IN (?, ?, ?) THEN ?
+				ELSE status
+			END,
+			exit_code = CASE
+				WHEN status IN (?, ?, ?) THEN ?
+				ELSE exit_code
+			END,
+			ended_at = CASE
+				WHEN status IN (?, ?, ?) THEN ?
+				ELSE ended_at
+			END,
+			updated_at = CASE
+				WHEN status IN (?, ?, ?) THEN ?
+				ELSE updated_at
+			END,
+			reason = CASE
+				WHEN status IN (?, ?, ?) THEN NULL
+				ELSE reason
+			END,
+			message = CASE
+				WHEN status IN (?, ?, ?) THEN NULL
+				ELSE message
+			END
+		WHERE job_id = ?`,
+		// status
+		nonterminalStatuses[0], nonterminalStatuses[1], nonterminalStatuses[2], "exited",
+		// exit_code
+		nonterminalStatuses[0], nonterminalStatuses[1], nonterminalStatuses[2], exitCode,
+		// ended_at
+		nonterminalStatuses[0], nonterminalStatuses[1], nonterminalStatuses[2], endedAt,
+		// updated_at
+		nonterminalStatuses[0], nonterminalStatuses[1], nonterminalStatuses[2], endedAt,
+		// reason
+		nonterminalStatuses[0], nonterminalStatuses[1], nonterminalStatuses[2],
+		// message
+		nonterminalStatuses[0], nonterminalStatuses[1], nonterminalStatuses[2],
+		// WHERE
+		jobID,
+	)
+	if err != nil {
+		return fmt.Errorf("update job: %w", err)
+	}
+
+	rows, _ := result.RowsAffected()
+	if rows == 0 {
+		return fmt.Errorf("unknown job id: %s", jobID)
+	}
+	return nil
+}
+
+func isTerminal(status string) bool {
+	for _, s := range terminalStatuses {
+		if status == s {
+			return true
+		}
+	}
+	return false
+}

shellctl-go/internal/runner_exit/runner_exit_test.go

@@ -0,0 +1,162 @@
+package runner_exit
+
+import (
+	"database/sql"
+	"path/filepath"
+	"testing"
+
+	_ "modernc.org/sqlite"
+)
+
+func setupTestDB(t *testing.T, dir string) string {
+	t.Helper()
+	dbPath := filepath.Join(dir, "shellctl.db")
+	db, err := sql.Open("sqlite", "file:"+dbPath)
+	if err != nil {
+		t.Fatalf("open sqlite: %v", err)
+	}
+	defer db.Close()
+
+	_, err = db.Exec(`
+		CREATE TABLE jobs (
+			job_id       TEXT PRIMARY KEY,
+			script_path  TEXT NOT NULL,
+			output_path  TEXT NOT NULL,
+			cwd          TEXT NOT NULL,
+			terminal_cols INTEGER NOT NULL DEFAULT 200,
+			terminal_rows INTEGER NOT NULL DEFAULT 50,
+			status       TEXT NOT NULL DEFAULT 'created',
+			session_name TEXT NOT NULL,
+			pane_target  TEXT NOT NULL,
+			exit_code    INTEGER,
+			reason       TEXT,
+			message      TEXT,
+			created_at   TEXT NOT NULL,
+			started_at   TEXT,
+			ended_at     TEXT,
+			updated_at   TEXT NOT NULL
+		)
+	`)
+	if err != nil {
+		t.Fatalf("create table: %v", err)
+	}
+
+	_, err = db.Exec(`INSERT INTO jobs (job_id, script_path, output_path, cwd, status, session_name, pane_target, created_at, updated_at)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		"test-job", "s", "o", "/tmp", "running", "sess", "pane", "2025-01-01T00:00:00Z", "2025-01-01T00:00:00Z")
+	if err != nil {
+		t.Fatalf("insert job: %v", err)
+	}
+
+	return dbPath
+}
+
+func TestRecordRunnerExitRunning(t *testing.T) {
+	dir := t.TempDir()
+	stateDir := dir
+	setupTestDB(t, dir)
+
+	err := RecordRunnerExit(stateDir, "test-job", 0, "2025-01-15T12:00:00Z", 5000)
+	if err != nil {
+		t.Fatalf("RecordRunnerExit: %v", err)
+	}
+
+	// Verify the row was updated
+	db, _ := sql.Open("sqlite", "file:"+filepath.Join(dir, "shellctl.db"))
+	defer db.Close()
+
+	var status string
+	var exitCode int
+	db.QueryRow("SELECT status, exit_code FROM jobs WHERE job_id = ?", "test-job").Scan(&status, &exitCode)
+
+	if status != "exited" {
+		t.Errorf("expected status=exited, got %s", status)
+	}
+	if exitCode != 0 {
+		t.Errorf("expected exit_code=0, got %d", exitCode)
+	}
+}
+
+func TestRecordRunnerExitNonZeroCode(t *testing.T) {
+	dir := t.TempDir()
+	setupTestDB(t, dir)
+
+	err := RecordRunnerExit(dir, "test-job", 42, "2025-01-15T12:00:00Z", 5000)
+	if err != nil {
+		t.Fatalf("RecordRunnerExit: %v", err)
+	}
+
+	db, _ := sql.Open("sqlite", "file:"+filepath.Join(dir, "shellctl.db"))
+	defer db.Close()
+
+	var exitCode int
+	db.QueryRow("SELECT exit_code FROM jobs WHERE job_id = ?", "test-job").Scan(&exitCode)
+
+	if exitCode != 42 {
+		t.Errorf("expected exit_code=42, got %d", exitCode)
+	}
+}
+
+func TestRecordRunnerExitJobNotFound(t *testing.T) {
+	dir := t.TempDir()
+	setupTestDB(t, dir)
+
+	err := RecordRunnerExit(dir, "nonexistent-job", 0, "2025-01-15T12:00:00Z", 5000)
+	if err == nil {
+		t.Error("expected error for nonexistent job")
+	}
+}
+
+func TestRecordRunnerExitDBNotFound(t *testing.T) {
+	dir := t.TempDir()
+
+	err := RecordRunnerExit(dir, "test-job", 0, "2025-01-15T12:00:00Z", 5000)
+	if err == nil {
+		t.Error("expected error when database doesn't exist")
+	}
+}
+
+func TestRecordRunnerExitTerminalIdempotent(t *testing.T) {
+	dir := t.TempDir()
+	dbPath := setupTestDB(t, dir)
+
+	// Manually set job to terminal state
+	db, _ := sql.Open("sqlite", "file:"+dbPath)
+	db.Exec(`UPDATE jobs SET status='terminated', exit_code=137, ended_at='2025-01-01T00:01:00Z' WHERE job_id='test-job'`)
+	db.Close()
+
+	// Should not overwrite
+	err := RecordRunnerExit(dir, "test-job", 0, "2025-01-15T12:00:00Z", 5000)
+	if err != nil {
+		t.Fatalf("RecordRunnerExit on terminal: %v", err)
+	}
+
+	db, _ = sql.Open("sqlite", "file:"+dbPath)
+	defer db.Close()
+
+	var status string
+	var exitCode int
+	db.QueryRow("SELECT status, exit_code FROM jobs WHERE job_id = ?", "test-job").Scan(&status, &exitCode)
+
+	if status != "terminated" {
+		t.Errorf("expected status=terminated (preserved), got %s", status)
+	}
+	if exitCode != 137 {
+		t.Errorf("expected exit_code=137 (preserved), got %d", exitCode)
+	}
+}
+
+func TestIsTerminal(t *testing.T) {
+	terminal := []string{"exited", "terminated", "failed", "lost"}
+	for _, s := range terminal {
+		if !isTerminal(s) {
+			t.Errorf("%s should be terminal", s)
+		}
+	}
+	nonTerminal := []string{"created", "starting", "running"}
+	for _, s := range nonTerminal {
+		if isTerminal(s) {
+			t.Errorf("%s should not be terminal", s)
+		}
+	}
+}

shellctl-go/internal/sanitize/sanitize.go

@@ -0,0 +1,193 @@
+// Package sanitize implements a streaming PTY output sanitizer.
+//
+// It strips ANSI escape sequences (CSI, OSC), normalizes carriage-return
+// progress updates into the final visible line, and performs incremental
+// UTF-8 decoding—mirroring the Python shellctl_runtime/sanitize.py.
+package sanitize
+
+import (
+	"bufio"
+	"io"
+	"os"
+	"unicode/utf8"
+)
+
+// escapeState tracks the ANSI escape sequence parser state.
+type escapeState int
+
+const (
+	stateNormal escapeState = iota
+	stateEsc
+	stateCSI
+	stateOSC
+	stateOSCEsc
+)
+
+// PtySanitizer incrementally converts PTY bytes into stable, readable UTF-8.
+type PtySanitizer struct {
+	lineBuffer []byte
+	pendingCR  bool
+	state      escapeState
+}
+
+// New returns a fresh PtySanitizer.
+func New() *PtySanitizer {
+	return &PtySanitizer{}
+}
+
+// Feed consumes one chunk of decoded text and returns newly stable output.
+func (s *PtySanitizer) Feed(text []byte) []byte {
+	var out []byte
+	for len(text) > 0 {
+		r, size := utf8.DecodeRune(text)
+		if r == utf8.RuneError && size <= 1 {
+			// Replace invalid byte with U+FFFD
+			text = text[1:]
+			r = utf8.RuneError
+		} else {
+			text = text[size:]
+		}
+		out = s.consumeRune(r, out)
+	}
+	return out
+}
+
+// Flush returns any remaining buffered content at end-of-stream.
+func (s *PtySanitizer) Flush() []byte {
+	s.state = stateNormal
+	s.pendingCR = false
+	result := s.lineBuffer
+	s.lineBuffer = nil
+	return result
+}
+
+func (s *PtySanitizer) consumeRune(r rune, out []byte) []byte {
+	switch s.state {
+	case stateNormal:
+		if r == '\x1b' {
+			s.state = stateEsc
+			return out
+		}
+		return s.consumeVisible(r, out)
+
+	case stateEsc:
+		switch r {
+		case '[':
+			s.state = stateCSI
+		case ']':
+			s.state = stateOSC
+		default:
+			s.state = stateNormal
+			if r != '\x1b' && isPrintable(r) {
+				return s.consumeVisible(r, out)
+			}
+		}
+		return out
+
+	case stateCSI:
+		// CSI sequence ends at a byte in the range 0x40–0x7E
+		if r >= '@' && r <= '~' {
+			s.state = stateNormal
+		}
+		return out
+
+	case stateOSC:
+		if r == '\x07' {
+			s.state = stateNormal
+		} else if r == '\x1b' {
+			s.state = stateOSCEsc
+		}
+		return out
+
+	case stateOSCEsc:
+		if r == '\\' {
+			s.state = stateNormal
+		} else {
+			s.state = stateOSC
+		}
+		return out
+	}
+	return out
+}
+
+func (s *PtySanitizer) consumeVisible(r rune, out []byte) []byte {
+	if s.pendingCR {
+		if r == '\n' {
+			out = append(out, s.lineBuffer...)
+			out = append(out, '\n')
+			s.lineBuffer = nil
+			s.pendingCR = false
+			return out
+		}
+		// CR without LF: overwrite line buffer (progress update)
+		s.lineBuffer = nil
+		s.pendingCR = false
+	}
+
+	if r == '\r' {
+		s.pendingCR = true
+		return out
+	}
+	if r == '\n' {
+		out = append(out, s.lineBuffer...)
+		out = append(out, '\n')
+		s.lineBuffer = nil
+		return out
+	}
+
+	// Append rune to line buffer
+	var buf [utf8.UTFMax]byte
+	n := utf8.EncodeRune(buf[:], r)
+	s.lineBuffer = append(s.lineBuffer, buf[:n]...)
+	return out
+}
+
+func isPrintable(r rune) bool {
+	// Match Python's str.isprintable: exclude C0/C1 control chars
+	return r >= 0x20 && r != 0x7f
+}
+
+// Run executes the streaming sanitizer as a stdin→stdout filter.
+// If readyFile is non-empty, it is touched before reading begins.
+func Run(readyFile string, stdin io.Reader, stdout io.Writer) error {
+	if readyFile != "" {
+		f, err := os.Create(readyFile)
+		if err != nil {
+			return err
+		}
+		f.Close()
+	}
+
+	sanitizer := New()
+	reader := bufio.NewReaderSize(stdin, 65536)
+	writer := bufio.NewWriter(stdout)
+	defer writer.Flush()
+
+	buf := make([]byte, 65536)
+	for {
+		n, err := reader.Read(buf)
+		if n > 0 {
+			out := sanitizer.Feed(buf[:n])
+			if len(out) > 0 {
+				if _, werr := writer.Write(out); werr != nil {
+					return werr
+				}
+				writer.Flush()
+			}
+		}
+		if err != nil {
+			if err == io.EOF {
+				break
+			}
+			return err
+		}
+	}
+
+	tail := sanitizer.Flush()
+	if len(tail) > 0 {
+		if _, err := writer.Write(tail); err != nil {
+			return err
+		}
+	}
+	return writer.Flush()
+}

shellctl-go/internal/sanitize/sanitize_test.go

@@ -0,0 +1,83 @@
+package sanitize
+
+import (
+	"testing"
+)
+
+func TestPlainText(t *testing.T) {
+	s := New()
+	out := s.Feed([]byte("hello\nworld\n"))
+	out = append(out, s.Flush()...)
+	expected := "hello\nworld\n"
+	if string(out) != expected {
+		t.Errorf("got %q, want %q", string(out), expected)
+	}
+}
+
+func TestStripCSI(t *testing.T) {
+	// ESC [ 31 m = red color, ESC [ 0 m = reset
+	input := []byte("\x1b[31mred\x1b[0m\n")
+	s := New()
+	out := s.Feed(input)
+	out = append(out, s.Flush()...)
+	expected := "red\n"
+	if string(out) != expected {
+		t.Errorf("got %q, want %q", string(out), expected)
+	}
+}
+
+func TestCarriageReturnOverwrite(t *testing.T) {
+	// Progress: "50%" CR "100%" LF -> only "100%" visible
+	input := []byte("50%\r100%\n")
+	s := New()
+	out := s.Feed(input)
+	out = append(out, s.Flush()...)
+	expected := "100%\n"
+	if string(out) != expected {
+		t.Errorf("got %q, want %q", string(out), expected)
+	}
+}
+
+func TestCRLF(t *testing.T) {
+	input := []byte("line1\r\nline2\r\n")
+	s := New()
+	out := s.Feed(input)
+	out = append(out, s.Flush()...)
+	expected := "line1\nline2\n"
+	if string(out) != expected {
+		t.Errorf("got %q, want %q", string(out), expected)
+	}
+}
+
+func TestOSCSequence(t *testing.T) {
+	// OSC: ESC ] ... BEL
+	input := []byte("\x1b]0;title\x07visible\n")
+	s := New()
+	out := s.Feed(input)
+	out = append(out, s.Flush()...)
+	expected := "visible\n"
+	if string(out) != expected {
+		t.Errorf("got %q, want %q", string(out), expected)
+	}
+}
+
+func TestFlushUnterminatedLine(t *testing.T) {
+	s := New()
+	out := s.Feed([]byte("no newline"))
+	out = append(out, s.Flush()...)
+	expected := "no newline"
+	if string(out) != expected {
+		t.Errorf("got %q, want %q", string(out), expected)
+	}
+}
+
+func TestInvalidUTF8(t *testing.T) {
+	// 0xFF is not valid UTF-8, should be replaced
+	s := New()
+	out := s.Feed([]byte{0xFF, 'a', '\n'})
+	out = append(out, s.Flush()...)
+	expected := "\uFFFDa\n"
+	if string(out) != expected {
+		t.Errorf("got %q, want %q", string(out), expected)
+	}
+}

shellctl-go/internal/server/api.go

@@ -0,0 +1,278 @@
+package server
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+	"runtime/debug"
+	"strconv"
+	"strings"
+	"time"
+)
+
+// Handler creates the HTTP handler (mux) for the shellctl API.
+func Handler(svc *Service, config *Config) http.Handler {
+	mux := http.NewServeMux()
+
+	auth := authMiddleware(config.AuthToken)
+
+	mux.HandleFunc("GET /healthz", handleHealthz)
+	mux.HandleFunc("POST /v1/jobs/run", auth(handleRunJob(svc)))
+	mux.HandleFunc("POST /v1/jobs/{job_id}/wait", auth(handleWaitJob(svc)))
+	mux.HandleFunc("GET /v1/jobs/{job_id}/log/tail", auth(handleTailJob(svc, config)))
+	mux.HandleFunc("GET /v1/jobs/{job_id}", auth(handleJobStatus(svc)))
+	mux.HandleFunc("GET /v1/jobs", auth(handleListJobs(svc, config)))
+	mux.HandleFunc("POST /v1/jobs/{job_id}/input", auth(handleInputJob(svc)))
+	mux.HandleFunc("POST /v1/jobs/{job_id}/terminate", auth(handleTerminateJob(svc, config)))
+	mux.HandleFunc("DELETE /v1/jobs/{job_id}", auth(handleDeleteJob(svc, config)))
+
+	return requestLoggingMiddleware(recoveryMiddleware(mux))
+}
+
+func handleHealthz(w http.ResponseWriter, r *http.Request) {
+	writeJSON(w, http.StatusOK, HealthResponse{Status: HealthStatus})
+}
+
+func handleRunJob(svc *Service) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		var req RunJobRequest
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			writeError(w, 400, "invalid_request", "Invalid JSON body")
+			return
+		}
+		if req.Script == "" {
+			writeError(w, 400, "invalid_request", "script is required")
+			return
+		}
+		// Validate env
+		if req.Env != nil {
+			for name, value := range req.Env {
+				if name == "" {
+					writeError(w, 422, "validation_error", "env names must be non-empty")
+					return
+				}
+				if strings.Contains(name, "=") {
+					writeError(w, 422, "validation_error", fmt.Sprintf("env name must not contain '=': %q", name))
+					return
+				}
+				if strings.Contains(name, "\x00") || strings.Contains(value, "\x00") {
+					writeError(w, 422, "validation_error", "env entries must not contain NUL")
+					return
+				}
+			}
+		}
+
+		result, err := svc.RunJob(&req)
+		if err != nil {
+			writeServerError(w, err)
+			return
+		}
+		writeJSON(w, http.StatusOK, result)
+	}
+}
+
+func handleWaitJob(svc *Service) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		jobID := r.PathValue("job_id")
+		var req WaitJobRequest
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			writeError(w, 400, "invalid_request", "Invalid JSON body")
+			return
+		}
+		result, err := svc.WaitJob(jobID, &req)
+		if err != nil {
+			writeServerError(w, err)
+			return
+		}
+		writeJSON(w, http.StatusOK, result)
+	}
+}
+
+func handleTailJob(svc *Service, config *Config) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		jobID := r.PathValue("job_id")
+		outputLimit := config.DefaultOutputLimitBytes
+		if v := r.URL.Query().Get("output_limit"); v != "" {
+			if parsed, err := strconv.Atoi(v); err == nil && parsed > 0 {
+				outputLimit = parsed
+			}
+		}
+		if outputLimit > config.MaxOutputLimitBytes {
+			outputLimit = config.MaxOutputLimitBytes
+		}
+		result, err := svc.TailJob(jobID, outputLimit)
+		if err != nil {
+			writeServerError(w, err)
+			return
+		}
+		writeJSON(w, http.StatusOK, result)
+	}
+}
+
+func handleJobStatus(svc *Service) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		jobID := r.PathValue("job_id")
+		view, err := svc.GetJobStatus(jobID)
+		if err != nil {
+			writeServerError(w, err)
+			return
+		}
+		writeJSON(w, http.StatusOK, view)
+	}
+}
+
+func handleListJobs(svc *Service, config *Config) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		var status *JobStatusName
+		if v := r.URL.Query().Get("status"); v != "" {
+			s := JobStatusName(v)
+			status = &s
+		}
+		limit := config.DefaultListLimit
+		if v := r.URL.Query().Get("limit"); v != "" {
+			if parsed, err := strconv.Atoi(v); err == nil && parsed > 0 {
+				limit = parsed
+			}
+		}
+		if limit > config.MaxListLimit {
+			limit = config.MaxListLimit
+		}
+
+		result, err := svc.ListJobs(status, limit)
+		if err != nil {
+			writeServerError(w, err)
+			return
+		}
+		writeJSON(w, http.StatusOK, result)
+	}
+}
+
+func handleInputJob(svc *Service) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		jobID := r.PathValue("job_id")
+		var req InputJobRequest
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			writeError(w, 400, "invalid_request", "Invalid JSON body")
+			return
+		}
+		result, err := svc.SendInput(jobID, &req)
+		if err != nil {
+			writeServerError(w, err)
+			return
+		}
+		writeJSON(w, http.StatusOK, result)
+	}
+}
+
+func handleTerminateJob(svc *Service, config *Config) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		jobID := r.PathValue("job_id")
+		var req TerminateJobRequest
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			// Allow empty body with defaults
+			req.GraceSeconds = config.DefaultTerminateGraceSeconds
+		}
+		if req.GraceSeconds == 0 {
+			req.GraceSeconds = config.DefaultTerminateGraceSeconds
+		}
+		view, err := svc.TerminateJob(jobID, req.GraceSeconds)
+		if err != nil {
+			writeServerError(w, err)
+			return
+		}
+		writeJSON(w, http.StatusOK, view)
+	}
+}
+
+func handleDeleteJob(svc *Service, config *Config) http.HandlerFunc {
+	return func(w http.ResponseWriter, r *http.Request) {
+		jobID := r.PathValue("job_id")
+		force := r.URL.Query().Get("force") == "true"
+		graceSeconds := config.DefaultTerminateGraceSeconds
+		if v := r.URL.Query().Get("grace_seconds"); v != "" {
+			if parsed, err := strconv.ParseFloat(v, 64); err == nil {
+				graceSeconds = parsed
+			}
+		}
+		result, err := svc.DeleteJob(jobID, force, graceSeconds)
+		if err != nil {
+			writeServerError(w, err)
+			return
+		}
+		writeJSON(w, http.StatusOK, result)
+	}
+}
+
+// Middleware
+
+// statusRecorder wraps ResponseWriter to capture the status code.
+type statusRecorder struct {
+	http.ResponseWriter
+	statusCode int
+}
+
+func (sr *statusRecorder) WriteHeader(code int) {
+	sr.statusCode = code
+	sr.ResponseWriter.WriteHeader(code)
+}
+
+func requestLoggingMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		start := time.Now()
+		rec := &statusRecorder{ResponseWriter: w, statusCode: 200}
+		next.ServeHTTP(rec, r)
+		log.Printf("%s %s -> %d (%s)", r.Method, r.URL.Path, rec.statusCode, time.Since(start).Round(time.Millisecond))
+	})
+}
+
+func recoveryMiddleware(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		defer func() {
+			if rec := recover(); rec != nil {
+				stack := debug.Stack()
+				log.Printf("PANIC %s %s: %v\n%s", r.Method, r.URL.Path, rec, stack)
+				writeError(w, 500, "internal_panic", fmt.Sprintf("internal server error: %v", rec))
+			}
+		}()
+		next.ServeHTTP(w, r)
+	})
+}
+
+func authMiddleware(token string) func(http.HandlerFunc) http.HandlerFunc {
+	return func(next http.HandlerFunc) http.HandlerFunc {
+		if token == "" {
+			return next // No auth configured
+		}
+		expected := "Bearer " + token
+		return func(w http.ResponseWriter, r *http.Request) {
+			if r.Header.Get("Authorization") != expected {
+				writeError(w, 401, "unauthorized", "Missing or invalid bearer token")
+				return
+			}
+			next(w, r)
+		}
+	}
+}
+
+// Response helpers
+
+func writeJSON(w http.ResponseWriter, status int, v any) {
+	w.Header().Set("Content-Type", "application/json")
+	w.WriteHeader(status)
+	json.NewEncoder(w).Encode(v)
+}
+
+func writeError(w http.ResponseWriter, status int, code, message string) {
+	writeJSON(w, status, ErrorResponse{Error: ErrorDetail{Code: code, Message: message}})
+}
+
+func writeServerError(w http.ResponseWriter, err error) {
+	if se, ok := err.(*ServerError); ok {
+		log.Printf("ERROR [%d] %s: %s", se.StatusCode, se.Code, se.Message)
+		writeError(w, se.StatusCode, se.Code, se.Message)
+		return
+	}
+	log.Printf("ERROR [500] internal_error: %v", err)
+	writeError(w, 500, "internal_error", err.Error())
+}

shellctl-go/internal/server/api_test.go

@@ -0,0 +1,162 @@
+package server
+
+import (
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+// We can't fully test the Service without tmux, but we can test the HTTP
+// layer wiring, error handling, and JSON serialization.
+
+func TestWriteJSON(t *testing.T) {
+	w := httptest.NewRecorder()
+	writeJSON(w, 200, HealthResponse{Status: "ok"})
+
+	if w.Code != 200 {
+		t.Errorf("expected 200, got %d", w.Code)
+	}
+	if w.Header().Get("Content-Type") != "application/json" {
+		t.Errorf("expected application/json, got %s", w.Header().Get("Content-Type"))
+	}
+
+	var result HealthResponse
+	json.NewDecoder(w.Body).Decode(&result)
+	if result.Status != "ok" {
+		t.Errorf("expected status=ok, got %s", result.Status)
+	}
+}
+
+func TestWriteError(t *testing.T) {
+	w := httptest.NewRecorder()
+	writeError(w, 400, "bad_request", "missing field")
+
+	if w.Code != 400 {
+		t.Errorf("expected 400, got %d", w.Code)
+	}
+
+	var result ErrorResponse
+	json.NewDecoder(w.Body).Decode(&result)
+	if result.Error.Code != "bad_request" {
+		t.Errorf("expected code=bad_request, got %s", result.Error.Code)
+	}
+	if result.Error.Message != "missing field" {
+		t.Errorf("expected message='missing field', got %s", result.Error.Message)
+	}
+}
+
+func TestWriteServerError(t *testing.T) {
+	w := httptest.NewRecorder()
+	err := NewServerError(404, "job_not_found", "Unknown job id")
+	writeServerError(w, err)
+
+	if w.Code != 404 {
+		t.Errorf("expected 404, got %d", w.Code)
+	}
+
+	var result ErrorResponse
+	json.NewDecoder(w.Body).Decode(&result)
+	if result.Error.Code != "job_not_found" {
+		t.Errorf("expected code=job_not_found, got %s", result.Error.Code)
+	}
+}
+
+func TestWriteServerErrorGeneric(t *testing.T) {
+	w := httptest.NewRecorder()
+	writeServerError(w, &json.SyntaxError{Offset: 5})
+
+	if w.Code != 500 {
+		t.Errorf("expected 500, got %d", w.Code)
+	}
+
+	var result ErrorResponse
+	json.NewDecoder(w.Body).Decode(&result)
+	if result.Error.Code != "internal_error" {
+		t.Errorf("expected code=internal_error, got %s", result.Error.Code)
+	}
+}
+
+func TestAuthMiddlewareNoToken(t *testing.T) {
+	// When no token configured, auth middleware should pass through
+	handler := authMiddleware("")(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(200)
+	})
+
+	req := httptest.NewRequest("GET", "/v1/jobs", nil)
+	w := httptest.NewRecorder()
+	handler(w, req)
+
+	if w.Code != 200 {
+		t.Errorf("expected 200 without auth, got %d", w.Code)
+	}
+}
+
+func TestAuthMiddlewareWithToken(t *testing.T) {
+	handler := authMiddleware("secret")(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(200)
+	})
+
+	// Without auth header
+	req := httptest.NewRequest("GET", "/v1/jobs", nil)
+	w := httptest.NewRecorder()
+	handler(w, req)
+	if w.Code != 401 {
+		t.Errorf("expected 401 without auth, got %d", w.Code)
+	}
+
+	// With correct auth header
+	req = httptest.NewRequest("GET", "/v1/jobs", nil)
+	req.Header.Set("Authorization", "Bearer secret")
+	w = httptest.NewRecorder()
+	handler(w, req)
+	if w.Code != 200 {
+		t.Errorf("expected 200 with correct auth, got %d", w.Code)
+	}
+
+	// With wrong auth header
+	req = httptest.NewRequest("GET", "/v1/jobs", nil)
+	req.Header.Set("Authorization", "Bearer wrong")
+	w = httptest.NewRecorder()
+	handler(w, req)
+	if w.Code != 401 {
+		t.Errorf("expected 401 with wrong auth, got %d", w.Code)
+	}
+}
+
+func TestHealthzHandler(t *testing.T) {
+	// Create a handler with a nil service (healthz doesn't use it)
+	mux := http.NewServeMux()
+	mux.HandleFunc("GET /healthz", handleHealthz)
+
+	req := httptest.NewRequest("GET", "/healthz", nil)
+	w := httptest.NewRecorder()
+	mux.ServeHTTP(w, req)
+
+	if w.Code != 200 {
+		t.Errorf("expected 200, got %d", w.Code)
+	}
+
+	body := w.Body.String()
+	if !strings.Contains(body, "ok") {
+		t.Errorf("expected body to contain 'ok', got %s", body)
+	}
+}
+
+func TestServerErrorFormat(t *testing.T) {
+	err := NewServerError(422, "validation_error", "bad input")
+	expected := "[422] validation_error: bad input"
+	if err.Error() != expected {
+		t.Errorf("expected %q, got %q", expected, err.Error())
+	}
+}
+
+func TestIsNotFound(t *testing.T) {
+	if !isNotFound(ErrJobNotFound) {
+		t.Error("ErrJobNotFound should be detected as not found")
+	}
+	if isNotFound(NewServerError(500, "internal_error", "x")) {
+		t.Error("500 error should not be detected as not found")
+	}
+}

shellctl-go/internal/server/config.go

@@ -0,0 +1,128 @@
+package server
+
+import (
+	"os"
+	"path/filepath"
+	"runtime"
+	"time"
+)
+
+const (
+	DefaultListen                       = "127.0.0.1:8765"
+	DefaultTimeoutSeconds               = 30.0
+	DefaultMaxWaitTimeoutSeconds        = 600.0
+	DefaultIdleFlushSeconds             = 0.5
+	DefaultTerminalCols                 = 200
+	DefaultTerminalRows                 = 50
+	DefaultOutputLimitBytes             = 16 * 1024
+	MaxOutputLimitBytes                 = 512 * 1024
+	DefaultListLimit                    = 50
+	MaxListLimit                        = 200
+	DefaultTerminateGraceSeconds        = 10.0
+	DefaultGCIntervalSeconds            = 60.0
+	DefaultGCFinishedJobRetentionSeconds = 300.0
+	DefaultPollInterval                 = 50 * time.Millisecond
+	DefaultPipeMonitorInterval          = 1 * time.Second
+	DefaultPipeReadyTimeout             = 10 * time.Second
+	DefaultSQLiteBusyTimeoutMs          = 5000
+	DefaultAuthTokenEnv                 = "SHELLCTL_AUTH_TOKEN"
+	HealthStatus                        = "ok"
+)
+
+// Config holds runtime configuration for the shellctl server.
+type Config struct {
+	Listen                       string
+	AuthToken                    string
+	StateDir                     string
+	RuntimeDir                   string
+	GCInterval                   time.Duration
+	GCFinishedJobRetention       time.Duration
+	DefaultTimeout               time.Duration
+	MaxWaitTimeout               time.Duration
+	IdleFlushDuration            time.Duration
+	DefaultCwd                   string
+	DefaultTerminalCols          int
+	DefaultTerminalRows          int
+	DefaultListLimit             int
+	MaxListLimit                 int
+	DefaultOutputLimitBytes      int
+	MaxOutputLimitBytes          int
+	DefaultTerminateGraceSeconds float64
+	PollInterval                 time.Duration
+	PipeMonitorInterval          time.Duration
+	PipeReadyTimeout             time.Duration
+	SQLiteBusyTimeoutMs          int
+	SanitizePtyCommand           []string
+	RunnerExitCommand            []string
+}
+
+// DefaultConfig returns a Config with sensible defaults.
+func DefaultConfig() *Config {
+	homeDir, _ := os.UserHomeDir()
+	stateDir := defaultStateDir()
+	runtimeDir := filepath.Join(stateDir, "runtime")
+
+	cfg := &Config{
+		Listen:                       DefaultListen,
+		StateDir:                     stateDir,
+		RuntimeDir:                   runtimeDir,
+		GCInterval:                   time.Duration(DefaultGCIntervalSeconds * float64(time.Second)),
+		GCFinishedJobRetention:       time.Duration(DefaultGCFinishedJobRetentionSeconds * float64(time.Second)),
+		DefaultTimeout:               time.Duration(DefaultTimeoutSeconds * float64(time.Second)),
+		MaxWaitTimeout:               time.Duration(DefaultMaxWaitTimeoutSeconds * float64(time.Second)),
+		IdleFlushDuration:            time.Duration(DefaultIdleFlushSeconds * float64(time.Second)),
+		DefaultCwd:                   homeDir,
+		DefaultTerminalCols:          DefaultTerminalCols,
+		DefaultTerminalRows:          DefaultTerminalRows,
+		DefaultListLimit:             DefaultListLimit,
+		MaxListLimit:                 MaxListLimit,
+		DefaultOutputLimitBytes:      DefaultOutputLimitBytes,
+		MaxOutputLimitBytes:          MaxOutputLimitBytes,
+		DefaultTerminateGraceSeconds: DefaultTerminateGraceSeconds,
+		PollInterval:                 DefaultPollInterval,
+		PipeMonitorInterval:          DefaultPipeMonitorInterval,
+		PipeReadyTimeout:             DefaultPipeReadyTimeout,
+		SQLiteBusyTimeoutMs:          DefaultSQLiteBusyTimeoutMs,
+		SanitizePtyCommand:           []string{"shellctl-sanitize-pty"},
+		RunnerExitCommand:            []string{"shellctl-runner-exit"},
+	}
+
+	// Auth token from environment if not set explicitly
+	if cfg.AuthToken == "" {
+		cfg.AuthToken = os.Getenv(DefaultAuthTokenEnv)
+	}
+
+	return cfg
+}
+
+// JobsDir returns the path to the jobs artifact directory.
+func (c *Config) JobsDir() string {
+	return filepath.Join(c.StateDir, "jobs")
+}
+
+// DBPath returns the path to the SQLite database.
+func (c *Config) DBPath() string {
+	return filepath.Join(c.StateDir, "shellctl.db")
+}
+
+// TmuxSocket returns the path to the dedicated tmux socket.
+func (c *Config) TmuxSocket() string {
+	return filepath.Join(c.RuntimeDir, "tmux.sock")
+}
+
+// RunnerPath returns the path to the installed runner script.
+func (c *Config) RunnerPath() string {
+	return filepath.Join(c.RuntimeDir, "bin", "shellctl-runner")
+}
+
+func defaultStateDir() string {
+	if runtime.GOOS == "darwin" {
+		home, _ := os.UserHomeDir()
+		return filepath.Join(home, ".local", "share", "shellctl")
+	}
+	if xdg := os.Getenv("XDG_DATA_HOME"); xdg != "" {
+		return filepath.Join(xdg, "shellctl")
+	}
+	home, _ := os.UserHomeDir()
+	return filepath.Join(home, ".local", "share", "shellctl")
+}

shellctl-go/internal/server/config_test.go

@@ -0,0 +1,56 @@
+package server
+
+import (
+	"testing"
+)
+
+func TestDefaultConfig(t *testing.T) {
+	cfg := DefaultConfig()
+	if cfg.Listen != DefaultListen {
+		t.Errorf("expected Listen=%s, got %s", DefaultListen, cfg.Listen)
+	}
+	if cfg.DefaultTerminalCols != DefaultTerminalCols {
+		t.Errorf("expected cols=%d, got %d", DefaultTerminalCols, cfg.DefaultTerminalCols)
+	}
+	if cfg.DefaultTerminalRows != DefaultTerminalRows {
+		t.Errorf("expected rows=%d, got %d", DefaultTerminalRows, cfg.DefaultTerminalRows)
+	}
+	if cfg.SQLiteBusyTimeoutMs != DefaultSQLiteBusyTimeoutMs {
+		t.Errorf("expected busy_timeout=%d, got %d", DefaultSQLiteBusyTimeoutMs, cfg.SQLiteBusyTimeoutMs)
+	}
+}
+
+func TestConfigPaths(t *testing.T) {
+	cfg := DefaultConfig()
+	cfg.StateDir = "/tmp/shellctl-test"
+	cfg.RuntimeDir = "/tmp/shellctl-test/runtime"
+
+	if cfg.JobsDir() != "/tmp/shellctl-test/jobs" {
+		t.Errorf("unexpected JobsDir: %s", cfg.JobsDir())
+	}
+	if cfg.DBPath() != "/tmp/shellctl-test/shellctl.db" {
+		t.Errorf("unexpected DBPath: %s", cfg.DBPath())
+	}
+	if cfg.TmuxSocket() != "/tmp/shellctl-test/runtime/tmux.sock" {
+		t.Errorf("unexpected TmuxSocket: %s", cfg.TmuxSocket())
+	}
+	if cfg.RunnerPath() != "/tmp/shellctl-test/runtime/bin/shellctl-runner" {
+		t.Errorf("unexpected RunnerPath: %s", cfg.RunnerPath())
+	}
+}
+
+func TestConfigAuthTokenFromEnv(t *testing.T) {
+	t.Setenv("SHELLCTL_AUTH_TOKEN", "my-secret-token")
+	cfg := DefaultConfig()
+	if cfg.AuthToken != "my-secret-token" {
+		t.Errorf("expected auth token from env, got %q", cfg.AuthToken)
+	}
+}
+
+func TestConfigNoAuthToken(t *testing.T) {
+	t.Setenv("SHELLCTL_AUTH_TOKEN", "")
+	cfg := DefaultConfig()
+	if cfg.AuthToken != "" {
+		t.Errorf("expected empty auth token, got %q", cfg.AuthToken)
+	}
+}

shellctl-go/internal/server/db.go

@@ -0,0 +1,368 @@
+package server
+
+import (
+	"database/sql"
+	"fmt"
+	"time"
+
+	_ "modernc.org/sqlite"
+)
+
+// JobStatusName represents the lifecycle states of a shellctl job.
+type JobStatusName string
+
+const (
+	StatusCreated    JobStatusName = "created"
+	StatusStarting   JobStatusName = "starting"
+	StatusRunning    JobStatusName = "running"
+	StatusExited     JobStatusName = "exited"
+	StatusTerminated JobStatusName = "terminated"
+	StatusFailed     JobStatusName = "failed"
+	StatusLost       JobStatusName = "lost"
+)
+
+// IsTerminal returns true if the status represents a final job state.
+func (s JobStatusName) IsTerminal() bool {
+	switch s {
+	case StatusExited, StatusTerminated, StatusFailed, StatusLost:
+		return true
+	}
+	return false
+}
+
+// JobRow represents a row in the jobs SQLite table.
+type JobRow struct {
+	JobID        string
+	ScriptPath   string
+	OutputPath   string
+	Cwd          string
+	TerminalCols int
+	TerminalRows int
+	Status       JobStatusName
+	SessionName  string
+	PaneTarget   string
+	ExitCode     *int
+	Reason       *string
+	Message      *string
+	CreatedAt    string
+	StartedAt    *string
+	EndedAt      *string
+	UpdatedAt    string
+}
+
+// DB wraps the SQLite database for shellctl job persistence.
+type DB struct {
+	db *sql.DB
+}
+
+// OpenDB opens (or creates) the shellctl SQLite database.
+func OpenDB(dbPath string, busyTimeoutMs int) (*DB, error) {
+	dsn := fmt.Sprintf("file:%s?_busy_timeout=%d&_journal_mode=WAL", dbPath, busyTimeoutMs)
+	db, err := sql.Open("sqlite", dsn)
+	if err != nil {
+		return nil, fmt.Errorf("open sqlite: %w", err)
+	}
+	db.SetMaxOpenConns(1)
+	db.SetConnMaxLifetime(0)
+	return &DB{db: db}, nil
+}
+
+// Close closes the database connection.
+func (d *DB) Close() error {
+	return d.db.Close()
+}
+
+// InitSchema creates the jobs table if it does not exist.
+func (d *DB) InitSchema() error {
+	_, err := d.db.Exec(`
+		CREATE TABLE IF NOT EXISTS jobs (
+			job_id       TEXT PRIMARY KEY,
+			script_path  TEXT NOT NULL,
+			output_path  TEXT NOT NULL,
+			cwd          TEXT NOT NULL,
+			terminal_cols INTEGER NOT NULL DEFAULT 200,
+			terminal_rows INTEGER NOT NULL DEFAULT 50,
+			status       TEXT NOT NULL DEFAULT 'created',
+			session_name TEXT NOT NULL,
+			pane_target  TEXT NOT NULL,
+			exit_code    INTEGER,
+			reason       TEXT,
+			message      TEXT,
+			created_at   TEXT NOT NULL,
+			started_at   TEXT,
+			ended_at     TEXT,
+			updated_at   TEXT NOT NULL
+		)
+	`)
+	return err
+}
+
+// InsertJob inserts a new job row. Returns false if the job_id already exists.
+func (d *DB) InsertJob(row *JobRow) (bool, error) {
+	_, err := d.db.Exec(`
+		INSERT INTO jobs (job_id, script_path, output_path, cwd, terminal_cols, terminal_rows,
+			status, session_name, pane_target, exit_code, reason, message,
+			created_at, started_at, ended_at, updated_at)
+		VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
+		row.JobID, row.ScriptPath, row.OutputPath, row.Cwd,
+		row.TerminalCols, row.TerminalRows,
+		string(row.Status), row.SessionName, row.PaneTarget,
+		row.ExitCode, row.Reason, row.Message,
+		row.CreatedAt, row.StartedAt, row.EndedAt, row.UpdatedAt,
+	)
+	if err != nil {
+		// SQLite UNIQUE constraint violation
+		if isUniqueViolation(err) {
+			return false, nil
+		}
+		return false, err
+	}
+	return true, nil
+}
+
+// GetJob retrieves a single job row by ID.
+func (d *DB) GetJob(jobID string) (*JobRow, error) {
+	row := d.db.QueryRow(`
+		SELECT job_id, script_path, output_path, cwd, terminal_cols, terminal_rows,
+			status, session_name, pane_target, exit_code, reason, message,
+			created_at, started_at, ended_at, updated_at
+		FROM jobs WHERE job_id = ?`, jobID)
+	return scanJobRow(row)
+}
+
+// ListJobs returns all jobs ordered by created_at descending, optionally filtered by status.
+func (d *DB) ListJobs(statuses []JobStatusName) ([]*JobRow, error) {
+	var rows *sql.Rows
+	var err error
+
+	if len(statuses) == 0 {
+		rows, err = d.db.Query(`SELECT job_id, script_path, output_path, cwd,
+			terminal_cols, terminal_rows, status, session_name, pane_target,
+			exit_code, reason, message, created_at, started_at, ended_at, updated_at
+			FROM jobs ORDER BY created_at DESC`)
+	} else {
+		args := make([]any, len(statuses))
+		placeholders := ""
+		for i, s := range statuses {
+			args[i] = string(s)
+			if i > 0 {
+				placeholders += ","
+			}
+			placeholders += "?"
+		}
+		query := fmt.Sprintf(`SELECT job_id, script_path, output_path, cwd,
+			terminal_cols, terminal_rows, status, session_name, pane_target,
+			exit_code, reason, message, created_at, started_at, ended_at, updated_at
+			FROM jobs WHERE status IN (%s) ORDER BY created_at DESC`, placeholders)
+		rows, err = d.db.Query(query, args...)
+	}
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	var result []*JobRow
+	for rows.Next() {
+		jr, err := scanJobRows(rows)
+		if err != nil {
+			return nil, err
+		}
+		result = append(result, jr)
+	}
+	return result, rows.Err()
+}
+
+// DeleteJob removes a job row by ID. Returns error if not found.
+func (d *DB) DeleteJob(jobID string) error {
+	result, err := d.db.Exec("DELETE FROM jobs WHERE job_id = ?", jobID)
+	if err != nil {
+		return err
+	}
+	n, _ := result.RowsAffected()
+	if n == 0 {
+		return ErrJobNotFound
+	}
+	return nil
+}
+
+// TransitionStatus performs a conditional CAS update on a job row.
+func (d *DB) TransitionStatus(jobID string, opts TransitionOpts) (*JobRow, error) {
+	now := FormatTimestamp(time.Now())
+
+	// Build WHERE clause
+	allowedList := make([]any, 0, len(opts.AllowedFrom)+len(opts.AllowOverrideFrom))
+	placeholders := ""
+	all := append(opts.AllowedFrom, opts.AllowOverrideFrom...)
+	for i, s := range all {
+		allowedList = append(allowedList, string(s))
+		if i > 0 {
+			placeholders += ","
+		}
+		placeholders += "?"
+	}
+
+	setClauses := `
+		status = ?,
+		updated_at = ?,
+		reason = ?,
+		message = ?`
+
+	args := []any{string(opts.Target), now, opts.Reason, opts.Message}
+
+	// started_at: set on first transition to starting/running
+	if opts.Target == StatusStarting || opts.Target == StatusRunning {
+		setClauses += `, started_at = CASE WHEN started_at IS NULL THEN ? ELSE started_at END`
+		args = append(args, now)
+	}
+
+	// ended_at + exit_code: set on terminal transitions
+	if opts.Target.IsTerminal() {
+		endedAt := opts.EndedAt
+		if endedAt == "" {
+			endedAt = now
+		}
+		setClauses += `, ended_at = CASE WHEN ended_at IS NULL THEN ? ELSE ended_at END`
+		args = append(args, endedAt)
+		// Set exit_code to 0 if not already set (normal exit without runner-exit callback)
+		setClauses += `, exit_code = CASE WHEN exit_code IS NULL THEN 0 ELSE exit_code END`
+	}
+
+	whereClause := fmt.Sprintf("job_id = ? AND status IN (%s)", placeholders)
+	args = append(args, jobID)
+	args = append(args, allowedList...)
+
+	if opts.RequireExitCodeNull {
+		whereClause += " AND exit_code IS NULL"
+	}
+
+	query := fmt.Sprintf("UPDATE jobs SET %s WHERE %s", setClauses, whereClause)
+	result, err := d.db.Exec(query, args...)
+	if err != nil {
+		return nil, fmt.Errorf("transition status: %w", err)
+	}
+
+	n, _ := result.RowsAffected()
+	_ = n // If 0 rows affected, we just return current state
+
+	return d.GetJob(jobID)
+}
+
+// RecordRunnerExit persists exit_code and ended_at for a drained job.
+// The update is idempotent for terminal rows: once a job reaches a terminal
+// state, this method returns nil without rewriting the row.
+func (d *DB) RecordRunnerExit(jobID string, exitCode int, endedAt string) error {
+	// Check if job exists and is already terminal
+	var status string
+	err := d.db.QueryRow("SELECT status FROM jobs WHERE job_id = ?", jobID).Scan(&status)
+	if err == sql.ErrNoRows {
+		return ErrJobNotFound
+	}
+	if err != nil {
+		return err
+	}
+	if JobStatusName(status).IsTerminal() {
+		return nil
+	}
+
+	nonterminal := []any{"created", "starting", "running"}
+
+	result, err := d.db.Exec(`
+		UPDATE jobs
+		SET status = CASE WHEN status IN (?, ?, ?) THEN ? ELSE status END,
+			exit_code = CASE WHEN status IN (?, ?, ?) THEN ? ELSE exit_code END,
+			ended_at = CASE WHEN status IN (?, ?, ?) AND ended_at IS NULL THEN ? ELSE ended_at END,
+			updated_at = CASE WHEN status IN (?, ?, ?) THEN ? ELSE updated_at END,
+			reason = CASE WHEN status IN (?, ?, ?) THEN NULL ELSE reason END,
+			message = CASE WHEN status IN (?, ?, ?) THEN NULL ELSE message END
+		WHERE job_id = ? AND status IN (?, ?, ?)`,
+		nonterminal[0], nonterminal[1], nonterminal[2], string(StatusExited),
+		// exit_code
+		nonterminal[0], nonterminal[1], nonterminal[2], exitCode,
+		// ended_at
+		nonterminal[0], nonterminal[1], nonterminal[2], endedAt,
+		// updated_at
+		nonterminal[0], nonterminal[1], nonterminal[2], endedAt,
+		// reason
+		nonterminal[0], nonterminal[1], nonterminal[2],
+		// message
+		nonterminal[0], nonterminal[1], nonterminal[2],
+		// WHERE
+		jobID,
+		nonterminal[0], nonterminal[1], nonterminal[2],
+	)
+	if err != nil {
+		return err
+	}
+	n, _ := result.RowsAffected()
+	if n == 0 {
+		return ErrJobNotFound
+	}
+	return nil
+}
+
+// TransitionOpts holds parameters for a CAS state transition.
+type TransitionOpts struct {
+	AllowedFrom         []JobStatusName
+	AllowOverrideFrom   []JobStatusName
+	Target              JobStatusName
+	RequireExitCodeNull bool
+	Reason              *string
+	Message             *string
+	EndedAt             string
+}
+
+func scanJobRow(row *sql.Row) (*JobRow, error) {
+	var jr JobRow
+	var status string
+	err := row.Scan(
+		&jr.JobID, &jr.ScriptPath, &jr.OutputPath, &jr.Cwd,
+		&jr.TerminalCols, &jr.TerminalRows,
+		&status, &jr.SessionName, &jr.PaneTarget,
+		&jr.ExitCode, &jr.Reason, &jr.Message,
+		&jr.CreatedAt, &jr.StartedAt, &jr.EndedAt, &jr.UpdatedAt,
+	)
+	if err == sql.ErrNoRows {
+		return nil, ErrJobNotFound
+	}
+	if err != nil {
+		return nil, err
+	}
+	jr.Status = JobStatusName(status)
+	return &jr, nil
+}
+
+func scanJobRows(rows *sql.Rows) (*JobRow, error) {
+	var jr JobRow
+	var status string
+	err := rows.Scan(
+		&jr.JobID, &jr.ScriptPath, &jr.OutputPath, &jr.Cwd,
+		&jr.TerminalCols, &jr.TerminalRows,
+		&status, &jr.SessionName, &jr.PaneTarget,
+		&jr.ExitCode, &jr.Reason, &jr.Message,
+		&jr.CreatedAt, &jr.StartedAt, &jr.EndedAt, &jr.UpdatedAt,
+	)
+	if err != nil {
+		return nil, err
+	}
+	jr.Status = JobStatusName(status)
+	return &jr, nil
+}
+
+func isUniqueViolation(err error) bool {
+	// modernc.org/sqlite returns error messages containing "UNIQUE constraint failed"
+	return err != nil && contains(err.Error(), "UNIQUE constraint failed")
+}
+
+func contains(s, substr string) bool {
+	return len(s) >= len(substr) && (s == substr || len(substr) == 0 ||
+		(len(s) > 0 && len(substr) > 0 && searchString(s, substr)))
+}
+
+func searchString(s, substr string) bool {
+	for i := 0; i <= len(s)-len(substr); i++ {
+		if s[i:i+len(substr)] == substr {
+			return true
+		}
+	}
+	return false
+}

shellctl-go/internal/server/db_test.go

@@ -0,0 +1,309 @@
+package server
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestJobStatusIsTerminal(t *testing.T) {
+	terminal := []JobStatusName{StatusExited, StatusTerminated, StatusFailed, StatusLost}
+	for _, s := range terminal {
+		if !s.IsTerminal() {
+			t.Errorf("%s should be terminal", s)
+		}
+	}
+	nonTerminal := []JobStatusName{StatusCreated, StatusStarting, StatusRunning}
+	for _, s := range nonTerminal {
+		if s.IsTerminal() {
+			t.Errorf("%s should not be terminal", s)
+		}
+	}
+}
+
+func TestOpenDBAndInitSchema(t *testing.T) {
+	dir := t.TempDir()
+	dbPath := filepath.Join(dir, "shellctl.db")
+
+	db, err := OpenDB(dbPath, 5000)
+	if err != nil {
+		t.Fatalf("OpenDB: %v", err)
+	}
+	defer db.Close()
+
+	if err := db.InitSchema(); err != nil {
+		t.Fatalf("InitSchema: %v", err)
+	}
+
+	// Verify table exists by inserting and reading back
+	row := &JobRow{
+		JobID:        "test-job-1",
+		ScriptPath:   "jobs/test-job-1/script",
+		OutputPath:   "jobs/test-job-1/output.log",
+		Cwd:          "/tmp",
+		TerminalCols: 80,
+		TerminalRows: 24,
+		Status:       StatusCreated,
+		SessionName:  "shellctl-test-job-1",
+		PaneTarget:   "shellctl-test-job-1:0.0",
+		CreatedAt:    "2025-01-01T00:00:00Z",
+		UpdatedAt:    "2025-01-01T00:00:00Z",
+	}
+
+	ok, err := db.InsertJob(row)
+	if err != nil {
+		t.Fatalf("InsertJob: %v", err)
+	}
+	if !ok {
+		t.Error("expected insert to succeed (ok=true)")
+	}
+
+	// Duplicate insert should return ok=false
+	ok, err = db.InsertJob(row)
+	if err != nil {
+		t.Fatalf("InsertJob duplicate: %v", err)
+	}
+	if ok {
+		t.Error("expected duplicate insert to return ok=false")
+	}
+}
+
+func TestGetJob(t *testing.T) {
+	dir := t.TempDir()
+	db := setupTestDB(t, dir)
+	defer db.Close()
+
+	insertTestJob(t, db, "job-get-1", StatusCreated)
+
+	row, err := db.GetJob("job-get-1")
+	if err != nil {
+		t.Fatalf("GetJob: %v", err)
+	}
+	if row.JobID != "job-get-1" {
+		t.Errorf("expected job_id=job-get-1, got %s", row.JobID)
+	}
+	if row.Status != StatusCreated {
+		t.Errorf("expected status=created, got %s", row.Status)
+	}
+	if row.TerminalCols != 80 {
+		t.Errorf("expected cols=80, got %d", row.TerminalCols)
+	}
+}
+
+func TestGetJobNotFound(t *testing.T) {
+	dir := t.TempDir()
+	db := setupTestDB(t, dir)
+	defer db.Close()
+
+	_, err := db.GetJob("nonexistent")
+	if err != ErrJobNotFound {
+		t.Errorf("expected ErrJobNotFound, got %v", err)
+	}
+}
+
+func TestListJobs(t *testing.T) {
+	dir := t.TempDir()
+	db := setupTestDB(t, dir)
+	defer db.Close()
+
+	insertTestJob(t, db, "job-list-1", StatusRunning)
+	insertTestJob(t, db, "job-list-2", StatusExited)
+	insertTestJob(t, db, "job-list-3", StatusCreated)
+
+	// List all
+	rows, err := db.ListJobs(nil)
+	if err != nil {
+		t.Fatalf("ListJobs: %v", err)
+	}
+	if len(rows) != 3 {
+		t.Errorf("expected 3 jobs, got %d", len(rows))
+	}
+
+	// List by status
+	rows, err = db.ListJobs([]JobStatusName{StatusRunning})
+	if err != nil {
+		t.Fatalf("ListJobs filtered: %v", err)
+	}
+	if len(rows) != 1 {
+		t.Errorf("expected 1 running job, got %d", len(rows))
+	}
+	if rows[0].JobID != "job-list-1" {
+		t.Errorf("expected job-list-1, got %s", rows[0].JobID)
+	}
+}
+
+func TestDeleteJob(t *testing.T) {
+	dir := t.TempDir()
+	db := setupTestDB(t, dir)
+	defer db.Close()
+
+	insertTestJob(t, db, "job-del-1", StatusExited)
+
+	if err := db.DeleteJob("job-del-1"); err != nil {
+		t.Fatalf("DeleteJob: %v", err)
+	}
+
+	_, err := db.GetJob("job-del-1")
+	if err != ErrJobNotFound {
+		t.Errorf("expected ErrJobNotFound after delete, got %v", err)
+	}
+}
+
+func TestDeleteJobNotFound(t *testing.T) {
+	dir := t.TempDir()
+	db := setupTestDB(t, dir)
+	defer db.Close()
+
+	err := db.DeleteJob("nonexistent")
+	if err != ErrJobNotFound {
+		t.Errorf("expected ErrJobNotFound, got %v", err)
+	}
+}
+
+func TestTransitionStatus(t *testing.T) {
+	dir := t.TempDir()
+	db := setupTestDB(t, dir)
+	defer db.Close()
+
+	insertTestJob(t, db, "job-trans-1", StatusCreated)
+
+	// Transition created → starting
+	row, err := db.TransitionStatus("job-trans-1", TransitionOpts{
+		AllowedFrom: []JobStatusName{StatusCreated},
+		Target:      StatusStarting,
+	})
+	if err != nil {
+		t.Fatalf("TransitionStatus: %v", err)
+	}
+	if row.Status != StatusStarting {
+		t.Errorf("expected starting, got %s", row.Status)
+	}
+
+	// Transition starting → running
+	row, err = db.TransitionStatus("job-trans-1", TransitionOpts{
+		AllowedFrom:         []JobStatusName{StatusStarting},
+		Target:              StatusRunning,
+		RequireExitCodeNull: true,
+	})
+	if err != nil {
+		t.Fatalf("TransitionStatus: %v", err)
+	}
+	if row.Status != StatusRunning {
+		t.Errorf("expected running, got %s", row.Status)
+	}
+
+	// Transition running → exited
+	exitCode := 0
+	row, err = db.TransitionStatus("job-trans-1", TransitionOpts{
+		AllowedFrom: []JobStatusName{StatusRunning},
+		Target:      StatusExited,
+	})
+	if err != nil {
+		t.Fatalf("TransitionStatus: %v", err)
+	}
+	if row.Status != StatusExited {
+		t.Errorf("expected exited, got %s", row.Status)
+	}
+	if row.ExitCode == nil {
+		t.Error("expected exit_code to be set after exited transition")
+	} else if *row.ExitCode != exitCode {
+		t.Errorf("expected exit_code=0, got %d", *row.ExitCode)
+	}
+	if row.EndedAt == nil {
+		t.Error("expected ended_at to be set after terminal transition")
+	}
+}
+
+func TestRecordRunnerExit(t *testing.T) {
+	dir := t.TempDir()
+	db := setupTestDB(t, dir)
+	defer db.Close()
+
+	insertTestJob(t, db, "job-exit-1", StatusRunning)
+
+	if err := db.RecordRunnerExit("job-exit-1", 42, "2025-01-15T12:00:00Z"); err != nil {
+		t.Fatalf("RecordRunnerExit: %v", err)
+	}
+
+	row, _ := db.GetJob("job-exit-1")
+	if row.Status != StatusExited {
+		t.Errorf("expected exited, got %s", row.Status)
+	}
+	if row.ExitCode == nil || *row.ExitCode != 42 {
+		t.Errorf("expected exit_code=42, got %v", row.ExitCode)
+	}
+}
+
+func TestRecordRunnerExitIdempotent(t *testing.T) {
+	dir := t.TempDir()
+	db := setupTestDB(t, dir)
+	defer db.Close()
+
+	insertTestJob(t, db, "job-exit-2", StatusExited)
+	exitCode := 10
+	row := &JobRow{
+		JobID: "job-exit-2", ScriptPath: "x", OutputPath: "y", Cwd: "/tmp",
+		TerminalCols: 80, TerminalRows: 24, Status: StatusExited,
+		SessionName: "s", PaneTarget: "p", ExitCode: &exitCode,
+		CreatedAt: "2025-01-01T00:00:00Z", UpdatedAt: "2025-01-01T00:00:00Z",
+		EndedAt: strPtr("2025-01-01T00:01:00Z"),
+	}
+	db.db.Exec(`UPDATE jobs SET exit_code=?, ended_at=? WHERE job_id=?`,
+		exitCode, "2025-01-01T00:01:00Z", "job-exit-2")
+	_ = row
+
+	// Should not overwrite existing terminal state
+	err := db.RecordRunnerExit("job-exit-2", 99, "2025-01-01T00:02:00Z")
+	if err != nil {
+		t.Fatalf("RecordRunnerExit on terminal: %v", err)
+	}
+
+	got, _ := db.GetJob("job-exit-2")
+	if got.ExitCode != nil && *got.ExitCode != 10 {
+		t.Errorf("expected exit_code=10 (preserved), got %d", *got.ExitCode)
+	}
+}
+
+// Helpers
+
+func setupTestDB(t *testing.T, dir string) *DB {
+	t.Helper()
+	dbPath := filepath.Join(dir, "shellctl.db")
+	db, err := OpenDB(dbPath, 5000)
+	if err != nil {
+		t.Fatalf("OpenDB: %v", err)
+	}
+	if err := db.InitSchema(); err != nil {
+		t.Fatalf("InitSchema: %v", err)
+	}
+	return db
+}
+
+func insertTestJob(t *testing.T, db *DB, jobID string, status JobStatusName) {
+	t.Helper()
+	row := &JobRow{
+		JobID:        jobID,
+		ScriptPath:   "jobs/" + jobID + "/script",
+		OutputPath:   "jobs/" + jobID + "/output.log",
+		Cwd:          "/tmp",
+		TerminalCols: 80,
+		TerminalRows: 24,
+		Status:       status,
+		SessionName:  "shellctl-" + jobID,
+		PaneTarget:   "shellctl-" + jobID + ":0.0",
+		CreatedAt:    "2025-01-01T00:00:00Z",
+		UpdatedAt:    "2025-01-01T00:00:00Z",
+	}
+	ok, err := db.InsertJob(row)
+	if err != nil || !ok {
+		t.Fatalf("InsertJob(%s): err=%v ok=%v", jobID, err, ok)
+	}
+}
+
+func strPtr(s string) *string {
+	return &s
+}
+
+func TestMain(m *testing.M) {
+	os.Exit(m.Run())
+}

shellctl-go/internal/server/errors.go

@@ -0,0 +1,22 @@
+package server
+
+import "fmt"
+
+// ServerError is a structured error with HTTP status code and machine-readable code.
+type ServerError struct {
+	StatusCode int
+	Code       string
+	Message    string
+}
+
+func (e *ServerError) Error() string {
+	return fmt.Sprintf("[%d] %s: %s", e.StatusCode, e.Code, e.Message)
+}
+
+// Common sentinel errors.
+var ErrJobNotFound = &ServerError{StatusCode: 404, Code: "job_not_found", Message: "Unknown job id"}
+
+// NewServerError creates a new ServerError.
+func NewServerError(statusCode int, code, message string) *ServerError {
+	return &ServerError{StatusCode: statusCode, Code: code, Message: message}
+}

shellctl-go/internal/server/output.go

@@ -0,0 +1,167 @@
+package server
+
+import (
+	"fmt"
+	"os"
+	"unicode/utf8"
+)
+
+// OutputWindow represents a UTF-8-safe slice of output.log.
+type OutputWindow struct {
+	Output    string `json:"output"`
+	Offset    int    `json:"offset"`
+	Truncated bool   `json:"truncated"`
+}
+
+// ReadOutputWindow reads a forward UTF-8-safe slice from an output file.
+func ReadOutputWindow(path string, offset, limit int) (*OutputWindow, error) {
+	info, err := os.Stat(path)
+	if os.IsNotExist(err) {
+		if offset == 0 {
+			return &OutputWindow{Output: "", Offset: 0, Truncated: false}, nil
+		}
+		return nil, NewServerError(400, "invalid_offset", "offset exceeds current file size 0")
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	size := int(info.Size())
+	if offset > size {
+		return nil, NewServerError(400, "invalid_offset",
+			fmt.Sprintf("offset %d exceeds current file size %d", offset, size))
+	}
+	if offset == size {
+		return &OutputWindow{Output: "", Offset: offset, Truncated: false}, nil
+	}
+
+	f, err := os.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+
+	// Read up to limit+4 bytes to handle UTF-8 boundary
+	readSize := limit + 4
+	if offset+readSize > size {
+		readSize = size - offset
+	}
+
+	buf := make([]byte, readSize)
+	_, err = f.ReadAt(buf, int64(offset))
+	if err != nil {
+		return nil, err
+	}
+
+	// Advance past any UTF-8 continuation bytes at the start
+	startShift := advanceToUTF8Boundary(buf, 0)
+	data := buf[startShift:]
+	budget := limit - startShift
+	if budget < 0 {
+		budget = 0
+	}
+
+	// Find the longest valid UTF-8 prefix within budget
+	consumed := validUTF8PrefixLen(data, budget)
+	if consumed == 0 && len(data) > 0 {
+		// At least consume one complete rune
+		consumed = firstCompleteRuneLen(data)
+	}
+
+	outputBytes := data[:consumed]
+	newOffset := offset + startShift + consumed
+	truncated := newOffset < size
+
+	return &OutputWindow{
+		Output:    string(outputBytes),
+		Offset:    newOffset,
+		Truncated: truncated,
+	}, nil
+}
+
+// TailOutputWindow reads a UTF-8-safe tail snapshot from an output file.
+func TailOutputWindow(path string, limit int) (*OutputWindow, error) {
+	info, err := os.Stat(path)
+	if os.IsNotExist(err) {
+		return &OutputWindow{Output: "", Offset: 0, Truncated: false}, nil
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	size := int(info.Size())
+	if size == 0 {
+		return &OutputWindow{Output: "", Offset: 0, Truncated: false}, nil
+	}
+
+	start := size - limit
+	if start < 0 {
+		start = 0
+	}
+	paddedStart := start - 4
+	if paddedStart < 0 {
+		paddedStart = 0
+	}
+
+	f, err := os.Open(path)
+	if err != nil {
+		return nil, err
+	}
+	defer f.Close()
+
+	readLen := size - paddedStart
+	buf := make([]byte, readLen)
+	_, err = f.ReadAt(buf, int64(paddedStart))
+	if err != nil {
+		return nil, err
+	}
+
+	relativeStart := advanceToUTF8Boundary(buf, start-paddedStart)
+	payload := buf[relativeStart:]
+	consumed := validUTF8PrefixLen(payload, len(payload))
+	outputBytes := payload[:consumed]
+
+	return &OutputWindow{
+		Output:    string(outputBytes),
+		Offset:    paddedStart + relativeStart + consumed,
+		Truncated: false,
+	}, nil
+}
+
+// advanceToUTF8Boundary skips continuation bytes (10xxxxxx) at position start.
+func advanceToUTF8Boundary(data []byte, start int) int {
+	for start < len(data) && isUTF8Continuation(data[start]) {
+		start++
+	}
+	return start
+}
+
+// isUTF8Continuation returns true if byte is a UTF-8 continuation byte.
+func isUTF8Continuation(b byte) bool {
+	return (b & 0xC0) == 0x80
+}
+
+// validUTF8PrefixLen returns the length of the longest valid UTF-8 prefix up to maxLen.
+func validUTF8PrefixLen(data []byte, maxLen int) int {
+	if maxLen > len(data) {
+		maxLen = len(data)
+	}
+	end := maxLen
+	for end > 0 {
+		if utf8.Valid(data[:end]) {
+			return end
+		}
+		end--
+	}
+	return 0
+}
+
+// firstCompleteRuneLen returns the byte length of the first complete rune in data.
+func firstCompleteRuneLen(data []byte) int {
+	for end := 1; end <= len(data) && end <= 4; end++ {
+		if utf8.Valid(data[:end]) {
+			return end
+		}
+	}
+	return 0
+}

shellctl-go/internal/server/output_test.go

@@ -0,0 +1,253 @@
+package server
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+)
+
+func TestReadOutputWindowEmptyFile(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "output.log")
+	os.WriteFile(path, []byte{}, 0644)
+
+	w, err := ReadOutputWindow(path, 0, 1024)
+	if err != nil {
+		t.Fatalf("ReadOutputWindow: %v", err)
+	}
+	if w.Output != "" {
+		t.Errorf("expected empty output, got %q", w.Output)
+	}
+	if w.Offset != 0 {
+		t.Errorf("expected offset=0, got %d", w.Offset)
+	}
+	if w.Truncated {
+		t.Error("expected truncated=false")
+	}
+}
+
+func TestReadOutputWindowNonexistentFile(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "nonexistent.log")
+
+	w, err := ReadOutputWindow(path, 0, 1024)
+	if err != nil {
+		t.Fatalf("ReadOutputWindow: %v", err)
+	}
+	if w.Output != "" {
+		t.Errorf("expected empty output, got %q", w.Output)
+	}
+}
+
+func TestReadOutputWindowFullContent(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "output.log")
+	content := "hello\nworld\n"
+	os.WriteFile(path, []byte(content), 0644)
+
+	w, err := ReadOutputWindow(path, 0, 1024)
+	if err != nil {
+		t.Fatalf("ReadOutputWindow: %v", err)
+	}
+	if w.Output != content {
+		t.Errorf("expected %q, got %q", content, w.Output)
+	}
+	if w.Truncated {
+		t.Error("expected truncated=false")
+	}
+}
+
+func TestReadOutputWindowWithOffset(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "output.log")
+	content := "hello\nworld\n"
+	os.WriteFile(path, []byte(content), 0644)
+
+	w, err := ReadOutputWindow(path, 6, 1024)
+	if err != nil {
+		t.Fatalf("ReadOutputWindow: %v", err)
+	}
+	if w.Output != "world\n" {
+		t.Errorf("expected 'world\\n', got %q", w.Output)
+	}
+	if w.Offset != len(content) {
+		t.Errorf("expected offset=%d, got %d", len(content), w.Offset)
+	}
+}
+
+func TestReadOutputWindowTruncated(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "output.log")
+	content := "0123456789"
+	os.WriteFile(path, []byte(content), 0644)
+
+	// Read only 5 bytes
+	w, err := ReadOutputWindow(path, 0, 5)
+	if err != nil {
+		t.Fatalf("ReadOutputWindow: %v", err)
+	}
+	if w.Output != "01234" {
+		t.Errorf("expected '01234', got %q", w.Output)
+	}
+	if !w.Truncated {
+		t.Error("expected truncated=true")
+	}
+	if w.Offset != 5 {
+		t.Errorf("expected offset=5, got %d", w.Offset)
+	}
+}
+
+func TestReadOutputWindowOffsetExceedsSize(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "output.log")
+	os.WriteFile(path, []byte("short"), 0644)
+
+	_, err := ReadOutputWindow(path, 100, 1024)
+	if err == nil {
+		t.Error("expected error for offset > size")
+	}
+}
+
+func TestReadOutputWindowOffsetAtEnd(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "output.log")
+	content := "hello"
+	os.WriteFile(path, []byte(content), 0644)
+
+	w, err := ReadOutputWindow(path, len(content), 1024)
+	if err != nil {
+		t.Fatalf("ReadOutputWindow: %v", err)
+	}
+	if w.Output != "" {
+		t.Errorf("expected empty output at end, got %q", w.Output)
+	}
+	if w.Offset != len(content) {
+		t.Errorf("expected offset=%d, got %d", len(content), w.Offset)
+	}
+}
+
+func TestReadOutputWindowMultibyteUTF8(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "output.log")
+	content := "hello 世界\n"
+	os.WriteFile(path, []byte(content), 0644)
+
+	w, err := ReadOutputWindow(path, 0, 1024)
+	if err != nil {
+		t.Fatalf("ReadOutputWindow: %v", err)
+	}
+	if w.Output != content {
+		t.Errorf("expected %q, got %q", content, w.Output)
+	}
+}
+
+func TestReadOutputWindowMultibyteTruncated(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "output.log")
+	// "世" = 3 bytes (E4 B8 96), "界" = 3 bytes (E7 95 8C)
+	content := "世界"
+	os.WriteFile(path, []byte(content), 0644)
+
+	// Read only 4 bytes — should not split a multibyte char
+	w, err := ReadOutputWindow(path, 0, 4)
+	if err != nil {
+		t.Fatalf("ReadOutputWindow: %v", err)
+	}
+	// Should get "世" (3 bytes), not partial "界"
+	if w.Output != "世" {
+		t.Errorf("expected '世', got %q", w.Output)
+	}
+	if !w.Truncated {
+		t.Error("expected truncated=true")
+	}
+}
+
+func TestTailOutputWindow(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "output.log")
+	content := "line1\nline2\nline3\n"
+	os.WriteFile(path, []byte(content), 0644)
+
+	w, err := TailOutputWindow(path, 100)
+	if err != nil {
+		t.Fatalf("TailOutputWindow: %v", err)
+	}
+	if w.Output != content {
+		t.Errorf("expected %q, got %q", content, w.Output)
+	}
+}
+
+func TestTailOutputWindowLimited(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "output.log")
+	content := "0123456789"
+	os.WriteFile(path, []byte(content), 0644)
+
+	w, err := TailOutputWindow(path, 5)
+	if err != nil {
+		t.Fatalf("TailOutputWindow: %v", err)
+	}
+	if w.Output != "56789" {
+		t.Errorf("expected '56789', got %q", w.Output)
+	}
+}
+
+func TestTailOutputWindowNonexistent(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "nonexistent.log")
+
+	w, err := TailOutputWindow(path, 100)
+	if err != nil {
+		t.Fatalf("TailOutputWindow: %v", err)
+	}
+	if w.Output != "" {
+		t.Errorf("expected empty, got %q", w.Output)
+	}
+}
+
+func TestTailOutputWindowEmpty(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "empty.log")
+	os.WriteFile(path, []byte{}, 0644)
+
+	w, err := TailOutputWindow(path, 100)
+	if err != nil {
+		t.Fatalf("TailOutputWindow: %v", err)
+	}
+	if w.Output != "" {
+		t.Errorf("expected empty, got %q", w.Output)
+	}
+}
+
+func TestIsUTF8Continuation(t *testing.T) {
+	// ASCII bytes are not continuation bytes
+	if isUTF8Continuation('a') {
+		t.Error("'a' should not be a continuation byte")
+	}
+	// 0x80-0xBF are continuation bytes
+	if !isUTF8Continuation(0x80) {
+		t.Error("0x80 should be a continuation byte")
+	}
+	if !isUTF8Continuation(0xBF) {
+		t.Error("0xBF should be a continuation byte")
+	}
+	// 0xC0 is not a continuation byte (it's a lead byte)
+	if isUTF8Continuation(0xC0) {
+		t.Error("0xC0 should not be a continuation byte")
+	}
+}
+
+func TestAdvanceToUTF8Boundary(t *testing.T) {
+	// "世" = E4 B8 96, continuation bytes at index 1 and 2
+	data := []byte{0xE4, 0xB8, 0x96, 0x41} // 世 + A
+	// Start at 1, should advance to 3 (the 'A')
+	result := advanceToUTF8Boundary(data, 1)
+	if result != 3 {
+		t.Errorf("expected 3, got %d", result)
+	}
+	// Start at 0 (lead byte), should stay at 0
+	result = advanceToUTF8Boundary(data, 0)
+	if result != 0 {
+		t.Errorf("expected 0, got %d", result)
+	}
+}

shellctl-go/internal/server/runtime.go

@@ -0,0 +1,35 @@
+package server
+
+import (
+	"crypto/rand"
+	"encoding/hex"
+	"fmt"
+	"time"
+)
+
+// GenerateJobID produces a short random hex job identifier.
+func GenerateJobID() string {
+	b := make([]byte, 8)
+	_, _ = rand.Read(b)
+	return hex.EncodeToString(b)
+}
+
+// FormatTimestamp returns an ISO-8601 UTC timestamp string.
+func FormatTimestamp(t time.Time) string {
+	return t.UTC().Truncate(time.Second).Format("2006-01-02T15:04:05Z")
+}
+
+// ParseTimestamp parses an ISO-8601 UTC timestamp.
+func ParseTimestamp(s string) (time.Time, error) {
+	return time.Parse("2006-01-02T15:04:05Z", s)
+}
+
+// JobSessionName returns the tmux session name for a job.
+func JobSessionName(jobID string) string {
+	return fmt.Sprintf("shellctl-%s", jobID)
+}
+
+// JobPaneTarget returns the tmux pane target for a job.
+func JobPaneTarget(jobID string) string {
+	return fmt.Sprintf("%s:0.0", JobSessionName(jobID))
+}

shellctl-go/internal/server/runtime_test.go

@@ -0,0 +1,71 @@
+package server
+
+import (
+	"testing"
+	"time"
+)
+
+func TestGenerateJobID(t *testing.T) {
+	id1 := GenerateJobID()
+	id2 := GenerateJobID()
+	if id1 == id2 {
+		t.Error("expected different job IDs")
+	}
+	if len(id1) != 16 {
+		t.Errorf("expected 16-char hex ID, got %d chars: %s", len(id1), id1)
+	}
+}
+
+func TestGenerateJobIDUniqueness(t *testing.T) {
+	seen := make(map[string]bool, 1000)
+	for i := 0; i < 1000; i++ {
+		id := GenerateJobID()
+		if seen[id] {
+			t.Fatalf("collision at iteration %d: %s", i, id)
+		}
+		seen[id] = true
+	}
+}
+
+func TestFormatTimestamp(t *testing.T) {
+	ts := time.Date(2025, 1, 15, 12, 30, 45, 0, time.UTC)
+	formatted := FormatTimestamp(ts)
+	expected := "2025-01-15T12:30:45Z"
+	if formatted != expected {
+		t.Errorf("expected %q, got %q", expected, formatted)
+	}
+}
+
+func TestParseTimestamp(t *testing.T) {
+	parsed, err := ParseTimestamp("2025-01-15T12:30:45Z")
+	if err != nil {
+		t.Fatalf("parse error: %v", err)
+	}
+	if parsed.Year() != 2025 || parsed.Month() != 1 || parsed.Day() != 15 {
+		t.Errorf("unexpected parsed time: %v", parsed)
+	}
+	if parsed.Hour() != 12 || parsed.Minute() != 30 || parsed.Second() != 45 {
+		t.Errorf("unexpected parsed time: %v", parsed)
+	}
+}
+
+func TestParseTimestampInvalid(t *testing.T) {
+	_, err := ParseTimestamp("not-a-timestamp")
+	if err == nil {
+		t.Error("expected error for invalid timestamp")
+	}
+}
+
+func TestJobSessionName(t *testing.T) {
+	name := JobSessionName("abc123")
+	if name != "shellctl-abc123" {
+		t.Errorf("expected 'shellctl-abc123', got %q", name)
+	}
+}
+
+func TestJobPaneTarget(t *testing.T) {
+	target := JobPaneTarget("abc123")
+	if target != "shellctl-abc123:0.0" {
+		t.Errorf("expected 'shellctl-abc123:0.0', got %q", target)
+	}
+}

shellctl-go/internal/server/server.go

@@ -0,0 +1,6 @@
+// Package server implements the shellctl HTTP API and job lifecycle service.
+//
+// This is the Go equivalent of shellctl/server/ in the Python implementation.
+// It manages tmux-backed shell jobs with SQLite persistence and provides
+// REST endpoints for run, wait, input, terminate, and delete operations.
+package server

shellctl-go/internal/server/service.go

@@ -0,0 +1,902 @@
+package server
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+)
+
+// Service is the core job lifecycle manager backed by SQLite and tmux.
+type Service struct {
+	config       *Config
+	db           *DB
+	tmux         *TmuxController
+	startingJobs map[string]bool
+	mu           sync.Mutex
+	cancelGC     context.CancelFunc
+	cancelMon    context.CancelFunc
+}
+
+// NewService creates a new shellctl service.
+func NewService(config *Config) *Service {
+	return &Service{
+		config:       config,
+		tmux:         NewTmuxController(config),
+		startingJobs: make(map[string]bool),
+	}
+}
+
+// Initialize performs the full server startup (prepare + reconcile + gc).
+func (s *Service) Initialize() error {
+	if err := s.PrepareRuntime(); err != nil {
+		return err
+	}
+	if err := s.Reconcile(); err != nil {
+		return err
+	}
+	return s.GCOnce()
+}
+
+// PrepareRuntime sets up directories, DB schema, runner script, and tmux server.
+func (s *Service) PrepareRuntime() error {
+	if err := os.MkdirAll(s.config.StateDir, 0700); err != nil {
+		return err
+	}
+	if err := os.MkdirAll(s.config.RuntimeDir, 0700); err != nil {
+		return err
+	}
+	if err := os.MkdirAll(s.config.JobsDir(), 0700); err != nil {
+		return err
+	}
+	runnerDir := filepath.Dir(s.config.RunnerPath())
+	if err := os.MkdirAll(runnerDir, 0700); err != nil {
+		return err
+	}
+
+	db, err := OpenDB(s.config.DBPath(), s.config.SQLiteBusyTimeoutMs)
+	if err != nil {
+		return err
+	}
+	s.db = db
+
+	if err := s.db.InitSchema(); err != nil {
+		return err
+	}
+
+	s.installRunner()
+	return s.tmux.StartServer()
+}
+
+// Shutdown stops background goroutines and closes the database.
+func (s *Service) Shutdown() {
+	if s.cancelGC != nil {
+		s.cancelGC()
+	}
+	if s.cancelMon != nil {
+		s.cancelMon()
+	}
+	if s.db != nil {
+		s.db.Close()
+	}
+}
+
+// StartBackgroundGC starts the periodic GC goroutine.
+func (s *Service) StartBackgroundGC() {
+	ctx, cancel := context.WithCancel(context.Background())
+	s.cancelGC = cancel
+	go func() {
+		ticker := time.NewTicker(s.config.GCInterval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				s.GCOnce()
+			}
+		}
+	}()
+}
+
+// StartBackgroundPipeMonitor starts the periodic pipe health check goroutine.
+func (s *Service) StartBackgroundPipeMonitor() {
+	ctx, cancel := context.WithCancel(context.Background())
+	s.cancelMon = cancel
+	go func() {
+		ticker := time.NewTicker(s.config.PipeMonitorInterval)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				s.CheckRunningJobsPipeHealth()
+			}
+		}
+	}()
+}
+
+// RunJob creates and starts a new tmux-backed job, then waits for initial output.
+func (s *Service) RunJob(req *RunJobRequest) (*JobResult, error) {
+	log.Printf("RunJob: script=%d bytes, cwd=%v, env_keys=%d", len(req.Script), req.Cwd, len(req.Env))
+	cwd, err := s.resolveCwd(req.Cwd)
+	if err != nil {
+		log.Printf("RunJob: resolveCwd failed: %v", err)
+		return nil, err
+	}
+
+	cols := s.config.DefaultTerminalCols
+	rows := s.config.DefaultTerminalRows
+	if req.Terminal != nil {
+		cols = req.Terminal.Cols
+		rows = req.Terminal.Rows
+	}
+
+	timeout := s.config.DefaultTimeout
+	if req.Timeout > 0 {
+		timeout = time.Duration(req.Timeout * float64(time.Second))
+	}
+	outputLimit := s.config.DefaultOutputLimitBytes
+	if req.OutputLimit > 0 {
+		outputLimit = req.OutputLimit
+	}
+	idleFlush := s.config.IdleFlushDuration
+	if req.IdleFlushSeconds > 0 {
+		idleFlush = time.Duration(req.IdleFlushSeconds * float64(time.Second))
+	}
+
+	createdAt := FormatTimestamp(time.Now())
+
+	// Allocate job directory and insert DB row
+	jobID, jobDir, err := s.allocateJobDir()
+	if err != nil {
+		return nil, err
+	}
+
+	s.mu.Lock()
+	s.startingJobs[jobID] = true
+	s.mu.Unlock()
+
+	// Write script and env files
+	scriptPath := filepath.Join(jobDir, "script")
+	outputPath := filepath.Join(jobDir, "output.log")
+	envPath := filepath.Join(jobDir, ".job-env.json")
+
+	if err := os.WriteFile(scriptPath, []byte(req.Script), 0600); err != nil {
+		s.cleanupStarting(jobID, jobDir)
+		return nil, err
+	}
+	envJSON := "{}"
+	if req.Env != nil {
+		pairs := make([]string, 0, len(req.Env))
+		for k, v := range req.Env {
+			pairs = append(pairs, fmt.Sprintf("%q:%q", k, v))
+		}
+		envJSON = "{" + strings.Join(pairs, ",") + "}"
+	}
+	if err := os.WriteFile(envPath, []byte(envJSON), 0600); err != nil {
+		s.cleanupStarting(jobID, jobDir)
+		return nil, err
+	}
+	if err := os.WriteFile(outputPath, []byte{}, 0600); err != nil {
+		s.cleanupStarting(jobID, jobDir)
+		return nil, err
+	}
+
+	row := &JobRow{
+		JobID:        jobID,
+		ScriptPath:   fmt.Sprintf("jobs/%s/script", jobID),
+		OutputPath:   fmt.Sprintf("jobs/%s/output.log", jobID),
+		Cwd:          cwd,
+		TerminalCols: cols,
+		TerminalRows: rows,
+		Status:       StatusCreated,
+		SessionName:  JobSessionName(jobID),
+		PaneTarget:   JobPaneTarget(jobID),
+		CreatedAt:    createdAt,
+		UpdatedAt:    createdAt,
+	}
+
+	ok, err := s.db.InsertJob(row)
+	if err != nil {
+		s.cleanupStarting(jobID, jobDir)
+		return nil, err
+	}
+	if !ok {
+		s.cleanupStarting(jobID, jobDir)
+		return nil, NewServerError(500, "job_id_collision", "Failed to allocate a unique job id")
+	}
+
+	// Transition to starting
+	s.db.TransitionStatus(jobID, TransitionOpts{
+		AllowedFrom: []JobStatusName{StatusCreated},
+		Target:      StatusStarting,
+	})
+
+	// Create tmux session and enable output pipe
+	log.Printf("RunJob [%s]: starting job, cwd=%s", jobID, cwd)
+	startErr := s.startJob(jobID, jobDir, cwd, cols, rows)
+	if startErr != nil {
+		log.Printf("RunJob [%s]: start failed: %v", jobID, startErr)
+		reason := "start_failed"
+		msg := startErr.Error()
+		s.db.TransitionStatus(jobID, TransitionOpts{
+			AllowedFrom: []JobStatusName{StatusCreated, StatusStarting, StatusRunning},
+			Target:      StatusFailed,
+			Reason:      &reason,
+			Message:     &msg,
+		})
+		s.tmux.CleanupSession(jobID)
+	} else {
+		log.Printf("RunJob [%s]: started successfully", jobID)
+	}
+
+	s.mu.Lock()
+	delete(s.startingJobs, jobID)
+	s.mu.Unlock()
+
+	// Wait for initial output
+	return s.WaitJob(jobID, &WaitJobRequest{
+		Offset:           0,
+		Timeout:          timeout.Seconds(),
+		OutputLimit:      outputLimit,
+		IdleFlushSeconds: idleFlush.Seconds(),
+	})
+}
+
+func (s *Service) startJob(jobID, jobDir, cwd string, cols, rows int) error {
+	log.Printf("startJob [%s]: creating tmux session", jobID)
+	if err := s.tmux.CreateJobSession(jobID, jobDir, cwd, cols, rows); err != nil {
+		log.Printf("startJob [%s]: tmux session failed: %v", jobID, err)
+		return err
+	}
+
+	pipeReadyPath := filepath.Join(jobDir, ".pipe-ready")
+	log.Printf("startJob [%s]: enabling output pipe", jobID)
+	if err := s.tmux.EnableOutputPipe(jobID, jobDir, pipeReadyPath); err != nil {
+		log.Printf("startJob [%s]: pipe-pane failed: %v", jobID, err)
+		return err
+	}
+
+	// Wait for pipe ready handshake
+	if err := s.waitForPipeReady(jobID, pipeReadyPath); err != nil {
+		log.Printf("startJob [%s]: pipe-ready timeout: %v", jobID, err)
+		return err
+	}
+
+	// Open start gate
+	log.Printf("startJob [%s]: opening start gate", jobID)
+	gateFile := filepath.Join(jobDir, "start-gate")
+	if err := os.WriteFile(gateFile, []byte{}, 0600); err != nil {
+		return err
+	}
+
+	// Transition to running
+	s.db.TransitionStatus(jobID, TransitionOpts{
+		AllowedFrom:         []JobStatusName{StatusStarting},
+		Target:              StatusRunning,
+		RequireExitCodeNull: true,
+	})
+
+	// Clean up ready file
+	os.Remove(pipeReadyPath)
+	return nil
+}
+
+func (s *Service) waitForPipeReady(jobID, readyFile string) error {
+	deadline := time.Now().Add(s.config.PipeReadyTimeout)
+	for {
+		if _, err := os.Stat(readyFile); err == nil {
+			// Ready file exists, verify pipe is active
+			active, err := s.tmux.IsOutputPipeActive(jobID)
+			if err != nil {
+				return err
+			}
+			if active == nil {
+				return NewServerError(500, "pipe_failed",
+					"tmux pane disappeared after ready-file handshake")
+			}
+			if *active {
+				return nil
+			}
+		}
+
+		if time.Now().After(deadline) {
+			return NewServerError(500, "pipe_failed",
+				fmt.Sprintf("timed out waiting for pipe ready (%v)", s.config.PipeReadyTimeout))
+		}
+		time.Sleep(s.config.PollInterval)
+	}
+}
+
+// WaitJob blocks until output, completion, truncation, or timeout.
+func (s *Service) WaitJob(jobID string, req *WaitJobRequest) (*JobResult, error) {
+	row, err := s.db.GetJob(jobID)
+	if err != nil {
+		return nil, err
+	}
+
+	outputPath := s.outputLogPath(row)
+	timeout := time.Duration(req.Timeout * float64(time.Second))
+	outputLimit := req.OutputLimit
+	if outputLimit <= 0 {
+		outputLimit = s.config.DefaultOutputLimitBytes
+	}
+	idleFlush := time.Duration(req.IdleFlushSeconds * float64(time.Second))
+
+	deadline := time.Now().Add(timeout)
+	var lastGrowthAt *time.Time
+
+	for {
+		view, err := s.GetJobStatus(jobID)
+		if err != nil {
+			return nil, err
+		}
+
+		currentSize := fileSize(outputPath)
+
+		if currentSize > int64(req.Offset) {
+			now := time.Now()
+			lastGrowthAt = &now
+		}
+
+		if view.Done {
+			window, err := ReadOutputWindow(outputPath, req.Offset, outputLimit)
+			if err != nil {
+				return nil, err
+			}
+			return s.jobResultFromView(view, row, window), nil
+		}
+
+		if currentSize > int64(req.Offset) {
+			window, err := ReadOutputWindow(outputPath, req.Offset, outputLimit)
+			if err != nil {
+				return nil, err
+			}
+			if window.Truncated {
+				return s.jobResultFromView(view, row, window), nil
+			}
+			if lastGrowthAt != nil {
+				if time.Since(*lastGrowthAt) >= idleFlush {
+					return s.jobResultFromView(view, row, window), nil
+				}
+			}
+		}
+
+		if time.Now().After(deadline) {
+			var window *OutputWindow
+			if currentSize > int64(req.Offset) {
+				window, err = ReadOutputWindow(outputPath, req.Offset, outputLimit)
+				if err != nil {
+					return nil, err
+				}
+			} else {
+				window = &OutputWindow{Output: "", Offset: req.Offset, Truncated: false}
+			}
+			return s.jobResultFromView(view, row, window), nil
+		}
+
+		time.Sleep(s.config.PollInterval)
+	}
+}
+
+// TailJob returns the tail of a job's output.
+func (s *Service) TailJob(jobID string, outputLimit int) (*JobResult, error) {
+	row, err := s.db.GetJob(jobID)
+	if err != nil {
+		return nil, err
+	}
+	view, err := s.GetJobStatus(jobID)
+	if err != nil {
+		return nil, err
+	}
+	window, err := TailOutputWindow(s.outputLogPath(row), outputLimit)
+	if err != nil {
+		return nil, err
+	}
+	return s.jobResultFromView(view, row, window), nil
+}
+
+// GetJobStatus materializes the current status from SQLite + live tmux state.
+func (s *Service) GetJobStatus(jobID string) (*JobStatusView, error) {
+	sessionExists, pipeActive, err := s.liveRuntimeState(jobID)
+	if err != nil {
+		return nil, err
+	}
+	return s.materializeStatusView(jobID, sessionExists, pipeActive)
+}
+
+// ListJobs returns recent jobs, optionally filtered by status.
+func (s *Service) ListJobs(status *JobStatusName, limit int) (*ListJobsResponse, error) {
+	rows, err := s.db.ListJobs(nil)
+	if err != nil {
+		return nil, err
+	}
+
+	var items []JobInfo
+	for _, row := range rows {
+		view, err := s.GetJobStatus(row.JobID)
+		if err != nil {
+			if isNotFound(err) {
+				continue
+			}
+			return nil, err
+		}
+		if status != nil && view.Status != *status {
+			continue
+		}
+		items = append(items, JobInfo{
+			JobID:     view.JobID,
+			Status:    view.Status,
+			CreatedAt: view.CreatedAt,
+			StartedAt: view.StartedAt,
+			EndedAt:   view.EndedAt,
+		})
+		if len(items) >= limit {
+			break
+		}
+	}
+	return &ListJobsResponse{Jobs: items}, nil
+}
+
+// SendInput sends input to a running job and waits.
+func (s *Service) SendInput(jobID string, req *InputJobRequest) (*JobResult, error) {
+	view, err := s.GetJobStatus(jobID)
+	if err != nil {
+		return nil, err
+	}
+	if view.Done {
+		return nil, NewServerError(409, "job_not_running", fmt.Sprintf("Job %s is already terminal", jobID))
+	}
+
+	if err := s.tmux.SendInput(jobID, req.Text); err != nil {
+		// Check if job became terminal in the meantime
+		if se, ok := err.(*ServerError); ok && se.Code == "tmux_target_missing" {
+			view, _ = s.GetJobStatus(jobID)
+			if view != nil && view.Done {
+				return nil, NewServerError(409, "job_not_running", fmt.Sprintf("Job %s is already terminal", jobID))
+			}
+		}
+		return nil, err
+	}
+
+	return s.WaitJob(jobID, &WaitJobRequest{
+		Offset:           req.Offset,
+		Timeout:          req.Timeout,
+		OutputLimit:      req.OutputLimit,
+		IdleFlushSeconds: req.IdleFlushSeconds,
+	})
+}
+
+// TerminateJob terminates a running job.
+func (s *Service) TerminateJob(jobID string, graceSeconds float64) (*JobStatusView, error) {
+	view, err := s.GetJobStatus(jobID)
+	if err != nil {
+		return nil, err
+	}
+	if view.Done {
+		s.tmux.CleanupSession(jobID)
+		return view, nil
+	}
+
+	s.db.TransitionStatus(jobID, TransitionOpts{
+		AllowedFrom:       []JobStatusName{StatusCreated, StatusStarting, StatusRunning},
+		AllowOverrideFrom: []JobStatusName{StatusExited},
+		Target:            StatusTerminated,
+	})
+
+	s.tmux.SendInterrupt(jobID)
+	if graceSeconds > 0 {
+		time.Sleep(time.Duration(graceSeconds * float64(time.Second)))
+	}
+	s.tmux.CleanupSession(jobID)
+
+	return s.GetJobStatus(jobID)
+}
+
+// DeleteJob deletes a job and its artifacts.
+func (s *Service) DeleteJob(jobID string, force bool, graceSeconds float64) (*DeleteJobResponse, error) {
+	view, err := s.GetJobStatus(jobID)
+	if err != nil {
+		return nil, err
+	}
+	if !view.Done {
+		if !force {
+			return nil, NewServerError(409, "job_running", fmt.Sprintf("Job %s is still running", jobID))
+		}
+		s.TerminateJob(jobID, graceSeconds)
+	}
+
+	s.tmux.CleanupSession(jobID)
+	if err := s.db.DeleteJob(jobID); err != nil {
+		return nil, err
+	}
+	os.RemoveAll(filepath.Join(s.config.JobsDir(), jobID))
+	return &DeleteJobResponse{JobID: jobID, Deleted: true}, nil
+}
+
+// Reconcile synchronizes SQLite rows with live tmux state.
+func (s *Service) Reconcile() error {
+	rows, err := s.db.ListJobs(nil)
+	if err != nil {
+		return err
+	}
+	for _, row := range rows {
+		view, err := s.GetJobStatus(row.JobID)
+		if err != nil {
+			if isNotFound(err) {
+				continue
+			}
+			return err
+		}
+		if view.Done {
+			s.tmux.CleanupSession(row.JobID)
+		}
+	}
+	return nil
+}
+
+// GCOnce removes expired terminal jobs.
+func (s *Service) GCOnce() error {
+	cutoff := time.Now().Add(-s.config.GCFinishedJobRetention)
+	rows, err := s.db.ListJobs(nil)
+	if err != nil {
+		return err
+	}
+	for _, row := range rows {
+		view, err := s.GetJobStatus(row.JobID)
+		if err != nil {
+			if isNotFound(err) {
+				continue
+			}
+			return err
+		}
+		if !view.Done || view.EndedAt == nil {
+			continue
+		}
+		endedAt, err := ParseTimestamp(*view.EndedAt)
+		if err != nil {
+			continue
+		}
+		if endedAt.After(cutoff) {
+			continue
+		}
+		s.tmux.CleanupSession(row.JobID)
+		s.db.DeleteJob(row.JobID)
+		os.RemoveAll(filepath.Join(s.config.JobsDir(), row.JobID))
+	}
+	return nil
+}
+
+// CheckRunningJobsPipeHealth fails running jobs whose pipe died.
+func (s *Service) CheckRunningJobsPipeHealth() {
+	rows, _ := s.db.ListJobs([]JobStatusName{StatusRunning})
+	for _, row := range rows {
+		s.GetJobStatus(row.JobID)
+	}
+}
+
+func (s *Service) materializeStatusView(jobID string, sessionExists bool, pipeActive *bool) (*JobStatusView, error) {
+	row, err := s.db.GetJob(jobID)
+	if err != nil {
+		return nil, err
+	}
+
+	status := row.Status
+
+	if status.IsTerminal() {
+		// Ensure ended_at is set
+		if row.EndedAt == nil {
+			now := FormatTimestamp(time.Now())
+			s.db.TransitionStatus(jobID, TransitionOpts{
+				AllowedFrom: []JobStatusName{status},
+				Target:      status,
+				EndedAt:     now,
+			})
+			if r, err := s.db.GetJob(jobID); err == nil {
+				row = r
+			}
+		}
+	} else if row.ExitCode != nil {
+		// Already has exit code, transition to exited
+		if r, err := s.db.TransitionStatus(jobID, TransitionOpts{
+			AllowedFrom: []JobStatusName{StatusCreated, StatusStarting, StatusRunning},
+			Target:      StatusExited,
+		}); err == nil {
+			row = r
+		}
+	} else if exit := s.drainedNormalExitMetadata(jobID); exit != nil {
+		// Recover from drained exit artifacts
+		s.db.RecordRunnerExit(jobID, exit.exitCode, exit.endedAt)
+		if r, err := s.db.GetJob(jobID); err == nil {
+			row = r
+		}
+	} else if sessionExists {
+		if pipeActive != nil && !*pipeActive {
+			s.mu.Lock()
+			isStarting := s.startingJobs[jobID]
+			s.mu.Unlock()
+			if !isStarting {
+				reason := "pipe_failed"
+				msg := "The tmux output pipe stopped while the job was still running."
+				if r, err := s.db.TransitionStatus(jobID, TransitionOpts{
+					AllowedFrom: []JobStatusName{StatusCreated, StatusStarting, StatusRunning},
+					Target:      StatusFailed,
+					Reason:      &reason,
+					Message:     &msg,
+				}); err == nil {
+					row = r
+				}
+			}
+		} else if status == StatusCreated || status == StatusStarting {
+			s.mu.Lock()
+			isStarting := s.startingJobs[jobID]
+			s.mu.Unlock()
+			if !isStarting {
+				if r, err := s.db.TransitionStatus(jobID, TransitionOpts{
+					AllowedFrom:         []JobStatusName{StatusCreated, StatusStarting},
+					Target:              StatusRunning,
+					RequireExitCodeNull: true,
+				}); err == nil {
+					row = r
+				}
+			}
+		}
+	} else {
+		// No session
+		if s.normalExitCommitPending(jobID) {
+			// Wait for pipe drain finalizer
+		} else {
+			s.mu.Lock()
+			isStarting := s.startingJobs[jobID]
+			s.mu.Unlock()
+			if !(isStarting && (status == StatusCreated || status == StatusStarting)) {
+				reason := "tmux_session_missing"
+				msg := "The dedicated tmux session is no longer present."
+				if r, err := s.db.TransitionStatus(jobID, TransitionOpts{
+					AllowedFrom: []JobStatusName{StatusCreated, StatusStarting, StatusRunning},
+					Target:      StatusLost,
+					Reason:      &reason,
+					Message:     &msg,
+				}); err == nil {
+					row = r
+				}
+			}
+		}
+	}
+
+	return s.statusViewFromRow(row), nil
+}
+
+func (s *Service) liveRuntimeState(jobID string) (bool, *bool, error) {
+	exists, err := s.tmux.SessionExists(JobSessionName(jobID))
+	if err != nil {
+		return false, nil, err
+	}
+	if !exists {
+		return false, nil, nil
+	}
+	active, err := s.tmux.IsOutputPipeActive(jobID)
+	if err != nil {
+		return false, nil, err
+	}
+	if active == nil {
+		return false, nil, nil
+	}
+	return true, active, nil
+}
+
+type exitMetadata struct {
+	exitCode int
+	endedAt  string
+}
+
+func (s *Service) drainedNormalExitMetadata(jobID string) *exitMetadata {
+	jobDir := filepath.Join(s.config.JobsDir(), jobID)
+	drainedPath := filepath.Join(jobDir, ".pipe-drained")
+	exitCodePath := filepath.Join(jobDir, "runner-exit-code")
+	endedAtPath := filepath.Join(jobDir, "runner-ended-at")
+
+	if !fileExists(drainedPath) || !fileExists(exitCodePath) || !fileExists(endedAtPath) {
+		return nil
+	}
+
+	exitCodeRaw, err := os.ReadFile(exitCodePath)
+	if err != nil {
+		return nil
+	}
+	endedAtRaw, err := os.ReadFile(endedAtPath)
+	if err != nil {
+		return nil
+	}
+
+	exitCodeStr := strings.TrimSpace(string(exitCodeRaw))
+	endedAtStr := strings.TrimSpace(string(endedAtRaw))
+	if exitCodeStr == "" || endedAtStr == "" {
+		return nil
+	}
+
+	code, err := strconv.Atoi(exitCodeStr)
+	if err != nil {
+		return nil
+	}
+
+	return &exitMetadata{exitCode: code, endedAt: endedAtStr}
+}
+
+func (s *Service) normalExitCommitPending(jobID string) bool {
+	jobDir := filepath.Join(s.config.JobsDir(), jobID)
+	return fileExists(filepath.Join(jobDir, "runner-exit-code")) &&
+		fileExists(filepath.Join(jobDir, "runner-ended-at")) &&
+		!fileExists(filepath.Join(jobDir, ".pipe-drained")) &&
+		!fileExists(filepath.Join(jobDir, ".pipe-failed"))
+}
+
+func (s *Service) outputLogPath(row *JobRow) string {
+	return filepath.Join(s.config.StateDir, row.OutputPath)
+}
+
+func (s *Service) statusViewFromRow(row *JobRow) *JobStatusView {
+	outputPath := s.outputLogPath(row)
+	offset := int(fileSize(outputPath))
+	return &JobStatusView{
+		JobID:     row.JobID,
+		Status:    row.Status,
+		Done:      row.Status.IsTerminal(),
+		ExitCode:  row.ExitCode,
+		CreatedAt: row.CreatedAt,
+		StartedAt: row.StartedAt,
+		EndedAt:   row.EndedAt,
+		Offset:    offset,
+	}
+}
+
+func (s *Service) jobResultFromView(view *JobStatusView, row *JobRow, window *OutputWindow) *JobResult {
+	outputPath := s.outputLogPath(row)
+	absPath, _ := filepath.Abs(outputPath)
+	return &JobResult{
+		JobID:      view.JobID,
+		Done:       view.Done,
+		Status:     view.Status,
+		ExitCode:   view.ExitCode,
+		OutputPath: absPath,
+		Output:     window.Output,
+		Offset:     window.Offset,
+		Truncated:  window.Truncated,
+	}
+}
+
+func (s *Service) resolveCwd(rawCwd *string) (string, error) {
+	var cwd string
+	if rawCwd != nil && *rawCwd != "" {
+		cwd = *rawCwd
+	} else {
+		cwd = s.config.DefaultCwd
+	}
+
+	info, err := os.Stat(cwd)
+	if err != nil || !info.IsDir() {
+		return "", NewServerError(400, "invalid_cwd", fmt.Sprintf("cwd is not a directory: %s", cwd))
+	}
+	abs, _ := filepath.Abs(cwd)
+	return abs, nil
+}
+
+func (s *Service) allocateJobDir() (string, string, error) {
+	for i := 0; i < 20; i++ {
+		jobID := GenerateJobID()
+		jobDir := filepath.Join(s.config.JobsDir(), jobID)
+		if err := os.Mkdir(jobDir, 0700); err == nil {
+			return jobID, jobDir, nil
+		}
+	}
+	return "", "", NewServerError(500, "job_id_collision", "Failed to allocate a unique job id")
+}
+
+func (s *Service) cleanupStarting(jobID, jobDir string) {
+	s.mu.Lock()
+	delete(s.startingJobs, jobID)
+	s.mu.Unlock()
+	os.RemoveAll(jobDir)
+}
+
+func (s *Service) installRunner() {
+	script := s.runnerScriptSource()
+	os.WriteFile(s.config.RunnerPath(), []byte(script), 0755)
+}
+
+func (s *Service) runnerScriptSource() string {
+	// The runner script is a bash wrapper that waits for the start-gate,
+	// then exec's the user script via a Python bootstrap (matching the Python impl).
+	// For the Go version, we can simplify this to a pure bash/exec approach.
+	return `#!/usr/bin/env bash
+set -uo pipefail
+
+JOB_DIR="$1"
+JOB_ID="$2"
+CWD="$3"
+SCRIPT_PATH="$JOB_DIR/script"
+ENV_PATH="$JOB_DIR/.job-env.json"
+START_GATE="$JOB_DIR/start-gate"
+RUNNER_EXIT_CODE_PATH="$JOB_DIR/runner-exit-code"
+RUNNER_ENDED_AT_PATH="$JOB_DIR/runner-ended-at"
+
+write_atomic() {
+  local dest="$1"
+  local value="$2"
+  local tmp="${dest}.tmp.$$"
+  printf '%s\n' "$value" > "$tmp"
+  mv "$tmp" "$dest"
+}
+
+while [ ! -e "$START_GATE" ]; do
+  sleep 0.05
+done
+
+unset TMUX
+unset SHELLCTL_STATE_DIR
+unset SHELLCTL_RUNTIME_DIR
+unset SHELLCTL_TMUX_SOCKET
+unset SHELLCTL_RUNNER
+unset SHELLCTL_AUTH_TOKEN
+
+cd "$CWD" || exit 111
+
+# Apply env from JSON if jq is available
+if command -v jq >/dev/null 2>&1 && [ -f "$ENV_PATH" ]; then
+  while IFS='=' read -r key value; do
+    export "$key=$value"
+  done < <(jq -r 'to_entries[] | "\(.key)=\(.value)"' "$ENV_PATH" 2>/dev/null)
+fi
+
+# Ensure HOME directory exists (agent backend may set a per-agent HOME).
+if [ -n "$HOME" ] && [ ! -d "$HOME" ]; then
+  mkdir -p "$HOME" 2>/dev/null || true
+fi
+
+# Determine how to run the script
+first_line=$(head -n1 "$SCRIPT_PATH")
+if [[ "$first_line" == "#!"* ]]; then
+  chmod +x "$SCRIPT_PATH"
+  "$SCRIPT_PATH"
+else
+  sh "$SCRIPT_PATH"
+fi
+EXIT_CODE=$?
+
+ENDED_AT="$(date -u +%Y-%m-%dT%H:%M:%SZ)"
+
+write_atomic "$RUNNER_EXIT_CODE_PATH" "$EXIT_CODE"
+write_atomic "$RUNNER_ENDED_AT_PATH" "$ENDED_AT"
+
+exit "$EXIT_CODE"
+`
+}
+
+// Helpers
+
+func fileExists(path string) bool {
+	_, err := os.Stat(path)
+	return err == nil
+}
+
+func fileSize(path string) int64 {
+	info, err := os.Stat(path)
+	if err != nil {
+		return 0
+	}
+	return info.Size()
+}
+
+func isNotFound(err error) bool {
+	if se, ok := err.(*ServerError); ok {
+		return se.Code == "job_not_found"
+	}
+	return false
+}

shellctl-go/internal/server/tmux.go

@@ -0,0 +1,286 @@
+package server
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+)
+
+// TmuxController manages tmux sessions for shellctl jobs via a dedicated socket.
+type TmuxController struct {
+	config *Config
+}
+
+// NewTmuxController creates a new tmux controller.
+func NewTmuxController(config *Config) *TmuxController {
+	return &TmuxController{config: config}
+}
+
+// StartServer ensures the tmux server is running.
+func (t *TmuxController) StartServer() error {
+	_, err := t.runTmux("start-server")
+	return err
+}
+
+// ListSessions returns the set of active tmux session names.
+func (t *TmuxController) ListSessions() (map[string]bool, error) {
+	result, err := t.runTmuxNoCheck("list-sessions", "-F", "#{session_name}")
+	if err != nil {
+		return nil, err
+	}
+	if result.exitCode != 0 {
+		stderr := strings.TrimSpace(result.stderr)
+		if isTmuxTargetMissing(stderr) {
+			return make(map[string]bool), nil
+		}
+		return nil, NewServerError(500, "tmux_error", stderr)
+	}
+	sessions := make(map[string]bool)
+	for _, line := range strings.Split(strings.TrimSpace(result.stdout), "\n") {
+		line = strings.TrimSpace(line)
+		if line != "" {
+			sessions[line] = true
+		}
+	}
+	return sessions, nil
+}
+
+// SessionExists checks if a tmux session exists by name.
+func (t *TmuxController) SessionExists(sessionName string) (bool, error) {
+	sessions, err := t.ListSessions()
+	if err != nil {
+		return false, err
+	}
+	return sessions[sessionName], nil
+}
+
+// IsOutputPipeActive checks the #{pane_pipe} format variable for a job pane.
+// Returns: true=active, false=inactive, error with nil bool if pane missing.
+func (t *TmuxController) IsOutputPipeActive(jobID string) (*bool, error) {
+	result, err := t.runTmuxNoCheck(
+		"display-message", "-p", "-t", JobPaneTarget(jobID), "#{pane_pipe}",
+	)
+	if err != nil {
+		return nil, err
+	}
+	if result.exitCode != 0 {
+		stderr := strings.TrimSpace(result.stderr)
+		if isTmuxTargetMissing(stderr) {
+			return nil, nil // pane missing
+		}
+		return nil, NewServerError(500, "tmux_error", stderr)
+	}
+	active := strings.TrimSpace(result.stdout) == "1"
+	return &active, nil
+}
+
+// CreateJobSession creates a new tmux session for a job.
+func (t *TmuxController) CreateJobSession(jobID, jobDir, cwd string, cols, rows int) error {
+	runnerCmd := shellJoin([]string{
+		t.config.RunnerPath(), jobDir, jobID, cwd,
+	})
+	result, err := t.runTmuxNoCheck(
+		"-f", "/dev/null",
+		"new-session", "-d",
+		"-s", JobSessionName(jobID),
+		"-x", fmt.Sprintf("%d", cols),
+		"-y", fmt.Sprintf("%d", rows),
+		runnerCmd,
+	)
+	if err != nil {
+		return err
+	}
+	if result.exitCode != 0 {
+		return NewServerError(500, "tmux_new_session_failed",
+			strings.TrimSpace(result.stderr))
+	}
+	return nil
+}
+
+// EnableOutputPipe attaches the sanitize→output pipeline via tmux pipe-pane.
+func (t *TmuxController) EnableOutputPipe(jobID, jobDir string, readyFile string) error {
+	pipeCmd := t.buildPipeCommand(jobID, jobDir, readyFile)
+	result, err := t.runTmuxNoCheck(
+		"pipe-pane", "-o", "-t", JobPaneTarget(jobID), pipeCmd,
+	)
+	if err != nil {
+		return err
+	}
+	if result.exitCode != 0 {
+		return NewServerError(500, "pipe_failed",
+			strings.TrimSpace(result.stderr))
+	}
+	return nil
+}
+
+// SendInput sends text to a job's tmux pane via load-buffer + paste-buffer.
+func (t *TmuxController) SendInput(jobID, text string) error {
+	bufferName := fmt.Sprintf("shellctl-in-%s", jobID)
+
+	// Write input to a temp file
+	tmpFile, err := os.CreateTemp(t.config.RuntimeDir, fmt.Sprintf("shellctl-input-%s-", jobID))
+	if err != nil {
+		return err
+	}
+	tmpPath := tmpFile.Name()
+	defer os.Remove(tmpPath)
+
+	if _, err := tmpFile.WriteString(text); err != nil {
+		tmpFile.Close()
+		return err
+	}
+	tmpFile.Close()
+
+	// Load buffer
+	result, err := t.runTmuxNoCheck("load-buffer", "-b", bufferName, tmpPath)
+	if err != nil {
+		return err
+	}
+	if result.exitCode != 0 {
+		stderr := strings.TrimSpace(result.stderr)
+		if isTmuxTargetMissing(stderr) {
+			return NewServerError(409, "tmux_target_missing", stderr)
+		}
+		return NewServerError(500, "tmux_input_failed", stderr)
+	}
+
+	// Paste buffer
+	result, err = t.runTmuxNoCheck(
+		"paste-buffer", "-t", JobPaneTarget(jobID), "-b", bufferName,
+	)
+	// Always clean up buffer
+	t.runTmuxNoCheck("delete-buffer", "-b", bufferName)
+
+	if err != nil {
+		return err
+	}
+	if result.exitCode != 0 {
+		stderr := strings.TrimSpace(result.stderr)
+		if isTmuxTargetMissing(stderr) {
+			return NewServerError(409, "tmux_target_missing", stderr)
+		}
+		return NewServerError(500, "tmux_input_failed", stderr)
+	}
+	return nil
+}
+
+// SendInterrupt sends C-c to a job's pane.
+func (t *TmuxController) SendInterrupt(jobID string) error {
+	_, _ = t.runTmuxNoCheck("send-keys", "-t", JobPaneTarget(jobID), "C-c")
+	return nil
+}
+
+// CleanupSession kills the tmux session for a job (best-effort).
+func (t *TmuxController) CleanupSession(jobID string) {
+	t.runTmuxNoCheck("kill-session", "-t", JobSessionName(jobID))
+}
+
+func (t *TmuxController) buildPipeCommand(jobID, jobDir, readyFile string) string {
+	sanitizeCmd := shellJoin(append(t.config.SanitizePtyCommand, "--ready-file", readyFile))
+	runnerExitCmd := shellJoin(append(t.config.RunnerExitCommand,
+		"--state-dir", t.config.StateDir,
+		"--job-id", jobID,
+		"--sqlite-busy-timeout-ms", fmt.Sprintf("%d", t.config.SQLiteBusyTimeoutMs),
+	))
+
+	outputPath := shellQuote(filepath.Join(jobDir, "output.log"))
+	drainedPath := shellQuote(filepath.Join(jobDir, ".pipe-drained"))
+	errorLogPath := shellQuote(filepath.Join(jobDir, "pipe-error.log"))
+	failedPath := shellQuote(filepath.Join(jobDir, ".pipe-failed"))
+	exitCodePath := shellQuote(filepath.Join(jobDir, "runner-exit-code"))
+	endedAtPath := shellQuote(filepath.Join(jobDir, "runner-ended-at"))
+
+	parts := []string{
+		fmt.Sprintf("%s >> %s 2> %s", sanitizeCmd, outputPath, errorLogPath),
+		"sanitize_status=$?",
+		"runner_exit_status=0",
+		fmt.Sprintf(
+			`if [ "$sanitize_status" -eq 0 ]; then : > %s; if [ -s %s ] && [ -s %s ]; then %s --exit-code "$(cat %s)" --ended-at "$(cat %s)" 2>> %s; runner_exit_status=$?; if [ "$runner_exit_status" -ne 0 ]; then printf 'runner-exit failed with status %%s\n' "$runner_exit_status" >> %s; fi; fi; else : > %s; fi`,
+			drainedPath, exitCodePath, endedAtPath, runnerExitCmd,
+			exitCodePath, endedAtPath, errorLogPath, errorLogPath, failedPath,
+		),
+		`if [ "$sanitize_status" -ne 0 ]; then exit "$sanitize_status"; fi`,
+		`exit "$sanitize_status"`,
+	}
+	return strings.Join(parts, " ; ")
+}
+
+type tmuxResult struct {
+	stdout   string
+	stderr   string
+	exitCode int
+}
+
+func (t *TmuxController) runTmux(args ...string) (string, error) {
+	result, err := t.runTmuxNoCheck(args...)
+	if err != nil {
+		return "", err
+	}
+	if result.exitCode != 0 {
+		stderr := strings.TrimSpace(result.stderr)
+		if stderr == "" {
+			stderr = "tmux command failed"
+		}
+		return "", NewServerError(500, "tmux_error", stderr)
+	}
+	return result.stdout, nil
+}
+
+func (t *TmuxController) runTmuxNoCheck(args ...string) (*tmuxResult, error) {
+	cmdArgs := append([]string{"-S", t.config.TmuxSocket()}, args...)
+	cmd := exec.Command("tmux", cmdArgs...)
+
+	// Clear TMUX env to avoid nesting
+	env := os.Environ()
+	filtered := env[:0]
+	for _, e := range env {
+		if !strings.HasPrefix(e, "TMUX=") {
+			filtered = append(filtered, e)
+		}
+	}
+	cmd.Env = filtered
+
+	var stdout, stderr bytes.Buffer
+	cmd.Stdout = &stdout
+	cmd.Stderr = &stderr
+
+	err := cmd.Run()
+	exitCode := 0
+	if err != nil {
+		if exitErr, ok := err.(*exec.ExitError); ok {
+			exitCode = exitErr.ExitCode()
+		} else {
+			return nil, fmt.Errorf("exec tmux: %w", err)
+		}
+	}
+
+	return &tmuxResult{
+		stdout:   stdout.String(),
+		stderr:   stderr.String(),
+		exitCode: exitCode,
+	}, nil
+}
+
+func isTmuxTargetMissing(stderr string) bool {
+	lower := strings.ToLower(stderr)
+	return strings.Contains(lower, "can't find pane") ||
+		strings.Contains(lower, "can't find session") ||
+		strings.Contains(lower, "no server running") ||
+		strings.Contains(lower, "failed to connect") ||
+		strings.Contains(lower, "server exited unexpectedly")
+}
+
+func shellJoin(parts []string) string {
+	quoted := make([]string, len(parts))
+	for i, p := range parts {
+		quoted[i] = shellQuote(p)
+	}
+	return strings.Join(quoted, " ")
+}
+
+func shellQuote(s string) string {
+	return "'" + strings.ReplaceAll(s, "'", `'\''`) + "'"
+}

shellctl-go/internal/server/tmux_test.go

@@ -0,0 +1,59 @@
+package server
+
+import (
+	"testing"
+)
+
+func TestShellQuote(t *testing.T) {
+	tests := []struct {
+		input    string
+		expected string
+	}{
+		{"hello", "'hello'"},
+		{"", "''"},
+		{"a b c", "'a b c'"},
+		{"it's", "'it'\\''s'"},
+		{"/path/to/file", "'/path/to/file'"},
+	}
+
+	for _, tc := range tests {
+		got := shellQuote(tc.input)
+		if got != tc.expected {
+			t.Errorf("shellQuote(%q) = %q, want %q", tc.input, got, tc.expected)
+		}
+	}
+}
+
+func TestShellJoin(t *testing.T) {
+	got := shellJoin([]string{"echo", "hello world", "it's"})
+	expected := "'echo' 'hello world' 'it'\\''s'"
+	if got != expected {
+		t.Errorf("shellJoin = %q, want %q", got, expected)
+	}
+}
+
+func TestIsTmuxTargetMissing(t *testing.T) {
+	missingMsgs := []string{
+		"can't find pane: shellctl-abc",
+		"can't find session: shellctl-abc",
+		"no server running on /tmp/tmux.sock",
+		"failed to connect to server",
+		"server exited unexpectedly",
+	}
+	for _, msg := range missingMsgs {
+		if !isTmuxTargetMissing(msg) {
+			t.Errorf("expected %q to be detected as target missing", msg)
+		}
+	}
+
+	validMsgs := []string{
+		"some other error",
+		"permission denied",
+		"",
+	}
+	for _, msg := range validMsgs {
+		if isTmuxTargetMissing(msg) {
+			t.Errorf("expected %q to NOT be detected as target missing", msg)
+		}
+	}
+}

shellctl-go/internal/server/types.go

@@ -0,0 +1,100 @@
+package server
+
+// RunJobRequest is the HTTP request body for POST /v1/jobs/run.
+type RunJobRequest struct {
+	Script           string            `json:"script"`
+	Cwd              *string           `json:"cwd,omitempty"`
+	Env              map[string]string `json:"env,omitempty"`
+	Terminal         *TerminalSize     `json:"terminal,omitempty"`
+	Timeout          float64           `json:"timeout,omitempty"`
+	OutputLimit      int               `json:"output_limit,omitempty"`
+	IdleFlushSeconds float64           `json:"idle_flush_seconds,omitempty"`
+}
+
+// TerminalSize specifies the initial PTY geometry.
+type TerminalSize struct {
+	Cols int `json:"cols"`
+	Rows int `json:"rows"`
+}
+
+// WaitJobRequest is the HTTP request body for POST /v1/jobs/{job_id}/wait.
+type WaitJobRequest struct {
+	Timeout          float64 `json:"timeout"`
+	Offset           int     `json:"offset"`
+	OutputLimit      int     `json:"output_limit,omitempty"`
+	IdleFlushSeconds float64 `json:"idle_flush_seconds,omitempty"`
+}
+
+// InputJobRequest is the HTTP request body for POST /v1/jobs/{job_id}/input.
+type InputJobRequest struct {
+	Text             string  `json:"text"`
+	Timeout          float64 `json:"timeout,omitempty"`
+	Offset           int     `json:"offset"`
+	OutputLimit      int     `json:"output_limit,omitempty"`
+	IdleFlushSeconds float64 `json:"idle_flush_seconds,omitempty"`
+}
+
+// TerminateJobRequest is the HTTP request body for POST /v1/jobs/{job_id}/terminate.
+type TerminateJobRequest struct {
+	GraceSeconds float64 `json:"grace_seconds,omitempty"`
+}
+
+// JobResult is the unified response for output-oriented job APIs.
+type JobResult struct {
+	JobID      string        `json:"job_id"`
+	Done       bool          `json:"done"`
+	Status     JobStatusName `json:"status"`
+	ExitCode   *int          `json:"exit_code"`
+	OutputPath string        `json:"output_path"`
+	Output     string        `json:"output"`
+	Offset     int           `json:"offset"`
+	Truncated  bool          `json:"truncated"`
+}
+
+// JobStatusView is the materialized lifecycle view.
+type JobStatusView struct {
+	JobID     string        `json:"job_id"`
+	Status    JobStatusName `json:"status"`
+	Done      bool          `json:"done"`
+	ExitCode  *int          `json:"exit_code"`
+	CreatedAt string        `json:"created_at"`
+	StartedAt *string       `json:"started_at"`
+	EndedAt   *string       `json:"ended_at"`
+	Offset    int           `json:"offset"`
+}
+
+// JobInfo is a compact job listing record.
+type JobInfo struct {
+	JobID     string        `json:"job_id"`
+	Status    JobStatusName `json:"status"`
+	CreatedAt string        `json:"created_at"`
+	StartedAt *string       `json:"started_at,omitempty"`
+	EndedAt   *string       `json:"ended_at,omitempty"`
+}
+
+// ListJobsResponse is the response for GET /v1/jobs.
+type ListJobsResponse struct {
+	Jobs []JobInfo `json:"jobs"`
+}
+
+// DeleteJobResponse is the response for DELETE /v1/jobs/{job_id}.
+type DeleteJobResponse struct {
+	JobID   string `json:"job_id"`
+	Deleted bool   `json:"deleted"`
+}
+
+// HealthResponse is the health check response.
+type HealthResponse struct {
+	Status string `json:"status"`
+}
+
+// ErrorDetail is the machine-readable API error payload.
+type ErrorDetail struct {
+	Code    string `json:"code"`
+	Message string `json:"message"`
+}
+
+// ErrorResponse is the error envelope.
+type ErrorResponse struct {
+	Error ErrorDetail `json:"error"`
+}

shellctl-go/tests/acceptance_test.go

@@ -0,0 +1,590 @@
+//go:build integration
+
+// Package tests runs the same acceptance test suite against both the Python
+// and Go shellctl server implementations to verify API compatibility.
+//
+// Prerequisites:
+//
+//	docker compose -f tests/docker-compose.yml up --build -d
+//	go test -tags=integration ./tests/... -v
+//	docker compose -f tests/docker-compose.yml down
+package tests
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"os"
+	"strings"
+	"testing"
+	"time"
+)
+
+var (
+	pythonURL = envOrDefault("SHELLCTL_PYTHON_URL", "http://localhost:15004")
+	goURL     = envOrDefault("SHELLCTL_GO_URL", "http://localhost:15005")
+	authToken = envOrDefault("SHELLCTL_TEST_TOKEN", "test-token-123")
+
+	httpClient = &http.Client{Timeout: 120 * time.Second}
+)
+
+// target represents one server under test.
+type target struct {
+	name    string
+	baseURL string
+}
+
+func targets() []target {
+	return []target{
+		{name: "python", baseURL: pythonURL},
+		{name: "go", baseURL: goURL},
+	}
+}
+
+func TestMain(m *testing.M) {
+	// Warmup: wait for both servers to be ready before running tests
+	for _, tgt := range targets() {
+		if !waitForServer(tgt) {
+			fmt.Fprintf(os.Stderr, "ERROR: %s server not ready at %s\n", tgt.name, tgt.baseURL)
+			os.Exit(1)
+		}
+	}
+	// Warmup job: the Python server's first job can be very slow (tmux server
+	// bootstrap, lazy imports). Send a throwaway job to prime it before real tests.
+	for _, tgt := range targets() {
+		warmupJob(tgt)
+	}
+	os.Exit(m.Run())
+}
+
+func waitForServer(tgt target) bool {
+	for i := 0; i < 60; i++ {
+		req, _ := http.NewRequest("GET", tgt.baseURL+"/healthz", nil)
+		resp, err := httpClient.Do(req)
+		if err == nil {
+			resp.Body.Close()
+			if resp.StatusCode == 200 {
+				return true
+			}
+		}
+		time.Sleep(time.Second)
+	}
+	return false
+}
+
+// warmupJob sends a trivial job to prime the server (tmux bootstrap, lazy init).
+// It retries up to 3 times with a 180s timeout per attempt.
+func warmupJob(tgt target) {
+	warmupClient := &http.Client{Timeout: 180 * time.Second}
+	payload, _ := json.Marshal(map[string]any{
+		"script":  "echo warmup",
+		"timeout": 10,
+	})
+	for attempt := 0; attempt < 3; attempt++ {
+		req, _ := http.NewRequest("POST", tgt.baseURL+"/v1/jobs/run", bytes.NewReader(payload))
+		req.Header.Set("Content-Type", "application/json")
+		req.Header.Set("Authorization", "Bearer "+authToken)
+		resp, err := warmupClient.Do(req)
+		if err != nil {
+			fmt.Fprintf(os.Stderr, "WARN: %s warmup job attempt %d failed: %v\n", tgt.name, attempt+1, err)
+			time.Sleep(2 * time.Second)
+			continue
+		}
+		io.Copy(io.Discard, resp.Body)
+		resp.Body.Close()
+		if resp.StatusCode == 200 {
+			return
+		}
+		fmt.Fprintf(os.Stderr, "WARN: %s warmup job attempt %d got status %d\n", tgt.name, attempt+1, resp.StatusCode)
+		time.Sleep(2 * time.Second)
+	}
+	fmt.Fprintf(os.Stderr, "WARN: %s warmup job failed after 3 attempts, continuing anyway\n", tgt.name)
+}
+
+// --- Test Cases ---
+
+func TestHealthz(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			resp := doGet(t, tgt, "/healthz", false)
+			assertStatus(t, resp, 200)
+			body := readBody(t, resp)
+			var result map[string]string
+			json.Unmarshal(body, &result)
+			if result["status"] != "ok" {
+				t.Errorf("expected status=ok, got %q", result["status"])
+			}
+		})
+	}
+}
+
+func TestAuthRequired(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			// Request without auth should fail
+			req, _ := http.NewRequest("GET", tgt.baseURL+"/v1/jobs", nil)
+			resp, err := http.DefaultClient.Do(req)
+			if err != nil {
+				t.Fatalf("request failed: %v", err)
+			}
+			defer resp.Body.Close()
+			if resp.StatusCode != 401 {
+				t.Errorf("expected 401, got %d", resp.StatusCode)
+			}
+		})
+	}
+}
+
+func TestRunSimpleScript(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			result := runJob(t, tgt, map[string]any{
+				"script":  "echo hello-world",
+				"timeout": 10,
+			})
+			assertJobDone(t, result)
+			assertExitCode(t, result, 0)
+			output := result["output"].(string)
+			if !strings.Contains(output, "hello-world") {
+				t.Errorf("expected output to contain 'hello-world', got %q", output)
+			}
+		})
+	}
+}
+
+func TestRunWithEnv(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			result := runJob(t, tgt, map[string]any{
+				"script":  "echo $MY_VAR",
+				"env":     map[string]string{"MY_VAR": "test-value-42"},
+				"timeout": 10,
+			})
+			assertJobDone(t, result)
+			assertExitCode(t, result, 0)
+			output := result["output"].(string)
+			if !strings.Contains(output, "test-value-42") {
+				t.Errorf("expected output to contain 'test-value-42', got %q", output)
+			}
+		})
+	}
+}
+
+func TestRunWithCwd(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			result := runJob(t, tgt, map[string]any{
+				"script":  "pwd",
+				"cwd":     "/tmp",
+				"timeout": 10,
+			})
+			assertJobDone(t, result)
+			assertExitCode(t, result, 0)
+			output := result["output"].(string)
+			if !strings.Contains(output, "/tmp") {
+				t.Errorf("expected output to contain '/tmp', got %q", output)
+			}
+		})
+	}
+}
+
+func TestRunNonZeroExit(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			result := runJob(t, tgt, map[string]any{
+				"script":  "exit 42",
+				"timeout": 10,
+			})
+			assertJobDone(t, result)
+			assertExitCode(t, result, 42)
+		})
+	}
+}
+
+func TestWaitForOutput(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			// Run a script that delays output
+			result := runJob(t, tgt, map[string]any{
+				"script":  "sleep 0.5 && echo delayed-output",
+				"timeout": 5,
+			})
+			assertJobDone(t, result)
+			assertExitCode(t, result, 0)
+			output := result["output"].(string)
+			if !strings.Contains(output, "delayed-output") {
+				t.Errorf("expected 'delayed-output', got %q", output)
+			}
+		})
+	}
+}
+
+func TestWaitJobWithOffset(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			// Run a multi-line script
+			result := runJob(t, tgt, map[string]any{
+				"script":  "echo line1\necho line2\necho line3",
+				"timeout": 10,
+			})
+			assertJobDone(t, result)
+			jobID := result["job_id"].(string)
+			offset := int(result["offset"].(float64))
+
+			// Wait with offset should return empty (already at end)
+			waitResult := waitJob(t, tgt, jobID, map[string]any{
+				"offset":  offset,
+				"timeout": 1,
+			})
+			// Should return with empty output since we're already past all data
+			if waitResult["output"].(string) != "" {
+				// It might return empty or might not, depending on timing
+				// Just verify no error occurred
+			}
+		})
+	}
+}
+
+func TestTailJob(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			result := runJob(t, tgt, map[string]any{
+				"script":  "echo line1\necho line2\necho final-line",
+				"timeout": 10,
+			})
+			assertJobDone(t, result)
+			jobID := result["job_id"].(string)
+
+			// Tail the job
+			resp := doGet(t, tgt, fmt.Sprintf("/v1/jobs/%s/log/tail", jobID), true)
+			assertStatus(t, resp, 200)
+			body := readBody(t, resp)
+			var tailResult map[string]any
+			json.Unmarshal(body, &tailResult)
+			output := tailResult["output"].(string)
+			if !strings.Contains(output, "final-line") {
+				t.Errorf("tail should contain 'final-line', got %q", output)
+			}
+		})
+	}
+}
+
+func TestGetJobStatus(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			result := runJob(t, tgt, map[string]any{
+				"script":  "echo done",
+				"timeout": 10,
+			})
+			assertJobDone(t, result)
+			jobID := result["job_id"].(string)
+
+			resp := doGet(t, tgt, fmt.Sprintf("/v1/jobs/%s", jobID), true)
+			assertStatus(t, resp, 200)
+			body := readBody(t, resp)
+			var status map[string]any
+			json.Unmarshal(body, &status)
+
+			if status["job_id"] != jobID {
+				t.Errorf("expected job_id=%s, got %v", jobID, status["job_id"])
+			}
+			if status["done"] != true {
+				t.Errorf("expected done=true")
+			}
+			if status["status"] != "exited" {
+				t.Errorf("expected status=exited, got %v", status["status"])
+			}
+		})
+	}
+}
+
+func TestListJobs(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			// Run a job first
+			runJob(t, tgt, map[string]any{
+				"script":  "echo for-listing",
+				"timeout": 10,
+			})
+
+			resp := doGet(t, tgt, "/v1/jobs", true)
+			assertStatus(t, resp, 200)
+			body := readBody(t, resp)
+			var listResult map[string]any
+			json.Unmarshal(body, &listResult)
+			jobs := listResult["jobs"].([]any)
+			if len(jobs) == 0 {
+				t.Error("expected at least one job in listing")
+			}
+		})
+	}
+}
+
+func TestTerminateJob(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			// Start a long-running job
+			result := runJob(t, tgt, map[string]any{
+				"script":  "sleep 60",
+				"timeout": 1, // short timeout so run returns quickly
+			})
+			jobID := result["job_id"].(string)
+
+			// Terminate it
+			resp := doPost(t, tgt, fmt.Sprintf("/v1/jobs/%s/terminate", jobID),
+				map[string]any{"grace_seconds": 1}, true)
+			assertStatus(t, resp, 200)
+			body := readBody(t, resp)
+			var termResult map[string]any
+			json.Unmarshal(body, &termResult)
+
+			if termResult["done"] != true {
+				t.Errorf("expected done=true after terminate")
+			}
+			status := termResult["status"].(string)
+			if status != "terminated" && status != "exited" {
+				t.Errorf("expected terminal status, got %q", status)
+			}
+		})
+	}
+}
+
+func TestDeleteJob(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			result := runJob(t, tgt, map[string]any{
+				"script":  "echo to-delete",
+				"timeout": 10,
+			})
+			assertJobDone(t, result)
+			jobID := result["job_id"].(string)
+
+			// Delete it
+			req, _ := http.NewRequest("DELETE",
+				fmt.Sprintf("%s/v1/jobs/%s", tgt.baseURL, jobID), nil)
+			req.Header.Set("Authorization", "Bearer "+authToken)
+			resp, err := http.DefaultClient.Do(req)
+			if err != nil {
+				t.Fatalf("delete request failed: %v", err)
+			}
+			defer resp.Body.Close()
+			assertStatus(t, resp, 200)
+
+			// Should be gone now
+			getResp := doGet(t, tgt, fmt.Sprintf("/v1/jobs/%s", jobID), true)
+			if getResp.StatusCode != 404 {
+				t.Errorf("expected 404 after delete, got %d", getResp.StatusCode)
+			}
+			getResp.Body.Close()
+		})
+	}
+}
+
+func TestForceDeleteRunningJob(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			// Start a long-running job
+			result := runJob(t, tgt, map[string]any{
+				"script":  "sleep 60",
+				"timeout": 1,
+			})
+			jobID := result["job_id"].(string)
+
+			// Force delete
+			req, _ := http.NewRequest("DELETE",
+				fmt.Sprintf("%s/v1/jobs/%s?force=true&grace_seconds=1", tgt.baseURL, jobID), nil)
+			req.Header.Set("Authorization", "Bearer "+authToken)
+			resp, err := http.DefaultClient.Do(req)
+			if err != nil {
+				t.Fatalf("delete request failed: %v", err)
+			}
+			defer resp.Body.Close()
+			assertStatus(t, resp, 200)
+		})
+	}
+}
+
+func TestSendInput(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			// Start a script that reads from stdin
+			result := runJob(t, tgt, map[string]any{
+				"script":  "read line && echo got:$line",
+				"timeout": 2, // Will timeout waiting for input
+			})
+			jobID := result["job_id"].(string)
+
+			if result["done"] == true {
+				// Already finished (possible race), skip
+				t.Skip("job completed before input could be sent")
+			}
+
+			// Send input
+			offset := int(result["offset"].(float64))
+			resp := doPost(t, tgt, fmt.Sprintf("/v1/jobs/%s/input", jobID),
+				map[string]any{
+					"text":    "hello-input\n",
+					"offset":  offset,
+					"timeout": 5,
+				}, true)
+			assertStatus(t, resp, 200)
+			body := readBody(t, resp)
+			var inputResult map[string]any
+			json.Unmarshal(body, &inputResult)
+			output := inputResult["output"].(string)
+			if !strings.Contains(output, "got:hello-input") {
+				t.Logf("output after input: %q (may need more wait time)", output)
+			}
+		})
+	}
+}
+
+func TestMultilineOutput(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			script := "for i in $(seq 1 20); do echo \"line $i\"; done"
+			result := runJob(t, tgt, map[string]any{
+				"script":  script,
+				"timeout": 10,
+			})
+			assertJobDone(t, result)
+			assertExitCode(t, result, 0)
+			output := result["output"].(string)
+			if !strings.Contains(output, "line 1") {
+				t.Errorf("missing 'line 1' in output")
+			}
+			if !strings.Contains(output, "line 20") {
+				t.Errorf("missing 'line 20' in output")
+			}
+		})
+	}
+}
+
+func TestInvalidCwd(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			resp := doPost(t, tgt, "/v1/jobs/run", map[string]any{
+				"script":  "echo x",
+				"cwd":     "/nonexistent-dir-xyz",
+				"timeout": 5,
+			}, true)
+			if resp.StatusCode != 400 {
+				body := readBody(t, resp)
+				t.Errorf("expected 400, got %d: %s", resp.StatusCode, string(body))
+			} else {
+				resp.Body.Close()
+			}
+		})
+	}
+}
+
+func TestJobNotFound(t *testing.T) {
+	for _, tgt := range targets() {
+		t.Run(tgt.name, func(t *testing.T) {
+			resp := doGet(t, tgt, "/v1/jobs/nonexistent-id-xyz", true)
+			if resp.StatusCode != 404 {
+				t.Errorf("expected 404, got %d", resp.StatusCode)
+			}
+			resp.Body.Close()
+		})
+	}
+}
+
+// --- Helpers ---
+
+func runJob(t *testing.T, tgt target, payload map[string]any) map[string]any {
+	t.Helper()
+	resp := doPost(t, tgt, "/v1/jobs/run", payload, true)
+	assertStatus(t, resp, 200)
+	body := readBody(t, resp)
+	var result map[string]any
+	if err := json.Unmarshal(body, &result); err != nil {
+		t.Fatalf("[%s] failed to parse run response: %v\nbody: %s", tgt.name, err, string(body))
+	}
+	return result
+}
+
+func waitJob(t *testing.T, tgt target, jobID string, payload map[string]any) map[string]any {
+	t.Helper()
+	resp := doPost(t, tgt, fmt.Sprintf("/v1/jobs/%s/wait", jobID), payload, true)
+	assertStatus(t, resp, 200)
+	body := readBody(t, resp)
+	var result map[string]any
+	json.Unmarshal(body, &result)
+	return result
+}
+
+func doGet(t *testing.T, tgt target, path string, withAuth bool) *http.Response {
+	t.Helper()
+	req, _ := http.NewRequest("GET", tgt.baseURL+path, nil)
+	if withAuth {
+		req.Header.Set("Authorization", "Bearer "+authToken)
+	}
+	resp, err := httpClient.Do(req)
+	if err != nil {
+		t.Fatalf("[%s] GET %s failed: %v", tgt.name, path, err)
+	}
+	return resp
+}
+
+func doPost(t *testing.T, tgt target, path string, payload map[string]any, withAuth bool) *http.Response {
+	t.Helper()
+	body, _ := json.Marshal(payload)
+	req, _ := http.NewRequest("POST", tgt.baseURL+path, bytes.NewReader(body))
+	req.Header.Set("Content-Type", "application/json")
+	if withAuth {
+		req.Header.Set("Authorization", "Bearer "+authToken)
+	}
+	resp, err := httpClient.Do(req)
+	if err != nil {
+		t.Fatalf("[%s] POST %s failed: %v", tgt.name, path, err)
+	}
+	return resp
+}
+
+func readBody(t *testing.T, resp *http.Response) []byte {
+	t.Helper()
+	defer resp.Body.Close()
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("failed to read response body: %v", err)
+	}
+	return body
+}
+
+func assertStatus(t *testing.T, resp *http.Response, expected int) {
+	t.Helper()
+	if resp.StatusCode != expected {
+		body, _ := io.ReadAll(resp.Body)
+		resp.Body.Close()
+		t.Fatalf("expected status %d, got %d: %s", expected, resp.StatusCode, string(body))
+	}
+}
+
+func assertJobDone(t *testing.T, result map[string]any) {
+	t.Helper()
+	if result["done"] != true {
+		t.Errorf("expected done=true, got %v (status=%v)", result["done"], result["status"])
+	}
+}
+
+func assertExitCode(t *testing.T, result map[string]any, expected int) {
+	t.Helper()
+	exitCode, ok := result["exit_code"].(float64)
+	if !ok {
+		t.Errorf("exit_code is nil or not a number: %v", result["exit_code"])
+		return
+	}
+	if int(exitCode) != expected {
+		t.Errorf("expected exit_code=%d, got %d", expected, int(exitCode))
+	}
+}
+
+func envOrDefault(key, defaultVal string) string {
+	if v := os.Getenv(key); v != "" {
+		return v
+	}
+	return defaultVal
+}

shellctl-go/tests/docker-compose.yml

@@ -0,0 +1,41 @@
+# docker-compose for integration testing both shellctl implementations.
+#
+# Usage:
+#   docker compose -f tests/docker-compose.yml up --build -d
+#   go test -tags=integration ./tests/... -v
+#   docker compose -f tests/docker-compose.yml down
+
+services:
+  shellctl-python:
+    build:
+      context: ../../dify-agent
+      dockerfile: docker/local-sandbox/Dockerfile
+    ports:
+      - "15004:5004"
+    environment:
+      - SHELLCTL_AUTH_TOKEN=test-token-123
+    command: ["shellctl", "serve", "--listen", "0.0.0.0:5004"]
+    healthcheck:
+      test: ["CMD", "curl", "-sf", "http://localhost:5004/healthz"]
+      interval: 2s
+      timeout: 3s
+      retries: 10
+      start_period: 10s
+
+  shellctl-go:
+    build:
+      context: ..
+      dockerfile: docker/Dockerfile
+      additional_contexts:
+        agent: ../../dify-agent
+    ports:
+      - "15005:5004"
+    environment:
+      - SHELLCTL_AUTH_TOKEN=test-token-123
+    command: ["shellctl", "serve", "--listen", "0.0.0.0:5004"]
+    healthcheck:
+      test: ["CMD", "bash", "-c", "exec 3<>/dev/tcp/localhost/5004 && echo -e 'GET /healthz HTTP/1.0\\r\\nHost: localhost\\r\\n\\r\\n' >&3 && cat <&3 | grep -q ok"]
+      interval: 2s
+      timeout: 3s
+      retries: 10
+      start_period: 5s

web/app/components/base/chat/chat/__tests__/hooks.spec.tsx

@@ -2010,7 +2010,7 @@ describe('useChat', () => {
       })
 
       const lastResponse = result.current.chatList[1]
-      expect(lastResponse!.content).toBe('history top-level answer')
+      expect(lastResponse!.content).toBe('')
       expect(lastResponse!.agent_response_parts).toBeUndefined()
       expect(lastResponse!.workflow_run_id).toBe('history-workflow-run')
       expect(lastResponse!.workflowProcess).toBeUndefined()
@@ -2028,71 +2028,6 @@ describe('useChat', () => {
       ])
     })
 
-    it('should keep new agent streaming response parts when completed history has no answer yet', async () => {
-      let callbacks: HookCallbacks
-      vi.mocked(ssePost).mockImplementation(async (_url, _params, options) => {
-        callbacks = options as HookCallbacks
-      })
-
-      const onGetConversationMessages = vi.fn().mockResolvedValue({
-        data: [{
-          id: 'm-new-agent-empty-history',
-          answer: '',
-          message: [{ role: 'user', text: 'hi' }],
-          agent_thoughts: [
-            {
-              id: 'history-thought',
-              thought: 'history thought',
-              answer: '',
-              tool: '',
-              tool_input: '',
-              observation: '',
-              position: 1,
-            },
-          ],
-          created_at: Date.now(),
-          inputs: {},
-          query: 'hi',
-        }],
-      })
-
-      const { result } = renderHook(() => useChat(
-        undefined,
-        undefined,
-        undefined,
-        undefined,
-        undefined,
-        undefined,
-        undefined,
-        { isNewAgent: true },
-      ))
-
-      act(() => {
-        result.current.handleSend('test-url', { query: 'new agent empty history' }, {
-          onGetConversationMessages,
-        })
-      })
-
-      await act(async () => {
-        callbacks.onThought({ id: 'stream-thought', thought: 'stream thought' })
-        callbacks.onData(' streamed answer', true, { event: 'agent_message', messageId: 'm-new-agent-empty-history', conversationId: 'c-new-agent-empty-history' })
-        await callbacks.onCompleted()
-      })
-
-      const lastResponse = result.current.chatList[1]
-      expect(lastResponse!.content).toBe('')
-      expect(lastResponse!.agent_response_parts).toEqual([
-        { type: 'thought', thought: expect.objectContaining({ id: 'stream-thought', thought: 'stream thought' }) },
-        { type: 'message', content: ' streamed answer' },
-      ])
-      expect(lastResponse!.agent_thoughts).toEqual([
-        expect.objectContaining({
-          id: 'history-thought',
-          thought: 'history thought',
-        }),
-      ])
-    })
-
     it('should handle onCompleted using agent thought when thought matches answer', async () => {
       let callbacks: HookCallbacks
       vi.mocked(ssePost).mockImplementation(async (_url, _params, options) => {

web/app/components/base/chat/chat/answer/__tests__/agent-roster-response-content.spec.tsx

@@ -38,9 +38,7 @@ describe('AgentRosterResponseContent', () => {
     render(<AgentRosterResponseContent item={item} />)
 
     expect(screen.getByRole('button', { name: /workFinished/i })).toBeInTheDocument()
-    await waitFor(() => {
-      expect(screen.getByTestId('agent-roster-response-content')).toHaveTextContent('history answer')
-    })
+    expect(screen.queryByText('history answer')).not.toBeInTheDocument()
 
     fireEvent.click(screen.getByRole('button', { name: /workFinished/i }))
 
@@ -49,7 +47,6 @@ describe('AgentRosterResponseContent', () => {
     })
 
     expect(screen.queryByText('internal thought should not render')).not.toBeInTheDocument()
-    expect(screen.getAllByTestId('agent-content-markdown')).toHaveLength(1)
   })
 
   it('should render new agent response parts in event order when thoughts and messages interleave', async () => {

web/app/components/base/chat/chat/answer/agent-roster-response-content.tsx

@@ -259,11 +259,9 @@ function ToolProcessCard({
 function AgentThoughtsProcessList({
   item,
   responding,
-  suppressAnswer,
 }: {
   item: ChatItem
   responding?: boolean
-  suppressAnswer?: boolean
 }) {
   return (
     <div className="mt-2 flex flex-col gap-1">
@@ -273,7 +271,6 @@ function AgentThoughtsProcessList({
           thought={thought}
           responding={responding}
           defaultOpen={index === 0}
-          suppressAnswer={suppressAnswer}
         />
       ))}
     </div>
@@ -284,19 +281,17 @@ function AgentThoughtProcessItem({
   thought,
   responding,
   defaultOpen,
-  suppressAnswer,
 }: {
   thought: ThoughtItem
   responding?: boolean
   defaultOpen?: boolean
-  suppressAnswer?: boolean
 }) {
   const tools = getToolProcesses(thought, responding)
   const answer = thought.answer?.trim()
 
   return (
     <div className="flex flex-col gap-1">
-      {answer && !suppressAnswer && (
+      {answer && (
         <div className="px-2 py-2 body-md-regular text-text-primary" data-testid="agent-content-markdown">
           <Markdown content={thought.answer || ''} />
         </div>
@@ -370,11 +365,9 @@ function AgentResponsePartList({
 function AgentThoughtsProcessGroup({
   item,
   responding,
-  suppressAnswer,
 }: {
   item: ChatItem
   responding?: boolean
-  suppressAnswer?: boolean
 }) {
   const { t } = useTranslation('agentV2')
   const [open, setOpen] = useState(false)
@@ -392,7 +385,6 @@ function AgentThoughtsProcessGroup({
         <AgentThoughtsProcessList
           item={item}
           responding={responding}
-          suppressAnswer={suppressAnswer}
         />
       </div>
     )
@@ -417,26 +409,12 @@ function AgentThoughtsProcessGroup({
         <AgentThoughtsProcessList
           item={item}
           responding={responding}
-          suppressAnswer={suppressAnswer}
         />
       )}
     </div>
   )
 }
 
-function getLastAgentThoughtAnswer(agentThoughts: ChatItem['agent_thoughts']) {
-  if (!agentThoughts?.length)
-    return ''
-
-  for (let index = agentThoughts.length - 1; index >= 0; index--) {
-    const answer = agentThoughts[index]?.answer?.trim()
-    if (answer)
-      return agentThoughts[index]?.answer || ''
-  }
-
-  return ''
-}
-
 export function AgentRosterResponseContent({
   item,
   responding,
@@ -456,8 +434,6 @@ export function AgentRosterResponseContent({
     )
   }
 
-  const visibleContent = content || getLastAgentThoughtAnswer(agent_thoughts)
-
   return (
     <div className="flex w-full flex-col gap-1" data-testid="agent-roster-response-content">
       {!!item.agent_response_parts?.length && (
@@ -472,12 +448,11 @@ export function AgentRosterResponseContent({
             <AgentThoughtsProcessGroup
               item={item}
               responding={responding}
-              suppressAnswer={!!visibleContent}
             />
           )}
-          {visibleContent && (
+          {content && (
             <div className="px-2 py-2 body-md-regular text-text-primary" data-testid="agent-content-markdown">
-              <Markdown content={visibleContent} />
+              <Markdown content={content} />
             </div>
           )}
         </>

web/app/components/base/chat/chat/hooks.ts

@@ -134,20 +134,6 @@ function upsertAgentResponseThoughtPart(responseItem: ChatItemInTree, thought: T
   })
 }
 
-function getLastAgentThoughtAnswer(agentThoughts: ThoughtItem[]) {
-  for (let index = agentThoughts.length - 1; index >= 0; index--) {
-    const answer = agentThoughts[index]?.answer?.trim()
-    if (answer)
-      return agentThoughts[index]?.answer || ''
-  }
-
-  return ''
-}
-
-function hasAgentResponseMessagePart(responseItem: ChatItem) {
-  return responseItem.agent_response_parts?.some(part => part.type === 'message' && !!part.content.trim()) ?? false
-}
-
 function getHistoryAgentThoughts(responseItem: HistoryConversationMessage) {
   if (!Array.isArray(responseItem.agent_thoughts))
     return []
@@ -998,21 +984,19 @@ export const useChat = (
               return onConversationComplete?.(conversationIdRef.current, completedWorkflowRunId)
 
             const historyAgentThoughts = getHistoryAgentThoughts(newResponseItem)
+            const hasHistoryAgentThoughtAnswer = historyAgentThoughts.some(thought => thought.answer?.trim())
             const lastHistoryAgentThought = historyAgentThoughts.at(-1)
             const historyAnswer = newResponseItem.answer || ''
-            const historyAgentThoughtAnswer = options.isNewAgent ? getLastAgentThoughtAnswer(historyAgentThoughts) : ''
-            const completedAnswer = historyAnswer || historyAgentThoughtAnswer
-            const isUseAgentThought = !options.isNewAgent && lastHistoryAgentThought?.thought === historyAnswer
-            const shouldKeepStreamingResponseParts = options.isNewAgent && !completedAnswer.trim() && hasAgentResponseMessagePart(responseItem)
+            const isUseAgentThought = (lastHistoryAgentThought?.thought === historyAnswer) || (options.isNewAgent && hasHistoryAgentThoughtAnswer)
             const messageLog = Array.isArray(newResponseItem.message) ? newResponseItem.message : []
             const answerTokens = newResponseItem.answer_tokens ?? 0
             const messageTokens = newResponseItem.message_tokens ?? 0
             const providerResponseLatency = newResponseItem.provider_response_latency ?? 0
             const historyAnswerFiles = getHistoryAnswerFiles(newResponseItem)
             updateChatTreeNode(responseItem.id, {
-              content: shouldKeepStreamingResponseParts || isUseAgentThought ? '' : completedAnswer,
+              content: isUseAgentThought ? '' : historyAnswer,
               agent_thoughts: historyAgentThoughts,
-              agent_response_parts: shouldKeepStreamingResponseParts ? responseItem.agent_response_parts : undefined,
+              agent_response_parts: undefined,
               citation: newResponseItem.retriever_resources,
               reasoningContent: newResponseItem.metadata?.reasoning,
               reasoningFinished: true,