Fix Docker-in-LXC probe storm against slow Proxmox hosts

The guest Docker socket probe hung minipc hard enough to need a power
cycle (2026-08-20): ~100 orphaned pct exec children, load 133, sshd and
pveproxy starved. Three bugs chained, each fixed here:

1. Dispatcher re-issued a probe while the previous one was still
   executing. The poll cycle's enrichment context had expired, so
   ExecuteCommand dispatched, returned the context error 50ms later,
   and the next 3s cycle sent the identical command again — unbounded
   concurrency against a host that was slow to begin with. The
   monitoring dispatcher now takes a per-guest in-flight claim before
   dispatching probe or inventory commands (completed probes release
   it; abandoned ones hold it for a 2-minute window), and both dispatch
   paths bail out under a dead context.

2. The host agent never got the July process-leak fix: 45480a5cc
   landed only on pulse/v6-release, so main-line agents killed just the
   direct shell on timeout, orphaning pct exec → lxc-attach children
   and blocking Wait on their inherited pipes (10s timeouts reported as
   300s+ durations). Port it: run each command in its own process
   group, SIGKILL the group on cancel, bound Wait with WaitDelay, and
   treat ErrWaitDelay after a clean exit as success.

3. Server-side abandonment never reached the agent. ExecuteCommand and
   ReadFile now refuse to dispatch under an already-expired context,
   and send a best-effort cancel_command when they stop waiting; the
   agent cancels the in-flight execution (killing its process group)
   and reports "command canceled". Older agents ignore the unknown
   message type.

Also add a per-node circuit breaker: three consecutive command failures
on one node suspend all Docker probe/inventory dispatch to it on the
existing 1m→30m backoff schedule, so a host-level stall (NFS flapping)
stops the probing entirely instead of failing guest by guest.

Regression tests simulate the storm without hardware: a never-returning
executor is not re-issued across poll cycles, an expired context
dispatches nothing and records no failure, abandoned probes hold their
claim, the breaker blocks new guests on a failing node, and the agent
kills the whole process group on timeout and on server-issued cancel.

Contract-Neutral: monitor.go delta is three private struct fields holding Docker probe dispatch state; host-agent deletion/re-enrollment lifecycle untouched — contracts and all other proofs are staged

docs/release-control/v6/internal/subsystems/agent-lifecycle.md

