fix(dockeragent): bound the collect cycle with a watchdog and pin hung-daemon deadline behavior

Follow-up to the 2026-07-17 live docker-update exercise where a DiskUsage
roundtrip against a colima daemon parked 6+ minutes even though
dockerCallWithRetry wraps every call in a 20s context.WithTimeout.

Investigation result: context deadline propagation through moby client
v0.5.0 (request.go, API-version negotiation) and the otelhttp transport
wrapper is intact. Reproducing with a deliberately hung unix-socket
daemon aborts DiskUsage at the deadline in both hang shapes (pre-header
and mid-body), so there is no client-library bug to fix or file
upstream; the production stall's root cause remains environmental
(deadline timer never fired process-side).

Containment and diagnosis:
- buildReport now runs under dockerCollectCycleTimeout (5m) so a wedged
  cycle can never stall the module indefinitely, plus an independent
  watchdog timer that logs an error with a full goroutine dump if the
  cycle outlives even that deadline - capturing exactly the evidence
  that was missing from the original incident.
- hung_daemon_deadline_test.go pins that a context deadline aborts the
  real moby client against a hung unix-socket daemon (pre-header and
  mid-body stalls), guarding future moby/otelhttp upgrades.

The incident note referenced a dockerCollectCycleTimeout watchdog as
already added; it did not exist on any branch - this commit is that
containment, landed for real.

Contract-Neutral: dockeragent collect-cycle watchdog containment: timeout plumbing only, no collection-semantics or contract-surface delta

internal/dockeragent/collect.go

