Cancel abandoned typed agent operations

docs/AGENT_SECURITY.md

@@ -259,7 +259,7 @@ or generic command path.
 | Rootful Docker or Podman inventory | No direct collector access to a root-equivalent daemon socket | Implemented as a typed-helper summary-only fallback. Migration preserves container ID/name/image/state/status/creation inventory and marks the report `typed-helper-summary`; stats, secondary inventories, update checks, and actions remain unavailable | Unit and installer regressions cover the boundary, but no representative live Docker/Podman qualification exists. Full rootful parity remains an explicit default blocker; the legacy/root profile is not safe-profile evidence | `agent-lifecycle`: record fresh install, migration, restart, helper loss/recovery, bounds, and summary parity on representative rootful Docker and Podman; decide explicitly whether reduced telemetry is sufficient |
 | Collector-owned rootless Docker or Podman | Direct access only to one usable runtime socket owned by the `pulse-agent` UID whose daemon also attests rootless mode | Implemented, unqualified live. Discovery is deferred until the collector account exists and accepts exactly one readable/writable, non-symlink socket owned by that UID across Docker and Podman. Cross-user, root-owned, remote, unreadable, non-rootless-daemon, or ambiguous endpoints are rejected. An installer-owned exact pin survives daemon loss and update; the running collector can move between direct rootless monitoring and rootful typed-helper summary without regaining action authority | Does not yet justify container-runtime parity or a default change | `deployment-installability`: record fresh install, migration, restart, socket-loss, ambiguity, helper/direct recovery, and telemetry parity on both rootless Docker and rootless Podman |
 | Separate runner package update and package-cache cleanup | Root-owned `pulse-agent-runner`, host-bound action credential, typed request, postcondition, and durable receipt | The schema-v6 committed-main systemd receipt records a real verified apt-cache mutation, stale-fingerprint refusal, replay, nonce-bound readiness, effective-unit override rejection, exact credential rotation, bodyless self-revocation, and collector/helper continuity. A separate production Router regression exercises HTTPS issuance, WSS admission, encrypted token persistence, failed-rotation rollback, exact socket invalidation, two server restarts, old-secret rejection, and durable self-revoke | Qualified for the exercised systemd fixture paths and focused production Router lifecycle. Focused installer tests also cover atomic pending cancellation and fail-closed credential retention. The evidence does not cover every runner operation, representative providers, or an exact release candidate | `agent-lifecycle` and `api-contracts`: reproduce the combined systemd and production Router path from trusted RC artifacts with representative package-update success/failure/cancellation evidence |
-| Separate runner Proxmox guest and container lifecycle/update actions | Root-owned runner with closed typed protocols; never the monitoring collector | Implemented, unqualified on representative PVE and container-runtime targets | No live-provider action-parity claim and no default change | `agent-lifecycle`: record target-bound success, stale-state refusal, cancellation, reconnect/replay, and independent postconditions on disposable real targets |
+| Separate runner Proxmox guest and container lifecycle/update actions | Root-owned runner with closed typed protocols; never the monitoring collector | Implemented, unqualified on representative PVE and container-runtime targets. Focused protocol tests prove expired requests are not dispatched, timeout/caller abandonment sends one request-bound cancellation across every typed mutation family, and a canceled Proxmox mutation persists an indeterminate terminal receipt that survives runner reconnect and replays without a second mutation | This is local protocol/durability proof only: it makes no live-provider action-parity claim and does not change the default | `agent-lifecycle`: record target-bound success, stale-state refusal, provider-handoff cancellation, reconnect/replay, and independent postconditions on disposable real targets |
 | Appliance, non-systemd, Windows, and macOS host-agent profiles | Platform API where sufficient; otherwise an explicitly named legacy/full-trust profile | **Unavailable for safe-profile apply.** The installer fails closed instead of silently installing a root-equivalent profile | Excluded from the Linux safe-profile claim | `deployment-installability`: land a platform-specific service, filesystem, update, helper, migration, rollback, and live-proof contract before marking that platform supported |
 
 The current committed-main Linux evidence qualifies exact commit

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

@@ -6380,7 +6380,7 @@ originally authorized expected value when validating the admitted request
 digest. The generic receipt store continues to bind completion and replay to
 the exact admitted attempt/action/operation/digest/agent identity, while the
 drift observation records why mutation was refused and replan is required.
-Typed APT, storage-cleanup, and Docker result payloads carry an optional bounded
+Typed APT, storage-cleanup, Docker, and Proxmox result payloads carry an optional bounded
 `reason_code` for pre-mutation refusal. Codes distinguish invalid contracts,
 missing agent capabilities, target drift, unavailable inspection, package
 manager contention or health, and bounded preflight failure without exposing
@@ -6389,8 +6389,11 @@ the field and remain valid; the server projects that legacy absence to
 `preflight_refused`. A refusal code cannot accompany a started mutation or a
 successful result. An already-empty package cache is a verified no-op success,
 because the requested postcondition is already satisfied.
-Callback loss and a reopened server store reconcile both APT actions by query
-only; the original typed dispatch is never resent. Legacy APT v1 terminal
+Callback loss and a reopened server store reconcile APT, Docker, and Proxmox
+actions by exact identity query only; the original typed dispatch is never
+resent. Proxmox query validation recognizes every closed lifecycle verb,
+revalidates the terminal envelope and chronology, and rejects mismatched
+action, digest, or agent identity. Legacy APT v1 terminal
 payloads that predate additive package-manager health facts remain
 structurally valid receipts, but their verified claim is projected as health-
 unknown/inconclusive rather than confirmed.
@@ -7234,7 +7237,7 @@ qualify migration and failure rollback, typed-helper update activation,
 provider parity and documented degradation, and separately credentialed live
 action sessions. Unsupported platforms continue to fail closed on migration.
 
-### Command dispatch is context-honest and abandoned executions are canceled
+### Command and durable typed dispatch are context-honest and canceled when abandoned
 
 The agent command transport now refuses to dispatch work its caller has
 already stopped waiting for, and propagates abandonment to the agent
@@ -7242,33 +7245,54 @@ already stopped waiting for, and propagates abandonment to the agent
 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
+1. `agentexec.Server.ExecuteCommand`, `ReadFile`, and every durable typed
+   mutation dispatcher 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
+   operation timeout or caller-context cancellation), it sends exactly one
    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.
+   to their own per-command timeout; the protocol change is additive. Before
+   launching a handler, the runner records the request in a bounded,
+   connection-generation-scoped state table. Cancellation or connection
+   teardown before handler registration leaves a tombstone that registration
+   consumes atomically, so provider handoff cannot start after abandonment.
 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
+   `execute_command`/`read_file` executions and durable host update,
+   storage-cleanup, Proxmox guest lifecycle, and container lifecycle/update
+   operations 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.
+   never received it, and extends it with server-driven cancellation. A typed
+   mutation that reached provider handoff remains explicitly indeterminate;
+   cancellation never claims rollback or no effect. The runner commits its
+   bounded terminal receipt after cancellation when its durable store remains
+   healthy even when the original waiter has gone, and reconnect/replay
+   returns that exact receipt without a second mutation.
 
 Proofs: `internal/agentexec/server_websocket_test.go`
 (`TestExecuteCommand_ExpiredContextNeverDispatches`,
 `TestExecuteCommand_AbandonedCommandSendsCancel`),
+`internal/agentexec/server_websocket_test.go`
+(`TestTypedOperations_AbandonedDispatchSendsExactlyOneCancel`,
+`TestTypedOperation_TimeoutSendsCancelAndExpiredContextNeverDispatches`),
+`internal/hostagent/operation_receipt_websocket_integration_test.go`
+(`TestRealServerActionRunnerCancellationPersistsAndReplaysProxmoxReceiptAfterReconnect`),
 `internal/hostagent/command_client_test.go`
 (`TestCommandClient_handleCancelCommand_CancelsRegisteredRequest`,
-`TestCommandClient_handleCancelCommand_UnknownRequestIsNoOp`), and
+`TestCommandClient_handleCancelCommand_UnknownRequestIsNoOp`,
+`TestCommandClient_CancellationBeforeRegistrationIsConsumedAndConnectionScoped`,
+`TestCommandClient_StaleCleanupCannotEraseReusedRequestCancellation`),
+`internal/hostagent/proxmox_guest_lifecycle_test.go`
+(`TestProxmoxGuestLifecycleCancellationBeforeHandlerRegistrationSkipsProviderAndPersistsReceipt`), and
 `internal/hostagent/commands_execute_unix_test.go` (timeout and cancel
 kill the whole process group; WaitDelay unblocks inherited pipes).
 

frontend-modern/public/docs/AGENT_SECURITY.md

