From 8439ce6e6bafbb09a93b7f8db8e6786481a8b15c Mon Sep 17 00:00:00 2001 From: rcourtman Date: Wed, 10 Jun 2026 09:38:52 +0100 Subject: [PATCH] Dedupe AI runtime guest-family, tool-pipeline, and provider clones Clears the ten dupl pairs across internal/ai: - patrol_intelligence.go, tools_query.go, tools_storage.go: VM and LXC system-container paths collapse into generics over read-state view method subsets (gatherGuestIntelligenceFromViews, canonicalGuestGetResult + guestViewGetResult, addCanonicalGuestSearchMatches + addGuestViewSearchMatches, appendGuestDiskSummaries). - tools_file.go: append/write share executeFileMutation driven by fileMutationSpec (approval-command text, shell redirect, verification strategy stay per-action and verbatim). - tools_kubernetes.go: deployment restart / pod delete share executeKubernetesResourceAction driven by kubernetesResourceAction. - providers/anthropic.go + anthropic_oauth.go: message conversion shared via convertMessagesToAnthropic (the OAuth copy was annotated 'same as regular client'). - memory/changes.go + memory/remediation.go: history loading shared via the generic loadMemoryHistory in memory/paths.go (10 MiB cap, sort, missing-file semantics preserved via a found flag). - findings.go and unified/alerts.go: Finding/findingJSON and UnifiedFinding/unifiedFindingJSON are deliberate marshal-mirror twins (AlertIdentifier json:"-" vs alert_identifier round-trip); merging would break every public literal. Suppressed with nolint:dupl and enforced instead by new reflect-based mirror-sync tests. Contract Extension Points name the shared helpers and the mirror invariant. Full ./internal/ai/... test tree passes. --- .../v6/internal/subsystems/ai-runtime.md | 26 ++ internal/ai/findings.go | 13 +- internal/ai/findings_marshal_mirror_test.go | 56 +++ internal/ai/memory/changes.go | 37 +- internal/ai/memory/paths.go | 41 ++ internal/ai/memory/remediation.go | 36 +- internal/ai/patrol_intelligence.go | 104 ++--- internal/ai/providers/anthropic.go | 23 +- internal/ai/providers/anthropic_oauth.go | 55 +-- internal/ai/tools/tools_file.go | 231 +++------- internal/ai/tools/tools_kubernetes.go | 154 +++---- internal/ai/tools/tools_query.go | 424 ++++++++---------- internal/ai/tools/tools_storage.go | 240 ++++------ internal/ai/unified/alerts.go | 13 +- internal/ai/unified/alerts_mirror_test.go | 47 ++ 15 files changed, 672 insertions(+), 828 deletions(-) create mode 100644 internal/ai/findings_marshal_mirror_test.go create mode 100644 internal/ai/unified/alerts_mirror_test.go diff --git a/docs/release-control/v6/internal/subsystems/ai-runtime.md b/docs/release-control/v6/internal/subsystems/ai-runtime.md index c2646abb4..71919822c 100644 --- a/docs/release-control/v6/internal/subsystems/ai-runtime.md +++ b/docs/release-control/v6/internal/subsystems/ai-runtime.md @@ -55,6 +55,32 @@ leave the transcript without exposing hidden provider/tool metadata. ## Extension Points +Guest-family AI runtime code paths (VMs and LXC system containers) share +generic helpers instead of per-family copies. Patrol guest intelligence +gathers both families through `gatherGuestIntelligenceFromViews` in +`internal/ai/patrol_intelligence.go`; the Assistant query tool resolves +single-guest gets through `canonicalGuestGetResult` / `guestViewGetResult` +and search matches through `addCanonicalGuestSearchMatches` / +`addGuestViewSearchMatches` in `internal/ai/tools/tools_query.go`; per-guest +disk summaries flow through `appendGuestDiskSummaries` in +`internal/ai/tools/tools_storage.go`. A new guest family extends those +helpers (their view-constraint interfaces name the read-state methods they +need) rather than re-rolling a parallel loop. The same shape applies to +mutating tool pipelines: file append/write share `executeFileMutation` +driven by `fileMutationSpec` (`internal/ai/tools/tools_file.go`), and +namespaced kubectl actions share `executeKubernetesResourceAction` driven by +`kubernetesResourceAction` (`internal/ai/tools/tools_kubernetes.go`) — new +file or kubectl actions add a spec, keeping approval-command text stable, +instead of duplicating the approval/audit pipeline. Anthropic message +conversion is shared by the API-key and OAuth clients through +`convertMessagesToAnthropic` (`internal/ai/providers/anthropic.go`), and AI +memory history files load through the generic `loadMemoryHistory` +(`internal/ai/memory/paths.go`). `Finding`/`findingJSON` and +`UnifiedFinding`/`unifiedFindingJSON` are deliberate marshal-mirror twins — +do not merge them; the mirror invariants are enforced by +`TestFindingJSONMirrorStaysInSync` and +`TestUnifiedFindingJSONMirrorStaysInSync`. + Assistant frontend presentation changes under `frontend-modern/src/components/AI/Chat/` must keep live tool activity aligned with OpenCode's mutable tool-part pattern without losing Pulse's operator audit diff --git a/internal/ai/findings.go b/internal/ai/findings.go index c96af5c2e..4c977d956 100644 --- a/internal/ai/findings.go +++ b/internal/ai/findings.go @@ -150,7 +150,12 @@ type FindingLifecycleEvent struct { const maxFindingLifecycleEvents = 50 -// Finding represents an AI-discovered insight about infrastructure +// Finding represents an AI-discovered insight about infrastructure. +// findingJSON deliberately mirrors this struct field-for-field +// (marshal-mirror pattern); merging them would change the public literal API. +// TestFindingJSONMirrorStaysInSync enforces the mirror. +// +//nolint:dupl // deliberate marshal-mirror twin of findingJSON type Finding struct { ID string `json:"id"` Key string `json:"key,omitempty"` // Stable issue key for runbook matching @@ -215,6 +220,12 @@ type Finding struct { LastRegressionAt *time.Time `json:"last_regression_at,omitempty"` // Timestamp of most recent regression } +// findingJSON is the marshal mirror of Finding: identical fields, but +// AlertIdentifier round-trips as "alert_identifier" in persistence while the +// public struct hides it from generic encoders with `json:"-"`. The mirror is +// enforced by TestFindingJSONMirrorStaysInSync. +// +//nolint:dupl // deliberate marshal-mirror twin of Finding type findingJSON struct { ID string `json:"id"` Key string `json:"key,omitempty"` diff --git a/internal/ai/findings_marshal_mirror_test.go b/internal/ai/findings_marshal_mirror_test.go new file mode 100644 index 000000000..11645fad5 --- /dev/null +++ b/internal/ai/findings_marshal_mirror_test.go @@ -0,0 +1,56 @@ +package ai + +import ( + "reflect" + "testing" +) + +// TestFindingJSONMirrorStaysInSync enforces the deliberate marshal-mirror +// relationship between Finding and findingJSON: identical field names and +// types, identical json tags, with AlertIdentifier as the only sanctioned +// divergence (hidden via `json:"-"` on the public struct, round-tripped as +// "alert_identifier" by the mirror). +func TestFindingJSONMirrorStaysInSync(t *testing.T) { + assertMarshalMirrorInSync(t, + reflect.TypeOf(Finding{}), + reflect.TypeOf(findingJSON{}), + map[string]string{"AlertIdentifier": "alert_identifier,omitempty"}, + ) +} + +// assertMarshalMirrorInSync fails the test unless mirror has exactly the same +// fields (names, types, json tags) as public, modulo the sanctioned tag +// overrides keyed by field name. +func assertMarshalMirrorInSync(t *testing.T, public, mirror reflect.Type, tagOverrides map[string]string) { + t.Helper() + + mirrorFields := make(map[string]reflect.StructField, mirror.NumField()) + for i := 0; i < mirror.NumField(); i++ { + f := mirror.Field(i) + mirrorFields[f.Name] = f + } + + if public.NumField() != mirror.NumField() { + t.Errorf("%s has %d fields but %s has %d — keep the marshal mirror in sync", + public.Name(), public.NumField(), mirror.Name(), mirror.NumField()) + } + + for i := 0; i < public.NumField(); i++ { + pf := public.Field(i) + mf, ok := mirrorFields[pf.Name] + if !ok { + t.Errorf("%s.%s is missing from mirror %s", public.Name(), pf.Name, mirror.Name()) + continue + } + if pf.Type != mf.Type { + t.Errorf("%s.%s type %s diverged from mirror type %s", public.Name(), pf.Name, pf.Type, mf.Type) + } + wantTag := pf.Tag.Get("json") + if override, isOverride := tagOverrides[pf.Name]; isOverride { + wantTag = override + } + if got := mf.Tag.Get("json"); got != wantTag { + t.Errorf("%s.%s mirror json tag = %q, want %q", public.Name(), pf.Name, got, wantTag) + } + } +} diff --git a/internal/ai/memory/changes.go b/internal/ai/memory/changes.go index 097a1d79e..4896eaae7 100644 --- a/internal/ai/memory/changes.go +++ b/internal/ai/memory/changes.go @@ -5,9 +5,7 @@ package memory import ( "encoding/json" - "fmt" "os" - "sort" "sync" "sync/atomic" "time" @@ -338,37 +336,12 @@ func (d *ChangeDetector) saveToDisk() error { // loadFromDisk loads changes from JSON file func (d *ChangeDetector) loadFromDisk() error { - if d.dataDir == "" { - return nil - } - - path, err := memoryPersistencePath(d.dataDir, changeHistoryFileName) - if err != nil { - return err - } - if st, err := os.Stat(path); err == nil { - const maxOnDiskBytes = 10 << 20 // 10 MiB safety cap - if st.Size() > maxOnDiskBytes { - return fmt.Errorf("change history file too large (%d bytes)", st.Size()) - } - } - data, err := os.ReadFile(path) - if err != nil { - if os.IsNotExist(err) { - return nil - } - return err - } - - var changes []Change - if err := json.Unmarshal(data, &changes); err != nil { - return err - } - - // Sort by time - sort.Slice(changes, func(i, j int) bool { - return changes[i].DetectedAt.Before(changes[j].DetectedAt) + changes, ok, err := loadMemoryHistory(d.dataDir, changeHistoryFileName, "change history", func(a, b Change) bool { + return a.DetectedAt.Before(b.DetectedAt) }) + if err != nil || !ok { + return err + } d.changes = changes d.trimChanges() diff --git a/internal/ai/memory/paths.go b/internal/ai/memory/paths.go index c9d40a839..5b8fedbc5 100644 --- a/internal/ai/memory/paths.go +++ b/internal/ai/memory/paths.go @@ -1,7 +1,10 @@ package memory import ( + "encoding/json" "fmt" + "os" + "sort" "strings" "github.com/rcourtman/pulse-go-rewrite/internal/securityutil" @@ -34,3 +37,41 @@ func memoryPersistencePath(dataDir string, leaf string) (string, error) { } return securityutil.JoinStorageLeaf(normalizedDir, leaf) } + +// loadMemoryHistory loads a JSON history slice from dataDir/fileName sorted +// by less, enforcing the shared 10 MiB on-disk safety cap. The boolean result +// is false when persistence is disabled (empty dataDir) or the file does not +// exist; label feeds the too-large error message. +func loadMemoryHistory[T any](dataDir, fileName, label string, less func(a, b T) bool) ([]T, bool, error) { + if dataDir == "" { + return nil, false, nil + } + + path, err := memoryPersistencePath(dataDir, fileName) + if err != nil { + return nil, false, err + } + if st, err := os.Stat(path); err == nil { + const maxOnDiskBytes = 10 << 20 // 10 MiB safety cap + if st.Size() > maxOnDiskBytes { + return nil, false, fmt.Errorf("%s file too large (%d bytes)", label, st.Size()) + } + } + data, err := os.ReadFile(path) + if err != nil { + if os.IsNotExist(err) { + return nil, false, nil + } + return nil, false, err + } + + var items []T + if err := json.Unmarshal(data, &items); err != nil { + return nil, false, err + } + + sort.Slice(items, func(i, j int) bool { + return less(items[i], items[j]) + }) + return items, true, nil +} diff --git a/internal/ai/memory/remediation.go b/internal/ai/memory/remediation.go index 92e6d0f5a..6e90d5e36 100644 --- a/internal/ai/memory/remediation.go +++ b/internal/ai/memory/remediation.go @@ -4,7 +4,6 @@ import ( "encoding/json" "fmt" "os" - "sort" "sync" "time" @@ -312,37 +311,12 @@ func (r *RemediationLog) saveToDisk() error { // loadFromDisk loads records from JSON file func (r *RemediationLog) loadFromDisk() error { - if r.dataDir == "" { - return nil - } - - path, err := memoryPersistencePath(r.dataDir, remediationHistoryFileName) - if err != nil { - return err - } - if st, err := os.Stat(path); err == nil { - const maxOnDiskBytes = 10 << 20 // 10 MiB safety cap - if st.Size() > maxOnDiskBytes { - return fmt.Errorf("remediation history file too large (%d bytes)", st.Size()) - } - } - data, err := os.ReadFile(path) - if err != nil { - if os.IsNotExist(err) { - return nil - } - return err - } - - var records []RemediationRecord - if err := json.Unmarshal(data, &records); err != nil { - return err - } - - // Sort by timestamp - sort.Slice(records, func(i, j int) bool { - return records[i].Timestamp.Before(records[j].Timestamp) + records, ok, err := loadMemoryHistory(r.dataDir, remediationHistoryFileName, "remediation history", func(a, b RemediationRecord) bool { + return a.Timestamp.Before(b.Timestamp) }) + if err != nil || !ok { + return err + } r.records = records r.trimRecords() diff --git a/internal/ai/patrol_intelligence.go b/internal/ai/patrol_intelligence.go index 2356d459a..4f4cfdf13 100644 --- a/internal/ai/patrol_intelligence.go +++ b/internal/ai/patrol_intelligence.go @@ -183,6 +183,20 @@ func (p *PatrolService) gatherGuestIntelligence(ctx context.Context) map[string] return intel } +// patrolGuestView is the read-state view method subset patrol intelligence +// consumes from VMs and system containers. +type patrolGuestView interface { + Template() bool + Name() string + VMID() int + Node() string + Instance() string + SourceID() string + ID() string + Status() unifiedresources.ResourceStatus + IPAddresses() []string +} + // gatherGuestsFromReadState iterates VMs and Containers via ReadState typed views. // Keys are SourceID() (the legacy guest ID like "qemu/100") for compatibility with // downstream consumers (scope matching, triage, finding dedup). Status checks use @@ -193,28 +207,43 @@ func gatherGuestsFromReadState( intel map[string]*GuestIntelligence, nodeGuests map[string][]guestIPInfo, ) { - for _, vm := range rs.VMs() { - if vm.Template() { + gatherGuestIntelligenceFromViews(rs.VMs(), "vm", servicediscovery.ResourceTypeVM, discoveryIndex, intel, nodeGuests) + gatherGuestIntelligenceFromViews(rs.Containers(), "system-container", servicediscovery.ResourceTypeSystemContainer, discoveryIndex, intel, nodeGuests) +} + +// gatherGuestIntelligenceFromViews collects per-guest intelligence for one +// guest family (VMs or system containers), resolving discovery metadata by +// node first and instance second. +func gatherGuestIntelligenceFromViews[V patrolGuestView]( + guests []V, + guestType string, + discoveryResourceType servicediscovery.ResourceType, + discoveryIndex map[discoveryKey]*servicediscovery.ResourceDiscovery, + intel map[string]*GuestIntelligence, + nodeGuests map[string][]guestIPInfo, +) { + for _, g := range guests { + if g.Template() { continue } gi := &GuestIntelligence{ - Name: vm.Name(), - GuestType: "vm", + Name: g.Name(), + GuestType: guestType, } - vmidStr := strconv.Itoa(vm.VMID()) + vmidStr := strconv.Itoa(g.VMID()) if d, ok := discoveryIndex[discoveryKey{ - resourceType: servicediscovery.ResourceTypeVM, - targetID: vm.Node(), + resourceType: discoveryResourceType, + targetID: g.Node(), resourceID: vmidStr, }]; ok { gi.ServiceName = d.ServiceName gi.ServiceType = d.ServiceType } - if gi.ServiceName == "" && vm.Instance() != "" && vm.Instance() != vm.Node() { + if gi.ServiceName == "" && g.Instance() != "" && g.Instance() != g.Node() { if d, ok := discoveryIndex[discoveryKey{ - resourceType: servicediscovery.ResourceTypeVM, - targetID: vm.Instance(), + resourceType: discoveryResourceType, + targetID: g.Instance(), resourceID: vmidStr, }]; ok { gi.ServiceName = d.ServiceName @@ -225,60 +254,15 @@ func gatherGuestsFromReadState( // Use SourceID (legacy guest ID like "qemu/100") as the map key // to stay compatible with downstream consumers. Fall back to // unified ID() if SourceID is empty (defensive). - sourceID := vm.SourceID() + sourceID := g.SourceID() if sourceID == "" { - sourceID = vm.ID() + sourceID = g.ID() } intel[sourceID] = gi - if vm.Status() == unifiedresources.StatusOnline && len(vm.IPAddresses()) > 0 { - for _, ip := range vm.IPAddresses() { - nodeGuests[vm.Node()] = append(nodeGuests[vm.Node()], guestIPInfo{ - guestID: sourceID, - ip: ip, - }) - } - } - } - - for _, ct := range rs.Containers() { - if ct.Template() { - continue - } - gi := &GuestIntelligence{ - Name: ct.Name(), - GuestType: "system-container", - } - - vmidStr := strconv.Itoa(ct.VMID()) - if d, ok := discoveryIndex[discoveryKey{ - resourceType: servicediscovery.ResourceTypeSystemContainer, - targetID: ct.Node(), - resourceID: vmidStr, - }]; ok { - gi.ServiceName = d.ServiceName - gi.ServiceType = d.ServiceType - } - if gi.ServiceName == "" && ct.Instance() != "" && ct.Instance() != ct.Node() { - if d, ok := discoveryIndex[discoveryKey{ - resourceType: servicediscovery.ResourceTypeSystemContainer, - targetID: ct.Instance(), - resourceID: vmidStr, - }]; ok { - gi.ServiceName = d.ServiceName - gi.ServiceType = d.ServiceType - } - } - - sourceID := ct.SourceID() - if sourceID == "" { - sourceID = ct.ID() - } - intel[sourceID] = gi - - if ct.Status() == unifiedresources.StatusOnline && len(ct.IPAddresses()) > 0 { - for _, ip := range ct.IPAddresses() { - nodeGuests[ct.Node()] = append(nodeGuests[ct.Node()], guestIPInfo{ + if g.Status() == unifiedresources.StatusOnline && len(g.IPAddresses()) > 0 { + for _, ip := range g.IPAddresses() { + nodeGuests[g.Node()] = append(nodeGuests[g.Node()], guestIPInfo{ guestID: sourceID, ip: ip, }) diff --git a/internal/ai/providers/anthropic.go b/internal/ai/providers/anthropic.go index 3022d9979..76d439f7b 100644 --- a/internal/ai/providers/anthropic.go +++ b/internal/ai/providers/anthropic.go @@ -139,11 +139,14 @@ type anthropicErrorDetail struct { Message string `json:"message"` } -// Chat sends a chat request to the Anthropic API -func (c *AnthropicClient) Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error) { - // Convert messages to Anthropic format - messages := make([]anthropicMessage, 0, len(req.Messages)) - for _, m := range req.Messages { +// convertMessagesToAnthropic converts provider-neutral chat messages into +// Anthropic-format messages: system messages are dropped (Anthropic carries +// the system prompt as a top-level field), tool results become user +// tool_result content blocks, and assistant tool calls become tool_use +// blocks. Shared by the API-key and OAuth clients. +func convertMessagesToAnthropic(reqMessages []Message) ([]anthropicMessage, error) { + messages := make([]anthropicMessage, 0, len(reqMessages)) + for _, m := range reqMessages { // Anthropic doesn't use "system" role in messages array if m.Role == "system" { continue @@ -200,6 +203,16 @@ func (c *AnthropicClient) Chat(ctx context.Context, req ChatRequest) (*ChatRespo Content: m.Content, }) } + return messages, nil +} + +// Chat sends a chat request to the Anthropic API +func (c *AnthropicClient) Chat(ctx context.Context, req ChatRequest) (*ChatResponse, error) { + // Convert messages to Anthropic format + messages, err := convertMessagesToAnthropic(req.Messages) + if err != nil { + return nil, err + } // Use provided model or fall back to client default model := req.Model diff --git a/internal/ai/providers/anthropic_oauth.go b/internal/ai/providers/anthropic_oauth.go index a36ae18fb..395687a3a 100644 --- a/internal/ai/providers/anthropic_oauth.go +++ b/internal/ai/providers/anthropic_oauth.go @@ -402,58 +402,9 @@ func (c *AnthropicOAuthClient) Chat(ctx context.Context, req ChatRequest) (*Chat } // Convert messages to Anthropic format (same as regular client) - messages := make([]anthropicMessage, 0, len(req.Messages)) - for _, m := range req.Messages { - if m.Role == "system" { - continue - } - - if m.ToolResult != nil { - contentJSON, err := json.Marshal(m.ToolResult.Content) - if err != nil { - return nil, fmt.Errorf("failed to marshal tool result content for %s: %w", m.ToolResult.ToolUseID, err) - } - messages = append(messages, anthropicMessage{ - Role: "user", - Content: []anthropicContent{ - { - Type: "tool_result", - ToolUseID: m.ToolResult.ToolUseID, - Content: contentJSON, - IsError: m.ToolResult.IsError, - }, - }, - }) - continue - } - - if m.Role == "assistant" && len(m.ToolCalls) > 0 { - contentBlocks := make([]anthropicContent, 0) - if m.Content != "" { - contentBlocks = append(contentBlocks, anthropicContent{ - Type: "text", - Text: m.Content, - }) - } - for _, tc := range m.ToolCalls { - contentBlocks = append(contentBlocks, anthropicContent{ - Type: "tool_use", - ID: tc.ID, - Name: tc.Name, - Input: tc.Input, - }) - } - messages = append(messages, anthropicMessage{ - Role: "assistant", - Content: contentBlocks, - }) - continue - } - - messages = append(messages, anthropicMessage{ - Role: m.Role, - Content: m.Content, - }) + messages, err := convertMessagesToAnthropic(req.Messages) + if err != nil { + return nil, err } model := req.Model diff --git a/internal/ai/tools/tools_file.go b/internal/ai/tools/tools_file.go index b41ea8260..eef9c4850 100644 --- a/internal/ai/tools/tools_file.go +++ b/internal/ai/tools/tools_file.go @@ -230,161 +230,49 @@ func (e *PulseToolExecutor) executeFileRead(ctx context.Context, path, targetHos return NewJSONResult(response), nil } +// fileMutationSpec captures how the append and write actions diverge inside +// the shared file-mutation pipeline. Approval command/detail strings must stay +// stable: pre-approval consumption matches on the exact command text. +type fileMutationSpec struct { + action string // response/validation verb: "append" / "write" + approvalCommand string // exact approval command text + approvalDetail string // human approval description + redirect string // shell redirect: ">>" / ">" + failureVerb string // error wrap: "append to file" / "write file" + auditDetail string // audit log description + verify func(ctx context.Context, routing CommandRoutingResult, path, dockerContainer, content string) map[string]interface{} +} + // executeFileAppend appends content to a file func (e *PulseToolExecutor) executeFileAppend(ctx context.Context, path, content, targetHost, dockerContainer string, args map[string]interface{}) (CallToolResult, error) { - if e.agentServer == nil { - return NewErrorResult(fmt.Errorf("no agent server available")), nil - } - approvalID, _ := args["_approval_id"].(string) - approvalID = strings.TrimSpace(approvalID) - - if blocked, reason := safety.IsSensitivePath(path); blocked { - return NewToolResponseResult(NewToolBlockedError( - "SENSITIVE_PATH", - fmt.Sprintf("Refusing to write sensitive path '%s' (%s).", path, reason), - map[string]interface{}{ - "path": path, - "reason": reason, - "policy_boundary": "Credential files cannot be modified through AI tools.", - }, - )), nil - } - - // Validate routing context - block if targeting a host node when child resources exist - // This prevents accidentally writing files to the host when user meant to write to an container/VM - routingResult := e.validateRoutingContext(targetHost) - if routingResult.IsBlocked() { - return NewToolResponseResult(routingResult.RoutingError.ToToolResponse()), nil - } - - // Validate resource is in resolved context (write operation) - // With PULSE_STRICT_RESOLUTION=true, this blocks execution on undiscovered resources - validation := e.validateResolvedResource(targetHost, "append", true) - if validation.IsBlocked() { - // Hard validation failure - return consistent error envelope - return NewToolResponseResult(validation.StrictError.ToToolResponse()), nil - } - // Soft validation warnings are logged inside validateResolvedResource - - // Use full routing resolution - includes provenance for debugging - routing := e.resolveTargetForCommandFull(targetHost) - if routing.AgentID == "" { - if routing.TargetType == "container" || routing.TargetType == "vm" { - return NewTextResult(fmt.Sprintf("'%s' is a %s but no agent is available on its host node. Install Pulse Unified Agent on the node.", targetHost, routing.TargetType)), nil - } - return NewTextResult(fmt.Sprintf("No agent found for host '%s'. Check that the hostname is correct and an agent is connected.", targetHost)), nil - } - - // INVARIANT: If the target resolves to a child resource (container/VM), writes MUST execute - // inside that context via pct_exec/qm_guest_exec. No silent node fallback. - if err := e.validateWriteExecutionContext(targetHost, routing); err != nil { - return NewToolResponseResult(err.ToToolResponse()), nil - } - - approvalCommand := fmt.Sprintf("Append to file: %s", path) - approvalTargetID := fmt.Sprintf("host=%s|container=%s|path=%s", targetHost, dockerContainer, path) - - // Check if pre-approved (validated + single-use). - preApproved := consumeApprovalWithValidation(args, e.orgID, approvalCommand, "file", approvalTargetID) - requiresApproval := !e.isAutonomous && e.controlLevel == ControlLevelControlled - - // Skip approval checks if pre-approved or in autonomous mode - if !preApproved && !e.isAutonomous && e.controlLevel == ControlLevelControlled { - target := targetHost - if dockerContainer != "" { - target = fmt.Sprintf("%s (container: %s)", targetHost, dockerContainer) - } - approvalID := e.createApprovalRecord( - approvalCommand, - "file", - approvalTargetID, - target, - fmt.Sprintf("Append %d bytes to %s", len(content), path), - ) - return NewTextResult(formatFileApprovalNeeded(path, target, "append", len(content), approvalID)), nil - } - - // Use base64 encoding to safely transfer content - encoded := base64.StdEncoding.EncodeToString([]byte(content)) - var command string - if dockerContainer != "" { - // Append inside Docker container - escape the full inner command to avoid nested quote breakage. - innerCommand := fmt.Sprintf("echo %s | base64 -d >> %s", shellEscape(encoded), shellEscape(path)) - command = fmt.Sprintf("docker exec %s sh -c %s", shellEscape(dockerContainer), shellEscape(innerCommand)) - } else { - // For host/container/VM targets - agent handles sh -c wrapping for container/VM - command = fmt.Sprintf("echo %s | base64 -d >> %s", shellEscape(encoded), shellEscape(path)) - } - - result, err := e.executeCommandWithAudit( - ctx, - "pulse_file_edit", - func() string { - if dockerContainer != "" { - return fmt.Sprintf("%s:%s:%s", targetHost, dockerContainer, path) - } - return fmt.Sprintf("%s:%s", targetHost, path) - }(), - approvalID, - requiresApproval, - routing.AgentID, - agentexec.ExecuteCommandPayload{ - Command: command, - TargetType: routing.TargetType, - TargetID: routing.TargetID, - }, - "pulse_file_edit", - fmt.Sprintf("append %d bytes to %s", len(content), path), - ) - if err != nil { - return NewErrorResult(fmt.Errorf("failed to append to file: %w", err)), nil - } - - if result.ExitCode != 0 { - errMsg := result.Stderr - if errMsg == "" { - errMsg = result.Stdout - } - if dockerContainer != "" { - response := map[string]interface{}{ - "success": false, - "action": "append", - "path": path, - "host": targetHost, - "exit_code": result.ExitCode, - "error": errMsg, - } - setContainerResponseFields(response, dockerContainer) - return NewJSONResultWithIsError(response, true), nil - } - return NewJSONResultWithIsError(map[string]interface{}{ - "success": false, - "action": "append", - "path": path, - "host": targetHost, - "exit_code": result.ExitCode, - "error": errMsg, - }, true), nil - } - - verify := e.verifyFileTailHash(ctx, routing, path, dockerContainer, content) - - response := map[string]interface{}{ - "success": true, - "action": "append", - "path": path, - "host": targetHost, - "bytes_written": len(content), - "verification": verify, - } - setContainerResponseFields(response, dockerContainer) - // Include execution provenance for observability - response["execution"] = buildExecutionProvenance(targetHost, routing) - return NewJSONResult(response), nil + return e.executeFileMutation(ctx, path, content, targetHost, dockerContainer, args, fileMutationSpec{ + action: "append", + approvalCommand: fmt.Sprintf("Append to file: %s", path), + approvalDetail: fmt.Sprintf("Append %d bytes to %s", len(content), path), + redirect: ">>", + failureVerb: "append to file", + auditDetail: fmt.Sprintf("append %d bytes to %s", len(content), path), + verify: e.verifyFileTailHash, + }) } // executeFileWrite writes content to a file (overwrites) func (e *PulseToolExecutor) executeFileWrite(ctx context.Context, path, content, targetHost, dockerContainer string, args map[string]interface{}) (CallToolResult, error) { + return e.executeFileMutation(ctx, path, content, targetHost, dockerContainer, args, fileMutationSpec{ + action: "write", + approvalCommand: fmt.Sprintf("Write file: %s", path), + approvalDetail: fmt.Sprintf("Write %d bytes to %s", len(content), path), + redirect: ">", + failureVerb: "write file", + auditDetail: fmt.Sprintf("write %d bytes to %s", len(content), path), + verify: e.verifyFileSHA256, + }) +} + +// executeFileMutation runs the shared append/write pipeline: sensitive-path +// blocking, routing + resolved-resource validation, approval gating, base64 +// transfer, audited execution, and post-write verification. +func (e *PulseToolExecutor) executeFileMutation(ctx context.Context, path, content, targetHost, dockerContainer string, args map[string]interface{}, spec fileMutationSpec) (CallToolResult, error) { if e.agentServer == nil { return NewErrorResult(fmt.Errorf("no agent server available")), nil } @@ -412,7 +300,7 @@ func (e *PulseToolExecutor) executeFileWrite(ctx context.Context, path, content, // Validate resource is in resolved context (write operation) // With PULSE_STRICT_RESOLUTION=true, this blocks execution on undiscovered resources - validation := e.validateResolvedResource(targetHost, "write", true) + validation := e.validateResolvedResource(targetHost, spec.action, true) if validation.IsBlocked() { // Hard validation failure - return consistent error envelope return NewToolResponseResult(validation.StrictError.ToToolResponse()), nil @@ -434,11 +322,10 @@ func (e *PulseToolExecutor) executeFileWrite(ctx context.Context, path, content, return NewToolResponseResult(err.ToToolResponse()), nil } - approvalCommand := fmt.Sprintf("Write file: %s", path) approvalTargetID := fmt.Sprintf("host=%s|container=%s|path=%s", targetHost, dockerContainer, path) // Check if pre-approved (validated + single-use). - preApproved := consumeApprovalWithValidation(args, e.orgID, approvalCommand, "file", approvalTargetID) + preApproved := consumeApprovalWithValidation(args, e.orgID, spec.approvalCommand, "file", approvalTargetID) requiresApproval := !e.isAutonomous && e.controlLevel == ControlLevelControlled // Skip approval checks if pre-approved or in autonomous mode @@ -448,25 +335,25 @@ func (e *PulseToolExecutor) executeFileWrite(ctx context.Context, path, content, target = fmt.Sprintf("%s (container: %s)", targetHost, dockerContainer) } approvalID := e.createApprovalRecord( - approvalCommand, + spec.approvalCommand, "file", approvalTargetID, target, - fmt.Sprintf("Write %d bytes to %s", len(content), path), + spec.approvalDetail, ) - return NewTextResult(formatFileApprovalNeeded(path, target, "write", len(content), approvalID)), nil + return NewTextResult(formatFileApprovalNeeded(path, target, spec.action, len(content), approvalID)), nil } // Use base64 encoding to safely transfer content encoded := base64.StdEncoding.EncodeToString([]byte(content)) var command string if dockerContainer != "" { - // Write inside Docker container - escape the full inner command to avoid nested quote breakage. - innerCommand := fmt.Sprintf("echo %s | base64 -d > %s", shellEscape(encoded), shellEscape(path)) + // Mutate inside Docker container - escape the full inner command to avoid nested quote breakage. + innerCommand := fmt.Sprintf("echo %s | base64 -d %s %s", shellEscape(encoded), spec.redirect, shellEscape(path)) command = fmt.Sprintf("docker exec %s sh -c %s", shellEscape(dockerContainer), shellEscape(innerCommand)) } else { // For host/container/VM targets - agent handles sh -c wrapping for container/VM - command = fmt.Sprintf("echo %s | base64 -d > %s", shellEscape(encoded), shellEscape(path)) + command = fmt.Sprintf("echo %s | base64 -d %s %s", shellEscape(encoded), spec.redirect, shellEscape(path)) } result, err := e.executeCommandWithAudit( @@ -487,10 +374,10 @@ func (e *PulseToolExecutor) executeFileWrite(ctx context.Context, path, content, TargetID: routing.TargetID, }, "pulse_file_edit", - fmt.Sprintf("write %d bytes to %s", len(content), path), + spec.auditDetail, ) if err != nil { - return NewErrorResult(fmt.Errorf("failed to write file: %w", err)), nil + return NewErrorResult(fmt.Errorf("failed to %s: %w", spec.failureVerb, err)), nil } if result.ExitCode != 0 { @@ -498,33 +385,23 @@ func (e *PulseToolExecutor) executeFileWrite(ctx context.Context, path, content, if errMsg == "" { errMsg = result.Stdout } - if dockerContainer != "" { - response := map[string]interface{}{ - "success": false, - "action": "write", - "path": path, - "host": targetHost, - "exit_code": result.ExitCode, - "error": errMsg, - } - setContainerResponseFields(response, dockerContainer) - return NewJSONResultWithIsError(response, true), nil - } - return NewJSONResultWithIsError(map[string]interface{}{ + response := map[string]interface{}{ "success": false, - "action": "write", + "action": spec.action, "path": path, "host": targetHost, "exit_code": result.ExitCode, "error": errMsg, - }, true), nil + } + setContainerResponseFields(response, dockerContainer) + return NewJSONResultWithIsError(response, true), nil } - verify := e.verifyFileSHA256(ctx, routing, path, dockerContainer, content) + verify := spec.verify(ctx, routing, path, dockerContainer, content) response := map[string]interface{}{ "success": true, - "action": "write", + "action": spec.action, "path": path, "host": targetHost, "bytes_written": len(content), diff --git a/internal/ai/tools/tools_kubernetes.go b/internal/ai/tools/tools_kubernetes.go index 358e5eac0..207964b48 100644 --- a/internal/ai/tools/tools_kubernetes.go +++ b/internal/ai/tools/tools_kubernetes.go @@ -598,109 +598,69 @@ func (e *PulseToolExecutor) executeKubernetesScale(ctx context.Context, args map return NewTextResult(fmt.Sprintf("kubectl command failed (exit code %d):\n%s", result.ExitCode, output)), nil } +// kubernetesResourceAction captures how the namespaced kubectl actions +// (deployment restart, pod delete) diverge inside the shared +// approval/audit/execution pipeline. The kubectl command text doubles as the +// approval command, so command construction must stay stable per action. +type kubernetesResourceAction struct { + argKey string // resource argument name: "deployment" / "pod" + kindLabel string // approval target kind segment: "deployment" / "pod" + approvalAction string // formatKubernetesApprovalNeeded action: "restart" / "delete_pod" + command func(namespace, name string) string + approvalDetail func(name string) string + auditDetail func(name string) string + successMessage func(name, namespace, output string) string +} + // executeKubernetesRestart restarts a deployment via rollout restart func (e *PulseToolExecutor) executeKubernetesRestart(ctx context.Context, args map[string]interface{}) (CallToolResult, error) { - clusterArg, _ := args["cluster"].(string) - namespace, _ := args["namespace"].(string) - deployment, _ := args["deployment"].(string) - approvalID, _ := args["_approval_id"].(string) - approvalID = strings.TrimSpace(approvalID) - - if clusterArg == "" { - return NewErrorResult(fmt.Errorf("cluster is required")), nil - } - if deployment == "" { - return NewErrorResult(fmt.Errorf("deployment is required")), nil - } - if namespace == "" { - namespace = "default" - } - - // Validate identifiers - if err := validateKubernetesResourceID(namespace); err != nil { - return NewErrorResult(fmt.Errorf("invalid namespace: %w", err)), nil - } - if err := validateKubernetesResourceID(deployment); err != nil { - return NewErrorResult(fmt.Errorf("invalid deployment: %w", err)), nil - } - - // Check control level - if e.controlLevel == ControlLevelReadOnly { - return NewTextResult("Kubernetes control operations are not available in read-only mode."), nil - } - - agentID, cluster, err := e.findAgentForKubernetesCluster(clusterArg) - if err != nil { - return NewTextResult(err.Error()), nil - } - - // Build command - command := fmt.Sprintf("kubectl -n %s rollout restart deployment/%s", shellEscape(namespace), shellEscape(deployment)) - clusterScope := cluster.ID - if clusterScope == "" { - clusterScope = clusterArg - } - approvalTargetID := fmt.Sprintf("%s:%s:deployment:%s", clusterScope, namespace, deployment) - - // Check if pre-approved (validated + single-use). - preApproved := consumeApprovalWithValidation(args, e.orgID, command, "kubernetes", approvalTargetID) - requiresApproval := !e.isAutonomous && e.controlLevel == ControlLevelControlled - - // Request approval if needed - if !preApproved && !e.isAutonomous && e.controlLevel == ControlLevelControlled { - displayName := cluster.DisplayName - approvalID := e.createApprovalRecord(command, "kubernetes", approvalTargetID, displayName, fmt.Sprintf("Restart deployment %s", deployment)) - return NewTextResult(formatKubernetesApprovalNeeded("restart", deployment, namespace, displayName, command, approvalID)), nil - } - - if e.agentServer == nil { - return NewErrorResult(fmt.Errorf("no agent server available")), nil - } - - result, err := e.executeCommandWithAudit( - ctx, - "pulse_kubernetes", - fmt.Sprintf("%s:%s:deployment:%s", clusterScope, namespace, deployment), - approvalID, - requiresApproval, - agentID, - agentexec.ExecuteCommandPayload{ - Command: command, - TargetType: "agent", - TargetID: "", + return e.executeKubernetesResourceAction(ctx, args, kubernetesResourceAction{ + argKey: "deployment", + kindLabel: "deployment", + approvalAction: "restart", + command: func(namespace, name string) string { + return fmt.Sprintf("kubectl -n %s rollout restart deployment/%s", shellEscape(namespace), shellEscape(name)) }, - "pulse_kubernetes", - fmt.Sprintf("restart deployment %s", deployment), - ) - if err != nil { - return NewErrorResult(fmt.Errorf("failed to execute kubectl: %w", err)), nil - } - - output := result.Stdout - if result.Stderr != "" { - output += "\n" + result.Stderr - } - - if result.ExitCode == 0 { - return NewTextResult(fmt.Sprintf("✓ Successfully initiated rollout restart for deployment '%s' in namespace '%s'. Action complete - pods will restart gradually.\n%s", deployment, namespace, output)), nil - } - - return NewTextResult(fmt.Sprintf("kubectl command failed (exit code %d):\n%s", result.ExitCode, output)), nil + approvalDetail: func(name string) string { return fmt.Sprintf("Restart deployment %s", name) }, + auditDetail: func(name string) string { return fmt.Sprintf("restart deployment %s", name) }, + successMessage: func(name, namespace, output string) string { + return fmt.Sprintf("✓ Successfully initiated rollout restart for deployment '%s' in namespace '%s'. Action complete - pods will restart gradually.\n%s", name, namespace, output) + }, + }) } // executeKubernetesDeletePod deletes a pod func (e *PulseToolExecutor) executeKubernetesDeletePod(ctx context.Context, args map[string]interface{}) (CallToolResult, error) { + return e.executeKubernetesResourceAction(ctx, args, kubernetesResourceAction{ + argKey: "pod", + kindLabel: "pod", + approvalAction: "delete_pod", + command: func(namespace, name string) string { + return fmt.Sprintf("kubectl -n %s delete pod %s", shellEscape(namespace), shellEscape(name)) + }, + approvalDetail: func(name string) string { return fmt.Sprintf("Delete pod %s", name) }, + auditDetail: func(name string) string { return fmt.Sprintf("delete pod %s", name) }, + successMessage: func(name, namespace, output string) string { + return fmt.Sprintf("✓ Successfully deleted pod '%s' in namespace '%s'. If managed by a controller, a new pod will be created.\n%s", name, namespace, output) + }, + }) +} + +// executeKubernetesResourceAction runs the shared namespaced kubectl action +// pipeline: argument validation, control-level gating, cluster agent +// resolution, approval gating, and audited execution. +func (e *PulseToolExecutor) executeKubernetesResourceAction(ctx context.Context, args map[string]interface{}, spec kubernetesResourceAction) (CallToolResult, error) { clusterArg, _ := args["cluster"].(string) namespace, _ := args["namespace"].(string) - pod, _ := args["pod"].(string) + name, _ := args[spec.argKey].(string) approvalID, _ := args["_approval_id"].(string) approvalID = strings.TrimSpace(approvalID) if clusterArg == "" { return NewErrorResult(fmt.Errorf("cluster is required")), nil } - if pod == "" { - return NewErrorResult(fmt.Errorf("pod is required")), nil + if name == "" { + return NewErrorResult(fmt.Errorf("%s is required", spec.argKey)), nil } if namespace == "" { namespace = "default" @@ -710,8 +670,8 @@ func (e *PulseToolExecutor) executeKubernetesDeletePod(ctx context.Context, args if err := validateKubernetesResourceID(namespace); err != nil { return NewErrorResult(fmt.Errorf("invalid namespace: %w", err)), nil } - if err := validateKubernetesResourceID(pod); err != nil { - return NewErrorResult(fmt.Errorf("invalid pod: %w", err)), nil + if err := validateKubernetesResourceID(name); err != nil { + return NewErrorResult(fmt.Errorf("invalid %s: %w", spec.argKey, err)), nil } // Check control level @@ -725,12 +685,12 @@ func (e *PulseToolExecutor) executeKubernetesDeletePod(ctx context.Context, args } // Build command - command := fmt.Sprintf("kubectl -n %s delete pod %s", shellEscape(namespace), shellEscape(pod)) + command := spec.command(namespace, name) clusterScope := cluster.ID if clusterScope == "" { clusterScope = clusterArg } - approvalTargetID := fmt.Sprintf("%s:%s:pod:%s", clusterScope, namespace, pod) + approvalTargetID := fmt.Sprintf("%s:%s:%s:%s", clusterScope, namespace, spec.kindLabel, name) // Check if pre-approved (validated + single-use). preApproved := consumeApprovalWithValidation(args, e.orgID, command, "kubernetes", approvalTargetID) @@ -739,8 +699,8 @@ func (e *PulseToolExecutor) executeKubernetesDeletePod(ctx context.Context, args // Request approval if needed if !preApproved && !e.isAutonomous && e.controlLevel == ControlLevelControlled { displayName := cluster.DisplayName - approvalID := e.createApprovalRecord(command, "kubernetes", approvalTargetID, displayName, fmt.Sprintf("Delete pod %s", pod)) - return NewTextResult(formatKubernetesApprovalNeeded("delete_pod", pod, namespace, displayName, command, approvalID)), nil + approvalID := e.createApprovalRecord(command, "kubernetes", approvalTargetID, displayName, spec.approvalDetail(name)) + return NewTextResult(formatKubernetesApprovalNeeded(spec.approvalAction, name, namespace, displayName, command, approvalID)), nil } if e.agentServer == nil { @@ -750,7 +710,7 @@ func (e *PulseToolExecutor) executeKubernetesDeletePod(ctx context.Context, args result, err := e.executeCommandWithAudit( ctx, "pulse_kubernetes", - fmt.Sprintf("%s:%s:pod:%s", clusterScope, namespace, pod), + approvalTargetID, approvalID, requiresApproval, agentID, @@ -760,7 +720,7 @@ func (e *PulseToolExecutor) executeKubernetesDeletePod(ctx context.Context, args TargetID: "", }, "pulse_kubernetes", - fmt.Sprintf("delete pod %s", pod), + spec.auditDetail(name), ) if err != nil { return NewErrorResult(fmt.Errorf("failed to execute kubectl: %w", err)), nil @@ -772,7 +732,7 @@ func (e *PulseToolExecutor) executeKubernetesDeletePod(ctx context.Context, args } if result.ExitCode == 0 { - return NewTextResult(fmt.Sprintf("✓ Successfully deleted pod '%s' in namespace '%s'. If managed by a controller, a new pod will be created.\n%s", pod, namespace, output)), nil + return NewTextResult(spec.successMessage(name, namespace, output)), nil } return NewTextResult(fmt.Sprintf("kubectl command failed (exit code %d):\n%s", result.ExitCode, output)), nil diff --git a/internal/ai/tools/tools_query.go b/internal/ai/tools/tools_query.go index ec0b0fa91..187ad04e5 100644 --- a/internal/ai/tools/tools_query.go +++ b/internal/ai/tools/tools_query.go @@ -2911,6 +2911,188 @@ func canonicalGuestManagedID(resource unifiedresources.Resource) string { return "" } +// queryGuestView is the read-state view method subset the guest get/search +// paths consume; *unifiedresources.VMView and *unifiedresources.ContainerView +// both satisfy it. +type queryGuestView interface { + ID() string + Name() string + VMID() int + Node() string + Instance() string + Status() unifiedresources.ResourceStatus + CPUPercent() float64 + CPUs() int + MemoryPercent() float64 + MemoryUsed() int64 + MemoryTotal() int64 + Tags() []string + IPAddresses() []string + LastBackup() time.Time +} + +// addCanonicalGuestSearchMatches appends search matches for canonical guest +// resources of one kind ("vm" / "system-container"), shared by the +// resource-search guest arms. +func addCanonicalGuestSearchMatches( + kind string, + resources []unifiedresources.Resource, + queryLower, statusFilter string, + governance *governedQueryMetadataResolver, + connectedAgentHostnames map[string]bool, + matchesQuery func(query string, candidates ...string) bool, + addMatch func(ResourceMatch), +) { + for _, resource := range resources { + status := string(resource.Status) + if !statusMatchesFilter(status, statusFilter) { + continue + } + candidates := canonicalGuestSearchCandidates(kind, resource) + if !matchesQuery(queryLower, candidates...) { + continue + } + + vmid := 0 + if resource.Proxmox != nil && resource.Proxmox.VMID > 0 { + vmid = resource.Proxmox.VMID + } + node := canonicalGuestTarget(resource) + metadataCandidates := append([]string{resourceDisplayName(resource), resource.ID}, candidates...) + addMatch(ResourceMatch{ + GovernedResourceMetadata: governance.Resolve(metadataCandidates...), + Type: kind, + ID: resource.ID, + Name: resourceDisplayName(resource), + Status: status, + Node: node, + NodeHasAgent: connectedAgentHostnames[node], + Platform: canonicalResourcePlatform(resource), + VMID: vmid, + AgentConnected: resourceAgentConnected(resource, connectedAgentHostnames), + }) + } +} + +// addGuestViewSearchMatches appends search matches for read-state guest views +// of one kind, the legacy fallback when no canonical guest resources exist. +func addGuestViewSearchMatches[V queryGuestView]( + kind string, + guests []V, + queryLower, statusFilter string, + governance *governedQueryMetadataResolver, + connectedAgentHostnames map[string]bool, + matchesQuery func(query string, candidates ...string) bool, + addMatch func(ResourceMatch), +) { + for _, g := range guests { + status := string(g.Status()) + if !statusMatchesFilter(status, statusFilter) { + continue + } + // Build searchable candidates: name, ID, VMID, canonical type-prefixed VMID, IPs, tags + vmidStr := fmt.Sprintf("%d", g.VMID()) + candidates := []string{g.Name(), g.ID(), vmidStr, kind + vmidStr} + candidates = append(candidates, g.IPAddresses()...) + candidates = append(candidates, g.Tags()...) + + if !matchesQuery(queryLower, candidates...) { + continue + } + addMatch(ResourceMatch{ + GovernedResourceMetadata: governance.Resolve(g.Name(), g.ID(), vmidStr), + Type: kind, + ID: g.ID(), + Name: g.Name(), + Status: status, + Node: g.Node(), + NodeHasAgent: connectedAgentHostnames[g.Node()], + Platform: "proxmox", + VMID: g.VMID(), + AgentConnected: connectedAgentHostnames[g.Name()], + }) + } +} + +// notFoundResourceResult is the shared not-found envelope for resource gets. +func notFoundResourceResult(kind, resourceID string) CallToolResult { + return NewJSONResult(map[string]interface{}{ + "error": "not_found", + "resource_id": resourceID, + "type": kind, + }) +} + +// canonicalGuestGetResult resolves a single-guest get against canonical +// unified resources, registering resolved-context access on a hit. +func (e *PulseToolExecutor) canonicalGuestGetResult(kind, resourceID string, governance *governedQueryMetadataResolver) (CallToolResult, bool) { + resource, ok := findCanonicalGuestResource(e.unifiedResourceProvider, kind, resourceID) + if !ok { + return CallToolResult{}, false + } + response := canonicalGuestResponse(kind, resource, governance) + if reg, ok := canonicalGuestRegistration(kind, resource); ok { + e.registerResolvedResourceWithExplicitAccess(reg) + } + return NewJSONResult(response), true +} + +// guestViewGetResult resolves a single-guest get against read-state views +// (VMs or system containers), registering resolved-context access with the +// guest-exec adapter ("qm" / "pct") on a hit. +func guestViewGetResult[V queryGuestView](e *PulseToolExecutor, guests []V, kind, adapter, resourceID string, governance *governedQueryMetadataResolver) (CallToolResult, bool) { + for _, g := range guests { + if fmt.Sprintf("%d", g.VMID()) != resourceID && g.Name() != resourceID && g.ID() != resourceID { + continue + } + used := g.MemoryUsed() + total := g.MemoryTotal() + response := EmptyResourceResponse() + response.GovernedResourceMetadata = governance.Resolve(g.Name(), g.ID(), fmt.Sprintf("%d", g.VMID())) + response.Type = kind + response.ID = g.ID() + response.Name = g.Name() + response.Status = string(g.Status()) + response.Node = g.Node() + response.CPU = ResourceCPU{ + Percent: g.CPUPercent(), + Cores: g.CPUs(), + } + response.Memory = ResourceMemory{ + Percent: g.MemoryPercent(), + UsedGB: float64(used) / (1024 * 1024 * 1024), + TotalGB: float64(total) / (1024 * 1024 * 1024), + } + response.Tags = g.Tags() + if !g.LastBackup().IsZero() { + t := g.LastBackup() + response.LastBackup = &t + } + + // Register in resolved context WITH explicit access (single-resource get operation) + e.registerResolvedResourceWithExplicitAccess(ResourceRegistration{ + Kind: kind, + ProviderUID: fmt.Sprintf("%d", g.VMID()), // VMID is the stable provider ID + Name: g.Name(), + Aliases: []string{g.Name(), fmt.Sprintf("%d", g.VMID()), g.ID()}, + HostUID: g.Node(), + HostName: g.Node(), + VMID: g.VMID(), + Node: g.Node(), + LocationChain: []string{"node:" + g.Node(), kind + ":" + g.Name()}, + Executors: []ExecutorRegistration{{ + ExecutorID: g.Node(), + Adapter: adapter, + Actions: guestExecutorActions(), + Priority: 10, + }}, + }) + + return NewJSONResult(response.NormalizeCollections()), true + } + return CallToolResult{}, false +} + func canonicalGuestTarget(resource unifiedresources.Resource) string { if resource.Proxmox != nil { return strings.TrimSpace(resource.Proxmox.NodeName) @@ -4662,128 +4844,22 @@ func (e *PulseToolExecutor) executeGetResource(_ context.Context, args map[strin }), nil case "vm": - if resource, ok := findCanonicalGuestResource(e.unifiedResourceProvider, "vm", resourceID); ok { - response := canonicalGuestResponse("vm", resource, governance) - if reg, ok := canonicalGuestRegistration("vm", resource); ok { - e.registerResolvedResourceWithExplicitAccess(reg) - } - return NewJSONResult(response), nil + if result, ok := e.canonicalGuestGetResult("vm", resourceID, governance); ok { + return result, nil } - for _, vm := range rs.VMs() { - if fmt.Sprintf("%d", vm.VMID()) == resourceID || vm.Name() == resourceID || vm.ID() == resourceID { - used := vm.MemoryUsed() - total := vm.MemoryTotal() - response := EmptyResourceResponse() - response.GovernedResourceMetadata = governance.Resolve(vm.Name(), vm.ID(), fmt.Sprintf("%d", vm.VMID())) - response.Type = "vm" - response.ID = vm.ID() - response.Name = vm.Name() - response.Status = string(vm.Status()) - response.Node = vm.Node() - response.CPU = ResourceCPU{ - Percent: vm.CPUPercent(), - Cores: vm.CPUs(), - } - response.Memory = ResourceMemory{ - Percent: vm.MemoryPercent(), - UsedGB: float64(used) / (1024 * 1024 * 1024), - TotalGB: float64(total) / (1024 * 1024 * 1024), - } - response.Tags = vm.Tags() - if !vm.LastBackup().IsZero() { - t := vm.LastBackup() - response.LastBackup = &t - } - - // Register in resolved context WITH explicit access (single-resource get operation) - e.registerResolvedResourceWithExplicitAccess(ResourceRegistration{ - Kind: "vm", - ProviderUID: fmt.Sprintf("%d", vm.VMID()), // VMID is the stable provider ID - Name: vm.Name(), - Aliases: []string{vm.Name(), fmt.Sprintf("%d", vm.VMID()), vm.ID()}, - HostUID: vm.Node(), - HostName: vm.Node(), - VMID: vm.VMID(), - Node: vm.Node(), - LocationChain: []string{"node:" + vm.Node(), "vm:" + vm.Name()}, - Executors: []ExecutorRegistration{{ - ExecutorID: vm.Node(), - Adapter: "qm", - Actions: guestExecutorActions(), - Priority: 10, - }}, - }) - - return NewJSONResult(response.NormalizeCollections()), nil - } + if result, ok := guestViewGetResult(e, rs.VMs(), "vm", "qm", resourceID, governance); ok { + return result, nil } - return NewJSONResult(map[string]interface{}{ - "error": "not_found", - "resource_id": resourceID, - "type": "vm", - }), nil + return notFoundResourceResult("vm", resourceID), nil case "system-container": - if resource, ok := findCanonicalGuestResource(e.unifiedResourceProvider, "system-container", resourceID); ok { - response := canonicalGuestResponse("system-container", resource, governance) - if reg, ok := canonicalGuestRegistration("system-container", resource); ok { - e.registerResolvedResourceWithExplicitAccess(reg) - } - return NewJSONResult(response), nil + if result, ok := e.canonicalGuestGetResult("system-container", resourceID, governance); ok { + return result, nil } - for _, ct := range rs.Containers() { - if fmt.Sprintf("%d", ct.VMID()) == resourceID || ct.Name() == resourceID || ct.ID() == resourceID { - used := ct.MemoryUsed() - total := ct.MemoryTotal() - response := EmptyResourceResponse() - response.GovernedResourceMetadata = governance.Resolve(ct.Name(), ct.ID(), fmt.Sprintf("%d", ct.VMID())) - response.Type = "system-container" - response.ID = ct.ID() - response.Name = ct.Name() - response.Status = string(ct.Status()) - response.Node = ct.Node() - response.CPU = ResourceCPU{ - Percent: ct.CPUPercent(), - Cores: ct.CPUs(), - } - response.Memory = ResourceMemory{ - Percent: ct.MemoryPercent(), - UsedGB: float64(used) / (1024 * 1024 * 1024), - TotalGB: float64(total) / (1024 * 1024 * 1024), - } - response.Tags = ct.Tags() - if !ct.LastBackup().IsZero() { - t := ct.LastBackup() - response.LastBackup = &t - } - - // Register in resolved context WITH explicit access (single-resource get operation) - e.registerResolvedResourceWithExplicitAccess(ResourceRegistration{ - Kind: "system-container", - ProviderUID: fmt.Sprintf("%d", ct.VMID()), // VMID is the stable provider ID - Name: ct.Name(), - Aliases: []string{ct.Name(), fmt.Sprintf("%d", ct.VMID()), ct.ID()}, - HostUID: ct.Node(), - HostName: ct.Node(), - VMID: ct.VMID(), - Node: ct.Node(), - LocationChain: []string{"node:" + ct.Node(), "system-container:" + ct.Name()}, - Executors: []ExecutorRegistration{{ - ExecutorID: ct.Node(), - Adapter: "pct", - Actions: guestExecutorActions(), - Priority: 10, - }}, - }) - - return NewJSONResult(response.NormalizeCollections()), nil - } + if result, ok := guestViewGetResult(e, rs.Containers(), "system-container", "pct", resourceID, governance); ok { + return result, nil } - return NewJSONResult(map[string]interface{}{ - "error": "not_found", - "resource_id": resourceID, - "type": "system-container", - }), nil + return notFoundResourceResult("system-container", resourceID), nil case "app-container": if resource, containerID, ok := findCanonicalAppContainerResource(e.unifiedResourceProvider, resourceID); ok { @@ -5543,123 +5619,15 @@ func (e *PulseToolExecutor) executeSearchResources(_ context.Context, args map[s } if (typeFilter == "" || typeFilter == "vm") && len(vmResources) > 0 { - for _, resource := range vmResources { - status := string(resource.Status) - if !statusMatchesFilter(status, statusFilter) { - continue - } - candidates := canonicalGuestSearchCandidates("vm", resource) - if !matchesQuery(queryLower, candidates...) { - continue - } - - vmid := 0 - if resource.Proxmox != nil && resource.Proxmox.VMID > 0 { - vmid = resource.Proxmox.VMID - } - node := canonicalGuestTarget(resource) - metadataCandidates := append([]string{resourceDisplayName(resource), resource.ID}, candidates...) - addMatch(ResourceMatch{ - GovernedResourceMetadata: governance.Resolve(metadataCandidates...), - Type: "vm", - ID: resource.ID, - Name: resourceDisplayName(resource), - Status: status, - Node: node, - NodeHasAgent: connectedAgentHostnames[node], - Platform: canonicalResourcePlatform(resource), - VMID: vmid, - AgentConnected: resourceAgentConnected(resource, connectedAgentHostnames), - }) - } + addCanonicalGuestSearchMatches("vm", vmResources, queryLower, statusFilter, governance, connectedAgentHostnames, matchesQuery, addMatch) } else if typeFilter == "" || typeFilter == "vm" { - for _, vm := range rs.VMs() { - status := string(vm.Status()) - if !statusMatchesFilter(status, statusFilter) { - continue - } - // Build searchable candidates: name, ID, VMID, canonical type-prefixed VMID, IPs, tags - vmidStr := fmt.Sprintf("%d", vm.VMID()) - candidates := []string{vm.Name(), vm.ID(), vmidStr, "vm" + vmidStr} - candidates = append(candidates, vm.IPAddresses()...) - candidates = append(candidates, vm.Tags()...) - - if !matchesQuery(queryLower, candidates...) { - continue - } - addMatch(ResourceMatch{ - GovernedResourceMetadata: governance.Resolve(vm.Name(), vm.ID(), vmidStr), - Type: "vm", - ID: vm.ID(), - Name: vm.Name(), - Status: status, - Node: vm.Node(), - NodeHasAgent: connectedAgentHostnames[vm.Node()], - Platform: "proxmox", - VMID: vm.VMID(), - AgentConnected: connectedAgentHostnames[vm.Name()], - }) - } + addGuestViewSearchMatches("vm", rs.VMs(), queryLower, statusFilter, governance, connectedAgentHostnames, matchesQuery, addMatch) } if (typeFilter == "" || typeFilter == "system-container") && len(systemContainerResources) > 0 { - for _, resource := range systemContainerResources { - status := string(resource.Status) - if !statusMatchesFilter(status, statusFilter) { - continue - } - candidates := canonicalGuestSearchCandidates("system-container", resource) - if !matchesQuery(queryLower, candidates...) { - continue - } - - vmid := 0 - if resource.Proxmox != nil && resource.Proxmox.VMID > 0 { - vmid = resource.Proxmox.VMID - } - node := canonicalGuestTarget(resource) - metadataCandidates := append([]string{resourceDisplayName(resource), resource.ID}, candidates...) - addMatch(ResourceMatch{ - GovernedResourceMetadata: governance.Resolve(metadataCandidates...), - Type: "system-container", - ID: resource.ID, - Name: resourceDisplayName(resource), - Status: status, - Node: node, - NodeHasAgent: connectedAgentHostnames[node], - Platform: canonicalResourcePlatform(resource), - VMID: vmid, - AgentConnected: resourceAgentConnected(resource, connectedAgentHostnames), - }) - } + addCanonicalGuestSearchMatches("system-container", systemContainerResources, queryLower, statusFilter, governance, connectedAgentHostnames, matchesQuery, addMatch) } else if typeFilter == "" || typeFilter == "system-container" { - for _, ct := range rs.Containers() { - status := string(ct.Status()) - if !statusMatchesFilter(status, statusFilter) { - continue - } - // Build searchable candidates: name, ID, VMID, canonical type-prefixed VMID, IPs, tags - vmidStr := fmt.Sprintf("%d", ct.VMID()) - candidates := []string{ct.Name(), ct.ID(), vmidStr, "system-container" + vmidStr} - candidates = append(candidates, ct.IPAddresses()...) - candidates = append(candidates, ct.Tags()...) - - if !matchesQuery(queryLower, candidates...) { - continue - } - addMatch(ResourceMatch{ - GovernedResourceMetadata: governance.Resolve(ct.Name(), ct.ID(), vmidStr), - Type: "system-container", - ID: ct.ID(), - Name: ct.Name(), - Status: status, - Node: ct.Node(), - NodeHasAgent: connectedAgentHostnames[ct.Node()], - Platform: "proxmox", - VMID: ct.VMID(), - AgentConnected: connectedAgentHostnames[ct.Name()], - }) - } + addGuestViewSearchMatches("system-container", rs.Containers(), queryLower, statusFilter, governance, connectedAgentHostnames, matchesQuery, addMatch) } if typeFilter == "" || typeFilter == "docker-host" { diff --git a/internal/ai/tools/tools_storage.go b/internal/ai/tools/tools_storage.go index ddca85077..c5b108f11 100644 --- a/internal/ai/tools/tools_storage.go +++ b/internal/ai/tools/tools_storage.go @@ -1547,6 +1547,100 @@ func toolHostLabel(host *unifiedresources.HostView) string { return toolHostTargetID(host) } +// guestDiskView is the read-state view method subset the disk-summary path +// consumes; *unifiedresources.VMView and *unifiedresources.ContainerView both +// satisfy it. +type guestDiskView interface { + comparable + ID() string + SourceID() string + Name() string + VMID() int + Node() string + Instance() string + Disks() []unifiedresources.DiskInfo +} + +// appendGuestDiskSummaries appends per-guest disk summaries for one guest +// family (VMs or system containers), applying the shared resource/instance/ +// min-usage filters and usage derivation. +func appendGuestDiskSummaries[V guestDiskView](resources []ResourceDisksSummary, guests []V, kind, resourceFilter, instanceFilter string, minUsage float64) []ResourceDisksSummary { + var zero V + for _, g := range guests { + if g == zero { + continue + } + + guestVMID := g.VMID() + guestVMIDStr := strconv.Itoa(guestVMID) + guestSourceID := strings.TrimSpace(g.SourceID()) + if guestSourceID == "" { + guestSourceID = g.ID() + } + + // Apply filters + if resourceFilter != "" && guestSourceID != resourceFilter && guestVMIDStr != resourceFilter { + continue + } + if instanceFilter != "" && g.Instance() != instanceFilter { + continue + } + + guestDisks := g.Disks() + // Skip guests without disk data + if len(guestDisks) == 0 { + continue + } + + var disks []ResourceDiskInfo + maxUsage := 0.0 + + for _, disk := range guestDisks { + usage := disk.Usage + if usage <= 0 && disk.Total > 0 { + usage = (float64(disk.Used) / float64(disk.Total)) * 100 + } + if usage <= 0 && disk.Used > 0 { + usage = 100 + } + + if usage > maxUsage { + maxUsage = usage + } + + disks = append(disks, ResourceDiskInfo{ + Device: disk.Device, + Mountpoint: disk.Mountpoint, + Type: disk.Filesystem, + TotalBytes: disk.Total, + UsedBytes: disk.Used, + FreeBytes: disk.Free, + Usage: usage, + }) + } + + // Apply min_usage filter + if minUsage > 0 && maxUsage < minUsage { + continue + } + + if disks == nil { + disks = []ResourceDiskInfo{} + } + + resources = append(resources, ResourceDisksSummary{ + ID: guestSourceID, + VMID: guestVMID, + Name: g.Name(), + Type: kind, + Node: g.Node(), + Instance: g.Instance(), + Disks: disks, + }) + } + return resources +} + func (e *PulseToolExecutor) executeGetResourceDisks(_ context.Context, args map[string]interface{}) (CallToolResult, error) { rs, err := e.readStateForControl() if err != nil { @@ -1567,154 +1661,12 @@ func (e *PulseToolExecutor) executeGetResourceDisks(_ context.Context, args map[ // Process VMs if typeFilter == "" || typeFilter == "vm" { - for _, vm := range rs.VMs() { - if vm == nil { - continue - } - - vmID := vm.VMID() - vmIDStr := strconv.Itoa(vmID) - vmSourceID := strings.TrimSpace(vm.SourceID()) - if vmSourceID == "" { - vmSourceID = vm.ID() - } - - // Apply filters - if resourceFilter != "" && vmSourceID != resourceFilter && vmIDStr != resourceFilter { - continue - } - if instanceFilter != "" && vm.Instance() != instanceFilter { - continue - } - - vmDisks := vm.Disks() - // Skip VMs without disk data - if len(vmDisks) == 0 { - continue - } - - var disks []ResourceDiskInfo - maxUsage := 0.0 - - for _, disk := range vmDisks { - usage := disk.Usage - if usage <= 0 && disk.Total > 0 { - usage = (float64(disk.Used) / float64(disk.Total)) * 100 - } - if usage <= 0 && disk.Used > 0 { - usage = 100 - } - - if usage > maxUsage { - maxUsage = usage - } - - disks = append(disks, ResourceDiskInfo{ - Device: disk.Device, - Mountpoint: disk.Mountpoint, - Type: disk.Filesystem, - TotalBytes: disk.Total, - UsedBytes: disk.Used, - FreeBytes: disk.Free, - Usage: usage, - }) - } - - // Apply min_usage filter - if minUsage > 0 && maxUsage < minUsage { - continue - } - - if disks == nil { - disks = []ResourceDiskInfo{} - } - - resources = append(resources, ResourceDisksSummary{ - ID: vmSourceID, - VMID: vmID, - Name: vm.Name(), - Type: "vm", - Node: vm.Node(), - Instance: vm.Instance(), - Disks: disks, - }) - } + resources = appendGuestDiskSummaries(resources, rs.VMs(), "vm", resourceFilter, instanceFilter, minUsage) } // Process containers if typeFilter == "" || typeFilter == "system-container" { - for _, ct := range rs.Containers() { - if ct == nil { - continue - } - - ctID := ct.VMID() - ctIDStr := strconv.Itoa(ctID) - ctSourceID := strings.TrimSpace(ct.SourceID()) - if ctSourceID == "" { - ctSourceID = ct.ID() - } - - // Apply filters - if resourceFilter != "" && ctSourceID != resourceFilter && ctIDStr != resourceFilter { - continue - } - if instanceFilter != "" && ct.Instance() != instanceFilter { - continue - } - - ctDisks := ct.Disks() - // Skip containers without disk data - if len(ctDisks) == 0 { - continue - } - - var disks []ResourceDiskInfo - maxUsage := 0.0 - - for _, disk := range ctDisks { - usage := disk.Usage - if usage <= 0 && disk.Total > 0 { - usage = (float64(disk.Used) / float64(disk.Total)) * 100 - } - if usage <= 0 && disk.Used > 0 { - usage = 100 - } - - if usage > maxUsage { - maxUsage = usage - } - - disks = append(disks, ResourceDiskInfo{ - Device: disk.Device, - Mountpoint: disk.Mountpoint, - Type: disk.Filesystem, - TotalBytes: disk.Total, - UsedBytes: disk.Used, - FreeBytes: disk.Free, - Usage: usage, - }) - } - - // Apply min_usage filter - if minUsage > 0 && maxUsage < minUsage { - continue - } - - if disks == nil { - disks = []ResourceDiskInfo{} - } - - resources = append(resources, ResourceDisksSummary{ - ID: ctSourceID, - VMID: ctID, - Name: ct.Name(), - Type: "system-container", - Node: ct.Node(), - Instance: ct.Instance(), - Disks: disks, - }) - } + resources = appendGuestDiskSummaries(resources, rs.Containers(), "system-container", resourceFilter, instanceFilter, minUsage) } if resources == nil { diff --git a/internal/ai/unified/alerts.go b/internal/ai/unified/alerts.go index c7c5eb75b..f8efb62cb 100644 --- a/internal/ai/unified/alerts.go +++ b/internal/ai/unified/alerts.go @@ -66,7 +66,12 @@ type UnifiedFindingLifecycleEvent struct { } // UnifiedFinding represents a unified finding that can originate from -// threshold alerts, AI analysis, or other detection methods +// threshold alerts, AI analysis, or other detection methods. +// unifiedFindingJSON deliberately mirrors this struct field-for-field +// (marshal-mirror pattern); merging them would change the public literal API. +// TestUnifiedFindingJSONMirrorStaysInSync enforces the mirror. +// +//nolint:dupl // deliberate marshal-mirror twin of unifiedFindingJSON type UnifiedFinding struct { ID string `json:"id"` Source FindingSource `json:"source"` @@ -136,6 +141,12 @@ type UnifiedFinding struct { RemindAt *time.Time `json:"remind_at,omitempty"` } +// unifiedFindingJSON is the marshal mirror of UnifiedFinding: identical +// fields, but AlertIdentifier round-trips as "alert_identifier" in +// persistence while the public struct hides it with `json:"-"`. The mirror is +// enforced by TestUnifiedFindingJSONMirrorStaysInSync. +// +//nolint:dupl // deliberate marshal-mirror twin of UnifiedFinding type unifiedFindingJSON struct { ID string `json:"id"` Source FindingSource `json:"source"` diff --git a/internal/ai/unified/alerts_mirror_test.go b/internal/ai/unified/alerts_mirror_test.go new file mode 100644 index 000000000..4a6bf628b --- /dev/null +++ b/internal/ai/unified/alerts_mirror_test.go @@ -0,0 +1,47 @@ +package unified + +import ( + "reflect" + "testing" +) + +// TestUnifiedFindingJSONMirrorStaysInSync enforces the deliberate +// marshal-mirror relationship between UnifiedFinding and unifiedFindingJSON: +// identical field names and types, identical json tags, with AlertIdentifier +// as the only sanctioned divergence (hidden via `json:"-"` on the public +// struct, round-tripped as "alert_identifier" by the mirror). +func TestUnifiedFindingJSONMirrorStaysInSync(t *testing.T) { + public := reflect.TypeOf(UnifiedFinding{}) + mirror := reflect.TypeOf(unifiedFindingJSON{}) + tagOverrides := map[string]string{"AlertIdentifier": "alert_identifier,omitempty"} + + mirrorFields := make(map[string]reflect.StructField, mirror.NumField()) + for i := 0; i < mirror.NumField(); i++ { + f := mirror.Field(i) + mirrorFields[f.Name] = f + } + + if public.NumField() != mirror.NumField() { + t.Errorf("%s has %d fields but %s has %d — keep the marshal mirror in sync", + public.Name(), public.NumField(), mirror.Name(), mirror.NumField()) + } + + for i := 0; i < public.NumField(); i++ { + pf := public.Field(i) + mf, ok := mirrorFields[pf.Name] + if !ok { + t.Errorf("%s.%s is missing from mirror %s", public.Name(), pf.Name, mirror.Name()) + continue + } + if pf.Type != mf.Type { + t.Errorf("%s.%s type %s diverged from mirror type %s", public.Name(), pf.Name, pf.Type, mf.Type) + } + wantTag := pf.Tag.Get("json") + if override, isOverride := tagOverrides[pf.Name]; isOverride { + wantTag = override + } + if got := mf.Tag.Get("json"); got != wantTag { + t.Errorf("%s.%s mirror json tag = %q, want %q", public.Name(), pf.Name, got, wantTag) + } + } +}