dominicnunez · Sep 19, 2026 · Sep 12, 2026 · Sep 12, 2026 · Sep 19, 2026 · Sep 19, 2026
diff --git a/cmd/agentos/dashboard_linux.go b/cmd/agentos/dashboard_linux.go
@@ -64,7 +64,7 @@ func runDashboard(ctx context.Context, config bootstrap.Config, output io.Writer
 	if err := startDashboardBrowser(ctx, bootstrapPath); err != nil {
 		_, _ = fmt.Fprintln(output, "Automatic browser launch unavailable; use the private manual launch path above.")
 	}
-	return serveAll(ctx, []serverBinding{{server: server, listener: listener}})
+	return serveAll(ctx, []serverBinding{{server: server, listener: listener}}, nil)
 }
 
 func startDashboardBrowser(ctx context.Context, bootstrapPath string) error {

diff --git a/cmd/agentos/main.go b/cmd/agentos/main.go
@@ -123,12 +123,22 @@ func runServer(ctx context.Context, config bootstrap.Config, source secrets.Sour
 	if err != nil {
 		return err
 	}
-	defer func() { err = errors.Join(err, models.close()) }()
+	var service *app.Service
+	defer func() {
+		if service != nil {
+			service.StopExecutions()
+		}
+		err = errors.Join(err, models.close())
+		if service != nil {
+			stopCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+			defer cancel()
+			err = errors.Join(err, service.WaitForStops(stopCtx))
+		}
+	}()
 	planner, err := planning.NewModelPlanner(planningModel{adapter: models.planning})
 	if err != nil {
 		return err
 	}
-	var service *app.Service
 	var selectedPlanner planning.Planner = planner
 	if models.registry != nil && config.Routing.PlanningRequirements != nil {
 		route, err := newAuxiliaryRoute(models.registry, config.Routing.Planning, *config.Routing.PlanningRequirements)
@@ -145,6 +155,10 @@ func runServer(ctx context.Context, config bootstrap.Config, source secrets.Sour
 	} else {
 		service = app.NewWithModelAndPlanner(events.NewGateway(l), models.task, planner)
 	}
+	// Recovery can dispatch work before any HTTP listener exists. Observe
+	// runtime cancellation now, rather than waiting for serving to return.
+	stopOnCancel := context.AfterFunc(ctx, service.StopExecutions)
+	defer stopOnCancel()
 	if _, err := service.Recover(ctx); err != nil {
 		return fmt.Errorf("recover durable runtime before serving: %w", err)
 	}
@@ -214,7 +228,7 @@ func runServer(ctx context.Context, config bootstrap.Config, source secrets.Sour
 		log.Printf("Agent OS A2A gateway listening on %s", a2aListener.Addr())
 		bindings = append(bindings, serverBinding{server: a2aServer, listener: a2aListener, certFile: tlsCertFile, keyFile: tlsKeyFile})
 	}
-	return serveAll(ctx, bindings)
+	return serveAll(ctx, bindings, service.StopExecutions)
 }
 
 type inferenceAdmissionStore interface {
@@ -404,7 +418,7 @@ func configuredProvider(ctx context.Context, provider bootstrap.Provider, runtim
 }
 
 func serve(ctx context.Context, server *http.Server, listener net.Listener, certFile, keyFile string) error {
-	return serveAll(ctx, []serverBinding{{server: server, listener: listener, certFile: certFile, keyFile: keyFile}})
+	return serveAll(ctx, []serverBinding{{server: server, listener: listener, certFile: certFile, keyFile: keyFile}}, nil)
 }
 
 type serverBinding struct {
@@ -413,7 +427,7 @@ type serverBinding struct {
 	certFile, keyFile string
 }
 
-func serveAll(ctx context.Context, bindings []serverBinding) error {
+func serveAll(ctx context.Context, bindings []serverBinding, stopWork func()) error {
 	if ctx == nil || len(bindings) == 0 {
 		closeListeners(bindings)
 		return fmt.Errorf("runtime context, server, and listener are required")
@@ -428,6 +442,16 @@ func serveAll(ctx context.Context, bindings []serverBinding) error {
 			return fmt.Errorf("runtime context, server, and listener are required")
 		}
 	}
+	requestLifetime, cancelRequests := context.WithCancel(ctx)
+	defer cancelRequests()
+	wrappedServers := make(map[*http.Server]struct{}, len(bindings))
+	for _, binding := range bindings {
+		if _, found := wrappedServers[binding.server]; found {
+			continue
+		}
+		wrappedServers[binding.server] = struct{}{}
+		binding.server.Handler = cancelOnShutdown(binding.server.Handler, requestLifetime)
+	}
 	results := make(chan error, len(bindings))
 	for _, binding := range bindings {
 		binding := binding
@@ -449,6 +473,12 @@ func serveAll(ctx context.Context, bindings []serverBinding) error {
 		}
 	case <-ctx.Done():
 	}
+	cancelRequests()
+	// Stop dispatch and active supervisors before waiting for their HTTP
+	// requests. A listener failure must contain sibling listeners' work too.
+	if stopWork != nil {
+		stopWork()
+	}
 	shutdownCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
 	defer cancel()
 	for _, binding := range bindings {
@@ -468,6 +498,24 @@ func serveAll(ctx context.Context, bindings []serverBinding) error {
 	return result
 }
 
+func cancelOnShutdown(handler http.Handler, shutdown context.Context) http.Handler {
+	if handler == nil {
+		handler = http.DefaultServeMux
+	}
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requestCtx, cancel := context.WithCancel(r.Context())
+		stop := context.AfterFunc(shutdown, cancel)
+		if shutdown.Err() != nil {
+			cancel()
+		}
+		defer func() {
+			stop()
+			cancel()
+		}()
+		handler.ServeHTTP(w, r.WithContext(requestCtx))
+	})
+}
+
 func closeListeners(bindings []serverBinding) {
 	for _, binding := range bindings {
 		if binding.listener != nil {

diff --git a/cmd/agentos/main_test.go b/cmd/agentos/main_test.go
@@ -242,7 +242,7 @@ func TestServeAllStopsBothListenersWhenRuntimeContextIsCancelled(t *testing.T) {
 	}
 	ctx, cancel := context.WithCancel(context.Background())
 	done := make(chan error, 1)
-	go func() { done <- serveAll(ctx, bindings) }()
+	go func() { done <- serveAll(ctx, bindings, nil) }()
 	for _, listener := range listeners {
 		response, err := http.Get("http://" + listener.Addr().String()) //nolint:gosec,noctx // loopback listener owned by this test
 		if err != nil {

diff --git a/cmd/agentos/normalization_shutdown_linux_test.go b/cmd/agentos/normalization_shutdown_linux_test.go
@@ -0,0 +1,229 @@
+//go:build linux
+
+package main
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"strconv"
+	"sync"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/dominicnunez/agentos/internal/app"
+	"github.com/dominicnunez/agentos/internal/core"
+	"github.com/dominicnunez/agentos/internal/events"
+	"github.com/dominicnunez/agentos/internal/execution"
+	"github.com/dominicnunez/agentos/internal/gateway"
+	"github.com/dominicnunez/agentos/internal/inference"
+	"github.com/dominicnunez/agentos/internal/intake"
+	"github.com/dominicnunez/agentos/internal/ledger"
+	"github.com/dominicnunez/agentos/internal/modelinput"
+)
+
+type normalizationShutdownContextKey struct{}
+
+type shutdownNormalizationAdapter struct {
+	started, cancelled chan struct{}
+	once               sync.Once
+	preservedBaseValue bool
+}
+
+func (*shutdownNormalizationAdapter) Name() string { return "shutdown-normalization/test-model" }
+
+func (*shutdownNormalizationAdapter) Descriptor() execution.ModelDescriptor {
+	return execution.ModelDescriptor{Provider: "shutdown-normalization", Model: "test-model", ExecutionProfileVersion: "test-profile"}
+}
+
+func (m *shutdownNormalizationAdapter) Complete(ctx context.Context, _ string) (execution.ModelResponse, error) {
+	return m.waitForShutdown(ctx)
+}
+
+func (m *shutdownNormalizationAdapter) CompleteRequest(ctx context.Context, _ modelinput.Request) (execution.ModelResponse, error) {
+	return m.waitForShutdown(ctx)
+}
+
+func (m *shutdownNormalizationAdapter) waitForShutdown(ctx context.Context) (execution.ModelResponse, error) {
+	m.once.Do(func() { close(m.started) })
+	m.preservedBaseValue = ctx.Value(normalizationShutdownContextKey{}) == "base-context-value"
+	<-ctx.Done()
+	close(m.cancelled)
+	return execution.ModelResponse{}, context.Cause(ctx)
+}
+
+func TestShutdownCancelsActiveHTTPNormalizationBeforeDrain(t *testing.T) {
+	for _, trigger := range []string{"runtime-cancelled", "listener-failed"} {
+		t.Run(trigger, func(t *testing.T) {
+			uid, gid := syscall.Geteuid(), syscall.Getegid()
+			store, err := ledger.Open(":memory:")
+			if err != nil {
+				t.Fatal(err)
+			}
+			t.Cleanup(func() { _ = store.Close() })
+			now := time.Now().UTC()
+			if err := store.ActivateInferencePolicy(t.Context(), inference.Policy{
+				Version: inference.PolicyVersion, OrganizationID: "org-1", Provider: "shutdown-normalization", Model: "test-model",
+				ExecutionProfileVersion: "test-profile", Mode: inference.Local,
+				MaxInputTokensPerRequest: 262_144, MaxOutputTokensPerRequest: 262_144, MaxTokensPerWindow: 1_000_000,
+				ContinuityReserveTokens: 100_000, WindowDurationSeconds: 3600, MaxConcurrentRequests: 1, MaxAttemptsPerRequest: 1,
+				AuthorizedBy: "local-uid-" + strconv.Itoa(uid), AuthorizedAt: now.Add(-time.Minute), AuthorizationExpiresAt: now.Add(time.Hour),
+			}); err != nil {
+				t.Fatal(err)
+			}
+			adapter := &shutdownNormalizationAdapter{started: make(chan struct{}), cancelled: make(chan struct{})}
+			guarded, err := inference.NewGuardedAdapter(store, adapter)
+			if err != nil {
+				t.Fatal(err)
+			}
+			normalizer, err := intake.NewModelNormalizer(intakeModel{adapter: guarded})
+			if err != nil {
+				t.Fatal(err)
+			}
+			eventGateway := events.NewGateway(store)
+			runtime := app.New(eventGateway)
+			operator := intake.NewWithNormalizer(runtime, normalizer)
+			human, err := gateway.NewHuman(operator, gateway.LocalHuman{
+				UID: uid, ID: core.ID("local-uid-" + strconv.Itoa(uid)), OrganizationID: "org-1",
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			userMux := http.NewServeMux()
+			userMux.Handle("/v1/user/", human)
+			userServer := newHTTPServer("", userMux, nil)
+			userServer.ConnContext = localConnContext
+			userServer.BaseContext = func(net.Listener) context.Context {
+				return context.WithValue(context.Background(), normalizationShutdownContextKey{}, "base-context-value")
+			}
+			runtimeDir := filepath.Join(t.TempDir(), "runtime")
+			if err := os.Mkdir(runtimeDir, 0o700); err != nil {
+				t.Fatal(err)
+			}
+			socketPath := filepath.Join(runtimeDir, "user.sock")
+			userListener, err := listenLocalHuman(t.Context(), socketPath, uid, gid)
+			if err != nil {
+				t.Fatal(err)
+			}
+			siblingListener, err := (&net.ListenConfig{}).Listen(t.Context(), "tcp", "127.0.0.1:0")
+			if err != nil {
+				_ = userListener.Close()
+				t.Fatal(err)
+			}
+			siblingServer := newHTTPServer("", http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+				w.WriteHeader(http.StatusNoContent)
+			}), nil)
+			ctx, cancel := context.WithCancel(t.Context())
+			serveDone := make(chan error, 1)
+			go func() {
+				defer close(serveDone)
+				serveDone <- serveAll(ctx, []serverBinding{{server: siblingServer, listener: siblingListener}, {server: userServer, listener: userListener}}, runtime.StopExecutions)
+			}()
+			transport := &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
+				return (&net.Dialer{}).DialContext(ctx, "unix", socketPath)
+			}}
+			client := &http.Client{Transport: transport, Timeout: 15 * time.Second}
+			requestDone := make(chan struct {
+				status int
+				err    error
+			}, 1)
+			go func() {
+				defer close(requestDone)
+				body := bytes.NewBufferString(`{"conversation_id":"shutdown-normalization","message_id":"message-1","text":"Draft a release note"}`)
+				request, requestErr := http.NewRequestWithContext(context.Background(), http.MethodPost, "http://agentos.local/v1/user/messages", body)
+				if requestErr != nil {
+					requestDone <- struct {
+						status int
+						err    error
+					}{err: requestErr}
+					return
+				}
+				request.Header.Set("Content-Type", "application/json")
+				response, requestErr := client.Do(request)
+				if requestErr != nil {
+					requestDone <- struct {
+						status int
+						err    error
+					}{err: requestErr}
+					return
+				}
+				_, _ = io.Copy(io.Discard, response.Body)
+				_ = response.Body.Close()
+				requestDone <- struct {
+					status int
+					err    error
+				}{status: response.StatusCode}
+			}()
+			t.Cleanup(func() {
+				cancel()
+				_ = userServer.Close()
+				_ = siblingServer.Close()
+				client.CloseIdleConnections()
+				select {
+				case <-serveDone:
+				case <-time.After(12 * time.Second):
+					t.Error("server cleanup did not finish")
+				}
+				select {
+				case <-requestDone:
+				case <-time.After(2 * time.Second):
+					t.Error("request cleanup did not finish")
+				}
+			})
+			select {
+			case <-adapter.started:
+			case <-time.After(5 * time.Second):
+				t.Fatal("normalization did not reach the provider adapter")
+			}
+			if trigger == "runtime-cancelled" {
+				cancel()
+			} else if err := siblingListener.Close(); err != nil {
+				t.Fatal(err)
+			}
+			select {
+			case <-adapter.cancelled:
+			case <-time.After(2 * time.Second):
+				t.Fatal("HTTP shutdown did not cancel active normalization before drain")
+			}
+			if !adapter.preservedBaseValue {
+				t.Fatal("shutdown context discarded the server base context")
+			}
+			select {
+			case result := <-requestDone:
+				if result.err != nil || result.status != http.StatusServiceUnavailable {
+					t.Fatalf("normalization response status=%d err=%v", result.status, result.err)
+				}
+			case <-time.After(5 * time.Second):
+				t.Fatal("cancelled normalization request did not drain")
+			}
+			select {
+			case err := <-serveDone:
+				if trigger == "runtime-cancelled" && err != nil {
+					t.Fatalf("runtime shutdown: %v", err)
+				}
+				if trigger == "listener-failed" && !errors.Is(err, net.ErrClosed) {
+					t.Fatalf("listener failure was lost: %v", err)
+				}
+			case <-time.After(5 * time.Second):
+				t.Fatal("servers did not finish draining cancelled normalization")
+			}
+			stream, err := eventGateway.Events(t.Context(), "")
+			if err != nil {
+				t.Fatal(err)
+			}
+			counts := map[string]int{}
+			for _, event := range stream {
+				counts[event.EventType]++
+			}
+			if counts["INFERENCE_RESERVED"] != 1 || counts["INFERENCE_RECONCILED"] != 1 || counts["INTENT_NORMALIZATION_FAILED"] != 1 || counts["INTENT_DRAFTED"] != 0 {
+				t.Fatalf("shutdown normalization accounting=%v", counts)
+			}
+		})
+	}
+}