@@ -259,7 +259,7 @@ or generic command path.
 | Rootful Docker or Podman inventory | No direct collector access to a root-equivalent daemon socket | Implemented as a typed-helper summary-only fallback. Migration preserves container ID/name/image/state/status/creation inventory and marks the report `typed-helper-summary`; stats, secondary inventories, update checks, and actions remain unavailable | Unit and installer regressions cover the boundary, but no representative live Docker/Podman qualification exists. Full rootful parity remains an explicit default blocker; the legacy/root profile is not safe-profile evidence | `agent-lifecycle`: record fresh install, migration, restart, helper loss/recovery, bounds, and summary parity on representative rootful Docker and Podman; decide explicitly whether reduced telemetry is sufficient |
 | Collector-owned rootless Docker or Podman | Direct access only to one usable runtime socket owned by the `pulse-agent` UID whose daemon also attests rootless mode | Implemented, unqualified live. Discovery is deferred until the collector account exists and accepts exactly one readable/writable, non-symlink socket owned by that UID across Docker and Podman. Cross-user, root-owned, remote, unreadable, non-rootless-daemon, or ambiguous endpoints are rejected. An installer-owned exact pin survives daemon loss and update; the running collector can move between direct rootless monitoring and rootful typed-helper summary without regaining action authority | Does not yet justify container-runtime parity or a default change | `deployment-installability`: record fresh install, migration, restart, socket-loss, ambiguity, helper/direct recovery, and telemetry parity on both rootless Docker and rootless Podman |
 | Separate runner package update and package-cache cleanup | Root-owned `pulse-agent-runner`, host-bound action credential, typed request, postcondition, and durable receipt | The schema-v6 committed-main systemd receipt records a real verified apt-cache mutation, stale-fingerprint refusal, replay, nonce-bound readiness, effective-unit override rejection, exact credential rotation, bodyless self-revocation, and collector/helper continuity. A separate production Router regression exercises HTTPS issuance, WSS admission, encrypted token persistence, failed-rotation rollback, exact socket invalidation, two server restarts, old-secret rejection, and durable self-revoke | Qualified for the exercised systemd fixture paths and focused production Router lifecycle. Focused installer tests also cover atomic pending cancellation and fail-closed credential retention. The evidence does not cover every runner operation, representative providers, or an exact release candidate | `agent-lifecycle` and `api-contracts`: reproduce the combined systemd and production Router path from trusted RC artifacts with representative package-update success/failure/cancellation evidence |
-| Separate runner Proxmox guest and container lifecycle/update actions | Root-owned runner with closed typed protocols; never the monitoring collector | Implemented, unqualified on representative PVE and container-runtime targets | No live-provider action-parity claim and no default change | `agent-lifecycle`: record target-bound success, stale-state refusal, cancellation, reconnect/replay, and independent postconditions on disposable real targets |
+| Separate runner Proxmox guest and container lifecycle/update actions | Root-owned runner with closed typed protocols; never the monitoring collector | Implemented, unqualified on representative PVE and container-runtime targets. Focused protocol tests prove expired requests are not dispatched, timeout/caller abandonment sends one request-bound cancellation across every typed mutation family, and a canceled Proxmox mutation persists an indeterminate terminal receipt that survives runner reconnect and replays without a second mutation | This is local protocol/durability proof only: it makes no live-provider action-parity claim and does not change the default | `agent-lifecycle`: record target-bound success, stale-state refusal, provider-handoff cancellation, reconnect/replay, and independent postconditions on disposable real targets |
 | Appliance, non-systemd, Windows, and macOS host-agent profiles | Platform API where sufficient; otherwise an explicitly named legacy/full-trust profile | **Unavailable for safe-profile apply.** The installer fails closed instead of silently installing a root-equivalent profile | Excluded from the Linux safe-profile claim | `deployment-installability`: land a platform-specific service, filesystem, update, helper, migration, rollback, and live-proof contract before marking that platform supported |
 
 The current committed-main Linux evidence qualifies exact commit

internal/agentexec/apt_codec.go

@@ -178,6 +178,27 @@ func ValidateOperationQueryResultForIdentity(result operationreceipt.QueryResult
 			return fmt.Errorf("docker container update readback has invalid terminal chronology")
 		}
 		return nil
+	case "start", "stop", "shutdown", "reboot":
+		if result.Record.ResultKind != ProxmoxGuestLifecycleReceiptKind || result.Record.ResultVersion != ProxmoxGuestLifecycleReceiptVersion {
+			return fmt.Errorf("proxmox guest lifecycle query result envelope mismatch")
+		}
+		payload, err := DecodeProxmoxGuestLifecycleResultPayload(result.Record.Result)
+		if err != nil {
+			return err
+		}
+		if payload.RequestID != identity.AttemptID || payload.ActionID != identity.ActionID || payload.Operation != identity.OperationKind || payload.OperationVersion != identity.OperationVersion || payload.RequestDigest != identity.RequestDigest {
+			return operationreceipt.ErrBindingConflict
+		}
+		if payload.MutationStarted && payload.Before.ObservedAt.IsZero() {
+			return fmt.Errorf("proxmox guest lifecycle mutation lacks a durable preflight observation")
+		}
+		if !payload.Before.ObservedAt.IsZero() && result.Record.TerminalAt.Before(payload.Before.ObservedAt) {
+			return fmt.Errorf("proxmox guest lifecycle preflight has invalid terminal chronology")
+		}
+		if payload.ReadbackRan && (payload.After.ObservedAt.IsZero() || result.Record.TerminalAt.Before(payload.After.ObservedAt)) {
+			return fmt.Errorf("proxmox guest lifecycle readback has invalid terminal chronology")
+		}
+		return nil
 	default:
 		return fmt.Errorf("unsupported operation query kind %q", identity.OperationKind)
 	}

internal/agentexec/proxmox_guest_lifecycle_codec_test.go

@@ -4,6 +4,9 @@ import (
 	"encoding/json"
 	"strings"
 	"testing"
+	"time"
+
+	"github.com/rcourtman/pulse-go-rewrite/internal/operationreceipt"
 )
 
 func boundProxmoxGuestLifecycle(t *testing.T) ProxmoxGuestLifecyclePayload {
@@ -60,3 +63,57 @@ func TestProxmoxGuestLifecycleResultIsRequestAndReceiptBound(t *testing.T) {
 		t.Fatalf("mismatched receipt error = %v", err)
 	}
 }
+
+func TestProxmoxGuestLifecycleTerminalReceiptQuerySupportsEveryClosedVerb(t *testing.T) {
+	now := time.Now().UTC()
+	for _, operation := range []string{"start", "stop", "shutdown", "reboot"} {
+		t.Run(operation, func(t *testing.T) {
+			req := ProxmoxGuestLifecyclePayload{
+				RequestID: "attempt-" + operation, ActionID: "action-" + operation, Operation: operation,
+				GuestKind: "vm", VMID: 101, ExpectedStatus: "running", Timeout: 30,
+			}
+			if operation == "start" {
+				req.ExpectedStatus = "stopped"
+			}
+			if err := BindProxmoxGuestLifecyclePayload(&req); err != nil {
+				t.Fatal(err)
+			}
+			identity := ProxmoxGuestLifecycleOperationIdentity("agent-pve", req)
+			result := ProxmoxGuestLifecycleResultPayload{
+				RequestID: req.RequestID, ActionID: req.ActionID, Operation: req.Operation,
+				OperationVersion: req.OperationVersion, RequestDigest: req.RequestDigest,
+				GuestKind: req.GuestKind, VMID: req.VMID, ExecutionPhase: ProxmoxGuestPhaseMutate,
+				MutationStarted: true, Before: ProxmoxGuestLifecycleSnapshot{Status: req.ExpectedStatus, ObservedAt: now.Add(-time.Second)},
+				Error: "Proxmox guest lifecycle mutation was interrupted after dispatch; target state requires recovery inspection",
+			}
+			encoded, err := json.Marshal(result)
+			if err != nil {
+				t.Fatal(err)
+			}
+			record := operationreceipt.Record{
+				Identity: identity, State: operationreceipt.StateTerminal,
+				AcceptedAt: now.Add(-2 * time.Second), StartedAt: now.Add(-time.Second), TerminalAt: now,
+				ResultKind: ProxmoxGuestLifecycleReceiptKind, ResultVersion: ProxmoxGuestLifecycleReceiptVersion, Result: encoded,
+			}
+			query := operationreceipt.QueryResult{Version: operationreceipt.ProtocolVersion, Status: operationreceipt.QueryFoundTerminal, Record: &record}
+			if err := ValidateOperationQueryResultForIdentity(query, identity, now.Add(time.Second)); err != nil {
+				t.Fatalf("valid Proxmox terminal receipt query rejected: %v", err)
+			}
+
+			query.Record.ResultKind = DockerContainerLifecycleReceiptKind
+			if err := ValidateOperationQueryResultForIdentity(query, identity, now.Add(time.Second)); err == nil {
+				t.Fatal("wrong Proxmox receipt envelope was accepted")
+			}
+
+			query.Record.ResultKind = ProxmoxGuestLifecycleReceiptKind
+			result.Before.ObservedAt = time.Time{}
+			query.Record.Result, err = json.Marshal(result)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if err := ValidateOperationQueryResultForIdentity(query, identity, now.Add(time.Second)); err == nil || !strings.Contains(err.Error(), "preflight observation") {
+				t.Fatalf("missing durable preflight observation error = %v", err)
+			}
+		})
+	}
+}

