diff --git a/apps/backend/cmd/agentctl/main.go b/apps/backend/cmd/agentctl/main.go index 6302a4562c..0de057cd76 100644 --- a/apps/backend/cmd/agentctl/main.go +++ b/apps/backend/cmd/agentctl/main.go @@ -15,6 +15,7 @@ import ( "net/http" "os" "os/signal" + "strconv" "syscall" "time" @@ -125,8 +126,12 @@ func run(cfg *config.Config, log *logger.Logger) { } }() - // Wait for shutdown signal - waitForShutdown(log, func(ctx context.Context) { + // Monitor parent liveness via inherited pipe (standalone mode only). + // Returns a channel that closes when the parent dies. + parentDied := monitorParentLiveness(log) + + // Wait for shutdown signal (OS signal or parent death) + waitForShutdown(log, parentDied, func(ctx context.Context) { // Flush pending traces before stopping instances if err := shared.ShutdownTracing(ctx); err != nil { log.Error("error shutting down tracing", zap.Error(err)) @@ -141,11 +146,24 @@ func run(cfg *config.Config, log *logger.Logger) { }) } -// waitForShutdown waits for shutdown signal and calls cleanup -func waitForShutdown(log *logger.Logger, cleanup func(ctx context.Context)) { - quit := make(chan os.Signal, 1) - signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) - <-quit +// waitForShutdown waits for a shutdown trigger (OS signal or parent death) and +// runs the cleanup function. parentDied may be nil when no parent monitor is active. +func waitForShutdown(log *logger.Logger, parentDied <-chan struct{}, cleanup func(ctx context.Context)) { + sigCh := make(chan os.Signal, 1) + signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) + + if parentDied == nil { + // No parent monitor — wait for OS signal only. + sig := <-sigCh + log.Info("received signal", zap.String("signal", sig.String())) + } else { + select { + case sig := <-sigCh: + log.Info("received signal", zap.String("signal", sig.String())) + case <-parentDied: + log.Warn("parent process died, initiating shutdown") + } + } log.Info("shutting down agentctl...") @@ -156,3 +174,40 @@ func waitForShutdown(log *logger.Logger, cleanup func(ctx context.Context)) { log.Info("agentctl stopped") } + +// monitorParentLiveness watches a pipe inherited from the parent process. +// The parent (kandev backend) passes the read-end of a pipe via ExtraFiles +// and sets KANDEV_PARENT_PIPE_FD to the FD number. A goroutine blocks on +// reading the pipe. When the parent dies — even via SIGKILL — the kernel +// closes the write-end, the read returns, and the returned channel is closed. +// Returns nil when the env var is absent (Docker, manual start, remote executors). +func monitorParentLiveness(log *logger.Logger) <-chan struct{} { + fdStr := os.Getenv("KANDEV_PARENT_PIPE_FD") + if fdStr == "" { + return nil + } + + fd, err := strconv.Atoi(fdStr) + if err != nil { + log.Warn("invalid KANDEV_PARENT_PIPE_FD", zap.String("value", fdStr), zap.Error(err)) + return nil + } + + pipe := os.NewFile(uintptr(fd), "parent-liveness-pipe") + if pipe == nil { + log.Warn("failed to open parent liveness pipe", zap.Int("fd", fd)) + return nil + } + + log.Info("parent liveness monitor started", zap.Int("fd", fd)) + + ch := make(chan struct{}) + go func() { + buf := make([]byte, 1) + _, _ = pipe.Read(buf) // blocks until pipe breaks + _ = pipe.Close() + close(ch) + }() + + return ch +} diff --git a/apps/backend/cmd/agentctl/main_test.go b/apps/backend/cmd/agentctl/main_test.go new file mode 100644 index 0000000000..d8937b3171 --- /dev/null +++ b/apps/backend/cmd/agentctl/main_test.go @@ -0,0 +1,74 @@ +package main + +import ( + "fmt" + "os" + "testing" + "time" + + "github.com/kandev/kandev/internal/common/logger" +) + +func newTestLogger(t *testing.T) *logger.Logger { + t.Helper() + log, err := logger.NewLogger(logger.LoggingConfig{ + Level: "warn", + Format: "console", + OutputPath: "stdout", + }) + if err != nil { + t.Fatalf("failed to create logger: %v", err) + } + return log +} + +func TestMonitorParentLiveness_NoEnvVar(t *testing.T) { + t.Setenv("KANDEV_PARENT_PIPE_FD", "") + ch := monitorParentLiveness(newTestLogger(t)) + if ch != nil { + t.Fatal("expected nil channel when env var is absent") + } +} + +func TestMonitorParentLiveness_InvalidEnvVar(t *testing.T) { + t.Setenv("KANDEV_PARENT_PIPE_FD", "not-a-number") + ch := monitorParentLiveness(newTestLogger(t)) + if ch != nil { + t.Fatal("expected nil channel for invalid FD value") + } +} + +func TestMonitorParentLiveness_PipeBreak(t *testing.T) { + // Create a pipe: r is the read-end (what agentctl would inherit), + // w is the write-end (what the parent keeps open). + r, w, err := os.Pipe() + if err != nil { + t.Fatalf("os.Pipe: %v", err) + } + defer func() { _ = r.Close() }() + + // Point the env var at the read-end's FD. + t.Setenv("KANDEV_PARENT_PIPE_FD", fmt.Sprintf("%d", r.Fd())) + + ch := monitorParentLiveness(newTestLogger(t)) + if ch == nil { + t.Fatal("expected non-nil channel") + } + + // Channel should not be closed yet. + select { + case <-ch: + t.Fatal("channel closed before pipe break") + case <-time.After(50 * time.Millisecond): + } + + // Close write-end to simulate parent death. + _ = w.Close() + + select { + case <-ch: + // success + case <-time.After(2 * time.Second): + t.Fatal("channel not closed after pipe break") + } +} diff --git a/apps/backend/internal/agentctl/client/launcher/launcher.go b/apps/backend/internal/agentctl/client/launcher/launcher.go index d04ce1b2f2..f9a6eef5f1 100644 --- a/apps/backend/internal/agentctl/client/launcher/launcher.go +++ b/apps/backend/internal/agentctl/client/launcher/launcher.go @@ -33,6 +33,11 @@ type Launcher struct { // For clean shutdown stopping bool + + // Parent liveness pipe: write-end kept open by the launcher. + // When the backend dies (even SIGKILL), the kernel closes this FD, + // breaking the pipe. agentctl detects the break and self-terminates. + parentPipe *os.File } // Config holds configuration for the launcher. @@ -144,9 +149,19 @@ func (l *Launcher) Start(ctx context.Context) error { // Set process attributes: // - Pdeathsig on Linux: kernel sends SIGTERM to child when parent dies. - // - Setpgid: create new process group so Ctrl+C doesn't propagate directly. l.cmd.SysProcAttr = buildSysProcAttr() + // Parent liveness pipe: agentctl inherits the read-end (FD 3 via ExtraFiles) + // and blocks on it. When this process dies — even via SIGKILL — the kernel + // closes the write-end, breaking the pipe. agentctl detects the break and + // initiates graceful shutdown. This is the portable equivalent of Pdeathsig. + pipeRead, pipeWrite, err := os.Pipe() + if err != nil { + return fmt.Errorf("failed to create parent liveness pipe: %w", err) + } + l.cmd.ExtraFiles = []*os.File{pipeRead} // child FD 3 + l.cmd.Env = append(l.cmd.Env, "KANDEV_PARENT_PIPE_FD=3") + // Capture stdout and stderr stdout, err := l.cmd.StdoutPipe() if err != nil { @@ -159,9 +174,16 @@ func (l *Launcher) Start(ctx context.Context) error { // Start the process if err := l.cmd.Start(); err != nil { + _ = pipeRead.Close() + _ = pipeWrite.Close() return fmt.Errorf("failed to start agentctl: %w", err) } + // Close read-end in parent (child inherited it). Keep write-end open — + // it closes automatically when this process exits, signaling agentctl. + _ = pipeRead.Close() + l.parentPipe = pipeWrite + l.logger.Info("agentctl process started", zap.Int("pid", l.cmd.Process.Pid)) // Pipe stdout/stderr to logger in background @@ -177,6 +199,7 @@ func (l *Launcher) Start(ctx context.Context) error { if killErr := l.cmd.Process.Kill(); killErr != nil { l.logger.Warn("failed to kill agentctl process after failed health check", zap.Error(killErr)) } + l.closeParentPipeLocked() return fmt.Errorf("agentctl failed to become healthy: %w", err) } @@ -208,6 +231,10 @@ func (l *Launcher) Stop(ctx context.Context) error { l.logger.Info("stopping agentctl subprocess", zap.Int("pid", pid)) + // Close the liveness pipe first — this signals agentctl that the parent + // is shutting down, complementing the SIGTERM that follows. + l.closeParentPipe() + // Send graceful stop signal (SIGTERM on Unix, interrupt on Windows) if err := l.gracefulStop(pid); err != nil { return err @@ -231,6 +258,22 @@ func (l *Launcher) Stop(ctx context.Context) error { } } +// closeParentPipe closes and nils the parent liveness pipe if open. +// Safe for concurrent use. +func (l *Launcher) closeParentPipe() { + l.mu.Lock() + defer l.mu.Unlock() + l.closeParentPipeLocked() +} + +// closeParentPipeLocked is like closeParentPipe but assumes mu is already held. +func (l *Launcher) closeParentPipeLocked() { + if l.parentPipe != nil { + _ = l.parentPipe.Close() + l.parentPipe = nil + } +} + // checkPortAvailable verifies the given port is not in use. // It checks by attempting a wildcard bind (matching what agentctl does with ":port"). func checkPortAvailable(port int) error { diff --git a/apps/backend/internal/agentctl/client/launcher/launcher_test.go b/apps/backend/internal/agentctl/client/launcher/launcher_test.go new file mode 100644 index 0000000000..94a9b0b7ec --- /dev/null +++ b/apps/backend/internal/agentctl/client/launcher/launcher_test.go @@ -0,0 +1,50 @@ +//go:build !windows + +package launcher + +import ( + "os" + "testing" +) + +func TestCloseParentPipe_ClosesAndNils(t *testing.T) { + r, w, err := os.Pipe() + if err != nil { + t.Fatalf("os.Pipe: %v", err) + } + defer func() { _ = r.Close() }() + + l := &Launcher{parentPipe: w} + l.closeParentPipe() + + if l.parentPipe != nil { + t.Error("parentPipe should be nil after closeParentPipe") + } + + // Writing to the closed write-end should fail. + _, err = w.Write([]byte("x")) + if err == nil { + t.Error("expected error writing to closed pipe") + } +} + +func TestCloseParentPipe_NilIsNoop(t *testing.T) { + l := &Launcher{} + // Must not panic. + l.closeParentPipe() + if l.parentPipe != nil { + t.Error("parentPipe should remain nil") + } +} + +func TestCloseParentPipe_Idempotent(t *testing.T) { + _, w, err := os.Pipe() + if err != nil { + t.Fatalf("os.Pipe: %v", err) + } + + l := &Launcher{parentPipe: w} + l.closeParentPipe() + // Second call must not panic (double close). + l.closeParentPipe() +} diff --git a/apps/backend/internal/agentctl/client/launcher/platform_unix.go b/apps/backend/internal/agentctl/client/launcher/platform_unix.go index eb16bc57f2..26e6d7f4e5 100644 --- a/apps/backend/internal/agentctl/client/launcher/platform_unix.go +++ b/apps/backend/internal/agentctl/client/launcher/platform_unix.go @@ -20,8 +20,6 @@ func (l *Launcher) gracefulStop(pid int) error { } // forceKill sends SIGKILL to the agentctl process. -// agentctl shares the backend's process group, so we kill the single process -// rather than the group (which would kill the backend itself). func (l *Launcher) forceKill(pid int) { _ = syscall.Kill(pid, syscall.SIGKILL) } diff --git a/apps/web/e2e/fixtures/backend.ts b/apps/web/e2e/fixtures/backend.ts index a6d420876a..353e52809c 100644 --- a/apps/web/e2e/fixtures/backend.ts +++ b/apps/web/e2e/fixtures/backend.ts @@ -51,6 +51,45 @@ function killProcess(proc: ChildProcess): Promise { }); } +/** + * Kills an entire process group. Used for the backend process which is spawned + * with `detached: true` so it becomes a process group leader. Sending signals + * to the negative PID targets all processes in that group (backend + agentctl). + * The 7s grace period gives agentctl time to cascade cleanup to agent process groups. + */ +function killProcessGroup(proc: ChildProcess): Promise { + return new Promise((resolve) => { + if (!proc.pid) { + resolve(); + return; + } + + const pid = proc.pid; + + try { + process.kill(-pid, "SIGTERM"); + } catch { + // Process group may already be gone + resolve(); + return; + } + + const timeout = setTimeout(() => { + try { + process.kill(-pid, "SIGKILL"); + } catch { + // Already dead + } + resolve(); + }, 7_000); + + proc.on("exit", () => { + clearTimeout(timeout); + resolve(); + }); + }); +} + /** * Worker-scoped fixture that spawns an isolated backend process and * a dedicated Next.js frontend. Each Playwright worker gets its own @@ -113,6 +152,7 @@ export const backendFixture = base.extend({ const backendProc: ChildProcess = spawn(KANDEV_BIN, [], { env: backendEnv as unknown as NodeJS.ProcessEnv, stdio: ["ignore", "pipe", "pipe"], + detached: true, // Own process group so we can kill backend + agentctl together }); const logFile = debug ? fs.createWriteStream(`/tmp/e2e-backend-${backendPort}.log`) : null; @@ -167,9 +207,9 @@ export const backendFixture = base.extend({ try { await use({ port: backendPort, baseUrl, frontendPort, frontendUrl, tmpDir }); } finally { - // Shutdown frontend first, then backend + // Shutdown frontend first (simple process), then backend (process group) await killProcess(frontendProc); - await killProcess(backendProc); + await killProcessGroup(backendProc); // Cleanup temp directory — ignore errors (backend may still hold files briefly) try {