@@ -6428,3 +6428,41 @@ authors a `privilege` block in its report (`pkg/agents/host/report.go`
 server can present the profile descriptively. Uninstall removes the sudoers
 file and helpers. `scripts/installtests/install_sh_test.go`
 (`TestInstallSHLeastPrivilegeProfile`) pins the profile's invariants.
+
+### Command dispatch is context-honest and abandoned executions are canceled
+
+The agent command transport now refuses to dispatch work its caller has
+already stopped waiting for, and propagates abandonment to the agent
+(minipc probe-storm incident, 2026-08-20: probes dispatched under an
+expired poll context were re-issued every cycle while every previous copy
+kept running on the Proxmox host). Three coupled guarantees:
+
+1. `agentexec.Server.ExecuteCommand` and `ReadFile` fail with a
+   `not dispatched` error when the caller's context is already expired —
+   nothing crosses the WebSocket, so a caller polling on a dead deadline
+   cannot leave the agent executing commands nobody awaits.
+2. When the server stops waiting for a dispatched request (its own
+   command timeout or caller-context cancellation), it sends the new
+   best-effort server→agent `cancel_command` message
+   (`MsgTypeCancelCmd`, payload `CancelCommandPayload{request_id}`).
+   Agents that predate the message ignore the unknown type and fall back
+   to their own per-command timeout; the protocol change is additive.
+3. The unified agent's command client tracks in-flight
+   `execute_command`/`read_file` executions by request ID and, on
+   `cancel_command`, cancels that execution's context. Command execution
+   runs each command in its own process group (`Setpgid`; SIGKILL of the
+   group via `cmd.Cancel`, `taskkill /T` on Windows), bounds `Wait` with
+   a 5s `WaitDelay` so orphan-held pipes cannot hang it, treats
+   `exec.ErrWaitDelay` after a clean exit as success, and reports
+   cancellation as a distinct `command canceled` failure. This ports the
+   `pulse/v6-release` process-leak fix (45480a5cc) to main, which had
+   never received it, and extends it with server-driven cancellation.
+
+Proofs: `internal/agentexec/server_websocket_test.go`
+(`TestExecuteCommand_ExpiredContextNeverDispatches`,
+`TestExecuteCommand_AbandonedCommandSendsCancel`),
+`internal/hostagent/command_client_test.go`
+(`TestCommandClient_handleCancelCommand_CancelsRegisteredRequest`,
+`TestCommandClient_handleCancelCommand_UnknownRequestIsNoOp`), and
+`internal/hostagent/commands_execute_unix_test.go` (timeout and cancel
+kill the whole process group; WaitDelay unblocks inherited pipes).

docs/release-control/v6/internal/subsystems/monitoring.md

@@ -3080,3 +3080,38 @@ that evidence alone. Proofs:
 `internal/models/deepcopy_test.go` (`TestCloneHostIsolatesAgentPrivilege`),
 `internal/monitoring/agent_fleet_doctor_test.go`
 (`TestAgentFleetDiagnosticsSurfacesPrivilegeProfileWithoutDegradingHealth`).
+
+### Guest Docker command dispatch is single-flight, context-honest, and node-breakered
+
+The Proxmox guest Docker socket probe and inventory dispatcher in
+`internal/monitoring/docker_detection.go` owns dispatch discipline, not just
+result caching (minipc probe-storm incident, 2026-08-20):
+
+- **Single-flight per guest.** A dispatched probe or inventory command takes
+  an in-flight claim keyed by container ID (inventory under an
+  `inventory:` prefix). Overlapping poll cycles skip claimed guests, so a
+  slow `pct exec` can never be stacked with identical copies of itself. A
+  completed command (success or genuine failure) releases the claim; an
+  abandoned one (the wait ended with a context error, so the agent may
+  still be executing) holds it for a 2-minute window.
+- **No dispatch under a dead context.** `CheckContainersForDocker` and
+  `CollectProxmoxGuestDockerInventory` bail out before dispatching when the
+  enrichment context has already expired, preserving previous Docker
+  status; the parallel prober also re-checks the context deterministically
+  after semaphore acquisition.
+- **Abandonment is not evidence.** Context-canceled/deadline results feed
+  neither the per-guest failure backoff nor the node breaker — they say
+  nothing about the guest or node.
+- **Per-node circuit breaker.** Three consecutive completed command
+  failures on one node suspend all guest Docker command dispatch to that
+  node on the existing 1m→30m backoff schedule (a node-level stall such as
+  NFS flapping fails every guest, including newly appearing ones); any
+  completed success closes the breaker. Checker reconfiguration resets
+  claims, streaks, and breakers.
+
+Proofs in `internal/monitoring/monitor_docker_test.go`:
+`TestCheckContainersForDocker_InFlightProbeNotReissued`,
+`TestCheckContainersForDocker_ExpiredContextDoesNotDispatch`,
+`TestCheckContainersForDocker_AbandonedProbeHoldsClaimWithoutFailure`,
+`TestCheckContainersForDocker_NodeCircuitBreaker`,
+`TestCollectProxmoxGuestDockerInventory_InFlightAndExpiredContext`.

internal/agentexec/server.go

@@ -1546,6 +1546,17 @@ func (s *Server) ExecuteCommand(ctx context.Context, agentID string, cmd Execute
 		return nil, err
 	}
 
+	// Never dispatch under a context that has already expired: the send
+	// would succeed, this call would return the context error a moment
+	// later, and the agent would be left executing a command nobody is
+	// waiting for. A caller polling on a dead deadline can otherwise
+	// re-issue the same command every cycle while every previous copy is
+	// still running on the target host (minipc probe-storm incident,
+	// 2026-08-20).
+	if err := ctx.Err(); err != nil {
+		return nil, fmt.Errorf("command %q not dispatched: %w", cmd.RequestID, err)
+	}
+
 	startedAt := time.Now()
 
 	ac, ok := s.connectionForContext(ctx, agentID)
@@ -1653,12 +1664,14 @@ func (s *Server) ExecuteCommand(ctx context.Context, agentID string, cmd Execute
 			Msg("Agent command completed")
 		return &result, nil
 	case <-timer.C:
+		s.cancelAgentCommand(ac, cmd.RequestID)
 		execLog.Warn().
 			Dur("timeout", timeout).
 			Dur("duration", time.Since(startedAt)).
 			Msg("Agent command timed out")
 		return nil, fmt.Errorf("command timed out after %v", timeout)
 	case <-ctx.Done():
+		s.cancelAgentCommand(ac, cmd.RequestID)
 		execLog.Warn().
 			Err(ctx.Err()).
 			Dur("duration", time.Since(startedAt)).
@@ -1671,6 +1684,25 @@ func (s *Server) ExecuteCommand(ctx context.Context, agentID string, cmd Execute
 	}
 }
 
+// cancelAgentCommand tells an agent to abort an execute_command request the
+// server has stopped waiting for, so the agent can reap the command's process
+// tree instead of running it to its full timeout. Best effort: agents that
+// predate the cancel_command message ignore it and fall back to their own
+// per-command timeout.
+func (s *Server) cancelAgentCommand(ac *agentConn, requestID string) {
+	msg, err := NewMessage(MsgTypeCancelCmd, requestID, CancelCommandPayload{RequestID: requestID})
+	if err != nil {
+		log.Debug().Err(err).Str("request_id", requestID).Msg("Failed to encode command cancellation")
+		return
+	}
+	ac.writeMu.Lock()
+	err = s.sendMessage(ac.conn, msg)
+	ac.writeMu.Unlock()
+	if err != nil {
+		log.Debug().Err(err).Str("request_id", requestID).Msg("Failed to send command cancellation to agent")
+	}
+}
+
 // hostOperationPayload exposes the durable operation identity shared by the
 // typed host APT operation request payloads.
 type hostOperationPayload interface {
@@ -2094,6 +2126,12 @@ func (s *Server) ReadFile(ctx context.Context, agentID string, req ReadFilePaylo
 		return nil, err
 	}
 
+	// Same rule as ExecuteCommand: never dispatch work the caller has
+	// already stopped waiting for.
+	if err := ctx.Err(); err != nil {
+		return nil, fmt.Errorf("read_file %q not dispatched: %w", req.RequestID, err)
+	}
+
 	ac, ok := s.connectionForContext(ctx, agentID)
 	if !ok {
 		log.Warn().
@@ -2177,8 +2215,10 @@ func (s *Server) ReadFile(ctx context.Context, agentID string, req ReadFilePaylo
 			Msg("Agent read_file completed")
 		return &result, nil
 	case <-timer.C:
+		s.cancelAgentCommand(ac, req.RequestID)
 		return nil, fmt.Errorf("read_file timed out after %v", timeout)
 	case <-ctx.Done():
+		s.cancelAgentCommand(ac, req.RequestID)
 		return nil, fmt.Errorf("read_file %q on agent %q canceled: %w", req.RequestID, agentID, ctx.Err())
 	case <-ac.done:
 		return nil, fmt.Errorf("agent %s disconnected before read_file result", agentID)

internal/agentexec/server_websocket_test.go

@@ -1078,3 +1078,159 @@ func TestRevokedAdmissionInvalidatesStaleSocketBeforeDispatch(t *testing.T) {
 		t.Fatal("revoked session remained visible as connected")
 	}
 }
+
+// registerCancelTestAgent registers agent "a1" over the websocket harness and
+// returns after the registration ack has been read.
+func registerCancelTestAgent(t *testing.T, s *Server, tsURL string) *cancelTestConn {
+	t.Helper()
+	conn, _, err := dialAgentExecWebSocket(t, tsURL)
+	if err != nil {
+		t.Fatalf("Dial: %v", err)
+	}
+	t.Cleanup(func() { conn.Close() })
+	wsWriteMessage(t, conn, mustNewMessage(t, MsgTypeAgentRegister, "", AgentRegisterPayload{
+		AgentID:  "a1",
+		Hostname: "host1",
+		Version:  "1.2.3",
+		Platform: "linux",
+		Token:    "any",
+	}))
+	_ = wsReadRegisteredPayload(t, conn)
+	return &cancelTestConn{t: t, conn: conn}
+}
+
+type cancelTestConn struct {
+	t    *testing.T
+	conn *websocket.Conn
+}
+
+// nextMessage returns the next message from the server, or ok=false when
+// nothing arrives before the timeout.
+func (c *cancelTestConn) nextMessage(timeout time.Duration) (wsRawMessage, bool) {
+	c.t.Helper()
+	msg, err := wsReadRawMessageWithTimeout(c.conn, timeout)
+	if err != nil {
+		return wsRawMessage{}, false
+	}
+	return msg, true
+}
+
+// The probe-storm incident (minipc, 2026-08-20) started with the server
+// dispatching commands under a parent context that had already expired: the
+// send succeeded, ExecuteCommand returned "context deadline exceeded,
+// duration 0.05", and the agent was left running the command. An expired
+// context must fail the call before anything reaches the agent.
+func TestExecuteCommand_ExpiredContextNeverDispatches(t *testing.T) {
+	s := NewServer(allowAllTestTokens)
+	ts := newWSServer(t, s)
+	defer ts.Close()
+
+	agent := registerCancelTestAgent(t, s, ts.URL)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	_, err := s.ExecuteCommand(ctx, "a1", ExecuteCommandPayload{
+		RequestID: "req-expired",
+		Command:   "echo hi",
+		Timeout:   5,
+		Trusted:   true,
+	})
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("ExecuteCommand error = %v, want context.Canceled", err)
+	}
+	if !strings.Contains(err.Error(), "not dispatched") {
+		t.Fatalf("error should say the command was not dispatched, got %q", err)
+	}
+
+	// The agent must never see the command.
+	if msg, ok := agent.nextMessage(500 * time.Millisecond); ok && msg.Type == MsgTypeExecuteCmd {
+		t.Fatalf("agent received execute_command despite expired context")
+	}
+}
+
+// When the server stops waiting for a dispatched command (its own timeout or
+// the caller's context), it must tell the agent to abort the execution so the
+// process tree is reaped instead of running on for minutes.
+func TestExecuteCommand_AbandonedCommandSendsCancel(t *testing.T) {
+	cases := []struct {
+		name    string
+		timeout int // ExecuteCommandPayload.Timeout in seconds
+		abandon func(cancel context.CancelFunc)
+		wantErr string
+	}{
+		{
+			name:    "server timeout",
+			timeout: 1,
+			abandon: func(context.CancelFunc) {}, // let the 1s timer fire
+			wantErr: "timed out",
+		},
+		{
+			name:    "caller context canceled",
+			timeout: 30,
+			abandon: func(cancel context.CancelFunc) {
+				time.Sleep(200 * time.Millisecond)
+				cancel()
+			},
+			wantErr: "context canceled",
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			s := NewServer(allowAllTestTokens)
+			ts := newWSServer(t, s)
+			defer ts.Close()
+
+			agent := registerCancelTestAgent(t, s, ts.URL)
+
+			ctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			go tc.abandon(cancel)
+
+			requestID := "req-" + strings.ReplaceAll(tc.name, " ", "-")
+			_, err := s.ExecuteCommand(ctx, "a1", ExecuteCommandPayload{
+				RequestID: requestID,
+				Command:   "sleep 300",
+				Timeout:   tc.timeout,
+				Trusted:   true,
+			})
+			if err == nil || !strings.Contains(err.Error(), tc.wantErr) {
+				t.Fatalf("ExecuteCommand error = %v, want containing %q", err, tc.wantErr)
+			}
+
+			// The agent first receives the command, then the cancellation.
+			sawExecute := false
+			sawCancel := false
+			deadline := time.Now().Add(3 * time.Second)
+			for time.Now().Before(deadline) && !sawCancel {
+				msg, ok := agent.nextMessage(time.Until(deadline))
+				if !ok {
+					break
+				}
+				switch msg.Type {
+				case MsgTypeExecuteCmd:
+					sawExecute = true
+				case MsgTypeCancelCmd:
+					if msg.Payload == nil {
+						t.Fatalf("cancel_command payload missing")
+					}
+					var payload CancelCommandPayload
+					if err := json.Unmarshal(*msg.Payload, &payload); err != nil {
+						t.Fatalf("unmarshal cancel_command payload: %v", err)
+					}
+					if payload.RequestID != requestID {
+						t.Fatalf("cancel_command request_id = %q, want %q", payload.RequestID, requestID)
+					}
+					sawCancel = true
+				}
+			}
+			if !sawExecute {
+				t.Fatalf("agent never received execute_command")
+			}
+			if !sawCancel {
+				t.Fatalf("agent never received cancel_command after the server abandoned the request")
+			}
+		})
+	}
+}

internal/agentexec/types.go

@@ -36,6 +36,7 @@ const (
 	MsgTypeDeployPreflight          MessageType = "deploy_preflight"
 	MsgTypeDeployInstall            MessageType = "deploy_install"
 	MsgTypeDeployCancelJob          MessageType = "deploy_cancel"
+	MsgTypeCancelCmd                MessageType = "cancel_command"
 
 	// Agent -> Server messages (deploy)
 	MsgTypeDeployProgress MessageType = "deploy_progress"
@@ -103,6 +104,15 @@ type RegisteredPayload struct {
 	Message string `json:"message,omitempty"`
 }
 
+// CancelCommandPayload is sent by the server when it stops waiting for a
+// previously dispatched execute_command request (caller context expired or
+// the server-side timeout fired). Agents that understand it abort the running
+// command and reap its whole process tree; older agents ignore the message
+// and rely on their own per-command timeout.
+type CancelCommandPayload struct {
+	RequestID string `json:"request_id"`
+}
+
 // ExecuteCommandPayload is sent by server to request command execution
 type ExecuteCommandPayload struct {
 	RequestID     string                `json:"request_id"`

internal/hostagent/command_client_test.go

@@ -1,6 +1,7 @@
 package hostagent
 
 import (
+	"context"
 	"io"
 	"testing"
 
@@ -219,3 +220,42 @@ func TestCommandClientBuildWebSocketOrigin(t *testing.T) {
 		})
 	}
 }
+
+// A server-issued cancel_command for an in-flight request must cancel exactly
+// that request's execution context (minipc probe-storm regression: abandoned
+// probes previously ran to their full timeout on the agent).
+func TestCommandClient_handleCancelCommand_CancelsRegisteredRequest(t *testing.T) {
+	c := &CommandClient{logger: zerolog.Nop()}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	c.registerActiveCommand("req-1", cancel)
+	defer c.unregisterActiveCommand("req-1")
+
+	c.handleCancelCommand(cancelCommandPayload{RequestID: "req-1"})
+
+	select {
+	case <-ctx.Done():
+	default:
+		t.Fatalf("cancel_command did not cancel the registered request context")
+	}
+}
+
+// A cancel for a request that already finished (or never existed) must be a
+// no-op, not a panic or a cancellation of some other command.
+func TestCommandClient_handleCancelCommand_UnknownRequestIsNoOp(t *testing.T) {
+	c := &CommandClient{logger: zerolog.Nop()}
+
+	otherCtx, otherCancel := context.WithCancel(context.Background())
+	defer otherCancel()
+	c.registerActiveCommand("other", otherCancel)
+	defer c.unregisterActiveCommand("other")
+
+	c.handleCancelCommand(cancelCommandPayload{RequestID: "missing"})
+
+	select {
+	case <-otherCtx.Done():
+		t.Fatalf("cancel for unknown request canceled an unrelated command")
+	default:
+	}
+}

internal/hostagent/commands.go

@@ -45,6 +45,10 @@ const (
 	outputTruncatedMsg   = "\n... (output truncated)"
 )
 
+// How long Wait may block on inherited stdout/stderr pipes after the command
+// exits or is killed. Package var so tests can shorten it.
+var commandWaitDelay = 5 * time.Second
+
 var (
 	reconnectMaxDelay    = 5 * time.Minute
 	reconnectJitterRatio = 0.1
@@ -121,6 +125,13 @@ type CommandClient struct {
 	conn   *websocket.Conn
 	connMu sync.Mutex
 	done   chan struct{}
+
+	// In-flight execute_command requests by request ID, so a server-issued
+	// cancel_command can abort the execution (and its process group) instead
+	// of letting an abandoned command run to its full timeout. The server
+	// enforces unique in-flight request IDs on its side.
+	activeCommandsMu sync.Mutex
+	activeCommands   map[string]context.CancelFunc
 }
 
 // NewCommandClient creates a new command execution client
@@ -202,6 +213,7 @@ const (
 	msgTypeDeployInstall                  messageType = "deploy_install"
 	msgTypeDeployCancel                   messageType = "deploy_cancel"
 	msgTypeDeployProgress                 messageType = "deploy_progress"
+	msgTypeCancelCmd                      messageType = "cancel_command"
 )
 
 type wsMessage struct {
@@ -253,6 +265,12 @@ type commandResultPayload struct {
 	Duration  int64  `json:"duration_ms"`
 }
 
+// cancelCommandPayload is sent by the server when it stops waiting for a
+// previously dispatched execute_command request.
+type cancelCommandPayload struct {
+	RequestID string `json:"request_id"`
+}
+
 // Run starts the command client and maintains the WebSocket connection
 func (c *CommandClient) Run(ctx context.Context) error {
 	stopCh := c.stopChan()
@@ -639,6 +657,14 @@ func (c *CommandClient) handleMessages(ctx context.Context, conn *websocket.Conn
 			}
 			c.handleDeployCancel(payload)
 
+		case msgTypeCancelCmd:
+			var payload cancelCommandPayload
+			if err := json.Unmarshal(msg.Payload, &payload); err != nil {
+				c.logger.Error().Err(err).Msg("Failed to parse cancel_command payload")
+				continue
+			}
+			c.handleCancelCommand(payload)
+
 		default:
 			c.logger.Debug().Str("type", string(msg.Type)).Msg("Unknown message type")
 		}
@@ -1037,7 +1063,12 @@ func (c *CommandClient) handleExecuteCommand(ctx context.Context, conn *websocke
 		Str("target_id", payload.TargetID).
 		Msg("Executing command")
 
-	result := c.executeCommand(ctx, payload)
+	cmdCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+	c.registerActiveCommand(payload.RequestID, cancel)
+	defer c.unregisterActiveCommand(payload.RequestID)
+
+	result := c.executeCommand(cmdCtx, payload)
 	result.Duration = time.Since(startTime).Milliseconds()
 
 	// Send result back
@@ -1069,6 +1100,40 @@ func (c *CommandClient) handleExecuteCommand(ctx context.Context, conn *websocke
 	}
 }
 
+func (c *CommandClient) registerActiveCommand(requestID string, cancel context.CancelFunc) {
+	c.activeCommandsMu.Lock()
+	defer c.activeCommandsMu.Unlock()
+	if c.activeCommands == nil {
+		c.activeCommands = make(map[string]context.CancelFunc)
+	}
+	c.activeCommands[requestID] = cancel
+}
+
+func (c *CommandClient) unregisterActiveCommand(requestID string) {
+	c.activeCommandsMu.Lock()
+	defer c.activeCommandsMu.Unlock()
+	delete(c.activeCommands, requestID)
+}
+
+// handleCancelCommand aborts an in-flight execute_command request the server
+// has stopped waiting for. Canceling the command context SIGKILLs its whole
+// process group, so an abandoned `pct exec` against a wedged guest is reaped
+// instead of running to its full timeout and orphaning children.
+func (c *CommandClient) handleCancelCommand(payload cancelCommandPayload) {
+	c.activeCommandsMu.Lock()
+	cancel, ok := c.activeCommands[payload.RequestID]
+	c.activeCommandsMu.Unlock()
+
+	if ok {
+		c.logger.Info().Str("request_id", payload.RequestID).Msg("Canceling command at server request")
+		cancel()
+	} else {
+		// Common benign race: the result was already sent while the cancel
+		// was in flight.
+		c.logger.Debug().Str("request_id", payload.RequestID).Msg("No active command to cancel")
+	}
+}
+
 func wrapCommand(payload executeCommandPayload) string {
 	targetType := strings.ToLower(strings.TrimSpace(payload.TargetType))
 	targetID := strings.TrimSpace(payload.TargetID)
@@ -1221,6 +1286,17 @@ func (c *CommandClient) executeCommand(ctx context.Context, payload executeComma
 		cmd.Env = append(os.Environ(), "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:"+os.Getenv("PATH"))
 	}
 
+	// Kill the whole process tree on timeout or cancellation, not just the
+	// direct shell. Server-issued commands routinely spawn grandchildren
+	// (pct exec → lxc-attach → docker); killing only the shell orphans those
+	// and, when the target is wedged, accumulates hung processes on the host
+	// every poll cycle until the machine runs out of PIDs (delly incident
+	// 2026-07-07, minipc incident 2026-08-20).
+	configureCommandProcessGroup(cmd)
+	// Bound Wait even if an orphaned grandchild inherits our stdout/stderr
+	// pipes and keeps them open past process exit.
+	cmd.WaitDelay = commandWaitDelay
+
 	stdout := newCappedBuffer(maxCommandOutputSize)
 	stderr := newCappedBuffer(maxCommandOutputSize)
 	cmd.Stdout = stdout
@@ -1231,11 +1307,22 @@ func (c *CommandClient) executeCommand(ctx context.Context, payload executeComma
 	result.Stdout = stdout.String()
 	result.Stderr = stderr.String()
 
+	if errors.Is(err, exec.ErrWaitDelay) {
+		// The command itself exited successfully; only the inherited pipes
+		// were still open (e.g. a deliberately backgrounded child).
+		err = nil
+	}
+
 	if err != nil {
 		if cmdCtx.Err() == context.DeadlineExceeded {
 			result.Error = "command timed out"
 			result.ExitCode = -1
 			result.Success = false
+		} else if cmdCtx.Err() == context.Canceled {
+			// The server told us it stopped waiting for this request.
+			result.Error = "command canceled"
+			result.ExitCode = -1
+			result.Success = false
 		} else if exitErr, ok := err.(*exec.ExitError); ok {
 			result.ExitCode = exitErr.ExitCode()
 			result.Success = false

internal/hostagent/commands_execute_unix_test.go

@@ -0,0 +1,144 @@
+//go:build !windows
+
+package hostagent
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/rs/zerolog"
+)
+
+// waitForProcessGone fails the test if the given pid is still alive after the
+// deadline. Signal 0 probes for existence; ESRCH means the kill got it.
+func waitForProcessGone(t *testing.T, pid int, timeout time.Duration) {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	for {
+		err := syscall.Kill(pid, 0)
+		if err == syscall.ESRCH {
+			return
+		}
+		if time.Now().After(deadline) {
+			_ = syscall.Kill(pid, syscall.SIGKILL)
+			t.Fatalf("pid %d still alive; process group was not killed (kill probe err=%v)", pid, err)
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+}
+
+func readPidFile(t *testing.T, path string) int {
+	t.Helper()
+	pidBytes, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("reading grandchild pid file: %v", err)
+	}
+	pid, err := strconv.Atoi(strings.TrimSpace(string(pidBytes)))
+	if err != nil {
+		t.Fatalf("parsing grandchild pid %q: %v", pidBytes, err)
+	}
+	return pid
+}
+
+// Regression test for the delly fork-bomb incident (2026-07-07) and the minipc
+// probe storm (2026-08-20): timing out a command must kill the whole process
+// group, not just the direct shell. Before the fix, grandchildren (pct exec →
+// lxc-attach → docker) survived the timeout and accumulated forever, and
+// cmd.Wait() blocked on the stdout pipe the orphans kept open.
+func TestCommandClient_executeCommand_TimeoutKillsProcessGroup(t *testing.T) {
+	pidFile := filepath.Join(t.TempDir(), "grandchild.pid")
+
+	c := &CommandClient{}
+	payload := testApprovedCommandPayload(t, c, executeCommandPayload{
+		RequestID: "r1",
+		// The backgrounded sleep is a grandchild that inherits our stdout
+		// pipe; `wait` keeps the shell alive until the timeout fires.
+		Command: "sleep 300 & echo $! > " + pidFile + "; wait",
+		Timeout: 1,
+	})
+
+	start := time.Now()
+	result := c.executeCommand(context.Background(), payload)
+	if elapsed := time.Since(start); elapsed > 15*time.Second {
+		t.Fatalf("executeCommand took %v; Wait must not hang on pipes held by orphaned grandchildren", elapsed)
+	}
+	if result.Success || result.Error != "command timed out" {
+		t.Fatalf("expected timeout failure, got %#v", result)
+	}
+
+	waitForProcessGone(t, readPidFile(t, pidFile), 5*time.Second)
+}
+
+// A command that exits promptly but leaves a background child holding the
+// stdout pipe must not block Wait past WaitDelay, and the command's own
+// success must be preserved.
+func TestCommandClient_executeCommand_WaitDelayUnblocksInheritedPipes(t *testing.T) {
+	oldDelay := commandWaitDelay
+	commandWaitDelay = 500 * time.Millisecond
+	defer func() { commandWaitDelay = oldDelay }()
+
+	c := &CommandClient{}
+	payload := testApprovedCommandPayload(t, c, executeCommandPayload{
+		RequestID: "r1",
+		// No timeout fires here, so nothing kills the background child; it
+		// holds the inherited stdout pipe and only WaitDelay unblocks Wait.
+		Command: "sleep 30 & echo done",
+		Timeout: 20,
+	})
+
+	start := time.Now()
+	result := c.executeCommand(context.Background(), payload)
+	if elapsed := time.Since(start); elapsed > 10*time.Second {
+		t.Fatalf("executeCommand took %v; WaitDelay should have unblocked Wait", elapsed)
+	}
+	if !result.Success || result.ExitCode != 0 {
+		t.Fatalf("expected success, got %#v", result)
+	}
+	if !strings.Contains(result.Stdout, "done") {
+		t.Fatalf("stdout = %q, expected to contain %q", result.Stdout, "done")
+	}
+}
+
+// A server-issued cancel_command must stop the running command promptly, kill
+// its whole process group, and report a distinct "command canceled" failure
+// (regression for the minipc incident, where the agent kept probes running
+// for 5+ minutes after the server had abandoned them).
+func TestCommandClient_executeCommand_CancelKillsProcessGroup(t *testing.T) {
+	pidFile := filepath.Join(t.TempDir(), "grandchild.pid")
+
+	c := &CommandClient{logger: zerolog.Nop()}
+	payload := testApprovedCommandPayload(t, c, executeCommandPayload{
+		RequestID: "r1",
+		Command:   "sleep 300 & echo $! > " + pidFile + "; wait",
+		Timeout:   60,
+	})
+
+	cmdCtx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	c.registerActiveCommand(payload.RequestID, cancel)
+	defer c.unregisterActiveCommand(payload.RequestID)
+
+	go func() {
+		// Give the shell time to start, then cancel the way handleMessages
+		// does when a cancel_command arrives.
+		time.Sleep(300 * time.Millisecond)
+		c.handleCancelCommand(cancelCommandPayload{RequestID: payload.RequestID})
+	}()
+
+	start := time.Now()
+	result := c.executeCommand(cmdCtx, payload)
+	if elapsed := time.Since(start); elapsed > 15*time.Second {
+		t.Fatalf("executeCommand took %v; cancel must stop the command promptly", elapsed)
+	}
+	if result.Success || result.Error != "command canceled" {
+		t.Fatalf("expected canceled failure, got %#v", result)
+	}
+
+	waitForProcessGone(t, readPidFile(t, pidFile), 5*time.Second)
+}

internal/hostagent/commands_procgroup_unix.go

@@ -0,0 +1,31 @@
+//go:build !windows
+
+package hostagent
+
+import (
+	"os"
+	"os/exec"
+	"syscall"
+)
+
+// configureCommandProcessGroup runs cmd in its own process group and replaces
+// the default context-cancel behavior (kill only the direct child) with a kill
+// of the entire group. Without this, timing out a command like
+// `pct exec <vmid> -- sh -c '...'` kills the outer shell but orphans
+// lxc-attach and everything below it; against a wedged container those
+// orphans never exit and pile up on the host every poll cycle.
+func configureCommandProcessGroup(cmd *exec.Cmd) {
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+	cmd.Cancel = func() error {
+		if cmd.Process == nil {
+			return os.ErrProcessDone
+		}
+		// With Setpgid the child's pgid equals its pid; a negative pid
+		// signals the whole group.
+		err := syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
+		if err == syscall.ESRCH {
+			return os.ErrProcessDone
+		}
+		return err
+	}
+}

internal/hostagent/commands_procgroup_windows.go

@@ -0,0 +1,23 @@
+//go:build windows
+
+package hostagent
+
+import (
+	"os/exec"
+	"strconv"
+)
+
+// configureCommandProcessGroup arranges for the command's whole process tree
+// to be killed when its context is canceled. Windows has no process groups in
+// the POSIX sense; short of managing a job object, taskkill /T is the
+// standard way to terminate a tree, with a direct Kill as fallback for the
+// root process.
+func configureCommandProcessGroup(cmd *exec.Cmd) {
+	cmd.Cancel = func() error {
+		if cmd.Process == nil {
+			return nil
+		}
+		_ = exec.Command("taskkill", "/T", "/F", "/PID", strconv.Itoa(cmd.Process.Pid)).Run()
+		return cmd.Process.Kill()
+	}
+}

internal/monitoring/docker_detection.go

@@ -3,6 +3,7 @@ package monitoring
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"math"
 	"strconv"
@@ -32,6 +33,22 @@ const (
 	// longer exist (a live failing guest refreshes its entry at least once
 	// per backoff window) and are pruned.
 	proxmoxGuestDockerProbeFailurePruneAfter = 24 * time.Hour
+
+	// A dispatched probe/inventory command owns its guest until it completes.
+	// When the server abandons the wait instead (poll context expired before
+	// the agent answered), the claim is held for this window: the agent may
+	// still be running the pct exec, and re-dispatching the same command
+	// every poll cycle is how the minipc incident (2026-08-20) accumulated
+	// ~100 orphaned children and hung the host.
+	proxmoxGuestDockerProbeInFlightWindow = 2 * time.Minute
+
+	// After this many consecutive command failures on one Proxmox node the
+	// circuit breaker opens and no Docker probe or inventory command is
+	// dispatched to that node until a backoff window passes. Per-guest
+	// backoff alone is not enough: a node that is slow for host-level
+	// reasons (NFS flapping, load) fails every guest's probe, and newly
+	// appearing guests would still be probed immediately.
+	proxmoxGuestDockerNodeFailureThreshold = 3
 )
 
 // dockerProbeFailureState tracks consecutive Docker socket probe failures for
@@ -102,6 +119,11 @@ func (m *Monitor) CheckContainersForDocker(ctx context.Context, containers []mod
 		previousContainers[ct.ID] = ct
 	}
 
+	// A context that is already dead means every probe would be abandoned the
+	// moment it was dispatched — the agent would be left running pct exec for
+	// nobody. Preserve previous status and try again next cycle.
+	ctxAlive := ctx.Err() == nil
+
 	// Identify containers that need Docker checking
 	var needsCheck []containerDockerCheck
 	for i, ct := range containers {
@@ -115,7 +137,10 @@ func (m *Monitor) CheckContainersForDocker(ctx context.Context, containers []mod
 		}
 
 		// Check if this container needs Docker detection
-		reason := m.containerNeedsDockerCheck(ct, previousContainers, checkerConfiguredAt)
+		reason := ""
+		if ctxAlive {
+			reason = m.containerNeedsDockerCheck(ct, previousContainers, checkerConfiguredAt)
+		}
 		if reason != "" && len(allowedVMIDs) > 0 {
 			if _, ok := allowedVMIDs[ct.VMID]; !ok {
 				// Outside the explicit VMID allowlist: never probe.
@@ -126,6 +151,15 @@ func (m *Monitor) CheckContainersForDocker(ctx context.Context, containers []mod
 			// A recent probe failure is still inside its backoff window.
 			reason = ""
 		}
+		if reason != "" && m.dockerNodeProbeBreakerOpen(ct.Node) {
+			// The whole node is failing Docker commands; don't pile on.
+			reason = ""
+		}
+		if reason != "" && !m.tryClaimDockerProbe(ct.ID) {
+			// A previously dispatched probe for this guest is still running
+			// (or was abandoned and may still be running on the agent).
+			reason = ""
+		}
 		if reason != "" {
 			needsCheck = append(needsCheck, containerDockerCheck{
 				index:     i,
@@ -159,7 +193,10 @@ func (m *Monitor) CheckContainersForDocker(ctx context.Context, containers []mod
 			if result.hasDocker {
 				dockerCount++
 			}
-			m.clearDockerProbeFailure(containers[result.index])
+			ct := containers[result.index]
+			m.releaseDockerProbe(ct.ID, false)
+			m.clearDockerProbeFailure(ct)
+			m.recordDockerNodeProbeResult(ct.Node, false)
 		} else if result.err != nil {
 			// Check failed - preserve previous status if available
 			ct := containers[result.index]
@@ -167,7 +204,12 @@ func (m *Monitor) CheckContainersForDocker(ctx context.Context, containers []mod
 				containers[result.index].HasDocker = prev.HasDocker
 				containers[result.index].DockerCheckedAt = prev.DockerCheckedAt
 			}
-			m.recordDockerProbeFailure(ct, result.err)
+			abandoned := dockerProbeAbandoned(result.err)
+			m.releaseDockerProbe(ct.ID, abandoned)
+			if !abandoned {
+				m.recordDockerProbeFailure(ct, result.err)
+				m.recordDockerNodeProbeResult(ct.Node, true)
+			}
 		}
 	}
 
@@ -308,7 +350,8 @@ func (m *Monitor) clearDockerProbeFailure(ct models.Container) {
 }
 
 // pruneDockerProbeFailures removes failure entries for containers that have
-// not been probed in a long time (guests that were removed while failing).
+// not been probed in a long time (guests that were removed while failing),
+// along with expired in-flight claims and stale node breaker entries.
 func (m *Monitor) pruneDockerProbeFailures() {
 	m.dockerProbeFailureMu.Lock()
 	defer m.dockerProbeFailureMu.Unlock()
@@ -317,14 +360,131 @@ func (m *Monitor) pruneDockerProbeFailures() {
 			delete(m.dockerProbeFailures, id)
 		}
 	}
+	for node, state := range m.dockerNodeProbeFailures {
+		if time.Since(state.lastAt) > proxmoxGuestDockerProbeFailurePruneAfter {
+			delete(m.dockerNodeProbeFailures, node)
+		}
+	}
+	for key, at := range m.dockerProbesInFlight {
+		if time.Since(at) > proxmoxGuestDockerProbeFailurePruneAfter {
+			delete(m.dockerProbesInFlight, key)
+		}
+	}
 }
 
-// resetDockerProbeFailures clears all failure streaks, so a reconfigured
-// checker retries every guest immediately.
+// resetDockerProbeFailures clears all failure streaks, in-flight claims, and
+// node circuit breakers, so a reconfigured checker retries every guest
+// immediately.
 func (m *Monitor) resetDockerProbeFailures() {
 	m.dockerProbeFailureMu.Lock()
 	defer m.dockerProbeFailureMu.Unlock()
 	m.dockerProbeFailures = nil
+	m.dockerProbesInFlight = nil
+	m.dockerNodeProbeFailures = nil
+}
+
+// dockerProbeAbandoned reports whether a probe error means the server stopped
+// waiting (its own context died) rather than the probe genuinely completing
+// with a failure. Abandonment says nothing about the guest or the node, so it
+// must not feed the failure backoff or the node circuit breaker — but the
+// agent may still be running the command, so the in-flight claim is kept.
+func dockerProbeAbandoned(err error) bool {
+	return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)
+}
+
+// dockerInventoryClaimKey namespaces the in-flight claim for the inventory
+// command so it never collides with the socket probe's claim for the same
+// guest: they are different commands with different lifetimes.
+func dockerInventoryClaimKey(containerID string) string {
+	return "inventory:" + containerID
+}
+
+// tryClaimDockerProbe records that a probe/inventory command is being
+// dispatched for the given key (container ID, optionally prefixed per command
+// kind). It returns false while a previous dispatch still owns the key, which
+// is what prevents overlapping poll cycles from stacking identical pct exec
+// commands on the same guest.
+func (m *Monitor) tryClaimDockerProbe(key string) bool {
+	now := time.Now()
+	m.dockerProbeFailureMu.Lock()
+	defer m.dockerProbeFailureMu.Unlock()
+	if at, ok := m.dockerProbesInFlight[key]; ok && now.Sub(at) < proxmoxGuestDockerProbeInFlightWindow {
+		return false
+	}
+	if m.dockerProbesInFlight == nil {
+		m.dockerProbesInFlight = make(map[string]time.Time)
+	}
+	m.dockerProbesInFlight[key] = now
+	return true
+}
+
+// releaseDockerProbe ends a claim taken with tryClaimDockerProbe. A completed
+// command (success or genuine failure) releases immediately; an abandoned one
+// keeps the claim until the in-flight window expires, because the agent may
+// still be executing it.
+func (m *Monitor) releaseDockerProbe(key string, abandoned bool) {
+	if abandoned {
+		return
+	}
+	m.dockerProbeFailureMu.Lock()
+	defer m.dockerProbeFailureMu.Unlock()
+	delete(m.dockerProbesInFlight, key)
+}
+
+// recordDockerNodeProbeResult feeds the per-node circuit breaker. Any
+// completed Docker probe/inventory command counts: a success closes the
+// breaker for the node, a failure extends the streak.
+func (m *Monitor) recordDockerNodeProbeResult(node string, failed bool) {
+	node = strings.TrimSpace(node)
+	if node == "" {
+		return
+	}
+	m.dockerProbeFailureMu.Lock()
+	if !failed {
+		state, wasOpen := m.dockerNodeProbeFailures[node]
+		delete(m.dockerNodeProbeFailures, node)
+		m.dockerProbeFailureMu.Unlock()
+		if wasOpen && state.failures >= proxmoxGuestDockerNodeFailureThreshold {
+			log.Info().
+				Str("node", node).
+				Int("previousFailures", state.failures).
+				Msg("Docker probe circuit breaker closed for node after successful command")
+		}
+		return
+	}
+	if m.dockerNodeProbeFailures == nil {
+		m.dockerNodeProbeFailures = make(map[string]*dockerProbeFailureState)
+	}
+	state, ok := m.dockerNodeProbeFailures[node]
+	if !ok {
+		state = &dockerProbeFailureState{}
+		m.dockerNodeProbeFailures[node] = state
+	}
+	state.failures++
+	state.lastAt = time.Now()
+	failures := state.failures
+	m.dockerProbeFailureMu.Unlock()
+	if failures == proxmoxGuestDockerNodeFailureThreshold {
+		log.Warn().
+			Str("node", node).
+			Int("consecutiveFailures", failures).
+			Msg("Docker probe circuit breaker opened: suspending Docker command dispatch to node")
+	}
+}
+
+// dockerNodeProbeBreakerOpen reports whether the node's circuit breaker is
+// open, i.e. Docker probe/inventory commands must not be dispatched to it
+// right now. The suspension window grows with the failure streak on the same
+// schedule as the per-guest backoff.
+func (m *Monitor) dockerNodeProbeBreakerOpen(node string) bool {
+	m.dockerProbeFailureMu.Lock()
+	defer m.dockerProbeFailureMu.Unlock()
+	state, ok := m.dockerNodeProbeFailures[node]
+	if !ok || state.failures < proxmoxGuestDockerNodeFailureThreshold {
+		return false
+	}
+	backoff := dockerProbeFailureBackoff(state.failures - proxmoxGuestDockerNodeFailureThreshold + 1)
+	return time.Since(state.lastAt) < backoff
 }
 
 // checkDockerParallel checks Docker for multiple containers in parallel
@@ -350,6 +510,14 @@ func (m *Monitor) checkDockerParallel(ctx context.Context, checker DockerChecker
 				return
 			}
 
+			// A select with both cases ready picks one at random; never let
+			// an already-dead context slip through to a dispatch the caller
+			// will abandon instantly.
+			if err := ctx.Err(); err != nil {
+				results[idx] = containerDockerResult{index: chk.index, err: err}
+				return
+			}
+
 			// Perform the check
 			hasDocker, err := checker.CheckDockerInContainer(ctx, chk.container.Node, chk.container.VMID)
 			if err != nil {
@@ -394,6 +562,12 @@ func (m *Monitor) CollectProxmoxGuestDockerInventory(ctx context.Context, contai
 		return
 	}
 
+	// Same rule as the socket probe: never dispatch commands the caller has
+	// already stopped waiting for.
+	if ctx.Err() != nil {
+		return
+	}
+
 	candidates := make([]models.Container, 0, len(containers))
 	for _, ct := range containers {
 		if ct.Status != "running" || !ct.HasDocker || ct.VMID <= 0 || strings.TrimSpace(ct.Node) == "" || ct.IsOCI {
@@ -407,6 +581,14 @@ func (m *Monitor) CollectProxmoxGuestDockerInventory(ctx context.Context, contai
 				Msg("Skipping Proxmox LXC Docker inventory because a guest-local host agent is linked")
 			continue
 		}
+		if m.dockerNodeProbeBreakerOpen(ct.Node) {
+			continue
+		}
+		if !m.tryClaimDockerProbe(dockerInventoryClaimKey(ct.ID)) {
+			// A previous inventory command for this guest is still running
+			// (or was abandoned and may still be running on the agent).
+			continue
+		}
 		candidates = append(candidates, ct)
 	}
 
@@ -426,10 +608,20 @@ func (m *Monitor) CollectProxmoxGuestDockerInventory(ctx context.Context, contai
 		wg.Add(1)
 		go func(container models.Container) {
 			defer wg.Done()
+			claimKey := dockerInventoryClaimKey(container.ID)
 			select {
 			case sem <- struct{}{}:
 				defer func() { <-sem }()
 			case <-ctx.Done():
+				// Nothing was dispatched; free the claim immediately.
+				m.releaseDockerProbe(claimKey, false)
+				mu.Lock()
+				failed++
+				mu.Unlock()
+				return
+			}
+			if ctx.Err() != nil {
+				m.releaseDockerProbe(claimKey, false)
 				mu.Lock()
 				failed++
 				mu.Unlock()
@@ -438,6 +630,11 @@ func (m *Monitor) CollectProxmoxGuestDockerInventory(ctx context.Context, contai
 
 			report, ok, err := collector.CollectDockerInventory(ctx, container)
 			if err != nil {
+				abandoned := dockerProbeAbandoned(err)
+				m.releaseDockerProbe(claimKey, abandoned)
+				if !abandoned {
+					m.recordDockerNodeProbeResult(container.Node, true)
+				}
 				log.Debug().
 					Err(err).
 					Str("container", container.Name).
@@ -448,12 +645,14 @@ func (m *Monitor) CollectProxmoxGuestDockerInventory(ctx context.Context, contai
 				mu.Unlock()
 				return
 			}
+			m.releaseDockerProbe(claimKey, false)
 			if !ok {
 				mu.Lock()
 				skipped++
 				mu.Unlock()
 				return
 			}
+			m.recordDockerNodeProbeResult(container.Node, false)
 
 			enrichGuestDockerReportFromContainer(&report, container)
 

internal/monitoring/monitor.go

@@ -1192,6 +1192,18 @@ type Monitor struct {
 	// failures back off instead of re-running pct exec on every poll cycle.
 	dockerProbeFailureMu sync.Mutex
 	dockerProbeFailures  map[string]*dockerProbeFailureState
+	// Dispatch time of the Docker probe/inventory command currently owning
+	// each guest. A guest with a live claim is never re-probed, so overlapping
+	// poll cycles (or cycles whose enrichment context expired while the agent
+	// was still running the previous pct exec) cannot stack identical commands
+	// on the Proxmox host (minipc probe-storm incident, 2026-08-20).
+	dockerProbesInFlight map[string]time.Time
+	// Consecutive Docker probe/inventory failures per Proxmox node. Once a
+	// node accumulates enough back-to-back failures the circuit breaker
+	// suspends all Docker command dispatch to it, because the node itself
+	// (not one guest) is the likely problem — e.g. pct exec crawling under
+	// NFS flapping.
+	dockerNodeProbeFailures map[string]*dockerProbeFailureState
 	// Agent profile cache to avoid disk I/O on every report (refs #1094)
 	agentProfileCacheMu sync.RWMutex
 	agentProfileCache   *agentProfileCacheEntry

internal/monitoring/monitor_docker_test.go

@@ -1,7 +1,10 @@
 package monitoring
 
 import (
+	"context"
+	"errors"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -1982,3 +1985,327 @@ func TestApplyDockerReport_SameHostnameDifferentTokens(t *testing.T) {
 		t.Errorf("Expected 2 hosts in state, got %d", len(hosts))
 	}
 }
+
+// blockingDockerChecker simulates the field failure mode of the minipc
+// incident (2026-08-20): a loaded Proxmox host where pct exec is arbitrarily
+// slow, so probes are still executing when the next poll cycle starts.
+type blockingDockerChecker struct {
+	mu      sync.Mutex
+	calls   []int
+	started chan struct{} // closed when the first probe begins
+	release chan struct{} // probes block here until the test closes it
+	results map[int]bool
+	errs    map[int]error
+}
+
+func (c *blockingDockerChecker) CheckDockerInContainer(ctx context.Context, node string, vmid int) (bool, error) {
+	c.mu.Lock()
+	c.calls = append(c.calls, vmid)
+	first := len(c.calls) == 1
+	c.mu.Unlock()
+	if first {
+		close(c.started)
+	}
+	select {
+	case <-c.release:
+	case <-ctx.Done():
+		return false, ctx.Err()
+	}
+	if err, ok := c.errs[vmid]; ok {
+		return false, err
+	}
+	return c.results[vmid], nil
+}
+
+func (c *blockingDockerChecker) callCount() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return len(c.calls)
+}
+
+// Bug 1 regression (minipc incident): while a probe for a guest is still
+// executing, subsequent poll cycles must not re-dispatch the same probe. In
+// the field the dispatcher re-issued the identical command every ~3s with
+// fresh request IDs, so a slow host self-amplified until it ran out of
+// resources.
+func TestCheckContainersForDocker_InFlightProbeNotReissued(t *testing.T) {
+	state := models.NewState()
+	monitor := &Monitor{state: state}
+	checker := &blockingDockerChecker{
+		started: make(chan struct{}),
+		release: make(chan struct{}),
+		results: map[int]bool{101: true},
+	}
+	monitor.SetDockerChecker(checker)
+
+	containers := func() []models.Container {
+		return []models.Container{
+			{ID: "ct-1", VMID: 101, Name: "slow-guest", Node: "node1", Status: "running"},
+		}
+	}
+
+	firstDone := make(chan struct{})
+	go func() {
+		defer close(firstDone)
+		monitor.CheckContainersForDocker(context.Background(), containers())
+	}()
+
+	select {
+	case <-checker.started:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("first probe never started")
+	}
+
+	// Second and third poll cycles arrive while the first probe is still
+	// executing. Before the in-flight claim they would each dispatch the
+	// same probe again.
+	monitor.CheckContainersForDocker(context.Background(), containers())
+	monitor.CheckContainersForDocker(context.Background(), containers())
+	if got := checker.callCount(); got != 1 {
+		t.Fatalf("expected 1 in-flight probe, got %d dispatches", got)
+	}
+
+	close(checker.release)
+	select {
+	case <-firstDone:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("first poll cycle never finished")
+	}
+
+	// The probe completed, releasing the claim: the guest is probeable again
+	// once the normal cadence asks for it.
+	monitor.dockerProbeFailureMu.Lock()
+	_, held := monitor.dockerProbesInFlight["ct-1"]
+	monitor.dockerProbeFailureMu.Unlock()
+	if held {
+		t.Fatalf("expected in-flight claim to be released after probe completion")
+	}
+}
+
+// Bug 3 regression (server side): a poll cycle whose enrichment context has
+// already expired must not dispatch probes at all — in the field the server
+// sent the command and then logged "Agent command canceled ... duration 0.05"
+// while the agent kept executing it. An abandoned dispatch must also not
+// count as a probe failure against the guest.
+func TestCheckContainersForDocker_ExpiredContextDoesNotDispatch(t *testing.T) {
+	state := models.NewState()
+	checkedAt := time.Now().Add(-time.Hour)
+	state.UpdateContainers([]models.Container{
+		{ID: "ct-1", VMID: 101, Name: "guest", Node: "node1", Status: "running", HasDocker: true, DockerCheckedAt: checkedAt},
+	})
+	monitor := &Monitor{state: state}
+	checker := &mockDockerChecker{results: map[int]bool{101: false}}
+	monitor.SetDockerChecker(checker)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	result := monitor.CheckContainersForDocker(ctx, []models.Container{
+		{ID: "ct-1", VMID: 101, Name: "guest", Node: "node1", Status: "running"},
+	})
+
+	if got := checker.callCount(); got != 0 {
+		t.Fatalf("expected no probe dispatch under expired context, got %d", got)
+	}
+	if !result[0].HasDocker || !result[0].DockerCheckedAt.Equal(checkedAt) {
+		t.Fatalf("expected previous docker status preserved, got %+v", result[0])
+	}
+	monitor.dockerProbeFailureMu.Lock()
+	failures := len(monitor.dockerProbeFailures)
+	monitor.dockerProbeFailureMu.Unlock()
+	if failures != 0 {
+		t.Fatalf("expected no probe failures recorded for an undispatched cycle, got %d", failures)
+	}
+}
+
+// A probe the server abandoned mid-flight (context canceled while waiting on
+// the agent) may still be running on the agent. It must not be re-dispatched
+// while the in-flight window holds, and it must not feed the failure backoff
+// or the node circuit breaker — abandonment says nothing about the guest.
+func TestCheckContainersForDocker_AbandonedProbeHoldsClaimWithoutFailure(t *testing.T) {
+	state := models.NewState()
+	monitor := &Monitor{state: state}
+	checker := &mockDockerChecker{
+		errs: map[int]error{101: context.Canceled},
+	}
+	monitor.SetDockerChecker(checker)
+
+	containers := []models.Container{
+		{ID: "ct-1", VMID: 101, Name: "guest", Node: "node1", Status: "running"},
+	}
+
+	monitor.CheckContainersForDocker(context.Background(), containers)
+	if got := checker.callCount(); got != 1 {
+		t.Fatalf("expected 1 dispatch, got %d", got)
+	}
+
+	monitor.dockerProbeFailureMu.Lock()
+	failures := len(monitor.dockerProbeFailures)
+	nodeFailures := len(monitor.dockerNodeProbeFailures)
+	_, held := monitor.dockerProbesInFlight["ct-1"]
+	monitor.dockerProbeFailureMu.Unlock()
+	if failures != 0 || nodeFailures != 0 {
+		t.Fatalf("abandoned probe must not record failures (guest=%d node=%d)", failures, nodeFailures)
+	}
+	if !held {
+		t.Fatalf("abandoned probe must keep its in-flight claim")
+	}
+
+	// Next cycles: still claimed, no re-dispatch.
+	monitor.CheckContainersForDocker(context.Background(), containers)
+	monitor.CheckContainersForDocker(context.Background(), containers)
+	if got := checker.callCount(); got != 1 {
+		t.Fatalf("expected abandoned probe not to be re-issued, got %d dispatches", got)
+	}
+
+	// Once the in-flight window has passed the guest is probeable again.
+	monitor.dockerProbeFailureMu.Lock()
+	monitor.dockerProbesInFlight["ct-1"] = time.Now().Add(-2 * proxmoxGuestDockerProbeInFlightWindow)
+	monitor.dockerProbeFailureMu.Unlock()
+	monitor.CheckContainersForDocker(context.Background(), containers)
+	if got := checker.callCount(); got != 2 {
+		t.Fatalf("expected retry after in-flight window expiry, got %d dispatches", got)
+	}
+}
+
+// Node circuit breaker: once a node accumulates enough consecutive command
+// failures, no Docker probe is dispatched to it — even for a newly appearing
+// guest that has no per-guest failure history. Other nodes are unaffected,
+// and a success on the node closes the breaker.
+func TestCheckContainersForDocker_NodeCircuitBreaker(t *testing.T) {
+	state := models.NewState()
+	monitor := &Monitor{state: state}
+	checker := &mockDockerChecker{
+		errs: map[int]error{
+			101: errors.New("pct exec hung"),
+			102: errors.New("pct exec hung"),
+			103: errors.New("pct exec hung"),
+		},
+		results: map[int]bool{104: true, 201: true},
+	}
+	monitor.SetDockerChecker(checker)
+
+	// Cycle 1: three guests on node1 fail, opening the breaker.
+	monitor.CheckContainersForDocker(context.Background(), []models.Container{
+		{ID: "ct-1", VMID: 101, Name: "a", Node: "node1", Status: "running"},
+		{ID: "ct-2", VMID: 102, Name: "b", Node: "node1", Status: "running"},
+		{ID: "ct-3", VMID: 103, Name: "c", Node: "node1", Status: "running"},
+	})
+	if got := checker.callCount(); got != 3 {
+		t.Fatalf("expected 3 dispatches in first cycle, got %d", got)
+	}
+	if !monitor.dockerNodeProbeBreakerOpen("node1") {
+		t.Fatalf("expected node1 breaker open after %d consecutive failures", proxmoxGuestDockerNodeFailureThreshold)
+	}
+
+	// Cycle 2: a brand-new guest on node1 must not be probed while the
+	// breaker is open; a guest on node2 still is.
+	monitor.CheckContainersForDocker(context.Background(), []models.Container{
+		{ID: "ct-4", VMID: 104, Name: "new-on-bad-node", Node: "node1", Status: "running"},
+		{ID: "ct-5", VMID: 201, Name: "on-good-node", Node: "node2", Status: "running"},
+	})
+	checker.mu.Lock()
+	calls := append([]int(nil), checker.calls...)
+	checker.mu.Unlock()
+	if len(calls) != 4 || calls[3] != 201 {
+		t.Fatalf("expected only node2 guest probed while breaker open, got dispatches %v", calls)
+	}
+
+	// Breaker backoff expires: node1 is retried, and the success closes the
+	// breaker.
+	monitor.dockerProbeFailureMu.Lock()
+	monitor.dockerNodeProbeFailures["node1"].lastAt = time.Now().Add(-time.Hour)
+	monitor.dockerProbeFailureMu.Unlock()
+	monitor.CheckContainersForDocker(context.Background(), []models.Container{
+		{ID: "ct-4", VMID: 104, Name: "new-on-bad-node", Node: "node1", Status: "running"},
+	})
+	if got := checker.callCount(); got != 5 {
+		t.Fatalf("expected node1 retried after breaker backoff, got %d dispatches", got)
+	}
+	monitor.dockerProbeFailureMu.Lock()
+	_, stillTracked := monitor.dockerNodeProbeFailures["node1"]
+	monitor.dockerProbeFailureMu.Unlock()
+	if stillTracked {
+		t.Fatalf("expected breaker to close after a successful command on node1")
+	}
+}
+
+// blockingInventoryCollector mirrors blockingDockerChecker for the inventory
+// command path.
+type blockingInventoryCollector struct {
+	mu      sync.Mutex
+	calls   int
+	started chan struct{}
+	release chan struct{}
+}
+
+func (c *blockingInventoryCollector) CollectDockerInventory(ctx context.Context, container models.Container) (agentsdocker.Report, bool, error) {
+	c.mu.Lock()
+	c.calls++
+	first := c.calls == 1
+	c.mu.Unlock()
+	if first {
+		close(c.started)
+	}
+	select {
+	case <-c.release:
+	case <-ctx.Done():
+		return agentsdocker.Report{}, false, ctx.Err()
+	}
+	return agentsdocker.Report{}, false, nil
+}
+
+func (c *blockingInventoryCollector) callCount() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.calls
+}
+
+// The inventory command path has the same in-flight and expired-context rules
+// as the socket probe: no dispatch under a dead context, and no re-dispatch
+// while a previous inventory command for the guest is still executing.
+func TestCollectProxmoxGuestDockerInventory_InFlightAndExpiredContext(t *testing.T) {
+	state := models.NewState()
+	monitor := &Monitor{state: state}
+	collector := &blockingInventoryCollector{
+		started: make(chan struct{}),
+		release: make(chan struct{}),
+	}
+	monitor.SetDockerInventoryCollector(collector)
+
+	containers := []models.Container{
+		{ID: "ct-1", VMID: 101, Name: "guest", Node: "node1", Status: "running", HasDocker: true},
+	}
+
+	// Expired context: nothing dispatched.
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	monitor.CollectProxmoxGuestDockerInventory(ctx, containers)
+	if got := collector.callCount(); got != 0 {
+		t.Fatalf("expected no inventory dispatch under expired context, got %d", got)
+	}
+
+	firstDone := make(chan struct{})
+	go func() {
+		defer close(firstDone)
+		monitor.CollectProxmoxGuestDockerInventory(context.Background(), containers)
+	}()
+	select {
+	case <-collector.started:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("first inventory collection never started")
+	}
+
+	// While in flight, the next cycle must not re-dispatch.
+	monitor.CollectProxmoxGuestDockerInventory(context.Background(), containers)
+	if got := collector.callCount(); got != 1 {
+		t.Fatalf("expected 1 in-flight inventory collection, got %d dispatches", got)
+	}
+
+	close(collector.release)
+	select {
+	case <-firstDone:
+	case <-time.After(5 * time.Second):
+		t.Fatalf("first inventory cycle never finished")
+	}
+}