package tools import ( "context" "strings" "sync" "time" "github.com/rcourtman/pulse-go-rewrite/internal/agentcapabilities" "github.com/rcourtman/pulse-go-rewrite/internal/agentexec" "github.com/rcourtman/pulse-go-rewrite/internal/models" "github.com/rcourtman/pulse-go-rewrite/internal/recovery" "github.com/rcourtman/pulse-go-rewrite/internal/unifiedresources" "github.com/rcourtman/pulse-go-rewrite/pkg/reporting" "github.com/rs/zerolog/log" ) // UnifiedResourceProvider gives the executor access to the unified resource // registry so that tool handlers can read physical disks, Ceph clusters, etc. // from the canonical model instead of raw StateSnapshot fields. type UnifiedResourceProvider interface { GetByType(t unifiedresources.ResourceType) []unifiedresources.Resource } // ServerVersion is the version of the shared Assistant tool implementation const ServerVersion = "1.0.0" // StateProvider is a type alias for models.SnapshotProvider. // Kept for local convenience; all new code should use models.SnapshotProvider directly. type StateProvider = models.SnapshotProvider // RecoveryPointsProvider provides paged access to persisted recovery points. // Tool handlers should prefer this over legacy state backup arrays when present. type RecoveryPointsProvider interface { ListPoints(ctx context.Context, opts recovery.ListPointsOptions) ([]recovery.RecoveryPoint, int, error) } // CommandPolicy evaluates command security type CommandPolicy interface { Evaluate(command string) agentexec.PolicyDecision } // AgentServer executes commands on agents type AgentServer interface { GetConnectedAgents() []agentexec.ConnectedAgent ExecuteCommand(ctx context.Context, agentID string, cmd agentexec.ExecuteCommandPayload) (*agentexec.CommandResultPayload, error) } // MetadataUpdater updates resource metadata type MetadataUpdater interface { SetResourceURL(resourceType, resourceID, url string) error } // FindingsManager manages patrol findings type FindingsManager interface { ResolveFinding(findingID, note string) error DismissFinding(findingID, reason, note string) error } // MetricsHistoryProvider provides historical metrics for trend analysis type MetricsHistoryProvider interface { GetResourceMetrics(resourceID string, period time.Duration) ([]MetricPoint, error) GetAllMetricsSummary(period time.Duration) (map[string]ResourceMetricsSummary, error) } // BaselineProvider provides learned baselines for anomaly detection type BaselineProvider interface { GetBaseline(resourceID, metric string) *MetricBaseline GetAllBaselines() map[string]map[string]*MetricBaseline // resourceID -> metric -> baseline } // PatternProvider provides detected patterns and predictions type PatternProvider interface { GetPatterns() []Pattern GetPredictions() []Prediction } // AlertProvider provides active alerts type AlertProvider interface { GetActiveAlerts() []ActiveAlert GetRecentlyResolved(minutes int) []models.ResolvedAlert } // FindingsProvider provides patrol findings type FindingsProvider interface { GetActiveFindings() []Finding GetDismissedFindings() []Finding } // PatrolFindingCreator is set on the executor during a patrol run to allow // patrol-specific tools (patrol_report_finding, patrol_assess_finding, // patrol_resolve_finding, patrol_get_findings) to create, assess, resolve, // and query findings. // Outside of a patrol run this is nil, and the tools return a clear error. type PatrolFindingCreator interface { CreateFinding(input PatrolFindingInput) (findingID string, isNew bool, err error) ResolveFinding(findingID, reason string) error GetActiveFindings(resourceID, minSeverity string) []PatrolFindingInfo } // PatrolObserverProposer is attached only for a first-party Patrol detection // run. It accepts model-authored observer artifacts at proposed state; it does // not grant validation, installation, execution, or infrastructure mutation. type PatrolObserverProposer interface { ProposeObserver(input PatrolObserverProposalInput) (PatrolObserverProposalResult, error) } type PatrolObserverProposalInput struct { ObjectiveID string ExpectedRevision uint64 EvidenceFit string Interpretation string TriggerKind string ProbeJSON string WakeEvidence string RequirementsJSON string } type PatrolObserverProposalResult struct { ObjectiveID string `json:"objective_id"` Revision uint64 `json:"revision"` ObserverID string `json:"observer_id"` Version uint64 `json:"observer_version"` State string `json:"state"` ArtifactDigest string `json:"artifact_digest"` CoverageState string `json:"coverage_state"` CoverageReason string `json:"coverage_reason"` } // PatrolFindingAssessor is the additive explicit-verdict extension implemented // by the current Patrol adapter. Keeping it separate preserves compatibility // with narrow test and extension adapters that only implement legacy finding // creation/query/resolve behavior. type PatrolFindingAssessor interface { AssessFinding(input PatrolFindingAssessmentInput) error } // PatrolFindingAlreadyDecidedError is returned when the model tries to assess // a finding whose accepted report already created it in the same Patrol run. // The tool boundary treats this typed condition as an idempotent no-op rather // than a failed call: the report is the authoritative new-issue verdict. type PatrolFindingAlreadyDecidedError struct { FindingID string } func (e *PatrolFindingAlreadyDecidedError) Error() string { return "finding " + e.FindingID + " was already created by an accepted report in this patrol run" } // PatrolFindingAlreadyReportedError is returned when the model repeats a // stable finding report whose first accepted write happened in this Patrol // run. The tool boundary treats this as an idempotent success: one model turn // cannot turn its own report into a recurrence or refresh its heartbeat. type PatrolFindingAlreadyReportedError struct { FindingID string } func (e *PatrolFindingAlreadyReportedError) Error() string { return "finding " + e.FindingID + " was already reported in this patrol run" } // PatrolFindingOwnedByAlertsError is returned when a proposed finding merely // restates a current resource-down condition that Pulse's real-time alerting // layer already owns. The tool boundary acknowledges the model decision as an // idempotent no-op so it cannot create duplicate operator work. type PatrolFindingOwnedByAlertsError struct { FindingID string } func (e *PatrolFindingOwnedByAlertsError) Error() string { return "finding " + e.FindingID + " restates resource state owned by real-time alerts" } // PatrolFindingAssessmentInput is the explicit terminal verdict for an active // finding that was presented to the model during one Patrol run. type PatrolFindingAssessmentInput struct { FindingID string `json:"finding_id"` Verdict string `json:"verdict"` Evidence string `json:"evidence"` Reason string `json:"reason"` } // PatrolFindingsChecker tracks whether Patrol core or a patrol run has loaded // existing findings. Tools use this to prevent lifecycle writes from bypassing // the deduplication and assessment snapshot. type PatrolFindingsChecker interface { HasCheckedFindings() bool } // PatrolCompleteFindingsChecker distinguishes a complete exact-scope snapshot // from a filtered findings read. Patrol core establishes this snapshot before // a detection model runs so the provider does not have to perform deterministic // lifecycle bookkeeping before it can make an evidence-owned decision. type PatrolCompleteFindingsChecker interface { HasCompleteFindingSnapshot() bool } // PatrolFindingInput contains the structured parameters the LLM passes // to the patrol_report_finding tool. type PatrolFindingInput struct { Key string `json:"key"` Severity string `json:"severity"` Category string `json:"category"` ResourceID string `json:"resource_id"` ResourceName string `json:"resource_name"` ResourceType string `json:"resource_type"` Title string `json:"title"` Description string `json:"description"` Impact string `json:"impact,omitempty"` Recommendation string `json:"recommendation,omitempty"` Evidence string `json:"evidence,omitempty"` } // PatrolFindingInfo is a lightweight view of a finding returned by // PatrolFindingCreator.GetActiveFindings. type PatrolFindingInfo struct { ID string `json:"id"` Key string `json:"key,omitempty"` Severity string `json:"severity"` Category string `json:"category"` ResourceID string `json:"resource_id"` ResourceName string `json:"resource_name"` ResourceType string `json:"resource_type"` Title string `json:"title"` Description string `json:"description"` DetectedAt string `json:"detected_at"` } // BackupProvider provides backup information type BackupProvider interface { GetBackups() models.Backups GetPBSInstances() []models.PBSInstance } // ReplicationProvider provides replication job information. type ReplicationProvider interface { GetReplicationJobs() []models.ReplicationJob } // ConnectionHealthProvider provides instance connection health information. type ConnectionHealthProvider interface { GetConnectionHealth() map[string]bool } // GuestConfigProvider provides guest configuration data (VM/system container). type GuestConfigProvider interface { GetGuestConfig(guestType, instance, node string, vmID int) (map[string]interface{}, error) } // AppContainerConfigRequest describes a canonical configuration read against an // API-backed app-container resource. type AppContainerConfigRequest struct { OrgID string ResourceID string ProviderUID string Name string Host string Platform string } // AppContainerConfigResult captures canonical configuration/state metadata for // an API-backed app-container resource. type AppContainerConfigResult struct { ResourceID string ProviderUID string Name string Host string Platform string Status string Version string HumanVersion string Notes string CustomApp bool UpgradeAvailable bool ImageUpdatesAvailable bool ContainerCount int UsedHostIPs []string Images []string Ports []PortInfo Networks []NetworkInfo Mounts []MountInfo Containers []AppContainerConfigContainer } // AppContainerConfigProvider executes canonical configuration reads for // API-backed app-container resources such as TrueNAS-managed applications. type AppContainerConfigProvider interface { GetConfig(ctx context.Context, req AppContainerConfigRequest) (*AppContainerConfigResult, error) } // DiskHealthProvider provides disk health information from host agents type DiskHealthProvider interface { GetHosts() []*unifiedresources.HostView } // UpdatesProvider provides Docker update operations for Assistant tools type UpdatesProvider interface { GetPendingUpdates(hostID string) []ContainerUpdateInfo TriggerUpdateCheck(hostID string) (DockerCommandStatus, error) UpdateContainer(hostID, containerID, containerName string) (DockerCommandStatus, error) // GetCommandStatus reports the current status of a previously queued // update command so callers can verify the command actually applied. GetCommandStatus(commandID string) (DockerCommandStatus, bool) IsUpdateActionsEnabled() bool } // AppContainerActionRequest describes a canonical control action against an // API-backed app-container resource. type AppContainerActionRequest struct { OrgID string ResourceID string ProviderUID string Name string Host string Platform string Action string } // AppContainerActionResult captures the post-action view of an app-container // after a native provider operation completes. type AppContainerActionResult struct { ResourceID string ProviderUID string Name string Host string Platform string Action string Status string Output string } // AppContainerActionProvider executes canonical actions for API-backed // app-container resources such as TrueNAS-managed applications. type AppContainerActionProvider interface { ExecuteAction(ctx context.Context, req AppContainerActionRequest) (*AppContainerActionResult, error) } // TypedActionPlanner is the only model-facing infrastructure mutation seam. // It can persist a canonical lifecycle plan but exposes no executor or // transport authority to the model tool runtime. type TypedActionPlanner interface { PlanTypedAction(ctx context.Context, orgID string, req unifiedresources.ActionRequest) (*unifiedresources.ActionPlan, error) } // AppContainerReadRequest describes a canonical read-only diagnostic request // against an API-backed app-container resource. type AppContainerReadRequest struct { OrgID string ResourceID string ProviderUID string Name string Host string Platform string Container string Lines int } // AppContainerReadResult captures a bounded read-only diagnostic result for an // API-backed app-container. type AppContainerReadResult struct { ResourceID string ProviderUID string Name string Host string Platform string Container string Lines int Output string } // AppContainerReadProvider executes canonical read-only diagnostics for // API-backed app-container resources such as TrueNAS-managed applications. type AppContainerReadProvider interface { ReadLogs(ctx context.Context, req AppContainerReadRequest) (*AppContainerReadResult, error) } // DiscoveryProvider provides AI-powered infrastructure discovery type DiscoveryProvider interface { GetDiscovery(id string) (*ResourceDiscoveryInfo, error) GetDiscoveryByResource(resourceType, targetID, resourceID string) (*ResourceDiscoveryInfo, error) ListDiscoveries() ([]*ResourceDiscoveryInfo, error) ListDiscoveriesByType(resourceType string) ([]*ResourceDiscoveryInfo, error) ListDiscoveriesByTarget(targetID string) ([]*ResourceDiscoveryInfo, error) FormatForAIContext(discoveries []*ResourceDiscoveryInfo) string // TriggerDiscovery initiates discovery for a resource and returns the result TriggerDiscovery(ctx context.Context, resourceType, targetID, resourceID string, force bool) (*ResourceDiscoveryInfo, error) } // ResolvedResourceInfo contains the minimal information needed for tool validation. // This is an interface to avoid import cycles with the chat package. type ResolvedResourceInfo interface { GetResourceID() string GetResourceType() string GetTargetHost() string GetAgentID() string GetAdapter() string GetVMID() int GetNode() string GetAllowedActions() []string // New structured identity methods GetProviderUID() string GetKind() string GetAliases() []string } // ResourceRegistration contains all fields needed to register a discovered resource. // This structured approach replaces the long parameter list for clarity. type ResourceRegistration struct { // Identity Kind string // Technology/transport kind: "node", "vm", "system-container", "app-container", etc. (drives routing) ProviderUID string // Stable provider ID (container ID, VMID, pod UID) Name string // Primary display name Aliases []string // Additional names that resolve to this resource // Scope HostUID string HostName string ParentUID string ParentKind string ClusterUID string Namespace string // Legacy fields (for backwards compatibility) VMID int Node string LocationChain []string // Executor paths Executors []ExecutorRegistration } // ExecutorRegistration describes how an executor can reach a resource. type ExecutorRegistration struct { ExecutorID string Adapter string Actions []string Priority int } // ResolvedContextProvider provides session-scoped resource resolution. // Query and discovery tools add resources; action tools validate against them. // This interface is implemented by the chat package's ResolvedContext. type ResolvedContextProvider interface { // AddResolvedResource adds a resource that was found via query/discovery. // Uses the new structured registration format. AddResolvedResource(reg ResourceRegistration) // GetResolvedResourceByID retrieves a resource by its canonical ID (kind:provider_uid) GetResolvedResourceByID(resourceID string) (ResolvedResourceInfo, bool) // GetResolvedResourceByAlias retrieves a resource by any of its aliases GetResolvedResourceByAlias(alias string) (ResolvedResourceInfo, bool) // ValidateResourceForAction checks if a resource can perform an action // Returns the resource if valid, error if not found or action not allowed ValidateResourceForAction(resourceID, action string) (ResolvedResourceInfo, error) // HasAnyResources returns true if at least one resource has been discovered HasAnyResources() bool // WasRecentlyAccessed checks if a resource was accessed within the given time window. // Used for routing validation to distinguish "this turn" from "session-wide" context. WasRecentlyAccessed(resourceID string, window time.Duration) bool // GetRecentlyAccessedResources returns resource IDs accessed within the given time window. GetRecentlyAccessedResources(window time.Duration) []string // MarkExplicitAccess marks a resource as recently accessed, indicating user intent. // Call this for single-resource operations (get, explicit select) but NOT for bulk // operations (list, search) to avoid poisoning routing validation. MarkExplicitAccess(resourceID string) } // RecentAccessWindow is the time window used to determine "recently referenced" resources. // Resources accessed within this window are considered to be from the current turn/exchange. const RecentAccessWindow = 30 * time.Second // ResourceDiscoveryInfo represents discovered information about a resource type ResourceDiscoveryInfo struct { ID string `json:"id"` ResourceType string `json:"resource_type"` ResourceID string `json:"resource_id"` TargetID string `json:"target_id,omitempty"` AgentID string `json:"agent_id,omitempty"` Hostname string `json:"hostname"` ServiceType string `json:"service_type"` ServiceName string `json:"service_name"` ServiceVersion string `json:"service_version"` Category string `json:"category"` CLIAccess string `json:"cli_access"` Facts []DiscoveryFact `json:"facts"` ConfigPaths []string `json:"config_paths"` DataPaths []string `json:"data_paths"` LogPaths []string `json:"log_paths,omitempty"` // Log file paths or commands (e.g., journalctl) Ports []DiscoveryPortInfo `json:"ports"` BindMounts []DiscoveryMount `json:"bind_mounts,omitempty"` // For Docker: host->container path mappings SuggestedURL string `json:"suggested_url,omitempty"` UserNotes string `json:"user_notes,omitempty"` Confidence float64 `json:"confidence"` AIReasoning string `json:"ai_reasoning,omitempty"` DiscoveredAt time.Time `json:"discovered_at"` UpdatedAt time.Time `json:"updated_at"` } // DiscoveryPortInfo represents a listening port discovered on a resource type DiscoveryPortInfo struct { Port int `json:"port"` Protocol string `json:"protocol"` Process string `json:"process,omitempty"` Address string `json:"address,omitempty"` } // DiscoveryMount represents a bind mount (host path -> container path) type DiscoveryMount struct { ContainerName string `json:"container_name,omitempty"` // Docker container name (for Docker inside LXC/VM) Source string `json:"source"` // Host path (where to actually write files) Destination string `json:"destination"` // Container path (what the service sees) Type string `json:"type,omitempty"` // Mount type: bind, volume, tmpfs ReadOnly bool `json:"read_only,omitempty"` } // DiscoveryFact represents a discovered fact about a resource type DiscoveryFact struct { Category string `json:"category"` Key string `json:"key"` Value string `json:"value"` Source string `json:"source,omitempty"` Confidence float64 `json:"confidence,omitempty"` // 0-1 confidence for this fact } // ExecutorConfig holds all dependencies for the tool executor type ExecutorConfig struct { // Required providers StateProvider StateProvider Policy CommandPolicy AgentServer AgentServer // Optional providers - patrol context MetricsHistory MetricsHistoryProvider BaselineProvider BaselineProvider PatternProvider PatternProvider AlertProvider AlertProvider FindingsProvider FindingsProvider // Optional providers - infrastructure BackupProvider BackupProvider ReplicationProvider ReplicationProvider ConnectionHealth ConnectionHealthProvider RecoveryPointsProvider RecoveryPointsProvider GuestConfigProvider GuestConfigProvider AppContainerConfigProvider AppContainerConfigProvider DiskHealthProvider DiskHealthProvider UpdatesProvider UpdatesProvider // Optional providers - management MetadataUpdater MetadataUpdater FindingsManager FindingsManager AgentProfileManager AgentProfileManager // Optional providers - intelligence IncidentArchiveProvider IncidentArchiveProvider EventCorrelatorProvider EventCorrelatorProvider KnowledgeStoreProvider KnowledgeStoreProvider // Optional providers - discovery DiscoveryProvider DiscoveryProvider // Optional providers - unified resources UnifiedResourceProvider UnifiedResourceProvider AppContainerActionProvider AppContainerActionProvider AppContainerReadProvider AppContainerReadProvider ActionAuditStore unifiedresources.ResourceStore TypedActionPlanner TypedActionPlanner // Optional typed read access to current infrastructure state. // When provided, tool handlers should prefer this over models.StateSnapshot iteration. ReadState unifiedresources.ReadState // Control settings ControlLevel ControlLevel ProtectedGuests []string // VMIDs that AI cannot control OrgID string // Tenant/org scope for approval records // Optional report-narration providers, used by the pulse_summarize // tool to produce AI-narrated synthesis in chat. When the per-tenant // AI service is configured these are the same interfaces it exposes // to the reporting handler for PDF generation. Absent values cause // the tool to return heuristic narrative instead — identical to // what the report PDF carries when no AI is configured. Defined // here (rather than as separate setters) so chat.Config can wire // them at session construction time alongside the rest of the // executor's providers. ReportNarrator reporting.Narrator ReportFleetNarrator reporting.FleetNarrator ReportFindingsProvider reporting.FindingsProvider } // PulseToolExecutor implements ToolExecutor for Pulse-specific tools type PulseToolExecutor struct { // Core providers stateProvider StateProvider policy CommandPolicy agentServer AgentServer // Patrol context providers metricsHistory MetricsHistoryProvider baselineProvider BaselineProvider patternProvider PatternProvider alertProvider AlertProvider findingsProvider FindingsProvider // Infrastructure context providers backupProvider BackupProvider replicationProvider ReplicationProvider connectionHealth ConnectionHealthProvider // Paged recovery points access for snapshot/backup tools. recoveryPointsProvider RecoveryPointsProvider guestConfigProvider GuestConfigProvider appContainerConfigProvider AppContainerConfigProvider diskHealthProvider DiskHealthProvider updatesProvider UpdatesProvider // Management providers metadataUpdater MetadataUpdater findingsManager FindingsManager agentProfileManager AgentProfileManager // Intelligence providers incidentArchiveProvider IncidentArchiveProvider eventCorrelatorProvider EventCorrelatorProvider knowledgeStoreProvider KnowledgeStoreProvider // Discovery provider discoveryProvider DiscoveryProvider // Unified resources provider unifiedResourceProvider UnifiedResourceProvider appContainerActionProvider AppContainerActionProvider appContainerReadProvider AppContainerReadProvider actionAuditStore unifiedresources.ResourceStore typedActionPlanner TypedActionPlanner // onActionCompleted fires after a terminal-state action audit // (Completed or Failed, including refused-before-dispatch // failures) is persisted. Wired by the API layer through the // per-org chat-service init so the agent SSE stream can publish // action.completed events. nil is safe — the callback is // fire-and-forget on its own goroutine to keep the dispatch hot // path off any consumer's slowness. onActionCompleted func(unifiedresources.ActionAuditRecord) // Typed state reader. Nil means "legacy-only": tools must fall back to StateSnapshot access. readState unifiedresources.ReadState // Control settings controlLevel ControlLevel protectedGuests []string // Current execution context targetType string targetID string isAutonomous bool orgID string // hasExecuteAuthority is request-local transport authority. Control // settings describe operator policy; they never grant a chat-only token // permission to execute infrastructure mutations. hasExecuteAuthority bool executeAuthorityBound bool // denyInfrastructureMutations is the request-local execution // restriction for non-interactive read-only workloads (e.g. Patrol // investigations): every infrastructure-mutating invocation is // blocked by the registry before its handler runs, regardless of // control level or autonomy. Core-owned and never serialized; it is // deliberately separate from isAutonomous, which only suppresses // interactive questions and grants no mutation authority. denyInfrastructureMutations bool // pulseStateAllowlist restricts pulse-state mutations to the named // tools. Nil and an empty map both deny all. pulseStateAllowlist map[string]bool // executionProfile is the core-owned request posture; see // execution_profile.go. executionProfile ExecutionProfile // proposalCapture is the request-local typed-proposal sink for // investigation runs; nil outside them. Clones share the pointer // deliberately: one run, one capture. proposalCapture *ProposalCapture // Session-scoped resolved context for resource validation // This is set per-session by the agentic loop before tool execution resolvedContext ResolvedContextProvider // Telemetry callback for recording metrics // This is optional - if nil, no telemetry is recorded telemetryCallback TelemetryCallback // Patrol finding creator — set only during a patrol run, nil otherwise. // Enables patrol_report_finding, patrol_resolve_finding, patrol_get_findings tools. patrolFindingCreatorMu sync.RWMutex patrolFindingCreator PatrolFindingCreator patrolObserverProposerMu sync.RWMutex patrolObserverProposer PatrolObserverProposer // Report-narration providers, used by pulse_summarize when the // per-tenant AI service is configured. Absent values cause the tool // to fall back to heuristic narrative — identical to the report PDF // behaviour when AI is unconfigured. reportNarrator reporting.Narrator reportFleetNarrator reporting.FleetNarrator reportFindingsProvider reporting.FindingsProvider // Tool registry registry *ToolRegistry } // TelemetryCallback is called when the executor needs to record telemetry. // This allows the chat layer to handle metrics without import cycles. type TelemetryCallback interface { // RecordStrictResolutionBlock records when strict resolution blocks an action RecordStrictResolutionBlock(tool, action string) // RecordRoutingMismatchBlock records when routing validation blocks an operation // that targeted a parent host when a child resource was recently referenced. // targetKind: "node" (the kind being targeted) // childKind: "system-container", "vm", "app-container" (the kind of the more specific resource) RecordRoutingMismatchBlock(tool, targetKind, childKind string) } // NewPulseToolExecutor creates a new Pulse tool executor with the given configuration func NewPulseToolExecutor(cfg ExecutorConfig) *PulseToolExecutor { e := &PulseToolExecutor{ stateProvider: cfg.StateProvider, policy: cfg.Policy, agentServer: cfg.AgentServer, metricsHistory: cfg.MetricsHistory, baselineProvider: cfg.BaselineProvider, patternProvider: cfg.PatternProvider, alertProvider: cfg.AlertProvider, findingsProvider: cfg.FindingsProvider, backupProvider: cfg.BackupProvider, replicationProvider: cfg.ReplicationProvider, connectionHealth: cfg.ConnectionHealth, recoveryPointsProvider: cfg.RecoveryPointsProvider, guestConfigProvider: cfg.GuestConfigProvider, appContainerConfigProvider: cfg.AppContainerConfigProvider, diskHealthProvider: cfg.DiskHealthProvider, updatesProvider: cfg.UpdatesProvider, metadataUpdater: cfg.MetadataUpdater, findingsManager: cfg.FindingsManager, agentProfileManager: cfg.AgentProfileManager, incidentArchiveProvider: cfg.IncidentArchiveProvider, eventCorrelatorProvider: cfg.EventCorrelatorProvider, knowledgeStoreProvider: cfg.KnowledgeStoreProvider, discoveryProvider: cfg.DiscoveryProvider, unifiedResourceProvider: cfg.UnifiedResourceProvider, appContainerActionProvider: cfg.AppContainerActionProvider, appContainerReadProvider: cfg.AppContainerReadProvider, actionAuditStore: cfg.ActionAuditStore, typedActionPlanner: cfg.TypedActionPlanner, readState: cfg.ReadState, controlLevel: cfg.ControlLevel, protectedGuests: cfg.ProtectedGuests, orgID: normalizeExecutorOrgID(cfg.OrgID), reportNarrator: cfg.ReportNarrator, reportFleetNarrator: cfg.ReportFleetNarrator, reportFindingsProvider: cfg.ReportFindingsProvider, registry: NewToolRegistry(), } // Auto-wire backup, replication, and connection health adapters from // stateProvider when no explicit provider was injected. Each closure // captures stateProvider and fetches the specific field on demand — // this avoids importing the full StateGetter interface into adapters.go. if sp := e.stateProvider; sp != nil { getSnapshot := func() models.StateSnapshot { return sp.ReadSnapshot() } if e.backupProvider == nil { e.backupProvider = NewBackupToolAdapter( func() models.Backups { return getSnapshot().Backups }, func() []models.PBSInstance { return getSnapshot().PBSInstances }, ) } if e.replicationProvider == nil { e.replicationProvider = NewReplicationToolAdapter( func() []models.ReplicationJob { return getSnapshot().ReplicationJobs }, ) } if e.connectionHealth == nil { e.connectionHealth = NewConnectionHealthToolAdapter( func() map[string]bool { return getSnapshot().ConnectionHealth }, ) } } // Register all tools e.registerTools() return e } // Clone returns an executor instance with shared immutable providers and fresh // session-scoped mutable state. This prevents concurrent sessions from // overwriting each other's routing and approval context. func (e *PulseToolExecutor) Clone() *PulseToolExecutor { if e == nil { return nil } clone := &PulseToolExecutor{ stateProvider: e.stateProvider, policy: e.policy, agentServer: e.agentServer, metricsHistory: e.metricsHistory, baselineProvider: e.baselineProvider, patternProvider: e.patternProvider, alertProvider: e.alertProvider, findingsProvider: e.findingsProvider, backupProvider: e.backupProvider, replicationProvider: e.replicationProvider, connectionHealth: e.connectionHealth, recoveryPointsProvider: e.recoveryPointsProvider, guestConfigProvider: e.guestConfigProvider, appContainerConfigProvider: e.appContainerConfigProvider, diskHealthProvider: e.diskHealthProvider, updatesProvider: e.updatesProvider, metadataUpdater: e.metadataUpdater, findingsManager: e.findingsManager, agentProfileManager: e.agentProfileManager, incidentArchiveProvider: e.incidentArchiveProvider, eventCorrelatorProvider: e.eventCorrelatorProvider, knowledgeStoreProvider: e.knowledgeStoreProvider, discoveryProvider: e.discoveryProvider, unifiedResourceProvider: e.unifiedResourceProvider, appContainerActionProvider: e.appContainerActionProvider, appContainerReadProvider: e.appContainerReadProvider, actionAuditStore: e.actionAuditStore, typedActionPlanner: e.typedActionPlanner, readState: e.readState, controlLevel: e.controlLevel, protectedGuests: append([]string(nil), e.protectedGuests...), targetType: e.targetType, targetID: e.targetID, isAutonomous: e.isAutonomous, orgID: e.orgID, hasExecuteAuthority: e.hasExecuteAuthority, executeAuthorityBound: e.executeAuthorityBound, denyInfrastructureMutations: e.denyInfrastructureMutations, pulseStateAllowlist: clonePulseStateAllowlist(e.pulseStateAllowlist), executionProfile: e.executionProfile, proposalCapture: e.proposalCapture, telemetryCallback: e.telemetryCallback, reportNarrator: e.reportNarrator, reportFleetNarrator: e.reportFleetNarrator, reportFindingsProvider: e.reportFindingsProvider, registry: e.registry, } clone.patrolFindingCreator = e.GetPatrolFindingCreator() clone.patrolObserverProposer = e.GetPatrolObserverProposer() return clone } func normalizeExecutorOrgID(orgID string) string { normalized := strings.TrimSpace(orgID) if normalized == "" { return "default" } return normalized } func (e *PulseToolExecutor) getReadState() unifiedresources.ReadState { if readState := e.getCanonicalReadState(); readState != nil { return readState } if e.stateProvider == nil { return nil } // Compatibility bridge for legacy-only wiring: derive a typed ReadState view // from the latest snapshot so tool handlers can stay platform-agnostic. rr := unifiedresources.NewRegistry(nil) rr.IngestSnapshot(e.stateProvider.ReadSnapshot()) return rr } func (e *PulseToolExecutor) getCanonicalReadState() unifiedresources.ReadState { if e.readState != nil { return e.readState } if readState, ok := e.unifiedResourceProvider.(unifiedresources.ReadState); ok && readState != nil { return readState } return nil } func (e *PulseToolExecutor) hasReadState() bool { if e.readState != nil || e.stateProvider != nil { return true } readState, ok := e.unifiedResourceProvider.(unifiedresources.ReadState) return ok && readState != nil } func (e *PulseToolExecutor) hasCanonicalReadState() bool { return e.getCanonicalReadState() != nil } // SetContext sets the current execution context func (e *PulseToolExecutor) SetContext(targetType, targetID string, autonomous bool) { e.targetType = targetType e.targetID = targetID e.isAutonomous = autonomous } // SetAutonomousMode updates only the execution mode, preserving any // session-scoped target context already attached to this executor clone. func (e *PulseToolExecutor) SetAutonomousMode(enabled bool) { e.isAutonomous = enabled } // SetOrgID sets the org scope used when creating approval records. func (e *PulseToolExecutor) SetOrgID(orgID string) { e.orgID = normalizeExecutorOrgID(orgID) } // SetControlLevel updates the control level func (e *PulseToolExecutor) SetControlLevel(level ControlLevel) { e.controlLevel = level } // SetExecuteAuthority binds the current request's explicit ai:execute // authority to this executor clone. It is deliberately not serialized and // defaults false. func (e *PulseToolExecutor) SetExecuteAuthority(allowed bool) { e.hasExecuteAuthority = allowed e.executeAuthorityBound = true } // SetProtectedGuests updates the protected guests list func (e *PulseToolExecutor) SetProtectedGuests(vmids []string) { e.protectedGuests = vmids } // RegisterTool allows tests or extensions to add tools at runtime. func (e *PulseToolExecutor) RegisterTool(tool RegisteredTool) { e.registry.RegisterExtension(tool) } // Runtime setter methods for updating providers after creation // SetMetadataUpdater sets the metadata updater func (e *PulseToolExecutor) SetMetadataUpdater(updater MetadataUpdater) { e.metadataUpdater = updater } // SetFindingsManager sets the findings manager func (e *PulseToolExecutor) SetFindingsManager(manager FindingsManager) { e.findingsManager = manager } // SetMetricsHistory sets the metrics history provider func (e *PulseToolExecutor) SetMetricsHistory(provider MetricsHistoryProvider) { e.metricsHistory = provider } // SetRecoveryPointsProvider sets a paged recovery points provider used by snapshot/backup tools. func (e *PulseToolExecutor) SetRecoveryPointsProvider(provider RecoveryPointsProvider) { e.recoveryPointsProvider = provider } // SetBaselineProvider sets the baseline provider func (e *PulseToolExecutor) SetBaselineProvider(provider BaselineProvider) { e.baselineProvider = provider } // SetPatternProvider sets the pattern provider func (e *PulseToolExecutor) SetPatternProvider(provider PatternProvider) { e.patternProvider = provider } // SetAlertProvider sets the alert provider func (e *PulseToolExecutor) SetAlertProvider(provider AlertProvider) { e.alertProvider = provider } // SetFindingsProvider sets the findings provider func (e *PulseToolExecutor) SetFindingsProvider(provider FindingsProvider) { e.findingsProvider = provider } // SetBackupProvider sets the backup provider func (e *PulseToolExecutor) SetBackupProvider(provider BackupProvider) { e.backupProvider = provider } // SetGuestConfigProvider sets the guest config provider func (e *PulseToolExecutor) SetGuestConfigProvider(provider GuestConfigProvider) { e.guestConfigProvider = provider } // SetAppContainerConfigProvider sets the provider used for canonical native // app-container configuration reads. func (e *PulseToolExecutor) SetAppContainerConfigProvider(provider AppContainerConfigProvider) { e.appContainerConfigProvider = provider } // SetDiskHealthProvider sets the disk health provider func (e *PulseToolExecutor) SetDiskHealthProvider(provider DiskHealthProvider) { e.diskHealthProvider = provider } // SetAgentProfileManager sets the agent profile manager func (e *PulseToolExecutor) SetAgentProfileManager(manager AgentProfileManager) { e.agentProfileManager = manager } // SetUpdatesProvider sets the updates provider for Docker container updates func (e *PulseToolExecutor) SetUpdatesProvider(provider UpdatesProvider) { e.updatesProvider = provider } // SetIncidentArchiveProvider sets the read-only legacy incident archive provider func (e *PulseToolExecutor) SetIncidentArchiveProvider(provider IncidentArchiveProvider) { e.incidentArchiveProvider = provider } // SetEventCorrelatorProvider sets the event correlator provider func (e *PulseToolExecutor) SetEventCorrelatorProvider(provider EventCorrelatorProvider) { e.eventCorrelatorProvider = provider } // SetKnowledgeStoreProvider sets the knowledge store provider for notes func (e *PulseToolExecutor) SetKnowledgeStoreProvider(provider KnowledgeStoreProvider) { e.knowledgeStoreProvider = provider } // SetDiscoveryProvider sets the discovery provider for infrastructure discovery func (e *PulseToolExecutor) SetDiscoveryProvider(provider DiscoveryProvider) { e.discoveryProvider = provider } // SetUnifiedResourceProvider sets the unified resource provider func (e *PulseToolExecutor) SetUnifiedResourceProvider(provider UnifiedResourceProvider) { e.unifiedResourceProvider = provider } // SetAppContainerActionProvider sets the provider used for canonical native // app-container control actions. func (e *PulseToolExecutor) SetAppContainerActionProvider(provider AppContainerActionProvider) { e.appContainerActionProvider = provider } // SetAppContainerReadProvider sets the provider used for canonical native // app-container read-only diagnostics. func (e *PulseToolExecutor) SetAppContainerReadProvider(provider AppContainerReadProvider) { e.appContainerReadProvider = provider } // SetActionAuditStore sets the durable store used to persist action audit and lifecycle events. func (e *PulseToolExecutor) SetActionAuditStore(store unifiedresources.ResourceStore) { e.actionAuditStore = store } // SetTypedActionPlanner installs the canonical lifecycle planning boundary. func (e *PulseToolExecutor) SetTypedActionPlanner(planner TypedActionPlanner) { e.typedActionPlanner = planner } // SetOnActionCompleted installs a fire-and-forget callback that runs // after every terminal-state action audit is persisted (Completed or // Failed, including refused-before-dispatch failures with stable // `plan_drift:` / `resource_remediation_locked:` error prefixes). // Pass nil to disable. Used by the API layer to bridge action // completion into the agent SSE stream without coupling the tools // package to the api package. The callback runs on its own // goroutine; consumers must not assume immediate or in-order // delivery relative to subsequent dispatches. func (e *PulseToolExecutor) SetOnActionCompleted(cb func(unifiedresources.ActionAuditRecord)) { if e == nil { return } e.onActionCompleted = cb } // GetActionAuditStore returns the durable store used to persist action audit and lifecycle events. func (e *PulseToolExecutor) GetActionAuditStore() unifiedresources.ResourceStore { return e.actionAuditStore } // SetResolvedContext sets the session-scoped resolved context for resource validation. // This should be called by the agentic loop before executing tools for a session. func (e *PulseToolExecutor) SetResolvedContext(ctx ResolvedContextProvider) { e.resolvedContext = ctx } // SetTelemetryCallback sets the telemetry callback for recording metrics func (e *PulseToolExecutor) SetTelemetryCallback(cb TelemetryCallback) { e.telemetryCallback = cb } // SetPatrolFindingCreator sets (or clears) the patrol finding creator. // This must be set before a patrol run and cleared after. func (e *PulseToolExecutor) SetPatrolFindingCreator(creator PatrolFindingCreator) { e.patrolFindingCreatorMu.Lock() e.patrolFindingCreator = creator e.patrolFindingCreatorMu.Unlock() } // GetPatrolFindingCreator returns the current patrol finding creator (may be nil). func (e *PulseToolExecutor) GetPatrolFindingCreator() PatrolFindingCreator { e.patrolFindingCreatorMu.RLock() defer e.patrolFindingCreatorMu.RUnlock() return e.patrolFindingCreator } // PatrolFindingSnapshotEstablished reports whether the current Patrol adapter // has loaded the complete active-finding set for the exact run scope. Cloned // executors intentionally share the run adapter, so this remains true at the // provider boundary without copying lifecycle state. func (e *PulseToolExecutor) PatrolFindingSnapshotEstablished() bool { if e == nil { return false } checker, ok := e.GetPatrolFindingCreator().(PatrolCompleteFindingsChecker) return ok && checker.HasCompleteFindingSnapshot() } func (e *PulseToolExecutor) SetPatrolObserverProposer(proposer PatrolObserverProposer) { e.patrolObserverProposerMu.Lock() e.patrolObserverProposer = proposer e.patrolObserverProposerMu.Unlock() } func (e *PulseToolExecutor) GetPatrolObserverProposer() PatrolObserverProposer { e.patrolObserverProposerMu.RLock() defer e.patrolObserverProposerMu.RUnlock() return e.patrolObserverProposer } // GetResolvedContext returns the current resolved context (may be nil) func (e *PulseToolExecutor) GetResolvedContext() ResolvedContextProvider { return e.resolvedContext } // invocationPolicy is the request-scoped safety policy for this executor // instance: the session control level plus the deny-infrastructure // restriction. Clones snapshot the policy, so a per-request restriction // on one clone can never leak into concurrent sessions. func (e *PulseToolExecutor) invocationPolicy() InvocationPolicy { return InvocationPolicy{ ControlLevel: e.controlLevel, HasExecuteAuthority: e.hasExecuteAuthority, ExecuteAuthorityBound: e.executeAuthorityBound, DenyInfrastructureMutations: e.denyInfrastructureMutations, PulseStateAllowlist: clonePulseStateAllowlist(e.pulseStateAllowlist), Profile: e.executionProfile, } } // SetDenyInfrastructureMutations toggles the request-local restriction // that blocks every infrastructure-mutating invocation at the registry // boundary, before any handler runs. Intended for non-interactive // read-only workloads such as Patrol investigations. func (e *PulseToolExecutor) SetDenyInfrastructureMutations(deny bool) { e.denyInfrastructureMutations = deny } // ListTools returns the list of available tools func (e *PulseToolExecutor) ListTools() []Tool { tools := e.registry.ListTools(e.invocationPolicy()) if len(tools) == 0 { return tools } available := make([]Tool, 0, len(tools)) for _, tool := range tools { if e.isToolAvailable(tool.Name) { available = append(available, tool) } } return available } // ListToolGovernance returns the governed manifest for currently available tools. func (e *PulseToolExecutor) ListToolGovernance() []ToolGovernanceDescriptor { tools := e.registry.ListToolGovernance(e.invocationPolicy()) if len(tools) == 0 { return tools } available := make([]ToolGovernanceDescriptor, 0, len(tools)) for _, tool := range tools { if e.isToolAvailable(tool.Name) { available = append(available, tool) } } return available } func (e *PulseToolExecutor) isToolAvailable(name string) bool { switch name { // Check tool availability based on primary requirements case agentcapabilities.PulseQueryToolName: return e.hasReadState() case agentcapabilities.PulseMetricsToolName: return e.hasReadState() || e.metricsHistory != nil || e.baselineProvider != nil || e.patternProvider != nil case agentcapabilities.PulseStorageToolName: return e.hasReadState() || e.unifiedResourceProvider != nil || e.backupProvider != nil || e.diskHealthProvider != nil || e.recoveryPointsProvider != nil case agentcapabilities.PulseDockerToolName: return e.hasReadState() || e.updatesProvider != nil case agentcapabilities.PulseKubernetesToolName: return e.hasCanonicalReadState() case agentcapabilities.PulseAlertsToolName: return e.alertProvider != nil || e.findingsProvider != nil || e.findingsManager != nil || e.hasReadState() case agentcapabilities.PulseReadToolName: return e.agentServer != nil || (e.appContainerReadProvider != nil && e.hasReadState()) case agentcapabilities.PulseControlToolName: // Control only creates canonical plans. Execution transports neither // enable planning nor substitute for the installed lifecycle planner. return e.typedActionPlanner != nil && e.hasReadState() case agentcapabilities.PulseFileEditToolName: return e.agentServer != nil case agentcapabilities.PulseDiscoveryToolName: return e.discoveryProvider != nil case agentcapabilities.PulseKnowledgeToolName: return e.actionAuditStore != nil || e.knowledgeStoreProvider != nil || e.incidentArchiveProvider != nil || e.eventCorrelatorProvider != nil case agentcapabilities.PulsePMGToolName: return e.hasReadState() case agentcapabilities.PulseSummarizeToolName: return e.hasReadState() case agentcapabilities.PatrolReportFindingToolName, agentcapabilities.PatrolAssessFindingToolName, agentcapabilities.PatrolResolveFindingToolName, agentcapabilities.PatrolGetFindingsToolName: // Always available when registered; handler checks patrolFindingCreator at runtime return e.GetPatrolFindingCreator() != nil case agentcapabilities.PatrolProposeObserverToolName: return e.GetPatrolObserverProposer() != nil case agentcapabilities.PatrolProposeActionToolName, agentcapabilities.PatrolActionCapabilitiesToolName: // These investigation-only tools share the request-local proposal // capture: it supplies both trusted correlation and the tenant-bound // capability catalog. They must never appear without that boundary. return e.proposalCapture != nil default: return e.hasReadState() } } // ExecuteTool executes a tool and returns the result func (e *PulseToolExecutor) ExecuteTool(ctx context.Context, name string, args map[string]interface{}) (CallToolResult, error) { return e.ExecuteInvocation(ctx, ToolInvocation{Name: name, Arguments: args}) } // ExecuteInvocation runs one tool call carrying its full invocation // envelope (tool-use ID, name, arguments). The ID rides the context so // stateful capture can key on call identity even under the loop's // concurrent per-turn execution. func (e *PulseToolExecutor) ExecuteInvocation(ctx context.Context, inv ToolInvocation) (CallToolResult, error) { log.Debug(). Str("tool", inv.Name). Str("invocation_id", inv.ID). Msg("Executing Pulse tool") return e.registry.Execute(withInvocationID(ctx, inv.ID), e, inv.Name, inv.Arguments) } // registerTools registers all available tools func (e *PulseToolExecutor) registerTools() { // All registry tools are registered below in provider exposure order. // pulse_query - search, get, config, topology, list, health e.registerQueryTools() // pulse_metrics - performance, temperatures, network, diskio, disks, baselines, patterns e.registerMetricsTools() // pulse_storage - pools, config, backups, snapshots, ceph, replication, pbs_jobs, raid, disk_health, resource_disks e.registerStorageTools() // pulse_docker - read-only updates/check/services/tasks/swarm state e.registerDockerTools() // pulse_kubernetes - clusters, nodes, pods, deployments e.registerKubernetesTools() // pulse_alerts - list, findings, resolved, resolve, dismiss e.registerAlertsTools() // pulse_read - read-only operations (exec, file, find, tail, logs) // This is ALWAYS classified as ToolKindRead and never triggers VERIFYING e.registerReadTools() // pulse_control - plan typed resource capabilities only e.registerControlTools() // pulse_discovery - get, run, list discoveries e.registerDiscoveryTools() // pulse_knowledge - remember, recall, incidents, correlate, relationships e.registerKnowledgeTools() // pulse_pmg - status, mail_stats, queues, spam e.registerPMGTools() // patrol_report_finding, patrol_resolve_finding, patrol_get_findings // These are always registered but only functional when patrolFindingCreator is set. e.registerPatrolTools() e.registerProposeTools() // pulse_summarize - retrospective synthesis for one resource or a fleet e.registerSummarizeTools() }