internal/agentexec/server.go

@@ -2176,14 +2176,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)
+		s.cancelAgentRequest(ac, cmd.RequestID, "execute_command")
 		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)
+		s.cancelAgentRequest(ac, cmd.RequestID, "execute_command")
 		execLog.Warn().
 			Err(ctx.Err()).
 			Dur("duration", time.Since(startedAt)).
@@ -2196,22 +2196,22 @@ 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) {
+// cancelAgentRequest tells an agent to abort a request the server has stopped
+// waiting for. Typed runners bind this message to the in-flight request
+// context; already-started mutations remain indeterminate and are reconciled
+// through their durable receipts. Best effort: agents that predate the
+// cancel_command message ignore it and fall back to their own timeout.
+func (s *Server) cancelAgentRequest(ac *agentConn, requestID, operation 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")
+		log.Debug().Err(err).Str("request_id", requestID).Str("operation", operation).Msg("Failed to encode request 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")
+		log.Debug().Err(err).Str("request_id", requestID).Str("operation", operation).Msg("Failed to send request cancellation to agent")
 	}
 }
 
@@ -2263,6 +2263,12 @@ func prepareHostOperationRequest(s *Server, agentID string, requestID *string, b
 
 func dispatchHostOperation[Req hostOperationPayload, Res any](ctx context.Context, s *Server, agentID string, req Req, op hostOperationDispatch[Req, Res]) (*Res, error) {
 	requestID, actionID, operation, timeoutSeconds := req.hostOperationIdentity()
+	if ctx == nil {
+		return nil, fmt.Errorf("%s request %q not dispatched: context is required", op.label, requestID)
+	}
+	if err := ctx.Err(); err != nil {
+		return nil, fmt.Errorf("%s request %q not dispatched: %w", op.label, requestID, err)
+	}
 
 	ac, ok := s.connectionForContext(ctx, agentID)
 	if !ok {
@@ -2313,8 +2319,10 @@ func dispatchHostOperation[Req hostOperationPayload, Res any](ctx context.Contex
 		}
 		return &result, nil
 	case <-timer.C:
+		s.cancelAgentRequest(ac, requestID, operation)
 		return nil, fmt.Errorf("%s timed out after %s", op.label, time.Duration(timeoutSeconds)*time.Second)
 	case <-ctx.Done():
+		s.cancelAgentRequest(ac, requestID, operation)
 		return nil, ctx.Err()
 	case <-ac.done:
 		return nil, fmt.Errorf("agent %s disconnected before %s receipt", agentID, op.label)
@@ -2565,6 +2573,12 @@ func dispatchTypedDockerContainerOperation[Res any](
 	pending map[string]chan Res, label string,
 	validate func(Res) error,
 ) (*Res, error) {
+	if ctx == nil {
+		return nil, fmt.Errorf("%s request %q not dispatched: context is required", label, requestID)
+	}
+	if err := ctx.Err(); err != nil {
+		return nil, fmt.Errorf("%s request %q not dispatched: %w", label, requestID, err)
+	}
 	ac, ok := s.connectionForContext(ctx, agentID)
 	if !ok {
 		return nil, fmt.Errorf("agent %s not connected", agentID)
@@ -2613,8 +2627,10 @@ func dispatchTypedDockerContainerOperation[Res any](
 		}
 		return &result, nil
 	case <-timer.C:
+		s.cancelAgentRequest(ac, requestID, identity.OperationKind)
 		return nil, fmt.Errorf("%s timed out after %s", label, time.Duration(timeoutSeconds)*time.Second)
 	case <-ctx.Done():
+		s.cancelAgentRequest(ac, requestID, identity.OperationKind)
 		return nil, ctx.Err()
 	case <-ac.done:
 		return nil, fmt.Errorf("agent %s disconnected before %s receipt", agentID, label)
@@ -2835,10 +2851,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)
+		s.cancelAgentRequest(ac, req.RequestID, "read_file")
 		return nil, fmt.Errorf("read_file timed out after %v", timeout)
 	case <-ctx.Done():
-		s.cancelAgentCommand(ac, req.RequestID)
+		s.cancelAgentRequest(ac, req.RequestID, "read_file")
 		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

@@ -1833,6 +1833,24 @@ func registerCancelTestAgent(t *testing.T, s *Server, tsURL string) *cancelTestC
 	return &cancelTestConn{t: t, conn: conn}
 }
 
+func registerTypedCancelTestAgent(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: "typed-a1", Hostname: "typed-host1", Version: "1.2.3", Platform: "linux", Token: "typed-token",
+		RuntimeRole: RuntimeRoleActionRunner, ActionCapability: ActionCapabilityTypedV1,
+		OperationReceiptVersion: operationreceipt.ProtocolVersion,
+	}))
+	if ack := wsReadRegisteredPayload(t, conn); !ack.Success {
+		t.Fatalf("typed action runner registration failed: %s", ack.Message)
+	}
+	return &cancelTestConn{t: t, conn: conn}
+}
+
 type cancelTestConn struct {
 	t    *testing.T
 	conn *websocket.Conn
@@ -1968,3 +1986,155 @@ func TestExecuteCommand_AbandonedCommandSendsCancel(t *testing.T) {
 		})
 	}
 }
+
+func TestTypedOperations_AbandonedDispatchSendsExactlyOneCancel(t *testing.T) {
+	type dispatchFunc func(context.Context, *Server) error
+	cases := []struct {
+		name        string
+		messageType MessageType
+		dispatch    dispatchFunc
+	}{
+		{
+			name: "host update", messageType: MsgTypeHostUpdate,
+			dispatch: func(ctx context.Context, s *Server) error {
+				req := HostUpdatePayload{RequestID: "update.cancel.1", ActionID: "update", Operation: HostUpdateOperationInstall, ExpectedInventoryHash: "sha256:" + strings.Repeat("a", 64), Timeout: 30}
+				if err := BindHostUpdatePayload(&req); err != nil {
+					return err
+				}
+				_, err := s.ExecuteHostUpdate(ctx, "typed-a1", req)
+				return err
+			},
+		},
+		{
+			name: "storage cleanup", messageType: MsgTypeHostStorageCleanup,
+			dispatch: func(ctx context.Context, s *Server) error {
+				req := HostStorageCleanupPayload{RequestID: "cleanup.cancel.1", ActionID: "cleanup", Operation: HostStorageCleanupOperationPackageCache, ExpectedFingerprint: "sha256:" + strings.Repeat("b", 64), Timeout: 30}
+				if err := BindHostStorageCleanupPayload(&req); err != nil {
+					return err
+				}
+				_, err := s.ExecuteHostStorageCleanup(ctx, "typed-a1", req)
+				return err
+			},
+		},
+		{
+			name: "Proxmox guest lifecycle", messageType: MsgTypeProxmoxGuestLifecycle,
+			dispatch: func(ctx context.Context, s *Server) error {
+				req := ProxmoxGuestLifecyclePayload{RequestID: "pve.cancel.1", ActionID: "pve", Operation: "shutdown", GuestKind: "vm", VMID: 101, ExpectedStatus: "running", Timeout: 30}
+				if err := BindProxmoxGuestLifecyclePayload(&req); err != nil {
+					return err
+				}
+				_, err := s.ExecuteProxmoxGuestLifecycle(ctx, "typed-a1", req)
+				return err
+			},
+		},
+		{
+			name: "container lifecycle", messageType: MsgTypeDockerContainerLifecycle,
+			dispatch: func(ctx context.Context, s *Server) error {
+				req := DockerContainerLifecyclePayload{RequestID: "docker.cancel.1", ActionID: "docker", Operation: DockerContainerOperationRestart, Runtime: "docker", ContainerID: strings.Repeat("c", 12), ExpectedState: "running", Timeout: 30}
+				if err := BindDockerContainerLifecyclePayload(&req); err != nil {
+					return err
+				}
+				_, err := s.ExecuteDockerContainerLifecycle(ctx, "typed-a1", req)
+				return err
+			},
+		},
+		{
+			name: "container update", messageType: MsgTypeDockerContainerUpdate,
+			dispatch: func(ctx context.Context, s *Server) error {
+				req := DockerContainerUpdatePayload{RequestID: "docker-update.cancel.1", ActionID: "docker-update", Runtime: "docker", ContainerID: strings.Repeat("d", 12), ExpectedImageDigest: "sha256:" + strings.Repeat("e", 64), Timeout: 30}
+				if err := BindDockerContainerUpdatePayload(&req); err != nil {
+					return err
+				}
+				_, err := s.ExecuteDockerContainerUpdate(ctx, "typed-a1", req)
+				return err
+			},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			admission := AgentAdmission{TokenID: "typed-token", AgentID: "typed-a1", Hostname: "typed-host1", RuntimeRole: RuntimeRoleActionRunner, ActionCapability: ActionCapabilityTypedV1}
+			s := NewServerWithAdmissionValidator(func(token, _, _ string) (AgentAdmission, bool) {
+				return admission, token == admission.TokenID
+			}, func(AgentAdmission) bool { return true })
+			ts := newWSServer(t, s)
+			defer ts.Close()
+			agent := registerTypedCancelTestAgent(t, s, ts.URL)
+
+			ctx, cancel := context.WithCancel(context.Background())
+			done := make(chan error, 1)
+			go func() { done <- tc.dispatch(ctx, s) }()
+
+			request, ok := agent.nextMessage(3 * time.Second)
+			if !ok || request.Type != tc.messageType {
+				t.Fatalf("dispatched message = %+v, want %s", request, tc.messageType)
+			}
+			cancel()
+			if err := <-done; !errors.Is(err, context.Canceled) {
+				t.Fatalf("dispatch error = %v, want context.Canceled", err)
+			}
+
+			cancelMessage, ok := agent.nextMessage(3 * time.Second)
+			if !ok || cancelMessage.Type != MsgTypeCancelCmd {
+				t.Fatalf("cancellation message = %+v, want %s", cancelMessage, MsgTypeCancelCmd)
+			}
+			var payload CancelCommandPayload
+			if cancelMessage.Payload == nil || json.Unmarshal(*cancelMessage.Payload, &payload) != nil || payload.RequestID != request.ID {
+				t.Fatalf("cancellation payload = %+v, want request_id %q", payload, request.ID)
+			}
+			if duplicate, ok := agent.nextMessage(250 * time.Millisecond); ok && duplicate.Type == MsgTypeCancelCmd {
+				t.Fatalf("received duplicate cancellation: %+v", duplicate)
+			}
+		})
+	}
+}
+
+func TestTypedOperation_TimeoutSendsCancelAndExpiredContextNeverDispatches(t *testing.T) {
+	newTypedServer := func(t *testing.T) (*Server, *httptest.Server, *cancelTestConn) {
+		t.Helper()
+		admission := AgentAdmission{TokenID: "typed-token", AgentID: "typed-a1", Hostname: "typed-host1", RuntimeRole: RuntimeRoleActionRunner, ActionCapability: ActionCapabilityTypedV1}
+		s := NewServerWithAdmissionValidator(func(token, _, _ string) (AgentAdmission, bool) {
+			return admission, token == admission.TokenID
+		}, func(AgentAdmission) bool { return true })
+		ts := newWSServer(t, s)
+		t.Cleanup(ts.Close)
+		return s, ts, registerTypedCancelTestAgent(t, s, ts.URL)
+	}
+
+	t.Run("expired caller context", func(t *testing.T) {
+		s, _, agent := newTypedServer(t)
+		expired, expire := context.WithCancel(context.Background())
+		expire()
+		notDispatched := ProxmoxGuestLifecyclePayload{RequestID: "pve.expired.1", ActionID: "pve", Operation: "shutdown", GuestKind: "vm", VMID: 101, ExpectedStatus: "running", Timeout: 30}
+		if err := BindProxmoxGuestLifecyclePayload(&notDispatched); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := s.ExecuteProxmoxGuestLifecycle(expired, "typed-a1", notDispatched); !errors.Is(err, context.Canceled) || !strings.Contains(err.Error(), "not dispatched") {
+			t.Fatalf("expired dispatch error = %v", err)
+		}
+		if unexpected, ok := agent.nextMessage(250 * time.Millisecond); ok {
+			t.Fatalf("expired context reached action runner: %+v", unexpected)
+		}
+	})
+
+	t.Run("server timeout", func(t *testing.T) {
+		s, _, agent := newTypedServer(t)
+		timed := ProxmoxGuestLifecyclePayload{RequestID: "pve.timeout.1", ActionID: "pve", Operation: "shutdown", GuestKind: "vm", VMID: 101, ExpectedStatus: "running", Timeout: 1}
+		if err := BindProxmoxGuestLifecyclePayload(&timed); err != nil {
+			t.Fatal(err)
+		}
+		done := make(chan error, 1)
+		go func() { _, err := s.ExecuteProxmoxGuestLifecycle(context.Background(), "typed-a1", timed); done <- err }()
+		request, ok := agent.nextMessage(3 * time.Second)
+		if !ok || request.Type != MsgTypeProxmoxGuestLifecycle {
+			t.Fatalf("timeout request = %+v", request)
+		}
+		if err := <-done; err == nil || !strings.Contains(err.Error(), "timed out") {
+			t.Fatalf("timeout error = %v", err)
+		}
+		cancelMessage, ok := agent.nextMessage(3 * time.Second)
+		if !ok || cancelMessage.Type != MsgTypeCancelCmd {
+			t.Fatalf("timeout cancellation = %+v", cancelMessage)
+		}
+	})
+}

internal/agentexec/types.go

@@ -116,10 +116,10 @@ type RegisteredPayload struct {
 }
 
 // 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.
+// previously dispatched command or durable typed operation (caller context
+// expired or the server-side timeout fired). Typed operation cancellation is
+// best effort: a mutation that already started remains indeterminate and is
+// reconciled through its durable receipt.
 type CancelCommandPayload struct {
 	RequestID string `json:"request_id"`
 }

internal/hostagent/command_client_test.go

@@ -2,10 +2,12 @@ package hostagent
 
 import (
 	"context"
+	"fmt"
 	"io"
 	"reflect"
 	"testing"
 
+	"github.com/gorilla/websocket"
 	"github.com/rcourtman/pulse-go-rewrite/internal/agentexec"
 	"github.com/rs/zerolog"
 )
@@ -291,10 +293,10 @@ func TestCommandClient_handleCancelCommand_CancelsRegisteredRequest(t *testing.T
 
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	c.registerActiveCommand("req-1", cancel)
-	defer c.unregisterActiveCommand("req-1")
+	state, _ := c.registerActiveCommand(nil, "req-1", cancel)
+	defer c.finishCancellableRequest(nil, "req-1", state)
 
-	c.handleCancelCommand(cancelCommandPayload{RequestID: "req-1"})
+	c.handleCancelCommand(nil, cancelCommandPayload{RequestID: "req-1"})
 
 	select {
 	case <-ctx.Done():
@@ -310,10 +312,10 @@ func TestCommandClient_handleCancelCommand_UnknownRequestIsNoOp(t *testing.T) {
 
 	otherCtx, otherCancel := context.WithCancel(context.Background())
 	defer otherCancel()
-	c.registerActiveCommand("other", otherCancel)
-	defer c.unregisterActiveCommand("other")
+	state, _ := c.registerActiveCommand(nil, "other", otherCancel)
+	defer c.finishCancellableRequest(nil, "other", state)
 
-	c.handleCancelCommand(cancelCommandPayload{RequestID: "missing"})
+	c.handleCancelCommand(nil, cancelCommandPayload{RequestID: "missing"})
 
 	select {
 	case <-otherCtx.Done():
@@ -322,6 +324,137 @@ func TestCommandClient_handleCancelCommand_UnknownRequestIsNoOp(t *testing.T) {
 	}
 }
 
+func TestCommandClient_CancellationBeforeRegistrationIsConsumedAndConnectionScoped(t *testing.T) {
+	c := &CommandClient{logger: zerolog.Nop()}
+	firstConnection := &websocket.Conn{}
+	secondConnection := &websocket.Conn{}
+	const requestID = "typed-before-register"
+
+	firstState := c.noteCancellableRequest(firstConnection, requestID)
+	if firstState == nil {
+		t.Fatal("failed to admit first cancellable request")
+	}
+	c.handleCancelCommand(firstConnection, cancelCommandPayload{RequestID: requestID})
+	firstCtx, firstCancel := context.WithCancel(context.Background())
+	defer firstCancel()
+	_, firstRegistered := c.registerActiveCommand(firstConnection, requestID, firstCancel)
+	if firstRegistered {
+		t.Fatal("pre-registration cancellation did not stop provider handoff")
+	}
+	select {
+	case <-firstCtx.Done():
+	default:
+		t.Fatal("pre-registration cancellation was not consumed by registration")
+	}
+
+	secondState := c.noteCancellableRequest(secondConnection, requestID)
+	if secondState == nil {
+		t.Fatal("connection-scoped request id was incorrectly treated as a duplicate")
+	}
+	secondCtx, secondCancel := context.WithCancel(context.Background())
+	defer secondCancel()
+	_, secondRegistered := c.registerActiveCommand(secondConnection, requestID, secondCancel)
+	if !secondRegistered {
+		t.Fatal("cancellation leaked into a different WebSocket generation")
+	}
+	c.clearCancellableRequests(firstConnection)
+	select {
+	case <-secondCtx.Done():
+		t.Fatal("first-generation teardown canceled second-generation work")
+	default:
+	}
+	c.clearCancellableRequests(secondConnection)
+	select {
+	case <-secondCtx.Done():
+	default:
+		t.Fatal("connection teardown did not cancel its active request")
+	}
+
+	thirdConnection := &websocket.Conn{}
+	thirdState := c.noteCancellableRequest(thirdConnection, "teardown-before-register")
+	if thirdState == nil {
+		t.Fatal("failed to admit teardown-race request")
+	}
+	c.clearCancellableRequests(thirdConnection)
+	thirdCtx, thirdCancel := context.WithCancel(context.Background())
+	defer thirdCancel()
+	_, thirdRegistered := c.registerActiveCommand(thirdConnection, "teardown-before-register", thirdCancel)
+	if thirdRegistered {
+		t.Fatal("request crossed registration after its connection was torn down")
+	}
+	select {
+	case <-thirdCtx.Done():
+	default:
+		t.Fatal("teardown tombstone was not consumed at registration")
+	}
+}
+
+func TestCommandClient_CancellableRequestAdmissionIsBounded(t *testing.T) {
+	c := &CommandClient{logger: zerolog.Nop()}
+	conn := &websocket.Conn{}
+	states := make([]*cancellableRequestState, maxCancellableRequestsPerConnection)
+	for i := 0; i < maxCancellableRequestsPerConnection; i++ {
+		states[i] = c.noteCancellableRequest(conn, fmt.Sprintf("request-%d", i))
+		if states[i] == nil {
+			t.Fatalf("request %d was refused below the bound", i)
+		}
+	}
+	if c.noteCancellableRequest(conn, "over-capacity") != nil {
+		t.Fatal("over-capacity cancellable request was admitted")
+	}
+	c.clearCancellableRequests(conn)
+	for i := 0; i < maxCancellableRequestsPerConnection; i++ {
+		c.finishCancellableRequest(conn, fmt.Sprintf("request-%d", i), states[i])
+	}
+	if c.noteCancellableRequest(conn, "after-clear") == nil {
+		t.Fatal("cancellable request capacity did not recover after teardown")
+	}
+}
+
+func TestCommandClient_StaleCleanupCannotEraseReusedRequestCancellation(t *testing.T) {
+	c := &CommandClient{logger: zerolog.Nop()}
+	conn := &websocket.Conn{}
+	const requestID = "reused-request"
+
+	firstState := c.noteCancellableRequest(conn, requestID)
+	if firstState == nil {
+		t.Fatal("failed to admit first request generation")
+	}
+	firstCtx, firstCancel := context.WithCancel(context.Background())
+	registeredState, registered := c.registerActiveCommand(conn, requestID, firstCancel)
+	if !registered || registeredState != firstState {
+		t.Fatal("first request generation did not register")
+	}
+	// Model the handler's cleanup before its wrapper defer runs.
+	c.finishCancellableRequest(conn, requestID, registeredState)
+
+	secondState := c.noteCancellableRequest(conn, requestID)
+	if secondState == nil || secondState == firstState {
+		t.Fatal("failed to admit a distinct reused request generation")
+	}
+	// The stale outer cleanup from generation A must not delete generation B.
+	c.finishCancellableRequest(conn, requestID, firstState)
+	c.handleCancelCommand(conn, cancelCommandPayload{RequestID: requestID})
+	secondCtx, secondCancel := context.WithCancel(context.Background())
+	defer secondCancel()
+	registeredState, registered = c.registerActiveCommand(conn, requestID, secondCancel)
+	if registered || registeredState != secondState {
+		t.Fatal("reused request lost its pre-registration cancellation fence")
+	}
+	select {
+	case <-secondCtx.Done():
+	default:
+		t.Fatal("reused request cancellation was not consumed")
+	}
+	c.finishCancellableRequest(conn, requestID, secondState)
+	firstCancel()
+	select {
+	case <-firstCtx.Done():
+	default:
+		t.Fatal("first request cleanup did not retain its own cancel function")
+	}
+}
+
 func TestCommandClientActionRunnerMessageCatalogRejectsGenericAuthority(t *testing.T) {
 	for _, message := range []messageType{msgTypeExecuteCmd, msgTypeReadFile, msgTypeDeployPreflight, msgTypeDeployInstall, msgTypeDeployCancel} {
 		if allowedActionRunnerMessage(message) {

internal/hostagent/commands.go

@@ -127,12 +127,13 @@ type CommandClient struct {
 	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
+	// In-flight command and durable typed-operation requests by request ID, so
+	// a server-issued cancel_command can abort the local execution context
+	// instead of letting abandoned work run to its full timeout. Mutations that
+	// already crossed provider handoff remain indeterminate and are settled by
+	// their durable receipt. The server enforces unique in-flight request IDs.
+	activeCommandsMu    sync.Mutex
+	cancellableRequests map[cancellableRequestKey]*cancellableRequestState
 
 	// actionRunnerOnly is an immutable constructor-selected protocol ceiling.
 	// It permits only the closed typed action families and their receipt,
@@ -146,6 +147,18 @@ type CommandClient struct {
 	actionHealthWriter    func(bool) error
 }
 
+const maxCancellableRequestsPerConnection = 256
+
+type cancellableRequestKey struct {
+	connection *websocket.Conn
+	requestID  string
+}
+
+type cancellableRequestState struct {
+	cancel   context.CancelFunc
+	canceled bool
+}
+
 // NewCommandClient creates a new command execution client
 func NewCommandClient(cfg Config, agentID, hostname, platform, version string) *CommandClient {
 	logger := cfg.Logger.With().Str("component", "command-client").Logger()
@@ -185,6 +198,7 @@ func NewCommandClient(cfg Config, agentID, hostname, platform, version string) *
 		operationReceiptErr:   receiptErr,
 		logger:                logger,
 		done:                  make(chan struct{}),
+		cancellableRequests:   make(map[cancellableRequestKey]*cancellableRequestState),
 	}
 }
 
@@ -398,6 +412,7 @@ func (c *CommandClient) connectAndHandle(ctx context.Context) error {
 	c.connMu.Lock()
 	c.conn = conn
 	c.connMu.Unlock()
+	defer c.clearCancellableRequests(conn)
 
 	defer func() {
 		c.connMu.Lock()
@@ -600,6 +615,18 @@ func computeReconnectDelay(failures int) time.Duration {
 	return utils.ExponentialBackoff(reconnectDelay, reconnectMaxDelay, failures, reconnectJitterRatio, reconnectRandFloat64)
 }
 
+func (c *CommandClient) launchCancellableRequest(conn *websocket.Conn, requestID, operation string, handle func()) {
+	state := c.noteCancellableRequest(conn, requestID)
+	if state == nil {
+		c.logger.Warn().Str("request_id", requestID).Str("operation", operation).Msg("Dropping duplicate, invalid, or over-capacity cancellable request")
+		return
+	}
+	go func() {
+		defer c.finishCancellableRequest(conn, requestID, state)
+		handle()
+	}()
+}
+
 func (c *CommandClient) handleMessages(ctx context.Context, conn *websocket.Conn) error {
 	for {
 		select {
@@ -627,8 +654,9 @@ func (c *CommandClient) handleMessages(ctx context.Context, conn *websocket.Conn
 				continue
 			}
 
-			// Execute command in background
-			go c.handleExecuteCommand(ctx, conn, payload)
+			c.launchCancellableRequest(conn, payload.RequestID, "execute_command", func() {
+				c.handleExecuteCommand(ctx, conn, payload)
+			})
 
 		case msgTypeReadFile:
 			// Handle read_file similarly (uses cat command internally)
@@ -637,7 +665,9 @@ func (c *CommandClient) handleMessages(ctx context.Context, conn *websocket.Conn
 				c.logger.Error().Err(err).Msg("Failed to parse read_file payload")
 				continue
 			}
-			go c.handleExecuteCommand(ctx, conn, payload)
+			c.launchCancellableRequest(conn, payload.RequestID, "read_file", func() {
+				c.handleExecuteCommand(ctx, conn, payload)
+			})
 
 		case msgTypeHostUpdate:
 			payload, err := agentexec.DecodeHostUpdatePayload(msg.Payload)
@@ -645,7 +675,9 @@ func (c *CommandClient) handleMessages(ctx context.Context, conn *websocket.Conn
 				c.logger.Error().Err(err).Msg("Failed to parse host_update payload")
 				continue
 			}
-			go c.handleHostUpdate(ctx, conn, payload)
+			c.launchCancellableRequest(conn, payload.RequestID, payload.Operation, func() {
+				c.handleHostUpdate(ctx, conn, payload)
+			})
 
 		case msgTypeProxmoxGuestLifecycle:
 			payload, err := agentexec.DecodeProxmoxGuestLifecyclePayload(msg.Payload)
@@ -653,7 +685,9 @@ func (c *CommandClient) handleMessages(ctx context.Context, conn *websocket.Conn
 				c.logger.Warn().Err(err).Msg("Dropping invalid Proxmox guest lifecycle request")
 				continue
 			}
-			go c.handleProxmoxGuestLifecycle(ctx, conn, payload)
+			c.launchCancellableRequest(conn, payload.RequestID, payload.Operation, func() {
+				c.handleProxmoxGuestLifecycle(ctx, conn, payload)
+			})
 
 		case msgTypeActionPreflight:
 			payload, err := agentexec.DecodeActionPreflightPayload(msg.Payload)
@@ -677,7 +711,9 @@ func (c *CommandClient) handleMessages(ctx context.Context, conn *websocket.Conn
 				c.logger.Error().Err(err).Msg("Failed to parse host_storage_cleanup payload")
 				continue
 			}
-			go c.handleHostStorageCleanup(ctx, conn, payload)
+			c.launchCancellableRequest(conn, payload.RequestID, payload.Operation, func() {
+				c.handleHostStorageCleanup(ctx, conn, payload)
+			})
 
 		case msgTypeDockerContainerUpdate:
 			payload, err := agentexec.DecodeDockerContainerUpdatePayload(msg.Payload)
@@ -685,7 +721,9 @@ func (c *CommandClient) handleMessages(ctx context.Context, conn *websocket.Conn
 				c.logger.Warn().Err(err).Msg("Dropping invalid docker container update request")
 				continue
 			}
-			go c.handleDockerContainerUpdate(ctx, conn, payload)
+			c.launchCancellableRequest(conn, payload.RequestID, payload.Operation, func() {
+				c.handleDockerContainerUpdate(ctx, conn, payload)
+			})
 
 		case msgTypeDockerContainerLifecycle:
 			payload, err := agentexec.DecodeDockerContainerLifecyclePayload(msg.Payload)
@@ -693,7 +731,9 @@ func (c *CommandClient) handleMessages(ctx context.Context, conn *websocket.Conn
 				c.logger.Error().Err(err).Msg("Failed to parse docker container lifecycle payload")
 				continue
 			}
-			go c.handleDockerContainerLifecycle(ctx, conn, payload)
+			c.launchCancellableRequest(conn, payload.RequestID, payload.Operation, func() {
+				c.handleDockerContainerLifecycle(ctx, conn, payload)
+			})
 
 		case msgTypeOperationQuery:
 			query, err := operationreceipt.DecodeQuery(msg.Payload)
@@ -733,7 +773,7 @@ func (c *CommandClient) handleMessages(ctx context.Context, conn *websocket.Conn
 				c.logger.Error().Err(err).Msg("Failed to parse cancel_command payload")
 				continue
 			}
-			c.handleCancelCommand(payload)
+			c.handleCancelCommand(conn, payload)
 
 		default:
 			c.logger.Debug().Str("type", string(msg.Type)).Msg("Unknown message type")
@@ -753,7 +793,9 @@ func (c *CommandClient) handleHostUpdate(ctx context.Context, conn *websocket.Co
 		RequestID: strings.TrimSpace(payload.RequestID), ActionID: strings.TrimSpace(payload.ActionID),
 		ExecutionPhase: agentexec.HostUpdatePhasePreflight, Verification: agentexec.HostUpdateVerificationInconclusive,
 	}
-	if c.packageUpdates == nil {
+	if updateCtx.Err() != nil {
+		result.Error = "host update canceled before mutation dispatch"
+	} else if c.packageUpdates == nil {
 		result.Error = "host package update service is unavailable"
 	} else {
 		result = c.packageUpdates.Apply(updateCtx, payload)
@@ -887,9 +929,9 @@ func (c *CommandClient) beginHostAPTOperation(ctx context.Context, conn *websock
 		timeout = defaultTimeout
 	}
 	opCtx, cancel := context.WithTimeout(ctx, timeout)
-	c.registerActiveCommand(requestID, cancel)
+	state, _ := c.registerActiveCommand(conn, requestID, cancel)
 	return opCtx, func() {
-		c.unregisterActiveCommand(requestID)
+		c.finishCancellableRequest(conn, requestID, state)
 		cancel()
 	}, true
 }
@@ -941,7 +983,9 @@ func (c *CommandClient) handleHostStorageCleanup(ctx context.Context, conn *webs
 		RequestID: strings.TrimSpace(payload.RequestID), ActionID: strings.TrimSpace(payload.ActionID),
 		ExecutionPhase: agentexec.HostStorageCleanupPhasePreflight, Verification: agentexec.HostStorageCleanupVerificationInconclusive,
 	}
-	if c.storageCleanup == nil {
+	if cleanupCtx.Err() != nil {
+		result.Error = "host cleanup canceled before mutation dispatch"
+	} else if c.storageCleanup == nil {
 		result.Error = "host storage cleanup service is unavailable"
 	} else {
 		result = c.storageCleanup.Apply(cleanupCtx, payload)
@@ -971,15 +1015,17 @@ func (c *CommandClient) handleDockerContainerLifecycle(ctx context.Context, conn
 		timeout = 2 * time.Minute
 	}
 	operationCtx, cancel := context.WithTimeout(ctx, timeout)
-	c.registerActiveCommand(payload.RequestID, cancel)
-	defer c.unregisterActiveCommand(payload.RequestID)
+	state, _ := c.registerActiveCommand(conn, payload.RequestID, cancel)
+	defer c.finishCancellableRequest(conn, payload.RequestID, state)
 	defer cancel()
 	result := agentexec.DockerContainerLifecycleResultPayload{
 		RequestID: payload.RequestID, ActionID: payload.ActionID, Operation: payload.Operation,
 		OperationVersion: payload.OperationVersion, RequestDigest: payload.RequestDigest, ContainerID: payload.ContainerID,
 		ExecutionPhase: agentexec.DockerContainerPhasePreflight,
 	}
-	if c.dockerLifecycle == nil {
+	if operationCtx.Err() != nil {
+		result.Error = "container lifecycle canceled before mutation dispatch"
+	} else if c.dockerLifecycle == nil {
 		result.Error = "typed container lifecycle service is unavailable"
 	} else {
 		result = c.dockerLifecycle.Apply(operationCtx, payload)
@@ -1182,8 +1228,8 @@ func (c *CommandClient) handleExecuteCommand(ctx context.Context, conn *websocke
 
 	cmdCtx, cancel := context.WithCancel(ctx)
 	defer cancel()
-	c.registerActiveCommand(payload.RequestID, cancel)
-	defer c.unregisterActiveCommand(payload.RequestID)
+	state, _ := c.registerActiveCommand(conn, payload.RequestID, cancel)
+	defer c.finishCancellableRequest(conn, payload.RequestID, state)
 
 	result := c.executeCommand(cmdCtx, payload)
 	result.Duration = time.Since(startTime).Milliseconds()
@@ -1217,36 +1263,116 @@ 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)
+func (c *CommandClient) noteCancellableRequest(conn *websocket.Conn, requestID string) *cancellableRequestState {
+	requestID = strings.TrimSpace(requestID)
+	if requestID == "" || len(requestID) > 128 {
+		return nil
 	}
-	c.activeCommands[requestID] = cancel
-}
-
-func (c *CommandClient) unregisterActiveCommand(requestID string) {
+	key := cancellableRequestKey{connection: conn, requestID: requestID}
 	c.activeCommandsMu.Lock()
 	defer c.activeCommandsMu.Unlock()
-	delete(c.activeCommands, requestID)
+	if c.cancellableRequests == nil {
+		c.cancellableRequests = make(map[cancellableRequestKey]*cancellableRequestState)
+	}
+	if _, exists := c.cancellableRequests[key]; exists || len(c.cancellableRequests) >= maxCancellableRequestsPerConnection {
+		return nil
+	}
+	state := &cancellableRequestState{}
+	c.cancellableRequests[key] = state
+	return state
 }
 
-// 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) {
+// registerActiveCommand atomically consumes a cancellation that arrived after
+// the WebSocket reader admitted the request but before its handler goroutine
+// registered. false means the handler must not cross provider handoff.
+func (c *CommandClient) registerActiveCommand(conn *websocket.Conn, requestID string, cancel context.CancelFunc) (*cancellableRequestState, bool) {
+	key := cancellableRequestKey{connection: conn, requestID: strings.TrimSpace(requestID)}
 	c.activeCommandsMu.Lock()
-	cancel, ok := c.activeCommands[payload.RequestID]
+	if c.cancellableRequests == nil {
+		c.cancellableRequests = make(map[cancellableRequestKey]*cancellableRequestState)
+	}
+	state, exists := c.cancellableRequests[key]
+	if !exists {
+		if len(c.cancellableRequests) >= maxCancellableRequestsPerConnection {
+			c.activeCommandsMu.Unlock()
+			cancel()
+			return nil, false
+		}
+		state = &cancellableRequestState{}
+		c.cancellableRequests[key] = state
+	}
+	if state.cancel != nil {
+		c.activeCommandsMu.Unlock()
+		cancel()
+		return nil, false
+	}
+	state.cancel = cancel
+	canceled := state.canceled
+	c.activeCommandsMu.Unlock()
+	if canceled {
+		cancel()
+		return state, false
+	}
+	return state, true
+}
+
+func (c *CommandClient) finishCancellableRequest(conn *websocket.Conn, requestID string, state *cancellableRequestState) {
+	key := cancellableRequestKey{connection: conn, requestID: strings.TrimSpace(requestID)}
+	c.activeCommandsMu.Lock()
+	if current := c.cancellableRequests[key]; state != nil && current == state {
+		delete(c.cancellableRequests, key)
+	}
+	c.activeCommandsMu.Unlock()
+}
+
+func (c *CommandClient) clearCancellableRequests(conn *websocket.Conn) {
+	var cancels []context.CancelFunc
+	c.activeCommandsMu.Lock()
+	for key, state := range c.cancellableRequests {
+		if key.connection != conn {
+			continue
+		}
+		// Retain the canceled state until the already-launched handler consumes
+		// it or its wrapper exits. Deleting a not-yet-registered entry here would
+		// let that handler recreate the key after disconnect and cross provider
+		// handoff under a dead WebSocket generation.
+		state.canceled = true
+		if state != nil && state.cancel != nil {
+			cancels = append(cancels, state.cancel)
+		}
+	}
+	c.activeCommandsMu.Unlock()
+	for _, cancel := range cancels {
+		cancel()
+	}
+}
+
+// handleCancelCommand aborts an in-flight command or durable typed-operation
+// context the server has stopped waiting for. Generic commands reap their
+// process group; typed mutations persist an indeterminate terminal receipt
+// when cancellation arrives after provider handoff.
+func (c *CommandClient) handleCancelCommand(conn *websocket.Conn, payload cancelCommandPayload) {
+	key := cancellableRequestKey{connection: conn, requestID: strings.TrimSpace(payload.RequestID)}
+	c.activeCommandsMu.Lock()
+	state, ok := c.cancellableRequests[key]
+	var cancel context.CancelFunc
+	if ok && state != nil {
+		if state.cancel == nil {
+			state.canceled = true
+		} else {
+			cancel = state.cancel
+		}
+	}
 	c.activeCommandsMu.Unlock()
 
-	if ok {
+	if cancel != nil {
 		c.logger.Info().Str("request_id", payload.RequestID).Msg("Canceling command at server request")
 		cancel()
+	} else if ok {
+		c.logger.Info().Str("request_id", payload.RequestID).Msg("Remembering cancellation until request handler registers")
 	} else {
-		// Common benign race: the result was already sent while the cancel
-		// was in flight.
+		// Common benign race: the result was already sent while the cancellation
+		// was in flight. Unknown request IDs are never retained for future work.
 		c.logger.Debug().Str("request_id", payload.RequestID).Msg("No active command to cancel")
 	}
 }

internal/hostagent/commands_execute_unix_test.go

@@ -121,14 +121,14 @@ func TestCommandClient_executeCommand_CancelKillsProcessGroup(t *testing.T) {
 
 	cmdCtx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	c.registerActiveCommand(payload.RequestID, cancel)
-	defer c.unregisterActiveCommand(payload.RequestID)
+	state, _ := c.registerActiveCommand(nil, payload.RequestID, cancel)
+	defer c.finishCancellableRequest(nil, payload.RequestID, state)
 
 	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})
+		c.handleCancelCommand(nil, cancelCommandPayload{RequestID: payload.RequestID})
 	}()
 
 	start := time.Now()

internal/hostagent/docker_update.go

@@ -44,8 +44,8 @@ func (c *CommandClient) handleDockerContainerUpdate(ctx context.Context, conn *w
 		timeout = 15 * time.Minute
 	}
 	operationCtx, cancel := context.WithTimeout(ctx, timeout)
-	c.registerActiveCommand(payload.RequestID, cancel)
-	defer c.unregisterActiveCommand(payload.RequestID)
+	state, _ := c.registerActiveCommand(conn, payload.RequestID, cancel)
+	defer c.finishCancellableRequest(conn, payload.RequestID, state)
 	defer cancel()
 
 	result := c.runDockerContainerUpdate(operationCtx, payload)
@@ -75,6 +75,10 @@ func (c *CommandClient) runDockerContainerUpdate(ctx context.Context, payload ag
 		result.Error = "typed container update preflight refused"
 		return result
 	}
+	if ctx.Err() != nil {
+		result.Error = "container update canceled before mutation dispatch"
+		return result
+	}
 	if c.dockerUpdater == nil {
 		c.logger.Warn().Str("request_id", payload.RequestID).Msg("Docker update refused: no docker module bridge wired")
 		result.ReasonCode = agentexec.ActionRefusalCapabilityUnavailable

internal/hostagent/operation_receipt_websocket_integration_test.go

@@ -2,10 +2,12 @@ package hostagent
 
 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"net/http"
 	"net/http/httptest"
 	"os"
+	"path/filepath"
 	"strings"
 	"sync"
 	"testing"
@@ -233,3 +235,161 @@ func TestRealServerAndUnifiedAgentWebSocketDockerDuplicateReplayMutatesOnce(t *t
 		t.Fatalf("typed docker mutation calls = %d, want one", manager.calls)
 	}
 }
+
+func TestRealServerActionRunnerCancellationPersistsAndReplaysProxmoxReceiptAfterReconnect(t *testing.T) {
+	admission := agentexec.AgentAdmission{
+		TokenID: "runner-token", AgentID: "agent-pve", Hostname: "pve.example.test",
+		RuntimeRole: agentexec.RuntimeRoleActionRunner, ActionCapability: agentexec.ActionCapabilityTypedV1,
+	}
+	server := agentexec.NewServerWithAdmissionValidator(func(token, _, _ string) (agentexec.AgentAdmission, bool) {
+		return admission, token == admission.TokenID
+	}, func(agentexec.AgentAdmission) bool { return true })
+	httpServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, request *http.Request) {
+		if request.Method == http.MethodPatch && request.URL.Path == "/api/agents/action-runner/credential" {
+			w.WriteHeader(http.StatusNoContent)
+			return
+		}
+		server.HandleWebSocket(w, request)
+	}))
+	defer httpServer.Close()
+
+	stateDir := t.TempDir()
+	logger := zerolog.Nop()
+	mutationStarted := make(chan struct{})
+	var startOnce sync.Once
+	var mutationMu sync.Mutex
+	mutationCalls := 0
+	manager := newProxmoxGuestLifecycleManager()
+	manager.run = func(ctx context.Context, _ string, args ...string) ([]byte, error) {
+		if len(args) == 0 {
+			return nil, errors.New("missing Proxmox verb")
+		}
+		if args[0] == "status" {
+			return []byte("status: running"), nil
+		}
+		if args[0] != "shutdown" || len(args) != 2 || args[1] != "101" {
+			return nil, errors.New("unexpected Proxmox mutation")
+		}
+		mutationMu.Lock()
+		mutationCalls++
+		mutationMu.Unlock()
+		startOnce.Do(func() { close(mutationStarted) })
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}
+
+	startRunner := func(t *testing.T) (*CommandClient, context.CancelFunc, <-chan error) {
+		t.Helper()
+		client := NewActionRunnerClient(ActionRunnerClientConfig{
+			PulseURL: httpServer.URL, APIToken: admission.TokenID, StateDir: stateDir,
+			HealthPath: filepath.Join(stateDir, "health.json"), ActivationNonce: strings.Repeat("a", 32), Logger: &logger,
+		}, admission.AgentID, admission.Hostname, "test")
+		client.proxmoxGuestLifecycle = manager
+		ctx, cancel := context.WithCancel(context.Background())
+		done := make(chan error, 1)
+		go func() { done <- client.Run(ctx) }()
+		deadline := time.Now().Add(3 * time.Second)
+		for !server.IsAgentConnected(admission.AgentID) && time.Now().Before(deadline) {
+			time.Sleep(10 * time.Millisecond)
+		}
+		if !server.IsAgentConnected(admission.AgentID) {
+			cancel()
+			_ = client.Close()
+			t.Fatal("action runner did not connect")
+		}
+		return client, cancel, done
+	}
+	stopRunner := func(t *testing.T, client *CommandClient, cancel context.CancelFunc, done <-chan error) {
+		t.Helper()
+		cancel()
+		_ = client.Close()
+		select {
+		case <-done:
+		case <-time.After(3 * time.Second):
+			t.Fatal("action runner did not stop")
+		}
+	}
+
+	request := agentexec.ProxmoxGuestLifecyclePayload{
+		RequestID: "pve.cancel-replay.1", ActionID: "pve.cancel-replay", Operation: "shutdown",
+		GuestKind: "vm", VMID: 101, ExpectedStatus: "running", Timeout: 30,
+	}
+	if err := agentexec.BindProxmoxGuestLifecyclePayload(&request); err != nil {
+		t.Fatal(err)
+	}
+	identity := agentexec.ProxmoxGuestLifecycleOperationIdentity(admission.AgentID, request)
+
+	first, cancelFirst, firstDone := startRunner(t)
+	dispatchCtx, cancelDispatch := context.WithCancel(context.Background())
+	dispatchDone := make(chan error, 1)
+	go func() {
+		_, err := server.ExecuteProxmoxGuestLifecycle(dispatchCtx, admission.AgentID, request)
+		dispatchDone <- err
+	}()
+	select {
+	case <-mutationStarted:
+	case <-time.After(3 * time.Second):
+		t.Fatal("Proxmox mutation did not start")
+	}
+	cancelDispatch()
+	if err := <-dispatchDone; !errors.Is(err, context.Canceled) {
+		t.Fatalf("canceled dispatch error = %v, want context.Canceled", err)
+	}
+
+	var query operationreceipt.QueryResult
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		queryCtx, cancelQuery := context.WithTimeout(context.Background(), time.Second)
+		result, err := server.QueryAgentOperation(queryCtx, admission.AgentID, identity)
+		cancelQuery()
+		if err == nil && result.Status == operationreceipt.QueryFoundTerminal {
+			query = result
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	if query.Record == nil || query.Status != operationreceipt.QueryFoundTerminal {
+		t.Fatalf("terminal receipt not observed after cancellation: %+v", query)
+	}
+	var canceled agentexec.ProxmoxGuestLifecycleResultPayload
+	if err := json.Unmarshal(query.Record.Result, &canceled); err != nil {
+		t.Fatal(err)
+	}
+	if !canceled.MutationStarted || canceled.MutationCompleted || !strings.Contains(canceled.Error, "recovery inspection") {
+		t.Fatalf("canceled durable receipt = %+v", canceled)
+	}
+	stopRunner(t, first, cancelFirst, firstDone)
+
+	second, cancelSecond, secondDone := startRunner(t)
+	defer stopRunner(t, second, cancelSecond, secondDone)
+	replayed, err := server.ExecuteProxmoxGuestLifecycle(context.Background(), admission.AgentID, request)
+	if err != nil || replayed == nil || !replayed.MutationStarted || replayed.MutationCompleted || replayed.Error != canceled.Error {
+		t.Fatalf("replayed result = %+v, err=%v", replayed, err)
+	}
+	mutationMu.Lock()
+	if mutationCalls != 1 {
+		t.Fatalf("Proxmox mutations = %d, want one", mutationCalls)
+	}
+	mutationMu.Unlock()
+
+	for _, mutate := range []func(*operationreceipt.Identity){
+		func(conflict *operationreceipt.Identity) { conflict.ActionID = "different-action" },
+		func(conflict *operationreceipt.Identity) {
+			conflict.RequestDigest = "sha256:" + strings.Repeat("f", 64)
+		},
+	} {
+		conflict := identity
+		mutate(&conflict)
+		queryCtx, cancelQuery := context.WithTimeout(context.Background(), 250*time.Millisecond)
+		_, err := server.QueryAgentOperation(queryCtx, admission.AgentID, conflict)
+		cancelQuery()
+		if err == nil {
+			t.Fatalf("conflicting receipt identity was accepted: %+v", conflict)
+		}
+	}
+	wrongAgent := identity
+	wrongAgent.AgentID = "other-agent"
+	if _, err := server.QueryAgentOperation(context.Background(), admission.AgentID, wrongAgent); !errors.Is(err, operationreceipt.ErrBindingConflict) {
+		t.Fatalf("cross-agent receipt query error = %v", err)
+	}
+}

internal/hostagent/proxmox_guest_lifecycle.go

@@ -145,10 +145,19 @@ func (c *CommandClient) handleProxmoxGuestLifecycle(ctx context.Context, conn *w
 		return
 	}
 	operationCtx, cancel := context.WithTimeout(ctx, time.Duration(payload.Timeout)*time.Second)
-	c.registerActiveCommand(payload.RequestID, cancel)
-	defer c.unregisterActiveCommand(payload.RequestID)
+	state, registered := c.registerActiveCommand(conn, payload.RequestID, cancel)
+	defer c.finishCancellableRequest(conn, payload.RequestID, state)
 	defer cancel()
-	result := c.proxmoxGuestLifecycle.Apply(operationCtx, payload)
+	result := agentexec.ProxmoxGuestLifecycleResultPayload{
+		RequestID: payload.RequestID, ActionID: payload.ActionID, Operation: payload.Operation,
+		OperationVersion: payload.OperationVersion, RequestDigest: payload.RequestDigest,
+		GuestKind: payload.GuestKind, VMID: payload.VMID, ExecutionPhase: agentexec.ProxmoxGuestPhasePreflight,
+	}
+	if !registered || operationCtx.Err() != nil {
+		result.Error = "Proxmox guest lifecycle canceled before mutation dispatch"
+	} else {
+		result = c.proxmoxGuestLifecycle.Apply(operationCtx, payload)
+	}
 	encoded, err := json.Marshal(result)
 	if err != nil {
 		return

internal/hostagent/proxmox_guest_lifecycle_test.go

@@ -75,7 +75,7 @@ func TestProxmoxGuestLifecycleCancellationStopsMutationAndProducesBoundFailure(t
 
 func TestProxmoxGuestLifecycleTerminalReceiptReplaysWithoutSecondMutation(t *testing.T) {
 	payload := boundProxmoxPayload(t)
-	client := &CommandClient{agentID: "agent-pve", logger: zerolog.Nop(), activeCommands: make(map[string]context.CancelFunc)}
+	client := &CommandClient{agentID: "agent-pve", logger: zerolog.Nop(), cancellableRequests: make(map[cancellableRequestKey]*cancellableRequestState)}
 	receipts, err := operationreceipt.Open(filepath.Join(t.TempDir(), "receipts.db"), hostOperationReceiptConfig())
 	if err != nil {
 		t.Fatal(err)
@@ -132,3 +132,77 @@ func TestProxmoxGuestLifecycleTerminalReceiptReplaysWithoutSecondMutation(t *tes
 		t.Fatalf("mutations=%d, want 1", mutations)
 	}
 }
+
+func TestProxmoxGuestLifecycleCancellationBeforeHandlerRegistrationSkipsProviderAndPersistsReceipt(t *testing.T) {
+	payload := boundProxmoxPayload(t)
+	client := &CommandClient{agentID: "agent-pve", logger: zerolog.Nop(), cancellableRequests: make(map[cancellableRequestKey]*cancellableRequestState)}
+	receipts, err := operationreceipt.Open(filepath.Join(t.TempDir(), "receipts.db"), hostOperationReceiptConfig())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer receipts.Close()
+	client.operationReceipts = receipts
+	client.proxmoxGuestLifecycle = newProxmoxGuestLifecycleManager()
+	providerCalls := 0
+	client.proxmoxGuestLifecycle.run = func(context.Context, string, ...string) ([]byte, error) {
+		providerCalls++
+		return nil, errors.New("provider must not run after pre-registration cancellation")
+	}
+
+	upgrader := websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+	serverConnections := make(chan *websocket.Conn, 1)
+	releaseServer := make(chan struct{})
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, upgradeErr := upgrader.Upgrade(w, r, nil)
+		if upgradeErr != nil {
+			return
+		}
+		serverConnections <- conn
+		<-releaseServer
+		_ = conn.Close()
+	}))
+	defer server.Close()
+	remote, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(server.URL, "http"), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer remote.Close()
+	serverConn := <-serverConnections
+	defer func() { releaseServer <- struct{}{} }()
+
+	if client.noteCancellableRequest(serverConn, payload.RequestID) == nil {
+		t.Fatal("failed to admit cancellable Proxmox request")
+	}
+	handlerWaiting := make(chan struct{})
+	releaseHandler := make(chan struct{})
+	handlerDone := make(chan struct{})
+	go func() {
+		close(handlerWaiting)
+		<-releaseHandler
+		client.handleProxmoxGuestLifecycle(context.Background(), serverConn, payload)
+		close(handlerDone)
+	}()
+	<-handlerWaiting
+	client.handleCancelCommand(serverConn, cancelCommandPayload{RequestID: payload.RequestID})
+	close(releaseHandler)
+
+	var message wsMessage
+	if err := remote.ReadJSON(&message); err != nil {
+		t.Fatal(err)
+	}
+	<-handlerDone
+	var result agentexec.ProxmoxGuestLifecycleResultPayload
+	if message.Type != msgTypeProxmoxGuestLifecycleResult || json.Unmarshal(message.Payload, &result) != nil {
+		t.Fatalf("terminal cancellation message=%+v result=%+v", message, result)
+	}
+	if result.MutationStarted || result.ExecutionPhase != agentexec.ProxmoxGuestPhasePreflight || !strings.Contains(result.Error, "canceled before mutation") {
+		t.Fatalf("pre-registration cancellation receipt=%+v", result)
+	}
+	if providerCalls != 0 {
+		t.Fatalf("provider calls=%d, want zero", providerCalls)
+	}
+	query, err := receipts.Query(agentexec.ProxmoxGuestLifecycleOperationIdentity(client.agentID, payload))
+	if err != nil || query.Status != operationreceipt.QueryFoundTerminal || query.Record == nil {
+		t.Fatalf("durable cancellation query=%+v err=%v", query, err)
+	}
+}