diff --git a/README.md b/README.md index 353d12001..f5c17deb7 100644 --- a/README.md +++ b/README.md @@ -5,3 +5,47 @@ paired with an Electron + TypeScript frontend (`frontend/`). See [`docs/`](docs/README.md) for architecture and status — start with the Lifecycle Manager + Session Manager lane in [`docs/architecture.md`](docs/architecture.md). + +## Backend daemon + +The Go binary in [`backend/`](backend/) is the HTTP daemon — a loopback-only +sidecar the Electron supervisor will spawn (Phase 1c). Phase 1a landed the +skeleton: chi router, middleware stack (recoverer → request-id → logger → +real-ip), `/healthz` + `/readyz`, atomic `running.json` PID/port handshake, +graceful shutdown on SIGINT/SIGTERM. + +### Run + +```bash +cd backend +go run . # binds 127.0.0.1:3001 with all defaults +AO_PORT=3019 go run . # override per invocation +``` + +Health check: + +```bash +curl localhost:3001/healthz # {"status":"ok"} +curl localhost:3001/readyz # {"status":"ready"} +``` + +### Configuration (env only) + +The bind host is always `127.0.0.1`: the daemon is a loopback-only sidecar +and binding any other interface would be a security regression, so the host +is intentionally not env-configurable. + +| Var | Default | Purpose | +|---|---|---| +| `AO_PORT` | `3001` | bind port; fails fast if taken | +| `AO_REQUEST_TIMEOUT` | `60s` | per-request timeout (Go duration) | +| `AO_SHUTDOWN_TIMEOUT` | `10s` | graceful-shutdown hard cap | +| `AO_RUN_FILE` | `/agent-orchestrator/running.json` | PID + port handshake path | + +### Test + +```bash +cd backend +gofmt -l . && go build ./... && go vet ./... && go test -race ./... +``` + diff --git a/backend/go.mod b/backend/go.mod index 22a555cda..311cea288 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -1,3 +1,5 @@ module github.com/aoagents/agent-orchestrator/backend go 1.22 + +require github.com/go-chi/chi/v5 v5.1.0 diff --git a/backend/go.sum b/backend/go.sum new file mode 100644 index 000000000..823cdbb1a --- /dev/null +++ b/backend/go.sum @@ -0,0 +1,2 @@ +github.com/go-chi/chi/v5 v5.1.0 h1:acVI1TYaD+hhedDJ3r54HyA6sExp3HfXq7QWEEY/xMw= +github.com/go-chi/chi/v5 v5.1.0/go.mod h1:DslCQbL2OYiznFReuXYUmQ2hGd1aDpCnlMNITLSKoi8= diff --git a/backend/internal/config/config.go b/backend/internal/config/config.go new file mode 100644 index 000000000..d6765dba2 --- /dev/null +++ b/backend/internal/config/config.go @@ -0,0 +1,140 @@ +// Package config loads the daemon's runtime configuration. The HTTP daemon is +// a loopback-only sidecar: it binds 127.0.0.1, takes no public traffic, and +// reads everything it needs from the environment with sane defaults so it can +// boot with zero configuration in development. +package config + +import ( + "fmt" + "net" + "os" + "path/filepath" + "strconv" + "time" +) + +const ( + // LoopbackHost is the only host the daemon ever binds. There is deliberately + // no AO_HOST env var: the daemon has no auth/CORS/TLS and a stray + // AO_HOST=0.0.0.0 would turn it into a public no-auth service. The legacy + // TS server bound all-interfaces by accident and docs/CROSS_PLATFORM.md + // already calls that out as a bug; the Go rewrite fixes it by removing the + // knob entirely. If a non-default loopback (e.g. ::1, 127.0.0.2) is ever + // needed, add it back with an IsLoopback() validator — not a raw env read. + LoopbackHost = "127.0.0.1" + // DefaultPort is the single port the whole surface (REST, SSE, WS, static) + // is served from. Single-port keeps it same-origin: no CORS, one lifecycle. + DefaultPort = 3001 + // DefaultRequestTimeout bounds a single request. Long-lived surfaces (SSE, + // WS) are mounted outside this timeout; it guards the REST surface only. + DefaultRequestTimeout = 60 * time.Second + // DefaultShutdownTimeout is the hard cap on graceful shutdown. After this + // the process exits even if connections are still draining. + DefaultShutdownTimeout = 10 * time.Second +) + +// Config is the fully-resolved daemon configuration. It is immutable once +// built by Load. +type Config struct { + // Host is the bind address. Always loopback — see LoopbackHost. + Host string + // Port is the TCP port to bind. The daemon fails fast if it is taken. + Port int + // RequestTimeout bounds REST request handling. + RequestTimeout time.Duration + // ShutdownTimeout is the hard graceful-shutdown deadline. + ShutdownTimeout time.Duration + // RunFilePath is where the PID + port handshake file (running.json) is + // written so the Electron supervisor can discover and reap the daemon. + RunFilePath string +} + +// Addr returns the host:port the HTTP server binds. It uses net.JoinHostPort so +// the result is correct for IPv6 literals as well as IPv4 / hostnames. +func (c Config) Addr() string { + return net.JoinHostPort(c.Host, strconv.Itoa(c.Port)) +} + +// Load resolves configuration from the environment, applying defaults. It +// returns an error only for values that are present but malformed (e.g. a +// non-numeric AO_PORT); missing values fall back to defaults. +// +// Recognised variables: +// +// AO_PORT bind port (default 3001) +// AO_REQUEST_TIMEOUT per-request timeout (Go duration > 0, default 60s) +// AO_SHUTDOWN_TIMEOUT shutdown deadline (Go duration > 0, default 10s) +// AO_RUN_FILE running.json path (default /running.json) +// +// The bind host is not configurable: the daemon is loopback-only by design. +func Load() (Config, error) { + cfg := Config{ + Host: LoopbackHost, + Port: DefaultPort, + RequestTimeout: DefaultRequestTimeout, + ShutdownTimeout: DefaultShutdownTimeout, + } + + if raw := os.Getenv("AO_PORT"); raw != "" { + port, err := strconv.Atoi(raw) + if err != nil { + return Config{}, fmt.Errorf("invalid AO_PORT %q: %w", raw, err) + } + if port < 1 || port > 65535 { + return Config{}, fmt.Errorf("invalid AO_PORT %d: out of range 1-65535", port) + } + cfg.Port = port + } + + if raw := os.Getenv("AO_REQUEST_TIMEOUT"); raw != "" { + d, err := parsePositiveDuration("AO_REQUEST_TIMEOUT", raw) + if err != nil { + return Config{}, err + } + cfg.RequestTimeout = d + } + + if raw := os.Getenv("AO_SHUTDOWN_TIMEOUT"); raw != "" { + d, err := parsePositiveDuration("AO_SHUTDOWN_TIMEOUT", raw) + if err != nil { + return Config{}, err + } + cfg.ShutdownTimeout = d + } + + runFile, err := resolveRunFilePath() + if err != nil { + return Config{}, err + } + cfg.RunFilePath = runFile + + return cfg, nil +} + +// parsePositiveDuration rejects zero and negative durations: a zero +// RequestTimeout would expire every request instantly, and a non-positive +// ShutdownTimeout would defeat graceful shutdown. +func parsePositiveDuration(name, raw string) (time.Duration, error) { + d, err := time.ParseDuration(raw) + if err != nil { + return 0, fmt.Errorf("invalid %s %q: %w", name, raw, err) + } + if d <= 0 { + return 0, fmt.Errorf("invalid %s %q: must be > 0", name, raw) + } + return d, nil +} + +// resolveRunFilePath picks where running.json lives. An explicit AO_RUN_FILE +// wins; otherwise it sits under the per-user state directory so multiple repos +// share one supervisor handshake location. +func resolveRunFilePath() (string, error) { + if p, ok := os.LookupEnv("AO_RUN_FILE"); ok && p != "" { + return p, nil + } + dir, err := os.UserConfigDir() + if err != nil { + return "", fmt.Errorf("resolve state dir: %w", err) + } + return filepath.Join(dir, "agent-orchestrator", "running.json"), nil +} diff --git a/backend/internal/config/config_test.go b/backend/internal/config/config_test.go new file mode 100644 index 000000000..dfcb5b8af --- /dev/null +++ b/backend/internal/config/config_test.go @@ -0,0 +1,84 @@ +package config + +import ( + "testing" + "time" +) + +func TestLoadDefaults(t *testing.T) { + // Clear every recognised var so we observe pure defaults regardless of the + // surrounding environment. + for _, k := range []string{"AO_PORT", "AO_REQUEST_TIMEOUT", "AO_SHUTDOWN_TIMEOUT", "AO_RUN_FILE"} { + t.Setenv(k, "") + } + + cfg, err := Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + if cfg.Host != LoopbackHost { + t.Errorf("Host = %q, want %q", cfg.Host, LoopbackHost) + } + if cfg.Port != DefaultPort { + t.Errorf("Port = %d, want %d", cfg.Port, DefaultPort) + } + if cfg.RequestTimeout != DefaultRequestTimeout { + t.Errorf("RequestTimeout = %s, want %s", cfg.RequestTimeout, DefaultRequestTimeout) + } + if cfg.ShutdownTimeout != DefaultShutdownTimeout { + t.Errorf("ShutdownTimeout = %s, want %s", cfg.ShutdownTimeout, DefaultShutdownTimeout) + } + if cfg.RunFilePath == "" { + t.Error("RunFilePath is empty, want a resolved default path") + } +} + +func TestLoadOverrides(t *testing.T) { + t.Setenv("AO_PORT", "4002") + t.Setenv("AO_REQUEST_TIMEOUT", "5s") + t.Setenv("AO_SHUTDOWN_TIMEOUT", "3s") + t.Setenv("AO_RUN_FILE", "/tmp/ao-test-running.json") + + cfg, err := Load() + if err != nil { + t.Fatalf("Load: %v", err) + } + if cfg.Addr() != "127.0.0.1:4002" { + t.Errorf("Addr() = %q, want 127.0.0.1:4002", cfg.Addr()) + } + if cfg.RequestTimeout != 5*time.Second { + t.Errorf("RequestTimeout = %s, want 5s", cfg.RequestTimeout) + } + if cfg.ShutdownTimeout != 3*time.Second { + t.Errorf("ShutdownTimeout = %s, want 3s", cfg.ShutdownTimeout) + } + if cfg.RunFilePath != "/tmp/ao-test-running.json" { + t.Errorf("RunFilePath = %q, want /tmp/ao-test-running.json", cfg.RunFilePath) + } +} + +func TestLoadInvalid(t *testing.T) { + tests := []struct { + name string + env map[string]string + }{ + {"non-numeric port", map[string]string{"AO_PORT": "abc"}}, + {"port out of range", map[string]string{"AO_PORT": "70000"}}, + {"bad request timeout", map[string]string{"AO_REQUEST_TIMEOUT": "soon"}}, + {"bad shutdown timeout", map[string]string{"AO_SHUTDOWN_TIMEOUT": "later"}}, + {"zero request timeout", map[string]string{"AO_REQUEST_TIMEOUT": "0s"}}, + {"negative request timeout", map[string]string{"AO_REQUEST_TIMEOUT": "-1s"}}, + {"zero shutdown timeout", map[string]string{"AO_SHUTDOWN_TIMEOUT": "0s"}}, + {"negative shutdown timeout", map[string]string{"AO_SHUTDOWN_TIMEOUT": "-5s"}}, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + for k, v := range tc.env { + t.Setenv(k, v) + } + if _, err := Load(); err == nil { + t.Fatal("Load() = nil error, want error") + } + }) + } +} diff --git a/backend/internal/httpd/json.go b/backend/internal/httpd/json.go new file mode 100644 index 000000000..9b87461fb --- /dev/null +++ b/backend/internal/httpd/json.go @@ -0,0 +1,17 @@ +package httpd + +import ( + "encoding/json" + "net/http" +) + +// writeJSON serialises v as JSON with the given status. It is the single JSON +// writer for the skeleton; the typed error envelope (open item Q1.3) will build +// on this in a later phase. +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json; charset=utf-8") + w.WriteHeader(status) + // A write error here means the client went away mid-response; there is + // nothing useful to do but stop. + _ = json.NewEncoder(w).Encode(v) +} diff --git a/backend/internal/httpd/log.go b/backend/internal/httpd/log.go new file mode 100644 index 000000000..abb9462ec --- /dev/null +++ b/backend/internal/httpd/log.go @@ -0,0 +1,40 @@ +package httpd + +import ( + "log/slog" + "net/http" + "time" + + "github.com/go-chi/chi/v5/middleware" +) + +// requestLogger emits one structured access-log line per request via the +// daemon's slog logger. Chi's built-in middleware.Logger writes to stdout +// using stdlib log; reusing the daemon's slog keeps every line on stderr in +// the same key=value shape as the rest of the daemon (one stream for the +// Electron supervisor to capture, one format to grep). +// +// Status, bytes, and duration come from a wrapped ResponseWriter so the log +// is accurate even when the handler returns without calling WriteHeader. The +// request id is read off the context populated by middleware.RequestID, so +// this middleware must be mounted after it. +func requestLogger(log *slog.Logger) func(http.Handler) http.Handler { + return func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + ww := middleware.NewWrapResponseWriter(w, r.ProtoMajor) + start := time.Now() + defer func() { + log.Info("http request", + "id", middleware.GetReqID(r.Context()), + "method", r.Method, + "path", r.URL.Path, + "status", ww.Status(), + "bytes", ww.BytesWritten(), + "duration", time.Since(start), + "remote", r.RemoteAddr, + ) + }() + next.ServeHTTP(ww, r) + }) + } +} diff --git a/backend/internal/httpd/router.go b/backend/internal/httpd/router.go new file mode 100644 index 000000000..6e078b8df --- /dev/null +++ b/backend/internal/httpd/router.go @@ -0,0 +1,63 @@ +// Package httpd builds and runs the daemon's HTTP surface. Phase 1a is the +// skeleton: the middleware stack, liveness/readiness probes, and a graceful +// run loop. Route registration (/api/v1, /events, /mux, /) lands in later +// phases on top of the router this package builds. +package httpd + +import ( + "log/slog" + "net/http" + + "github.com/go-chi/chi/v5" + "github.com/go-chi/chi/v5/middleware" + + "github.com/aoagents/agent-orchestrator/backend/internal/config" +) + +// NewRouter builds the root router with the standard middleware stack and the +// health probes mounted. +// +// Middleware order (outermost first): +// +// Recoverer → turn a handler panic into 500 instead of crashing the daemon +// RequestID → attach a request id for correlation +// requestLogger → slog-backed access log, stderr, carries the request id +// RealIP → normalise client IP (loopback proxy from the dev server) +// +// The per-request Timeout from the decision table is deliberately NOT applied +// globally: it must wrap only the /api/v1 REST surface, never the long-lived +// SSE (/events) or WebSocket (/mux) surfaces, nor the always-must-answer health +// probes. It is therefore applied per-surface when those subrouters are mounted +// in Phase 1b; cfg.RequestTimeout carries the value through to that point. +func NewRouter(cfg config.Config, log *slog.Logger) chi.Router { + r := chi.NewRouter() + + r.Use(middleware.Recoverer) + r.Use(middleware.RequestID) + r.Use(requestLogger(log)) + r.Use(middleware.RealIP) + + mountHealth(r) + + return r +} + +// mountHealth registers the liveness and readiness probes the Electron +// supervisor polls before letting the renderer connect. +func mountHealth(r chi.Router) { + r.Get("/healthz", handleHealthz) + r.Get("/readyz", handleReadyz) +} + +// handleHealthz is the liveness probe: it answers 200 as long as the process is +// up and serving. It does no dependency checks by design. +func handleHealthz(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) +} + +// handleReadyz is the readiness probe. In the 1a skeleton the daemon is ready +// as soon as it is listening; later phases will gate this on dependency +// initialisation (e.g. store/event-bus warm-up). +func handleReadyz(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, http.StatusOK, map[string]string{"status": "ready"}) +} diff --git a/backend/internal/httpd/server.go b/backend/internal/httpd/server.go new file mode 100644 index 000000000..f42ed88aa --- /dev/null +++ b/backend/internal/httpd/server.go @@ -0,0 +1,113 @@ +package httpd + +import ( + "context" + "errors" + "fmt" + "log/slog" + "net" + "net/http" + "os" + "time" + + "github.com/aoagents/agent-orchestrator/backend/internal/config" + "github.com/aoagents/agent-orchestrator/backend/internal/runfile" +) + +// Server is the daemon's HTTP server together with its lifecycle: bind the +// loopback port, publish the running.json handshake, serve until the context +// is cancelled, then shut down gracefully and clean up the handshake file. +type Server struct { + cfg config.Config + log *slog.Logger + http *http.Server + listen net.Listener +} + +// New constructs a Server and binds the listener immediately so a port +// conflict fails fast — before any running.json is written. The caller owns +// the returned Server's lifecycle via Run. +func New(cfg config.Config, log *slog.Logger) (*Server, error) { + ln, err := net.Listen("tcp", cfg.Addr()) + if err != nil { + return nil, fmt.Errorf("bind %s (is a daemon already running?): %w", cfg.Addr(), err) + } + + srv := &Server{ + cfg: cfg, + log: log, + listen: ln, + http: &http.Server{ + Handler: NewRouter(cfg, log), + // ReadHeaderTimeout guards against slow-loris even on loopback; + // per-request body/handler timeouts are applied per-surface. + ReadHeaderTimeout: 10 * time.Second, + }, + } + return srv, nil +} + +// Addr returns the actual bound address (useful when the configured port was 0 +// and the OS chose one — primarily in tests). +func (s *Server) Addr() net.Addr { return s.listen.Addr() } + +// Run serves until ctx is cancelled (SIGINT/SIGTERM via signal.NotifyContext), +// then performs a graceful shutdown bounded by cfg.ShutdownTimeout. It writes +// running.json before serving and removes it on the way out. Run blocks until +// shutdown is complete. +func (s *Server) Run(ctx context.Context) error { + info := runfile.Info{ + PID: os.Getpid(), + Port: s.boundPort(), + StartedAt: time.Now().UTC(), + } + if err := runfile.Write(s.cfg.RunFilePath, info); err != nil { + s.listen.Close() + return fmt.Errorf("write run-file: %w", err) + } + defer func() { + if err := runfile.Remove(s.cfg.RunFilePath); err != nil { + s.log.Warn("failed to remove run-file", "path", s.cfg.RunFilePath, "err", err) + } + }() + + serveErr := make(chan error, 1) + go func() { + s.log.Info("daemon listening", "addr", s.Addr().String(), "pid", info.PID) + // Serve returns ErrServerClosed on a clean Shutdown; that is success. + if err := s.http.Serve(s.listen); err != nil && !errors.Is(err, http.ErrServerClosed) { + serveErr <- err + return + } + serveErr <- nil + }() + + select { + case err := <-serveErr: + // Serve died on its own (bind already happened, so this is a real + // runtime failure) before any shutdown signal. + return err + case <-ctx.Done(): + s.log.Info("shutdown signal received, draining connections", "timeout", s.cfg.ShutdownTimeout) + } + + shutdownCtx, cancel := context.WithTimeout(context.Background(), s.cfg.ShutdownTimeout) + defer cancel() + + if err := s.http.Shutdown(shutdownCtx); err != nil { + // The deadline elapsed with connections still open; force them closed. + s.log.Warn("graceful shutdown timed out, forcing close", "err", err) + _ = s.http.Close() + return fmt.Errorf("graceful shutdown exceeded %s: %w", s.cfg.ShutdownTimeout, err) + } + + s.log.Info("daemon stopped cleanly") + return <-serveErr +} + +func (s *Server) boundPort() int { + if tcp, ok := s.listen.Addr().(*net.TCPAddr); ok { + return tcp.Port + } + return s.cfg.Port +} diff --git a/backend/internal/httpd/server_test.go b/backend/internal/httpd/server_test.go new file mode 100644 index 000000000..2570397fc --- /dev/null +++ b/backend/internal/httpd/server_test.go @@ -0,0 +1,130 @@ +package httpd + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "path/filepath" + "testing" + "time" + + "github.com/aoagents/agent-orchestrator/backend/internal/config" + "github.com/aoagents/agent-orchestrator/backend/internal/runfile" +) + +func discardLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +func TestHealthProbes(t *testing.T) { + router := NewRouter(config.Config{}, discardLogger()) + srv := httptest.NewServer(router) + defer srv.Close() + + client := &http.Client{Timeout: 2 * time.Second} + for _, path := range []string{"/healthz", "/readyz"} { + resp, err := client.Get(srv.URL + path) + if err != nil { + t.Fatalf("GET %s: %v", path, err) + } + resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Errorf("GET %s = %d, want 200", path, resp.StatusCode) + } + if ct := resp.Header.Get("Content-Type"); ct != "application/json; charset=utf-8" { + t.Errorf("GET %s Content-Type = %q, want JSON", path, ct) + } + } +} + +// TestServerLifecycle exercises the full Run loop: bind an ephemeral port, +// publish running.json, serve a request, then cancel the context and confirm a +// clean shutdown that removes the handshake file. +func TestServerLifecycle(t *testing.T) { + runPath := filepath.Join(t.TempDir(), "running.json") + cfg := config.Config{ + Host: "127.0.0.1", + Port: 0, // let the OS pick a free port — no conflict with a real daemon + ShutdownTimeout: 5 * time.Second, + RunFilePath: runPath, + } + + srv, err := New(cfg, discardLogger()) + if err != nil { + t.Fatalf("New: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + runErr := make(chan error, 1) + go func() { runErr <- srv.Run(ctx) }() + + // Wait for the handshake file to confirm the server is up. + base := "http://" + srv.Addr().String() + waitForHealth(t, base) + + info, err := runfile.Read(runPath) + if err != nil { + t.Fatalf("read run-file: %v", err) + } + if info == nil { + t.Fatal("run-file not written while server running") + } + if info.Port == 0 { + t.Error("run-file recorded port 0; want the actual bound port") + } + + cancel() + + select { + case err := <-runErr: + if err != nil { + t.Fatalf("Run returned error on graceful shutdown: %v", err) + } + case <-time.After(10 * time.Second): + t.Fatal("Run did not return after context cancel") + } + + if after, _ := runfile.Read(runPath); after != nil { + t.Error("run-file still present after shutdown; want it removed") + } +} + +func waitForHealth(t *testing.T, base string) { + t.Helper() + // Per-request timeout so a stalled connect or hung handshake doesn't park + // the test for the full Go test timeout; the outer deadline only bounds + // the polling loop, not any single GET. + client := &http.Client{Timeout: 500 * time.Millisecond} + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + resp, err := client.Get(base + "/healthz") + if err == nil { + resp.Body.Close() + if resp.StatusCode == http.StatusOK { + return + } + } + time.Sleep(20 * time.Millisecond) + } + t.Fatal("server did not become healthy within timeout") +} + +// TestNewFailsOnPortConflict confirms a second bind of the same port fails +// fast rather than silently sharing it. +func TestNewFailsOnPortConflict(t *testing.T) { + cfg := config.Config{Host: "127.0.0.1", Port: 0, RunFilePath: filepath.Join(t.TempDir(), "r.json")} + + first, err := New(cfg, discardLogger()) + if err != nil { + t.Fatalf("first New: %v", err) + } + defer first.listen.Close() + + // Re-bind the exact port the first server took. + conflict := config.Config{Host: "127.0.0.1", Port: first.boundPort(), RunFilePath: cfg.RunFilePath} + if _, err := New(conflict, discardLogger()); err == nil { + t.Fatal("New on an already-bound port = nil error, want bind failure") + } +} diff --git a/backend/internal/runfile/process_unix.go b/backend/internal/runfile/process_unix.go new file mode 100644 index 000000000..efe957e18 --- /dev/null +++ b/backend/internal/runfile/process_unix.go @@ -0,0 +1,24 @@ +//go:build unix + +package runfile + +import ( + "errors" + "os" + "syscall" +) + +// processAlive probes existence with signal 0: kill(pid, 0) returns nil if the +// process exists and we can signal it, EPERM if it exists but is owned by +// another user, and ESRCH (or any other error from FindProcess) if it is gone. +func processAlive(pid int) bool { + proc, err := os.FindProcess(pid) + if err != nil { + return false + } + err = proc.Signal(syscall.Signal(0)) + if err == nil { + return true + } + return errors.Is(err, syscall.EPERM) +} diff --git a/backend/internal/runfile/process_windows.go b/backend/internal/runfile/process_windows.go new file mode 100644 index 000000000..1f8e78fee --- /dev/null +++ b/backend/internal/runfile/process_windows.go @@ -0,0 +1,21 @@ +//go:build windows + +package runfile + +import ( + "syscall" +) + +// processAlive opens the process with the minimum-rights query flag. On +// Windows, OpenProcess returns ERROR_INVALID_PARAMETER for a PID that no +// longer maps to a live process, and a usable handle when one is. We close +// the handle immediately; the only thing we needed was the open's outcome. +func processAlive(pid int) bool { + const PROCESS_QUERY_LIMITED_INFORMATION = 0x1000 + h, err := syscall.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, false, uint32(pid)) + if err != nil { + return false + } + _ = syscall.CloseHandle(h) + return true +} diff --git a/backend/internal/runfile/rename_unix.go b/backend/internal/runfile/rename_unix.go new file mode 100644 index 000000000..dd9dbd503 --- /dev/null +++ b/backend/internal/runfile/rename_unix.go @@ -0,0 +1,13 @@ +//go:build unix + +package runfile + +import "os" + +// atomicReplace renames src to dst, replacing dst if it exists. POSIX +// rename(2) is atomic and overwrites an existing destination by default, +// provided src and dst live on the same filesystem — which is always true +// here because the temp file is created in the target directory. +func atomicReplace(src, dst string) error { + return os.Rename(src, dst) +} diff --git a/backend/internal/runfile/rename_windows.go b/backend/internal/runfile/rename_windows.go new file mode 100644 index 000000000..031411eee --- /dev/null +++ b/backend/internal/runfile/rename_windows.go @@ -0,0 +1,28 @@ +//go:build windows + +package runfile + +import "syscall" + +// movefileReplaceExisting tells MoveFileEx to overwrite dst if it already +// exists. Mirrors MOVEFILE_REPLACE_EXISTING from the Win32 API; declared +// locally so we don't pull in golang.org/x/sys for a single constant. +const movefileReplaceExisting = 0x1 + +// atomicReplace renames src to dst, replacing dst if it exists. Go's +// os.Rename on Windows happens to do the same MoveFileEx call internally, +// but calling it directly makes the cross-platform contract explicit instead +// of leaning on a runtime implementation detail. The replace is atomic +// against concurrent readers — readers see either the old or the new file, +// never an empty or partially-written one. +func atomicReplace(src, dst string) error { + srcPtr, err := syscall.UTF16PtrFromString(src) + if err != nil { + return err + } + dstPtr, err := syscall.UTF16PtrFromString(dst) + if err != nil { + return err + } + return syscall.MoveFileEx(srcPtr, dstPtr, movefileReplaceExisting) +} diff --git a/backend/internal/runfile/runfile.go b/backend/internal/runfile/runfile.go new file mode 100644 index 000000000..7dafe1bef --- /dev/null +++ b/backend/internal/runfile/runfile.go @@ -0,0 +1,111 @@ +// Package runfile manages running.json — the PID + port handshake the Electron +// main process uses to discover, health-check, and reap the daemon. The daemon +// writes it on startup and removes it on graceful shutdown. On startup the +// daemon also checks for a stale entry left by a crashed predecessor so it can +// fail fast instead of fighting over the port. +package runfile + +import ( + "encoding/json" + "errors" + "fmt" + "os" + "path/filepath" + "time" +) + +// Info is the on-disk handshake payload. +type Info struct { + // PID is the daemon process id. + PID int `json:"pid"` + // Port is the loopback port the daemon bound. + Port int `json:"port"` + // StartedAt is when the daemon came up (RFC 3339). + StartedAt time.Time `json:"startedAt"` +} + +// Write atomically writes running.json at path, creating parent directories +// as needed. It writes to a temp file in the same directory and then calls +// atomicReplace — POSIX rename(2) on Unix, MoveFileEx with +// MOVEFILE_REPLACE_EXISTING on Windows — so a reader never observes a +// partial file and a stale running.json from a crashed predecessor is +// overwritten without an intermediate "no file" window. +func Write(path string, info Info) error { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return fmt.Errorf("create run-file dir: %w", err) + } + data, err := json.MarshalIndent(info, "", " ") + if err != nil { + return fmt.Errorf("marshal run-file: %w", err) + } + data = append(data, '\n') + + tmp, err := os.CreateTemp(filepath.Dir(path), ".running-*.json") + if err != nil { + return fmt.Errorf("create temp run-file: %w", err) + } + tmpName := tmp.Name() + defer os.Remove(tmpName) // no-op once the rename succeeds + + if _, err := tmp.Write(data); err != nil { + tmp.Close() + return fmt.Errorf("write temp run-file: %w", err) + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("close temp run-file: %w", err) + } + if err := atomicReplace(tmpName, path); err != nil { + return fmt.Errorf("replace run-file: %w", err) + } + return nil +} + +// Read loads running.json. A missing file returns (nil, nil) — that is the +// normal "no daemon recorded" state, not an error. +func Read(path string) (*Info, error) { + data, err := os.ReadFile(path) + if errors.Is(err, os.ErrNotExist) { + return nil, nil + } + if err != nil { + return nil, fmt.Errorf("read run-file: %w", err) + } + var info Info + if err := json.Unmarshal(data, &info); err != nil { + return nil, fmt.Errorf("parse run-file: %w", err) + } + return &info, nil +} + +// Remove deletes running.json. A missing file is not an error — graceful +// shutdown should be idempotent. +func Remove(path string) error { + if err := os.Remove(path); err != nil && !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("remove run-file: %w", err) + } + return nil +} + +// CheckStale inspects an existing run-file before the new daemon binds. It +// returns: +// +// - (nil, nil) no run-file, or one left by a dead process (safe to +// proceed; the caller should overwrite it); +// - (*Info, nil) a run-file whose recorded PID is still alive — a live +// daemon already owns the port, so the caller should fail fast. +// +// A run-file pointing at a dead PID is treated as stale and reported safe; the +// fresh Write will overwrite it. +func CheckStale(path string) (*Info, error) { + info, err := Read(path) + if err != nil { + return nil, err + } + if info == nil || info.PID <= 0 { + return nil, nil + } + if processAlive(info.PID) { + return info, nil + } + return nil, nil +} diff --git a/backend/internal/runfile/runfile_test.go b/backend/internal/runfile/runfile_test.go new file mode 100644 index 000000000..fbdf74e07 --- /dev/null +++ b/backend/internal/runfile/runfile_test.go @@ -0,0 +1,119 @@ +package runfile + +import ( + "os" + "path/filepath" + "testing" + "time" +) + +func TestWriteReadRoundTrip(t *testing.T) { + path := filepath.Join(t.TempDir(), "nested", "running.json") + want := Info{PID: 4242, Port: 3001, StartedAt: time.Now().UTC().Truncate(time.Second)} + + if err := Write(path, want); err != nil { + t.Fatalf("Write: %v", err) + } + got, err := Read(path) + if err != nil { + t.Fatalf("Read: %v", err) + } + if got == nil { + t.Fatal("Read returned nil for an existing file") + } + if got.PID != want.PID || got.Port != want.Port || !got.StartedAt.Equal(want.StartedAt) { + t.Errorf("round trip mismatch: got %+v, want %+v", *got, want) + } +} + +// TestWriteOverwritesExisting is the cross-platform overwrite check: a stale +// running.json from a crashed predecessor must be replaced cleanly. POSIX +// rename(2) handles this natively; Windows needs MoveFileEx with +// MOVEFILE_REPLACE_EXISTING — atomicReplace gives us both. +func TestWriteOverwritesExisting(t *testing.T) { + path := filepath.Join(t.TempDir(), "running.json") + + if err := Write(path, Info{PID: 1, Port: 3001}); err != nil { + t.Fatalf("first Write: %v", err) + } + if err := Write(path, Info{PID: 2, Port: 3002}); err != nil { + t.Fatalf("second Write (overwrite): %v", err) + } + + got, err := Read(path) + if err != nil { + t.Fatalf("Read: %v", err) + } + if got == nil || got.PID != 2 || got.Port != 3002 { + t.Errorf("after overwrite: got %+v, want PID=2 Port=3002", got) + } +} + +func TestReadMissingIsNotError(t *testing.T) { + got, err := Read(filepath.Join(t.TempDir(), "absent.json")) + if err != nil { + t.Fatalf("Read missing: %v", err) + } + if got != nil { + t.Errorf("Read missing = %+v, want nil", got) + } +} + +func TestRemoveIdempotent(t *testing.T) { + path := filepath.Join(t.TempDir(), "running.json") + if err := Remove(path); err != nil { + t.Errorf("Remove on missing file: %v", err) + } + if err := Write(path, Info{PID: 1, Port: 2}); err != nil { + t.Fatalf("Write: %v", err) + } + if err := Remove(path); err != nil { + t.Errorf("Remove existing: %v", err) + } + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Errorf("file still present after Remove") + } +} + +func TestCheckStaleDeadPID(t *testing.T) { + path := filepath.Join(t.TempDir(), "running.json") + // PID 0x7FFFFFFF is effectively guaranteed not to exist. + if err := Write(path, Info{PID: 0x7FFFFFFF, Port: 3001}); err != nil { + t.Fatalf("Write: %v", err) + } + live, err := CheckStale(path) + if err != nil { + t.Fatalf("CheckStale: %v", err) + } + if live != nil { + t.Errorf("CheckStale on dead PID = %+v, want nil (stale, safe to overwrite)", live) + } +} + +func TestCheckStaleLivePID(t *testing.T) { + path := filepath.Join(t.TempDir(), "running.json") + // This test process is unquestionably alive. + if err := Write(path, Info{PID: os.Getpid(), Port: 3001}); err != nil { + t.Fatalf("Write: %v", err) + } + live, err := CheckStale(path) + if err != nil { + t.Fatalf("CheckStale: %v", err) + } + if live == nil { + t.Fatal("CheckStale on live PID = nil, want the live Info") + } + if live.PID != os.Getpid() { + t.Errorf("live.PID = %d, want %d", live.PID, os.Getpid()) + } +} + +func TestCheckStaleNoFile(t *testing.T) { + live, err := CheckStale(filepath.Join(t.TempDir(), "absent.json")) + if err != nil { + t.Fatalf("CheckStale: %v", err) + } + if live != nil { + t.Errorf("CheckStale with no file = %+v, want nil", live) + } +} diff --git a/backend/main.go b/backend/main.go index 30a6e84c6..78a232927 100644 --- a/backend/main.go +++ b/backend/main.go @@ -1,7 +1,62 @@ +// Command backend is the Agent Orchestrator HTTP daemon: a loopback-only +// sidecar spawned and supervised by the Electron main process. Phase 1a brings +// up the server skeleton — config, 127.0.0.1 bind, middleware stack, health +// probes, the running.json handshake, and graceful shutdown. package main -import "fmt" +import ( + "context" + "fmt" + "log/slog" + "os" + "os/signal" + "syscall" + + "github.com/aoagents/agent-orchestrator/backend/internal/config" + "github.com/aoagents/agent-orchestrator/backend/internal/httpd" + "github.com/aoagents/agent-orchestrator/backend/internal/runfile" +) func main() { - fmt.Println("ao backend daemon starting") + if err := run(); err != nil { + fmt.Fprintln(os.Stderr, "ao backend daemon: "+err.Error()) + os.Exit(1) + } +} + +func run() error { + cfg, err := config.Load() + if err != nil { + return err + } + + log := newLogger() + + // Fail fast if a live daemon already owns the handshake file. A run-file + // left by a crashed predecessor (dead PID) is treated as stale and + // overwritten when the new server starts. + if live, err := runfile.CheckStale(cfg.RunFilePath); err != nil { + return fmt.Errorf("inspect run-file: %w", err) + } else if live != nil { + return fmt.Errorf("daemon already running (pid %d, port %d); refusing to start", live.PID, live.Port) + } + + srv, err := httpd.New(cfg, log) + if err != nil { + return err + } + + // signal.NotifyContext cancels ctx on SIGINT/SIGTERM, which drives the + // graceful shutdown inside Server.Run. + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + return srv.Run(ctx) +} + +// newLogger returns the daemon's slog logger. It writes to stderr so the +// Electron supervisor can capture it separately from any structured stdout +// protocol added later. +func newLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelDebug})) }