mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-07-09 16:00:59 +00:00
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.
This commit is contained in:
rcourtman
parent
b855b21c7f
commit
8439ce6e6b
15 changed files with 672 additions and 828 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"`
|
||||
|
|
|
|||
56
internal/ai/findings_marshal_mirror_test.go
Normal file
56
internal/ai/findings_marshal_mirror_test.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
})
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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),
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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" {
|
||||
|
|
|
|||
|
|
@ -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 {
|
||||
|
|
|
|||
|
|
@ -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"`
|
||||
|
|
|
|||
47
internal/ai/unified/alerts_mirror_test.go
Normal file
47
internal/ai/unified/alerts_mirror_test.go
Normal file
|
|
@ -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)
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue