From dc6b5c31975c76b4c28860f7b999f2bf0dc2a36f Mon Sep 17 00:00:00 2001 From: rcourtman Date: Fri, 10 Jul 2026 00:05:22 +0100 Subject: [PATCH] Verify AI Kubernetes scale and Docker update actions after execution Two AI write paths asserted unverified success while the Proxmox guest and Docker start/stop/restart handlers already do read-after-write checks. Bring both in line with that idiom: - pulse_kubernetes scale: re-read the deployment's spec/ready replicas via kubectl through the same agent (bounded settle-and-retry window) and return a JSON response with a verification block instead of "Action complete - no verification needed". - pulse_docker update: the docker agent already recreates the container, health-checks it, rolls back on failure, and acks a terminal command status; expose that status through a new Monitor lookup (GetDockerCommandStatus) plus UpdatesProvider.GetCommandStatus, and poll it within a bounded window. Responses now report verified success, verified failure (is_error), or an explicit inconclusive, never unverified success. kubernetes_control_test.go also carries a small in-flight fix from the parallel remediation-lock work (in-memory ActionAuditStore in the test helper) that these tests require to run on this tree. --- internal/ai/tools/adapters.go | 21 +++ internal/ai/tools/adapters_test.go | 31 +++- internal/ai/tools/data_types.go | 24 +-- internal/ai/tools/docker_updates_test.go | 138 ++++++++++++++++ internal/ai/tools/executor.go | 3 + internal/ai/tools/kubernetes_control_test.go | 156 ++++++++++++++++++- internal/ai/tools/shared_test.go | 5 + internal/ai/tools/tools_docker.go | 129 ++++++++++++++- internal/ai/tools/tools_kubernetes.go | 111 ++++++++++++- internal/ai/tools/tools_storage_test.go | 7 + internal/monitoring/docker_commands.go | 30 ++++ internal/monitoring/docker_commands_test.go | 47 ++++++ 12 files changed, 676 insertions(+), 26 deletions(-) diff --git a/internal/ai/tools/adapters.go b/internal/ai/tools/adapters.go index b2d7891f9..f7c7b8fd0 100644 --- a/internal/ai/tools/adapters.go +++ b/internal/ai/tools/adapters.go @@ -700,6 +700,7 @@ func (a *MetadataUpdaterToolAdapter) SetResourceURL(resourceType, resourceID, ur type UpdatesCommandRunner interface { QueueDockerCheckUpdatesCommand(hostID string) (models.DockerHostCommandStatus, error) QueueDockerContainerUpdateCommand(hostID, containerID, containerName string) (models.DockerHostCommandStatus, error) + GetDockerCommandStatus(commandID string) (models.DockerHostCommandStatus, bool) } // UpdatesConfig provides configuration for update operations @@ -849,6 +850,26 @@ func (a *UpdatesToolAdapter) UpdateContainer(hostID, containerID, containerName }, nil } +// GetCommandStatus implements UpdatesProvider +func (a *UpdatesToolAdapter) GetCommandStatus(commandID string) (DockerCommandStatus, bool) { + if a.commands == nil { + return DockerCommandStatus{}, false + } + + cmdStatus, ok := a.commands.GetDockerCommandStatus(commandID) + if !ok { + return DockerCommandStatus{}, false + } + + return DockerCommandStatus{ + ID: cmdStatus.ID, + Type: cmdStatus.Type, + Status: cmdStatus.Status, + Message: cmdStatus.Message, + FailureReason: cmdStatus.FailureReason, + }, true +} + // IsUpdateActionsEnabled implements UpdatesProvider func (a *UpdatesToolAdapter) IsUpdateActionsEnabled() bool { if a.config == nil { diff --git a/internal/ai/tools/adapters_test.go b/internal/ai/tools/adapters_test.go index 6a4f0c1ff..5b6651550 100644 --- a/internal/ai/tools/adapters_test.go +++ b/internal/ai/tools/adapters_test.go @@ -112,10 +112,11 @@ func (f *fakeMetadataUpdater) SetResourceURL(resourceType, resourceID, url strin } type fakeUpdatesCommandRunner struct { - checkStatus models.DockerHostCommandStatus - updateStatus models.DockerHostCommandStatus - checkErr error - updateErr error + checkStatus models.DockerHostCommandStatus + updateStatus models.DockerHostCommandStatus + checkErr error + updateErr error + commandStatus map[string]models.DockerHostCommandStatus } func (f *fakeUpdatesCommandRunner) QueueDockerCheckUpdatesCommand(_ string) (models.DockerHostCommandStatus, error) { @@ -126,6 +127,11 @@ func (f *fakeUpdatesCommandRunner) QueueDockerContainerUpdateCommand(_ string, _ return f.updateStatus, f.updateErr } +func (f *fakeUpdatesCommandRunner) GetDockerCommandStatus(commandID string) (models.DockerHostCommandStatus, bool) { + status, ok := f.commandStatus[commandID] + return status, ok +} + type fakeUpdatesConfig struct { enabled bool } @@ -538,6 +544,23 @@ func TestUpdatesToolAdapter(t *testing.T) { t.Fatalf("unexpected update status: %+v err=%v", status, err) } + if _, ok := adapter.GetCommandStatus("cmd2"); ok { + t.Fatal("expected command status miss when runner has no record") + } + runner.commandStatus = map[string]models.DockerHostCommandStatus{ + "cmd2": {ID: "cmd2", Type: "update", Status: "failed", Message: "update failed", FailureReason: "container crashed"}, + } + cmdStatus, ok := adapter.GetCommandStatus("cmd2") + if !ok { + t.Fatal("expected command status hit") + } + if cmdStatus.Status != "failed" || cmdStatus.FailureReason != "container crashed" { + t.Fatalf("unexpected command status mapping: %+v", cmdStatus) + } + if _, ok := (&UpdatesToolAdapter{}).GetCommandStatus("cmd2"); ok { + t.Fatal("expected command status miss when commands runner missing") + } + if trimContainerName("/redis") != "redis" || trimContainerName("plain") != "plain" { t.Fatal("unexpected trim result") } diff --git a/internal/ai/tools/data_types.go b/internal/ai/tools/data_types.go index 5c9889940..fd311eec6 100644 --- a/internal/ai/tools/data_types.go +++ b/internal/ai/tools/data_types.go @@ -1298,10 +1298,11 @@ type ContainerUpdateInfo struct { // DockerCommandStatus represents the status of a queued Docker command type DockerCommandStatus struct { - ID string `json:"id"` - Type string `json:"type"` - Status string `json:"status"` - Message string `json:"message"` + ID string `json:"id"` + Type string `json:"type"` + Status string `json:"status"` + Message string `json:"message"` + FailureReason string `json:"failure_reason,omitempty"` } // DockerUpdatesResponse is returned by pulse_list_docker_updates @@ -1334,13 +1335,14 @@ type DockerCheckUpdatesResponse struct { // DockerUpdateContainerResponse is returned by pulse_update_docker_container type DockerUpdateContainerResponse struct { - Success bool `json:"success"` - TargetID string `json:"target_id"` - ContainerID string `json:"container_id"` - ContainerName string `json:"container_name"` - CommandID string `json:"command_id"` - Message string `json:"message"` - Command DockerCommandStatus `json:"command"` + Success bool `json:"success"` + TargetID string `json:"target_id"` + ContainerID string `json:"container_id"` + ContainerName string `json:"container_name"` + CommandID string `json:"command_id"` + Message string `json:"message"` + Command DockerCommandStatus `json:"command"` + Verification map[string]interface{} `json:"verification,omitempty"` } // ========== Kubernetes Types ========== diff --git a/internal/ai/tools/docker_updates_test.go b/internal/ai/tools/docker_updates_test.go index 27fbee8a7..c775196dc 100644 --- a/internal/ai/tools/docker_updates_test.go +++ b/internal/ai/tools/docker_updates_test.go @@ -135,6 +135,9 @@ func TestExecuteUpdateDockerContainer_RetriesTransientError(t *testing.T) { origSleep := dockerUpdateQueueSleepFn dockerUpdateQueueSleepFn = func(context.Context, time.Duration) error { return nil } t.Cleanup(func() { dockerUpdateQueueSleepFn = origSleep }) + origVerifySleep := dockerUpdateVerifySleepFn + dockerUpdateVerifySleepFn = func(context.Context, time.Duration) error { return nil } + t.Cleanup(func() { dockerUpdateVerifySleepFn = origVerifySleep }) state := models.StateSnapshot{ DockerHosts: []models.DockerHost{ @@ -157,6 +160,12 @@ func TestExecuteUpdateDockerContainer_RetriesTransientError(t *testing.T) { Type: "update", Status: "queued", }, nil).Once() + updatesProvider.On("GetCommandStatus", "cmd-update").Return(DockerCommandStatus{ + ID: "cmd-update", + Type: "update", + Status: "completed", + Message: "Container nginx updated successfully", + }, true) exec := NewPulseToolExecutor(ExecutorConfig{ UpdatesProvider: updatesProvider, @@ -176,3 +185,132 @@ func TestExecuteUpdateDockerContainer_RetriesTransientError(t *testing.T) { assert.Equal(t, "cmd-update", resp.CommandID) updatesProvider.AssertExpectations(t) } + +func newDockerUpdateExecutor(t *testing.T, updatesProvider *mockUpdatesProvider) *PulseToolExecutor { + t.Helper() + + origVerifySleep := dockerUpdateVerifySleepFn + dockerUpdateVerifySleepFn = func(context.Context, time.Duration) error { return nil } + t.Cleanup(func() { dockerUpdateVerifySleepFn = origVerifySleep }) + + state := models.StateSnapshot{ + DockerHosts: []models.DockerHost{ + { + ID: "host1", + Hostname: "dock1", + DisplayName: "Dock One", + Containers: []models.DockerContainer{ + {ID: "c1", Name: "/nginx"}, + }, + }, + }, + } + + return NewPulseToolExecutor(ExecutorConfig{ + UpdatesProvider: updatesProvider, + StateProvider: &mockStateProvider{state: state}, + ControlLevel: ControlLevelAutonomous, + }) +} + +func TestExecuteUpdateDockerContainer_VerifiedSuccess(t *testing.T) { + updatesProvider := &mockUpdatesProvider{} + updatesProvider.On("IsUpdateActionsEnabled").Return(true).Once() + updatesProvider.On("UpdateContainer", "host1", "c1", "nginx").Return(DockerCommandStatus{ + ID: "cmd-update", + Type: "update_container", + Status: "queued", + }, nil).Once() + // The command is dispatched and in progress first, then completes. + updatesProvider.On("GetCommandStatus", "cmd-update").Return(DockerCommandStatus{ + ID: "cmd-update", Status: "dispatched", + }, true).Once() + updatesProvider.On("GetCommandStatus", "cmd-update").Return(DockerCommandStatus{ + ID: "cmd-update", Status: "in_progress", Message: "Pulling image nginx...", + }, true).Once() + updatesProvider.On("GetCommandStatus", "cmd-update").Return(DockerCommandStatus{ + ID: "cmd-update", Status: "completed", Message: "Container nginx updated successfully", + }, true).Once() + + exec := newDockerUpdateExecutor(t, updatesProvider) + result, err := exec.executeUpdateDockerContainer(context.Background(), map[string]interface{}{ + "host": "dock1", + "container": "c1", + }) + require.NoError(t, err) + require.False(t, result.IsError) + + var resp DockerUpdateContainerResponse + require.NoError(t, json.Unmarshal([]byte(result.Content[0].Text), &resp)) + assert.True(t, resp.Success) + assert.Contains(t, resp.Message, "Update verified") + require.NotNil(t, resp.Verification) + assert.Equal(t, true, resp.Verification["confirmed"]) + assert.Equal(t, "verified", resp.Verification["outcome"]) + assert.Equal(t, "command_status", resp.Verification["method"]) + updatesProvider.AssertExpectations(t) +} + +func TestExecuteUpdateDockerContainer_VerifiedFailure(t *testing.T) { + updatesProvider := &mockUpdatesProvider{} + updatesProvider.On("IsUpdateActionsEnabled").Return(true).Once() + updatesProvider.On("UpdateContainer", "host1", "c1", "nginx").Return(DockerCommandStatus{ + ID: "cmd-update", + Type: "update_container", + Status: "queued", + }, nil).Once() + updatesProvider.On("GetCommandStatus", "cmd-update").Return(DockerCommandStatus{ + ID: "cmd-update", + Status: "failed", + FailureReason: "New container crashed immediately (exit code 1)", + }, true).Once() + + exec := newDockerUpdateExecutor(t, updatesProvider) + result, err := exec.executeUpdateDockerContainer(context.Background(), map[string]interface{}{ + "host": "dock1", + "container": "c1", + }) + require.NoError(t, err) + assert.True(t, result.IsError) + + var resp DockerUpdateContainerResponse + require.NoError(t, json.Unmarshal([]byte(result.Content[0].Text), &resp)) + assert.False(t, resp.Success) + assert.Contains(t, resp.Message, "failed") + assert.Contains(t, resp.Message, "crashed immediately") + require.NotNil(t, resp.Verification) + assert.Equal(t, false, resp.Verification["confirmed"]) + assert.Equal(t, "failed", resp.Verification["outcome"]) + updatesProvider.AssertExpectations(t) +} + +func TestExecuteUpdateDockerContainer_InconclusiveAfterWindow(t *testing.T) { + updatesProvider := &mockUpdatesProvider{} + updatesProvider.On("IsUpdateActionsEnabled").Return(true).Once() + updatesProvider.On("UpdateContainer", "host1", "c1", "nginx").Return(DockerCommandStatus{ + ID: "cmd-update", + Type: "update_container", + Status: "queued", + }, nil).Once() + // The update never reaches a terminal state within the window. + updatesProvider.On("GetCommandStatus", "cmd-update").Return(DockerCommandStatus{ + ID: "cmd-update", Status: "in_progress", Message: "Pulling image nginx...", + }, true) + + exec := newDockerUpdateExecutor(t, updatesProvider) + result, err := exec.executeUpdateDockerContainer(context.Background(), map[string]interface{}{ + "host": "dock1", + "container": "c1", + }) + require.NoError(t, err) + require.False(t, result.IsError) + + var resp DockerUpdateContainerResponse + require.NoError(t, json.Unmarshal([]byte(result.Content[0].Text), &resp)) + assert.True(t, resp.Success) // queueing succeeded; completion is explicitly unverified + assert.Contains(t, resp.Message, "NOT yet verified") + require.NotNil(t, resp.Verification) + assert.Equal(t, false, resp.Verification["confirmed"]) + assert.Equal(t, "inconclusive", resp.Verification["outcome"]) + updatesProvider.AssertNumberOfCalls(t, "GetCommandStatus", dockerUpdateVerifyMaxAttempts) +} diff --git a/internal/ai/tools/executor.go b/internal/ai/tools/executor.go index 3cf6f8d92..5e4402d70 100644 --- a/internal/ai/tools/executor.go +++ b/internal/ai/tools/executor.go @@ -206,6 +206,9 @@ type UpdatesProvider interface { GetPendingUpdates(hostID string) []ContainerUpdateInfo TriggerUpdateCheck(hostID string) (DockerCommandStatus, error) UpdateContainer(hostID, containerID, containerName string) (DockerCommandStatus, error) + // GetCommandStatus reports the current status of a previously queued + // update command so callers can verify the command actually applied. + GetCommandStatus(commandID string) (DockerCommandStatus, bool) IsUpdateActionsEnabled() bool } diff --git a/internal/ai/tools/kubernetes_control_test.go b/internal/ai/tools/kubernetes_control_test.go index b3cc154b7..a8b251549 100644 --- a/internal/ai/tools/kubernetes_control_test.go +++ b/internal/ai/tools/kubernetes_control_test.go @@ -2,7 +2,9 @@ package tools import ( "context" + "encoding/json" "testing" + "time" "github.com/rcourtman/pulse-go-rewrite/internal/agentcapabilities" "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" @@ -18,7 +20,13 @@ func newConfiguredKubernetesExecutor(snapshot models.StateSnapshot, mutate func( adapter := unifiedresources.NewMonitorAdapter(nil) adapter.PopulateFromSnapshot(snapshot) - cfg := ExecutorConfig{UnifiedResourceProvider: adapter} + // Autonomous dispatches fail closed when no audit store is wired + // (unknown remediation-lock state), so these routing/control tests + // always get an in-memory store. + cfg := ExecutorConfig{ + UnifiedResourceProvider: adapter, + ActionAuditStore: unifiedresources.NewMemoryStore(), + } if mutate != nil { mutate(&cfg) } @@ -233,7 +241,7 @@ func TestExecuteKubernetesScale(t *testing.T) { assert.Contains(t, result.Content[0].Text, "scale") }) - t.Run("ExecuteSuccess", func(t *testing.T) { + t.Run("ExecuteSuccessVerified", func(t *testing.T) { mockAgent := &mockAgentServer{ agents: []agentexec.ConnectedAgent{{AgentID: "agent-1", Hostname: "k8s-host"}}, } @@ -244,6 +252,14 @@ func TestExecuteKubernetesScale(t *testing.T) { ExitCode: 0, Stdout: "deployment.apps/nginx scaled", }, nil) + // Read-after-write: the scale handler re-reads the deployment state. + mockAgent.On("ExecuteCommand", mock.Anything, "agent-1", mock.MatchedBy(func(cmd agentexec.ExecuteCommandPayload) bool { + return cmd.Command == "kubectl -n 'default' get deployment 'nginx' -o 'jsonpath={.spec.replicas} {.status.readyReplicas}'" && + cmd.TargetType == "agent" + })).Return(&agentexec.CommandResultPayload{ + ExitCode: 0, + Stdout: "3 3", + }, nil) state := models.StateSnapshot{ KubernetesClusters: []models.KubernetesCluster{ @@ -260,8 +276,140 @@ func TestExecuteKubernetesScale(t *testing.T) { "replicas": 3, }) require.NoError(t, err) - assert.Contains(t, result.Content[0].Text, "Successfully scaled") - assert.Contains(t, result.Content[0].Text, "nginx") + require.False(t, result.IsError) + + var resp map[string]interface{} + require.NoError(t, json.Unmarshal([]byte(result.Content[0].Text), &resp)) + assert.Equal(t, true, resp["success"]) + assert.Equal(t, "nginx", resp["deployment"]) + verification, ok := resp["verification"].(map[string]interface{}) + require.True(t, ok, "expected verification in response") + assert.Equal(t, true, verification["confirmed"]) + assert.Equal(t, "kubectl_get", verification["method"]) + observed, ok := verification["observed"].(map[string]interface{}) + require.True(t, ok, "expected observed state in verification") + assert.Equal(t, float64(3), observed["replicas"]) + assert.Equal(t, float64(3), observed["ready_replicas"]) + mockAgent.AssertExpectations(t) + }) + + t.Run("ExecuteSuccessVerificationMismatch", func(t *testing.T) { + origSleep := kubernetesVerifySleepFn + kubernetesVerifySleepFn = func(context.Context, time.Duration) error { return nil } + t.Cleanup(func() { kubernetesVerifySleepFn = origSleep }) + + mockAgent := &mockAgentServer{ + agents: []agentexec.ConnectedAgent{{AgentID: "agent-1", Hostname: "k8s-host"}}, + } + mockAgent.On("ExecuteCommand", mock.Anything, "agent-1", mock.MatchedBy(func(cmd agentexec.ExecuteCommandPayload) bool { + return cmd.Command == "kubectl -n 'default' scale deployment 'nginx' --replicas=3" + })).Return(&agentexec.CommandResultPayload{ + ExitCode: 0, + Stdout: "deployment.apps/nginx scaled", + }, nil).Once() + // The re-read keeps observing a different desired replica count. + mockAgent.On("ExecuteCommand", mock.Anything, "agent-1", mock.MatchedBy(func(cmd agentexec.ExecuteCommandPayload) bool { + return cmd.Command == "kubectl -n 'default' get deployment 'nginx' -o 'jsonpath={.spec.replicas} {.status.readyReplicas}'" + })).Return(&agentexec.CommandResultPayload{ + ExitCode: 0, + Stdout: "1 1", + }, nil).Times(kubernetesVerifyMaxAttempts) + + state := models.StateSnapshot{ + KubernetesClusters: []models.KubernetesCluster{ + {ID: "c1", Name: "cluster-1", AgentID: "agent-1"}, + }, + } + exec := newConfiguredKubernetesExecutor(state, func(cfg *ExecutorConfig) { + cfg.AgentServer = mockAgent + cfg.ControlLevel = ControlLevelAutonomous + }) + result, err := exec.executeKubernetesScale(ctx, map[string]interface{}{ + "cluster": "cluster-1", + "deployment": "nginx", + "replicas": 3, + }) + require.NoError(t, err) + + var resp map[string]interface{} + require.NoError(t, json.Unmarshal([]byte(result.Content[0].Text), &resp)) + verification, ok := resp["verification"].(map[string]interface{}) + require.True(t, ok, "expected verification in response") + assert.Equal(t, false, verification["confirmed"]) + observed, ok := verification["observed"].(map[string]interface{}) + require.True(t, ok, "expected observed state in verification") + assert.Equal(t, float64(1), observed["replicas"]) + mockAgent.AssertExpectations(t) + }) + + t.Run("ExecuteSuccessVerificationReadError", func(t *testing.T) { + mockAgent := &mockAgentServer{ + agents: []agentexec.ConnectedAgent{{AgentID: "agent-1", Hostname: "k8s-host"}}, + } + mockAgent.On("ExecuteCommand", mock.Anything, "agent-1", mock.MatchedBy(func(cmd agentexec.ExecuteCommandPayload) bool { + return cmd.Command == "kubectl -n 'default' scale deployment 'nginx' --replicas=3" + })).Return(&agentexec.CommandResultPayload{ + ExitCode: 0, + Stdout: "deployment.apps/nginx scaled", + }, nil).Once() + mockAgent.On("ExecuteCommand", mock.Anything, "agent-1", mock.MatchedBy(func(cmd agentexec.ExecuteCommandPayload) bool { + return cmd.Command == "kubectl -n 'default' get deployment 'nginx' -o 'jsonpath={.spec.replicas} {.status.readyReplicas}'" + })).Return((*agentexec.CommandResultPayload)(nil), assert.AnError).Once() + + state := models.StateSnapshot{ + KubernetesClusters: []models.KubernetesCluster{ + {ID: "c1", Name: "cluster-1", AgentID: "agent-1"}, + }, + } + exec := newConfiguredKubernetesExecutor(state, func(cfg *ExecutorConfig) { + cfg.AgentServer = mockAgent + cfg.ControlLevel = ControlLevelAutonomous + }) + result, err := exec.executeKubernetesScale(ctx, map[string]interface{}{ + "cluster": "cluster-1", + "deployment": "nginx", + "replicas": 3, + }) + require.NoError(t, err) + + var resp map[string]interface{} + require.NoError(t, json.Unmarshal([]byte(result.Content[0].Text), &resp)) + // The scale itself succeeded; verification must report unconfirmed, not error out. + assert.Equal(t, true, resp["success"]) + verification, ok := resp["verification"].(map[string]interface{}) + require.True(t, ok, "expected verification in response") + assert.Equal(t, false, verification["confirmed"]) + assert.NotEmpty(t, verification["note"]) + mockAgent.AssertExpectations(t) + }) + + t.Run("ExecuteFailureNoVerification", func(t *testing.T) { + mockAgent := &mockAgentServer{ + agents: []agentexec.ConnectedAgent{{AgentID: "agent-1", Hostname: "k8s-host"}}, + } + mockAgent.On("ExecuteCommand", mock.Anything, "agent-1", mock.MatchedBy(func(cmd agentexec.ExecuteCommandPayload) bool { + return cmd.Command == "kubectl -n 'default' scale deployment 'nginx' --replicas=3" + })).Return(&agentexec.CommandResultPayload{ + ExitCode: 1, + Stderr: "Error from server (NotFound): deployments.apps \"nginx\" not found", + }, nil).Once() + + state := models.StateSnapshot{ + KubernetesClusters: []models.KubernetesCluster{ + {ID: "c1", Name: "cluster-1", AgentID: "agent-1"}, + }, + } + exec := newConfiguredKubernetesExecutor(state, func(cfg *ExecutorConfig) { + cfg.AgentServer = mockAgent + cfg.ControlLevel = ControlLevelAutonomous + }) + result, err := exec.executeKubernetesScale(ctx, map[string]interface{}{ + "cluster": "cluster-1", + "deployment": "nginx", + "replicas": 3, + }) + require.NoError(t, err) + assert.Contains(t, result.Content[0].Text, "kubectl command failed") mockAgent.AssertExpectations(t) }) } diff --git a/internal/ai/tools/shared_test.go b/internal/ai/tools/shared_test.go index 60843e485..1f4870e13 100644 --- a/internal/ai/tools/shared_test.go +++ b/internal/ai/tools/shared_test.go @@ -134,6 +134,11 @@ func (m *mockUpdatesProvider) UpdateContainer(hostID, containerID, containerName return args.Get(0).(DockerCommandStatus), args.Error(1) } +func (m *mockUpdatesProvider) GetCommandStatus(commandID string) (DockerCommandStatus, bool) { + args := m.Called(commandID) + return args.Get(0).(DockerCommandStatus), args.Bool(1) +} + func (m *mockUpdatesProvider) IsUpdateActionsEnabled() bool { args := m.Called() return args.Bool(0) diff --git a/internal/ai/tools/tools_docker.go b/internal/ai/tools/tools_docker.go index ec9bf2df7..6d63ba613 100644 --- a/internal/ai/tools/tools_docker.go +++ b/internal/ai/tools/tools_docker.go @@ -22,9 +22,31 @@ const ( // dockerUpdateQueueRetryMaxDelay caps exponential backoff for queue retries. dockerUpdateQueueRetryMaxDelay = 250 * time.Millisecond + + // dockerUpdateVerifyMaxAttempts bounds post-queue verification polling of the + // update command status. Combined with dockerUpdateVerifyPollInterval this + // gives the agent one report cycle (~30s) to pick the command up plus time + // for small-image updates to finish; slower updates report as inconclusive. + dockerUpdateVerifyMaxAttempts = 30 + + // dockerUpdateVerifyPollInterval is the delay between verification polls. + dockerUpdateVerifyPollInterval = 2 * time.Second ) -var dockerUpdateQueueSleepFn = sleepWithContext +// Terminal docker host command statuses, mirroring the DockerCommandStatus* +// constants in internal/monitoring. Duplicated as literals because the tools +// package sits below monitoring in the wiring (monitoring satisfies the +// UpdatesCommandRunner interface defined here). +const ( + dockerCommandStatusCompleted = "completed" + dockerCommandStatusFailed = "failed" + dockerCommandStatusExpired = "expired" +) + +var ( + dockerUpdateQueueSleepFn = sleepWithContext + dockerUpdateVerifySleepFn = sleepWithContext +) // registerDockerTools registers the pulse_docker tool func (e *PulseToolExecutor) registerDockerTools() { @@ -528,17 +550,116 @@ func (e *PulseToolExecutor) executeUpdateDockerContainer(ctx context.Context, ar return NewTextResult(fmt.Sprintf("Failed to queue update command: %v", err)), nil } + // The agent executes the update asynchronously: it pulls the image, + // recreates the container, verifies the replacement is running and healthy + // (rolling back otherwise), then acknowledges a terminal command status. + // Poll that status so we report verified success, verified failure, or an + // honest inconclusive - never unverified success. + verify := e.verifyDockerUpdateCommand(ctx, cmdStatus.ID, containerName) + outcome, _ := verify["outcome"].(string) + + var message string + switch outcome { + case "verified": + message = fmt.Sprintf("Update verified: container '%s' was updated, recreated, and is running again.", containerName) + case "failed": + reason := "" + if observed, ok := verify["observed"].(map[string]interface{}); ok { + if fr, ok := observed["failure_reason"].(string); ok && fr != "" { + reason = fr + } else if msg, ok := observed["message"].(string); ok { + reason = msg + } + } + if reason == "" { + reason = "the agent reported the update command as failed" + } + message = fmt.Sprintf("Update of container '%s' failed: %s", containerName, reason) + default: + message = fmt.Sprintf("Update command queued for container '%s', but it had not reached a terminal state within the verification window. Completion is NOT yet verified - re-check with pulse_docker action 'updates' or inspect the container before reporting success.", containerName) + } + response := DockerUpdateContainerResponse{ - Success: true, + Success: outcome != "failed", TargetID: dockerHost.ID, ContainerID: container.ID, ContainerName: containerName, CommandID: cmdStatus.ID, - Message: fmt.Sprintf("Update command queued for container '%s'. The agent will pull the latest image and recreate the container.", containerName), + Message: message, Command: cmdStatus, + Verification: verify, } - return NewJSONResult(response), nil + return NewJSONResultWithIsError(response, outcome == "failed"), nil +} + +// verifyDockerUpdateCommand polls the queued update command's status until it +// reaches a terminal state or the bounded verification window elapses. The +// returned map follows the same verification shape as +// verifyDockerContainerState / verifyGuestAction: confirmed only on observed +// terminal success, with an explicit outcome of "verified", "failed", or +// "inconclusive". +func (e *PulseToolExecutor) verifyDockerUpdateCommand(ctx context.Context, commandID, containerName string) map[string]interface{} { + verification := map[string]interface{}{ + "confirmed": false, + "method": "command_status", + "command_id": commandID, + } + + if commandID == "" { + verification["outcome"] = "inconclusive" + verification["note"] = "queue returned no command ID, so the update outcome cannot be tracked" + return verification + } + + var lastObserved map[string]interface{} + + for attempt := 1; attempt <= dockerUpdateVerifyMaxAttempts; attempt++ { + status, ok := e.updatesProvider.GetCommandStatus(commandID) + if ok { + lastObserved = map[string]interface{}{"status": status.Status} + if status.Message != "" { + lastObserved["message"] = status.Message + } + if status.FailureReason != "" { + lastObserved["failure_reason"] = status.FailureReason + } + + switch status.Status { + case dockerCommandStatusCompleted: + verification["confirmed"] = true + verification["outcome"] = "verified" + verification["observed"] = lastObserved + verification["note"] = fmt.Sprintf("agent confirmed container '%s' was recreated and is running", containerName) + return verification + case dockerCommandStatusFailed, dockerCommandStatusExpired: + verification["outcome"] = "failed" + verification["observed"] = lastObserved + return verification + } + } + + if attempt == dockerUpdateVerifyMaxAttempts { + break + } + if err := dockerUpdateVerifySleepFn(ctx, dockerUpdateVerifyPollInterval); err != nil { + verification["outcome"] = "inconclusive" + verification["note"] = "verification canceled before the update reached a terminal state" + if lastObserved != nil { + verification["observed"] = lastObserved + } + return verification + } + } + + verification["outcome"] = "inconclusive" + if lastObserved != nil { + verification["observed"] = lastObserved + verification["note"] = "update command did not reach a terminal state within the verification window; completion is unverified" + } else { + verification["note"] = "update command status was not trackable within the verification window; completion is unverified" + } + return verification } // Helper methods for Docker updates diff --git a/internal/ai/tools/tools_kubernetes.go b/internal/ai/tools/tools_kubernetes.go index 74915b561..ac4197b50 100644 --- a/internal/ai/tools/tools_kubernetes.go +++ b/internal/ai/tools/tools_kubernetes.go @@ -3,13 +3,27 @@ package tools import ( "context" "fmt" + "strconv" "strings" + "time" "github.com/rcourtman/pulse-go-rewrite/internal/agentcapabilities" "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" ) +const ( + // kubernetesVerifyMaxAttempts bounds read-after-write verification reads + // after a kubectl control action, mirroring verifyDockerContainerState. + kubernetesVerifyMaxAttempts = 3 + + // kubernetesVerifySettleDelay is the pause between verification attempts, + // letting the API server settle before re-reading. + kubernetesVerifySettleDelay = 500 * time.Millisecond +) + +var kubernetesVerifySleepFn = sleepWithContext + type kubernetesClusterTarget struct { ID string AgentID string @@ -591,11 +605,102 @@ func (e *PulseToolExecutor) executeKubernetesScale(ctx context.Context, args map output += "\n" + result.Stderr } - if result.ExitCode == 0 { - return NewTextResult(fmt.Sprintf("✓ Successfully scaled deployment '%s' to %d replicas in namespace '%s'. Action complete - no verification needed.\n%s", deployment, replicas, namespace, output)), nil + if result.ExitCode != 0 { + return NewTextResult(fmt.Sprintf("kubectl command failed (exit code %d):\n%s", result.ExitCode, output)), nil } - return NewTextResult(fmt.Sprintf("kubectl command failed (exit code %d):\n%s", result.ExitCode, output)), nil + // Read-after-write: re-read the deployment's replica state through the + // same agent instead of asserting unverified success. + verify := e.verifyKubernetesScale(ctx, agentID, namespace, deployment, replicas) + verify["ok"] = true + + response := map[string]interface{}{ + "success": true, + "action": "scale", + "cluster": cluster.DisplayName, + "namespace": namespace, + "deployment": deployment, + "requested_replicas": replicas, + "command": command, + "exit_code": result.ExitCode, + "output": output, + "verification": verify, + } + return NewJSONResult(response), nil +} + +// verifyKubernetesScale re-reads the deployment's replica state after a scale +// command, mirroring the read-after-write idiom of verifyGuestAction and +// verifyDockerContainerState. Confirmed means the desired replica count is +// persisted on the deployment spec; readiness is reported observationally +// since pods may still be starting or terminating. +func (e *PulseToolExecutor) verifyKubernetesScale(ctx context.Context, agentID, namespace, deployment string, expectedReplicas int) map[string]interface{} { + verifyCmd := fmt.Sprintf("kubectl -n %s get deployment %s -o %s", + shellEscape(namespace), shellEscape(deployment), shellEscape("jsonpath={.spec.replicas} {.status.readyReplicas}")) + + var lastOut string + var lastExit int + for attempt := 1; attempt <= kubernetesVerifyMaxAttempts; attempt++ { + res, err := e.agentServer.ExecuteCommand(ctx, agentID, agentexec.ExecuteCommandPayload{ + Command: verifyCmd, + TargetType: "agent", + }) + if err != nil { + return map[string]interface{}{"confirmed": false, "method": "kubectl_get", "command": verifyCmd, "note": err.Error()} + } + lastExit = res.ExitCode + lastOut = strings.TrimSpace(res.Stdout + "\n" + res.Stderr) + + if res.ExitCode == 0 { + fields := strings.Fields(lastOut) + if len(fields) >= 1 { + observedReplicas, parseErr := strconv.Atoi(fields[0]) + if parseErr == nil { + readyReplicas := 0 + if len(fields) >= 2 { + if ready, readyErr := strconv.Atoi(fields[1]); readyErr == nil { + readyReplicas = ready + } + } + observed := map[string]interface{}{"replicas": observedReplicas, "ready_replicas": readyReplicas} + if observedReplicas == expectedReplicas { + return map[string]interface{}{ + "confirmed": true, + "method": "kubectl_get", + "command": verifyCmd, + "expected": map[string]interface{}{"replicas": expectedReplicas}, + "observed": observed, + } + } + if attempt == kubernetesVerifyMaxAttempts { + return map[string]interface{}{ + "confirmed": false, + "method": "kubectl_get", + "command": verifyCmd, + "expected": map[string]interface{}{"replicas": expectedReplicas}, + "observed": observed, + } + } + } + } + } + + if attempt == kubernetesVerifyMaxAttempts { + break + } + if err := kubernetesVerifySleepFn(ctx, kubernetesVerifySettleDelay); err != nil { + return map[string]interface{}{"confirmed": false, "method": "kubectl_get", "command": verifyCmd, "note": "context canceled", "raw": lastOut, "exit_code": lastExit} + } + } + + return map[string]interface{}{ + "confirmed": false, + "method": "kubectl_get", + "command": verifyCmd, + "expected": map[string]interface{}{"replicas": expectedReplicas}, + "raw": lastOut, + "exit_code": lastExit, + } } // kubernetesResourceAction captures how the namespaced kubectl actions diff --git a/internal/ai/tools/tools_storage_test.go b/internal/ai/tools/tools_storage_test.go index 67e22ed43..f1f0cbfda 100644 --- a/internal/ai/tools/tools_storage_test.go +++ b/internal/ai/tools/tools_storage_test.go @@ -61,6 +61,13 @@ func (s *stubUpdatesProvider) UpdateContainer(hostID, containerID, containerName return s.updateStatus, s.updateErr } +func (s *stubUpdatesProvider) GetCommandStatus(commandID string) (DockerCommandStatus, bool) { + if s.updateStatus.ID == commandID { + return s.updateStatus, true + } + return DockerCommandStatus{}, false +} + func (s *stubUpdatesProvider) IsUpdateActionsEnabled() bool { return s.enabled } diff --git a/internal/monitoring/docker_commands.go b/internal/monitoring/docker_commands.go index 856a02fb2..60a622870 100644 --- a/internal/monitoring/docker_commands.go +++ b/internal/monitoring/docker_commands.go @@ -366,6 +366,36 @@ func (m *Monitor) QueueDockerCheckUpdatesCommand(hostID string) (models.DockerHo return cmd.status, nil } +// GetDockerCommandStatus returns the current status of a docker host command +// by ID. Active commands are looked up directly; terminal commands +// (completed/failed) are cleared from the active map on acknowledgement but +// persist as the host's last command in state, so those are found there. +func (m *Monitor) GetDockerCommandStatus(commandID string) (models.DockerHostCommandStatus, bool) { + m.mu.Lock() + defer m.mu.Unlock() + + commandID = strings.TrimSpace(commandID) + if commandID == "" { + return models.DockerHostCommandStatus{}, false + } + + if hostID, ok := m.dockerCommandIndex[commandID]; ok { + if cmd, ok := m.dockerCommands[hostID]; ok && cmd.status.ID == commandID { + return cmd.status, true + } + } + + if m.state != nil { + for _, host := range m.state.GetDockerHosts() { + if host.Command != nil && host.Command.ID == commandID { + return *host.Command, true + } + } + } + + return models.DockerHostCommandStatus{}, false +} + func (m *Monitor) getDockerCommandPayload(hostID string) (map[string]any, *models.DockerHostCommandStatus) { m.mu.Lock() defer m.mu.Unlock() diff --git a/internal/monitoring/docker_commands_test.go b/internal/monitoring/docker_commands_test.go index 355823118..3fab59677 100644 --- a/internal/monitoring/docker_commands_test.go +++ b/internal/monitoring/docker_commands_test.go @@ -1215,6 +1215,53 @@ func TestQueueDockerContainerUpdateCommand(t *testing.T) { } } +func TestGetDockerCommandStatus(t *testing.T) { + t.Parallel() + + monitor := newTestMonitorForCommands(t) + host := models.DockerHost{ + ID: "host-status", + Hostname: "node-status", + Status: "online", + } + monitor.state.UpsertDockerHost(host) + + if _, ok := monitor.GetDockerCommandStatus("nonexistent"); ok { + t.Fatal("expected lookup miss for unknown command ID") + } + if _, ok := monitor.GetDockerCommandStatus(""); ok { + t.Fatal("expected lookup miss for empty command ID") + } + + cmdStatus, err := monitor.QueueDockerContainerUpdateCommand(host.ID, "container-1", "nginx") + if err != nil { + t.Fatalf("Failed to queue update command: %v", err) + } + + // Active command is found in the in-memory command map. + got, ok := monitor.GetDockerCommandStatus(cmdStatus.ID) + if !ok { + t.Fatal("expected active command to be found") + } + if got.Status != DockerCommandStatusQueued { + t.Fatalf("expected status %q, got %q", DockerCommandStatusQueued, got.Status) + } + + // Terminal acknowledgement clears the active map but persists the last + // command on host state; the lookup must still resolve it there. + if _, _, _, err := monitor.AcknowledgeDockerHostCommand(cmdStatus.ID, host.ID, DockerCommandStatusCompleted, "Container nginx updated successfully"); err != nil { + t.Fatalf("Failed to acknowledge command: %v", err) + } + + got, ok = monitor.GetDockerCommandStatus(cmdStatus.ID) + if !ok { + t.Fatal("expected completed command to be found via host state") + } + if got.Status != DockerCommandStatusCompleted { + t.Fatalf("expected status %q, got %q", DockerCommandStatusCompleted, got.Status) + } +} + func TestQueueDockerContainerUpdateCommand_BlockedBySecurityPosture(t *testing.T) { t.Parallel()