@@ -24,6 +24,16 @@ import (
 
 // buildReport gathers all system and container metrics into a single report
 func (a *Agent) buildReport(ctx context.Context) (agentsdocker.Report, error) {
+	// Cycle-level containment: per-call deadlines below should bound every
+	// docker call, but one live exercise saw a call park for minutes with its
+	// deadline never firing, so the whole cycle gets its own ceiling plus an
+	// independent watchdog that dumps goroutines if even this deadline fails
+	// to abort the cycle.
+	ctx, cancel := context.WithTimeout(ctx, dockerCollectCycleTimeout)
+	defer cancel()
+	stopWatchdog := startCollectCycleWatchdog(a.logger, dockerCollectCycleTimeout+dockerCollectWatchdogGrace)
+	defer stopWatchdog()
+
 	info, err := dockerCallWithRetry(ctx, dockerInfoCallTimeout, func(callCtx context.Context) (systemtypes.Info, error) {
 		return a.docker.Info(callCtx)
 	})

internal/dockeragent/hardening.go

@@ -5,9 +5,12 @@ import (
 	"errors"
 	"fmt"
 	"net"
+	"runtime"
 	"strings"
 	"syscall"
 	"time"
+
+	"github.com/rs/zerolog"
 )
 
 const (
@@ -19,10 +22,48 @@ const (
 	dockerUpdateCallTimeout        = 2 * time.Minute
 	dockerUpdateOverallTimeout     = 15 * time.Minute
 
+	// dockerCollectCycleTimeout bounds one whole collection cycle
+	// (buildReport). Every docker call inside the cycle already carries its
+	// own per-call timeout, so a healthy cycle finishes in seconds; this
+	// ceiling only trips when a call stalls without its deadline aborting it
+	// (observed once on 2026-07-17: a DiskUsage roundtrip parked 6+ minutes
+	// on a colima daemon with the 20s per-call deadline never firing).
+	dockerCollectCycleTimeout = 5 * time.Minute
+
+	// dockerCollectWatchdogGrace is how long past dockerCollectCycleTimeout
+	// the watchdog waits before concluding that the cycle context deadline
+	// failed to abort the cycle and dumping goroutines for diagnosis.
+	dockerCollectWatchdogGrace = 30 * time.Second
+
 	dockerCallRetryAttempts = 2
 	dockerRetryBaseDelay    = 200 * time.Millisecond
 )
 
+// startCollectCycleWatchdog arms an independent timer that fires only if a
+// collection cycle is still running after budget. The cycle context deadline
+// should abort the cycle well before then, so the watchdog firing means
+// context cancellation was swallowed or a runtime timer was lost; it logs the
+// full goroutine stack so a recurrence of the 2026-07-17 stall is diagnosable
+// in the field. The returned stop function must be called when the cycle ends.
+func startCollectCycleWatchdog(logger zerolog.Logger, budget time.Duration) (stop func()) {
+	done := make(chan struct{})
+	go func() {
+		timer := time.NewTimer(budget)
+		defer timer.Stop()
+		select {
+		case <-done:
+		case <-timer.C:
+			buf := make([]byte, 64*1024)
+			n := runtime.Stack(buf, true)
+			logger.Error().
+				Dur("budget", budget).
+				Str("goroutines", string(buf[:n])).
+				Msg("Docker collect cycle still running past its watchdog budget; the cycle context deadline did not abort it")
+		}
+	}()
+	return func() { close(done) }
+}
+
 func dockerCallWithRetry[T any](ctx context.Context, timeout time.Duration, call func(context.Context) (T, error)) (T, error) {
 	return dockerCallWithRetryAttempts(ctx, timeout, dockerCallRetryAttempts, call)
 }

internal/dockeragent/hardening_test.go

@@ -1,14 +1,37 @@
 package dockeragent
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"strings"
+	"sync"
 	"syscall"
 	"testing"
 	"time"
+
+	"github.com/rs/zerolog"
 )
 
+// syncBuffer is a goroutine-safe bytes.Buffer for capturing watchdog log
+// output written from another goroutine.
+type syncBuffer struct {
+	mu  sync.Mutex
+	buf bytes.Buffer
+}
+
+func (b *syncBuffer) Write(p []byte) (int, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.Write(p)
+}
+
+func (b *syncBuffer) String() string {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.buf.String()
+}
+
 func TestDockerCallWithRetryAttempts_RetriesTransientErrors(t *testing.T) {
 	t.Parallel()
 
@@ -80,3 +103,40 @@ func TestAnnotateDockerConnectionError(t *testing.T) {
 		t.Fatalf("expected timeout hint, got %q", timeoutErr.Error())
 	}
 }
+
+func TestCollectCycleWatchdogFiresWhenNotStopped(t *testing.T) {
+	var out syncBuffer
+	logger := zerolog.New(&out)
+
+	stop := startCollectCycleWatchdog(logger, 20*time.Millisecond)
+	defer stop()
+
+	deadline := time.Now().Add(2 * time.Second)
+	for time.Now().Before(deadline) {
+		if strings.Contains(out.String(), "watchdog budget") {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	logged := out.String()
+	if !strings.Contains(logged, "watchdog budget") {
+		t.Fatalf("watchdog did not log after budget elapsed: %q", logged)
+	}
+	if !strings.Contains(logged, "goroutine") {
+		t.Fatalf("watchdog log does not include a goroutine dump: %q", logged)
+	}
+}
+
+func TestCollectCycleWatchdogSilentWhenStopped(t *testing.T) {
+	var out syncBuffer
+	logger := zerolog.New(&out)
+
+	stop := startCollectCycleWatchdog(logger, 30*time.Millisecond)
+	stop()
+
+	time.Sleep(100 * time.Millisecond)
+	if logged := out.String(); logged != "" {
+		t.Fatalf("watchdog logged after being stopped: %q", logged)
+	}
+}

internal/dockeragent/hung_daemon_deadline_test.go

@@ -0,0 +1,181 @@
+package dockeragent
+
+// Regression pins for the 2026-07-17 live-exercise stall: collectStorageUsage
+// sat 6+ minutes inside moby DiskUsage even though dockerCallWithRetry wraps
+// every call in context.WithTimeout. Investigation could not reproduce a
+// deadline being swallowed anywhere in the moby client stack; these tests pin
+// that a context deadline aborts the real moby client (request.go + otelhttp
+// transport + API-version negotiation) against a hung unix-socket daemon, in
+// both hang shapes, so a future moby/otelhttp upgrade cannot silently regress
+// it:
+//
+//   - hang before response headers (daemon computing /system/df forever)
+//   - hang mid-body (daemon streams partial JSON then stalls)
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/moby/moby/client"
+)
+
+// hungSocketServer listens on a unix socket, accepts connections, discards
+// whatever the client writes, and never responds.
+func startHungSocketServer(t *testing.T) string {
+	t.Helper()
+
+	dir, err := os.MkdirTemp("", "pulsedu")
+	if err != nil {
+		t.Fatalf("mkdtemp: %v", err)
+	}
+	t.Cleanup(func() { _ = os.RemoveAll(dir) })
+
+	sockPath := filepath.Join(dir, "docker.sock")
+	ln, err := net.Listen("unix", sockPath)
+	if err != nil {
+		t.Fatalf("listen unix: %v", err)
+	}
+	t.Cleanup(func() { _ = ln.Close() })
+
+	go func() {
+		for {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go func(c net.Conn) {
+				defer c.Close()
+				buf := make([]byte, 4096)
+				for {
+					if _, err := c.Read(buf); err != nil {
+						return
+					}
+					// Read requests forever, never write a byte back.
+				}
+			}(conn)
+		}
+	}()
+
+	return sockPath
+}
+
+// startStallingHTTPServer answers /_ping so API-version negotiation succeeds,
+// then for /system/df writes response headers plus a partial JSON body and
+// stalls forever.
+func startStallingHTTPServer(t *testing.T) string {
+	t.Helper()
+
+	dir, err := os.MkdirTemp("", "pulsedu")
+	if err != nil {
+		t.Fatalf("mkdtemp: %v", err)
+	}
+	t.Cleanup(func() { _ = os.RemoveAll(dir) })
+
+	sockPath := filepath.Join(dir, "docker.sock")
+	ln, err := net.Listen("unix", sockPath)
+	if err != nil {
+		t.Fatalf("listen unix: %v", err)
+	}
+
+	hang := make(chan struct{})
+	mux := http.NewServeMux()
+	mux.HandleFunc("/_ping", func(w http.ResponseWriter, _ *http.Request) {
+		w.Header().Set("Api-Version", "1.51")
+		w.WriteHeader(http.StatusOK)
+	})
+	mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) {
+		// Any versioned endpoint (e.g. /v1.51/system/df): stream a partial
+		// body, then stall.
+		w.Header().Set("Content-Type", "application/json")
+		w.WriteHeader(http.StatusOK)
+		_, _ = fmt.Fprint(w, `{"LayersSize": 1, "Images": [`)
+		if f, ok := w.(http.Flusher); ok {
+			f.Flush()
+		}
+		<-hang
+	})
+
+	srv := &http.Server{Handler: mux}
+	go func() { _ = srv.Serve(ln) }()
+	t.Cleanup(func() {
+		close(hang)
+		_ = srv.Close()
+	})
+
+	return sockPath
+}
+
+func newTestMobyClient(t *testing.T, sockPath string) dockerClient {
+	t.Helper()
+
+	cli, err := newMobyDockerClient(
+		client.WithHost("unix://"+sockPath),
+		client.WithAPIVersionNegotiation(),
+	)
+	if err != nil {
+		t.Fatalf("new moby client: %v", err)
+	}
+	t.Cleanup(func() { _ = cli.Close() })
+	return cli
+}
+
+// diskUsageViaProductionPath mirrors collectStorageUsage's exact wrapper and
+// options.
+func diskUsageViaProductionPath(ctx context.Context, cli dockerClient, timeout time.Duration) error {
+	_, err := dockerCallWithRetryAttempts(ctx, timeout, 1, func(callCtx context.Context) (client.DiskUsageResult, error) {
+		return cli.DiskUsage(callCtx, dockerDiskUsageOptions{
+			Containers: true,
+			Images:     true,
+			BuildCache: true,
+			Volumes:    true,
+			Verbose:    true,
+		})
+	})
+	return err
+}
+
+func TestDiskUsageDeadlineAbortsHungSocketBeforeHeaders(t *testing.T) {
+	sockPath := startHungSocketServer(t)
+	cli := newTestMobyClient(t, sockPath)
+
+	const timeout = 2 * time.Second
+	start := time.Now()
+	err := diskUsageViaProductionPath(context.Background(), cli, timeout)
+	elapsed := time.Since(start)
+
+	if err == nil {
+		t.Fatalf("expected error from hung daemon socket, got nil after %v", elapsed)
+	}
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("expected context.DeadlineExceeded, got %v (after %v)", err, elapsed)
+	}
+	if elapsed > timeout+3*time.Second {
+		t.Fatalf("deadline did not abort the call promptly: elapsed %v for %v timeout", elapsed, timeout)
+	}
+	t.Logf("pre-header hang aborted after %v with: %v", elapsed, err)
+}
+
+func TestDiskUsageDeadlineAbortsHungSocketMidBody(t *testing.T) {
+	sockPath := startStallingHTTPServer(t)
+	cli := newTestMobyClient(t, sockPath)
+
+	const timeout = 2 * time.Second
+	start := time.Now()
+	err := diskUsageViaProductionPath(context.Background(), cli, timeout)
+	elapsed := time.Since(start)
+
+	if err == nil {
+		t.Fatalf("expected error from mid-body stall, got nil after %v", elapsed)
+	}
+	if elapsed > timeout+3*time.Second {
+		t.Fatalf("deadline did not abort the mid-body stall promptly: elapsed %v for %v timeout (err=%v)", elapsed, timeout, err)
+	}
+	t.Logf("mid-body stall aborted after %v with: %v", elapsed, err)
+}