896 lines
35 KiB
Go
896 lines
35 KiB
Go
package agent
|
||
|
||
import (
|
||
"context"
|
||
"crypto/rand"
|
||
"crypto/sha256"
|
||
"encoding/hex"
|
||
"fmt"
|
||
"log/slog"
|
||
"sort"
|
||
"strings"
|
||
"sync"
|
||
"time"
|
||
|
||
"github.com/example/glpi-ai-agent/internal/config"
|
||
"github.com/example/glpi-ai-agent/internal/knowledge"
|
||
"github.com/example/glpi-ai-agent/internal/learning"
|
||
"github.com/example/glpi-ai-agent/internal/metrics"
|
||
"github.com/example/glpi-ai-agent/internal/model"
|
||
"github.com/example/glpi-ai-agent/internal/queue"
|
||
"github.com/example/glpi-ai-agent/internal/state"
|
||
)
|
||
|
||
type GLPI interface {
|
||
Ping(context.Context) error
|
||
ValidateContract(context.Context) error
|
||
ListRecentTickets(context.Context, int, string) ([]model.Ticket, error)
|
||
GetTicket(context.Context, int64) (model.Ticket, error)
|
||
GetFollowups(context.Context, int64) ([]model.Followup, error)
|
||
SetCategory(context.Context, int64, int64) error
|
||
AddFollowup(context.Context, int64, string, bool) error
|
||
GetCategories(context.Context) ([]model.Category, error)
|
||
}
|
||
type AI interface {
|
||
Ping(context.Context) error
|
||
Analyse(context.Context, model.Ticket, []model.Category, []model.KnowledgeHit, []model.KnowledgeHit, model.ContextSnapshot) (model.Decision, error)
|
||
}
|
||
type ContextCollector interface {
|
||
Collect(context.Context, model.Ticket) model.ContextSnapshot
|
||
}
|
||
type Service struct {
|
||
cfg config.Config
|
||
glpi GLPI
|
||
ai AI
|
||
knowledge *knowledge.Store
|
||
learning *learning.Store
|
||
state *state.Store
|
||
q *queue.Queue
|
||
metrics *metrics.Metrics
|
||
policy Policy
|
||
context ContextCollector
|
||
locks sync.Map
|
||
catMu sync.RWMutex
|
||
categories []model.Category
|
||
catAt time.Time
|
||
}
|
||
|
||
func New(cfg config.Config, g GLPI, ai AI, k *knowledge.Store, l *learning.Store, s *state.Store, q *queue.Queue, m *metrics.Metrics, contextCollector ContextCollector) *Service {
|
||
return &Service{cfg: cfg, glpi: g, ai: ai, knowledge: k, learning: l, state: s, q: q, metrics: m, context: contextCollector, policy: NewPolicy(cfg.AutoCategory, cfg.AutoReply, cfg.CategoryConfidence, cfg.ReplyConfidence, cfg.KnowledgeMinScore, cfg.KnowledgeRetrievalFloor, cfg.KnowledgeEvidenceRetrievalWeight, cfg.KnowledgeEvidenceAIWeight, cfg.KnowledgeEvidenceCategoryWeight, cfg.KnowledgeAllowedSources, cfg.KnowledgeAutoReplySources, cfg.CommunicationLanguage, cfg.CommunicationStyle, cfg.CommunicationSalutation, cfg.CommunicationClosing, cfg.CommunicationSignature, cfg.AIContentLabelEnabled, cfg.ContextBlockReplyOnError, cfg.ContextBlockReplyOnIncident, cfg.ContextRelevanceMinScore)}
|
||
}
|
||
func (s *Service) Queue() *queue.Queue { return s.q }
|
||
func (s *Service) Start(ctx context.Context) {
|
||
go s.healthLoop(ctx)
|
||
go s.pollLoop(ctx)
|
||
for i := 0; i < s.cfg.Workers; i++ {
|
||
go s.worker(ctx, i)
|
||
}
|
||
}
|
||
func (s *Service) pollLoop(ctx context.Context) {
|
||
ticker := time.NewTicker(s.cfg.GLPIPollInterval)
|
||
defer ticker.Stop()
|
||
s.poll(ctx)
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
s.poll(ctx)
|
||
}
|
||
}
|
||
}
|
||
func (s *Service) poll(ctx context.Context) {
|
||
tickets, err := s.glpi.ListRecentTickets(ctx, s.cfg.GLPIPollLimit, s.cfg.GLPITicketFilter)
|
||
s.metrics.Polls.Add(1)
|
||
s.metrics.SetLastPoll(time.Now())
|
||
if err != nil {
|
||
s.metrics.Errors.Add(1)
|
||
slog.Error("GLPI poll failed", "error", err)
|
||
return
|
||
}
|
||
for _, t := range tickets {
|
||
version := sourceVersion(t)
|
||
if !s.state.Seen(t.ID, version) {
|
||
if s.q.Enqueue(t.ID) {
|
||
s.metrics.QueueDepth.Store(int64(s.q.Len()))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
func (s *Service) healthLoop(ctx context.Context) {
|
||
check := func() {
|
||
c, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||
defer cancel()
|
||
gerr := s.glpi.Ping(c)
|
||
oerr := s.ai.Ping(c)
|
||
s.metrics.SetHealth(gerr == nil, oerr == nil)
|
||
}
|
||
check()
|
||
ticker := time.NewTicker(30 * time.Second)
|
||
defer ticker.Stop()
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case <-ticker.C:
|
||
check()
|
||
}
|
||
}
|
||
}
|
||
func (s *Service) worker(ctx context.Context, n int) {
|
||
for {
|
||
id, ok := s.q.Next(ctx)
|
||
if !ok {
|
||
return
|
||
}
|
||
s.metrics.QueueDepth.Store(int64(s.q.Len()))
|
||
if err := s.Process(ctx, id); err != nil {
|
||
slog.Error("ticket processing failed", "worker", n, "ticket_id", id, "error", err)
|
||
}
|
||
s.q.Done(id)
|
||
s.metrics.QueueDepth.Store(int64(s.q.Len()))
|
||
}
|
||
}
|
||
func (s *Service) Process(ctx context.Context, id int64) error {
|
||
muAny, _ := s.locks.LoadOrStore(id, &sync.Mutex{})
|
||
mu := muAny.(*sync.Mutex)
|
||
mu.Lock()
|
||
defer mu.Unlock()
|
||
start := time.Now()
|
||
run := model.RunRecord{RunID: newRunID(), TicketID: id, StartedAt: start, DryRun: s.cfg.DryRun, Outcome: "error"}
|
||
finish := func(err error) {
|
||
run.FinishedAt = time.Now()
|
||
if err != nil {
|
||
run.Error = err.Error()
|
||
s.metrics.Errors.Add(1)
|
||
}
|
||
if e := s.state.Append(run); e != nil {
|
||
slog.Error("persist run failed", "error", e)
|
||
}
|
||
}
|
||
t, err := s.glpi.GetTicket(ctx, id)
|
||
if err != nil {
|
||
run.Reason = "ticket_load_failed"
|
||
finish(err)
|
||
return err
|
||
}
|
||
run.TicketName = t.Name
|
||
run.SourceVersion = sourceVersion(t)
|
||
run.CategoryBefore = t.CategoryID
|
||
run.ExecutionChecks = append(run.ExecutionChecks, model.RuleCheck{Code: "execution_ticket_loaded", Group: "eligibility", Label: "Ticket konnte geladen werden", Status: "pass", Actual: "ja", Expected: "ja"})
|
||
alreadySeen := s.state.Seen(t.ID, run.SourceVersion)
|
||
run.ExecutionChecks = append(run.ExecutionChecks, model.RuleCheck{Code: "execution_not_already_processed", Group: "eligibility", Label: "Diese Ticketversion wurde noch nicht verarbeitet", Status: passFail(!alreadySeen), Blocking: alreadySeen, Actual: boolText(!alreadySeen), Expected: "ja"})
|
||
if alreadySeen {
|
||
run.Outcome = "skipped"
|
||
run.Reason = "already_processed"
|
||
s.metrics.Skipped.Add(1)
|
||
finish(nil)
|
||
return nil
|
||
}
|
||
statusAllowed := s.statusAllowed(t.StatusID)
|
||
run.ExecutionChecks = append(run.ExecutionChecks, model.RuleCheck{Code: "execution_status_allowed", Group: "eligibility", Label: "Ticketstatus ist zur Verarbeitung freigegeben", Status: passFail(statusAllowed), Blocking: !statusAllowed, Actual: fmt.Sprintf("Status #%d", t.StatusID), Expected: fmt.Sprintf("einer von %v", s.cfg.GLPIAllowedStatusIDs)})
|
||
if !statusAllowed {
|
||
run.Outcome = "skipped"
|
||
run.Reason = "status_not_allowed"
|
||
s.metrics.Skipped.Add(1)
|
||
finish(nil)
|
||
return nil
|
||
}
|
||
followups, err := s.glpi.GetFollowups(ctx, id)
|
||
if err != nil {
|
||
run.Reason = "followup_check_failed"
|
||
finish(err)
|
||
return err
|
||
}
|
||
canReply := len(followups) == 0
|
||
run.ExecutionChecks = append(run.ExecutionChecks, model.RuleCheck{Code: "execution_no_existing_followup", Group: "execution", Label: "Ticket hat noch keine Antwort / kein Followup", Status: passFail(canReply), Blocking: !canReply, Actual: fmt.Sprintf("%d Followups", len(followups)), Expected: "0 Followups"})
|
||
categories, err := s.getCategories(ctx)
|
||
if err != nil {
|
||
run.Reason = "categories_failed"
|
||
finish(err)
|
||
return err
|
||
}
|
||
run.CategoryBeforeName = categoryName(categories, t.CategoryID)
|
||
promptCats := shortlistCategories(t, categories, s.cfg.CategoryPromptLimit)
|
||
auditTopK := s.cfg.KnowledgeAuditTopK
|
||
if auditTopK <= 0 {
|
||
auditTopK = s.cfg.KnowledgeTopK
|
||
if auditTopK <= 0 {
|
||
auditTopK = 10
|
||
}
|
||
}
|
||
llmTopK := s.cfg.KnowledgeTopK
|
||
if llmTopK <= 0 {
|
||
llmTopK = 6
|
||
}
|
||
allRetrievalHits, err := s.knowledge.Search(ctx, t.Name+"\n"+stripHTML(t.Content), 0, categories)
|
||
if err != nil {
|
||
run.Reason = "knowledge_search_failed"
|
||
finish(err)
|
||
return err
|
||
}
|
||
categoryRetrievalHits := knowledge.FilterHitsBySources(allRetrievalHits, s.cfg.KnowledgeCategorySources, auditTopK)
|
||
retrievalHits := knowledge.FilterHitsBySources(allRetrievalHits, s.cfg.KnowledgeAllowedSources, auditTopK)
|
||
categoryLLMHits, _ := selectKnowledgeCandidates(categoryRetrievalHits, llmTopK, s.cfg.KnowledgeRetrievalFloor, s.cfg.KnowledgeCandidateMaxGap)
|
||
llmHits, candidateCutoff := selectKnowledgeCandidates(retrievalHits, llmTopK, s.cfg.KnowledgeRetrievalFloor, s.cfg.KnowledgeCandidateMaxGap)
|
||
// Reply selection is independent from category classification. Do not expose
|
||
// reply candidates when a reply is already impossible (for example because
|
||
// a followup exists or AUTO_REPLY is disabled). This also gives the Ollama
|
||
// client an explicit category-only mode.
|
||
replyLLMHits := llmHits
|
||
if !canReply || !s.cfg.AutoReply {
|
||
replyLLMHits = nil
|
||
}
|
||
run.KnowledgeLLMCandidates = len(replyLLMHits)
|
||
run.KnowledgeCandidateCutoff = candidateCutoff
|
||
run.KnowledgeCandidateMaxGap = s.cfg.KnowledgeCandidateMaxGap
|
||
run.KnowledgeAuditTopK = auditTopK
|
||
llmCandidateIDs := knowledgeHitIDSet(replyLLMHits)
|
||
contextData := model.ContextSnapshot{}
|
||
if s.context != nil && s.cfg.ContextEnabled {
|
||
s.metrics.ContextFetches.Add(1)
|
||
contextData = s.context.Collect(ctx, t)
|
||
run.ContextChanges = len(contextData.Changes)
|
||
run.ContextIncidents = len(contextData.MajorIncidents)
|
||
run.ContextIssues = len(contextData.ServiceIssues)
|
||
run.ContextDevices = len(contextData.UserDevices)
|
||
run.ContextWarnings = append([]string(nil), contextData.Warnings...)
|
||
run.ContextDetails = auditContextDetails(contextData, 5)
|
||
if contextData.Incomplete {
|
||
s.metrics.ContextErrors.Add(1)
|
||
}
|
||
}
|
||
decision, err := s.ai.Analyse(ctx, t, promptCats, categoryLLMHits, replyLLMHits, contextData)
|
||
if err != nil {
|
||
run.Reason = "ai_failed"
|
||
finish(err)
|
||
return err
|
||
}
|
||
// The classifier provides an independent category recommendation. Knowledge
|
||
// explicitly mapped to that category receives a deterministic post-retrieval
|
||
// alignment signal before the final policy gate.
|
||
hits := s.knowledge.RerankForCategory(retrievalHits, decision.Category.ID)
|
||
if len(hits) > 0 {
|
||
run.KnowledgeCandidates = auditKnowledgeCandidates(hits, s.cfg.KnowledgeMinScore, auditTopK, llmCandidateIDs, candidateCutoff, s.cfg.KnowledgeRetrievalFloor, llmTopK)
|
||
run.KnowledgeTopID = hits[0].Doc.ID
|
||
run.KnowledgeTopTitle = hits[0].Doc.Title
|
||
run.KnowledgeScore = hits[0].Score
|
||
run.KnowledgeSemanticScore = hits[0].SemanticScore
|
||
run.KnowledgeTitleScore = hits[0].TitleScore
|
||
run.KnowledgeLexicalScore = hits[0].LexicalScore
|
||
run.KnowledgeKeywordScore = hits[0].KeywordScore
|
||
run.KnowledgeCategoryScore = hits[0].CategoryScore
|
||
run.KnowledgeBestChunk = hits[0].BestChunkExcerpt
|
||
run.KnowledgeBestQueryChunk = hits[0].BestQueryExcerpt
|
||
run.KnowledgeQueryChunks = hits[0].QueryChunkCount
|
||
run.KnowledgeDocumentChunks = hits[0].DocumentChunkCount
|
||
run.KnowledgeThreshold = s.cfg.KnowledgeMinScore
|
||
if hits[0].Doc.MinScore > run.KnowledgeThreshold {
|
||
run.KnowledgeThreshold = hits[0].Doc.MinScore
|
||
}
|
||
}
|
||
result, err := s.policy.Evaluate(t, decision, categories, hits, contextData)
|
||
if err != nil {
|
||
run.Reason = "policy_rejected"
|
||
finish(err)
|
||
return err
|
||
}
|
||
run.AIReason = result.AIReason
|
||
run.Reason = result.AIReason // backwards compatible audit field
|
||
run.AIRecommendedCategoryID = result.CategoryRecommendationID
|
||
run.AIRecommendedCategoryName = result.CategoryRecommendationName
|
||
run.AICategoryConfidence = result.CategoryConfidence
|
||
run.CategoryThreshold = result.CategoryThreshold
|
||
run.CategoryDecision = result.CategoryDecision
|
||
run.CategoryProposed = result.CategoryID
|
||
run.CategoryWouldChange = result.ChangeCategory
|
||
run.AIReplyRecommended = result.ReplyRecommendation
|
||
run.AIReplyConfidence = result.ReplyConfidence
|
||
run.ReplyThreshold = result.ReplyThreshold
|
||
run.AIKnowledgeID = result.ReplyKnowledgeID
|
||
run.ReplyDecision = result.ReplyDecision
|
||
run.ReplyProposed = result.Reply
|
||
run.KnowledgeID = result.KnowledgeID
|
||
if result.KnowledgeThreshold > 0 {
|
||
run.KnowledgeThreshold = result.KnowledgeThreshold
|
||
}
|
||
run.KnowledgeEvidenceScore = result.KnowledgeEvidenceScore
|
||
run.KnowledgeRetrievalFloor = result.KnowledgeRetrievalFloor
|
||
run.KnowledgeCategoryAligned = result.KnowledgeCategoryAligned
|
||
run.CategoryChecks = append([]model.RuleCheck(nil), result.CategoryChecks...)
|
||
run.ReplyChecks = append([]model.RuleCheck(nil), result.ReplyChecks...)
|
||
run.ExecutionChecks = append(run.ExecutionChecks, model.RuleCheck{Code: "execution_dry_run", Group: "execution", Label: "Live-Schreibmodus aktiv", Status: map[bool]string{true: "info", false: "pass"}[s.cfg.DryRun], Blocking: false, Actual: map[bool]string{true: "DRY RUN", false: "LIVE"}[s.cfg.DryRun], Expected: "LIVE für tatsächliche Änderungen", Detail: "Im DRY RUN werden freigegebene Aktionen nur simuliert."})
|
||
run.PolicyReason = result.CategoryDecision + "; " + result.ReplyDecision
|
||
if !canReply {
|
||
// An existing followup is the authoritative execution-level reason why
|
||
// no reply can be proposed, regardless of the model/policy recommendation.
|
||
run.ReplyProposed = false
|
||
run.ReplyDecision = "reply_existing_followup"
|
||
run.PolicyReason = result.CategoryDecision + "; " + run.ReplyDecision
|
||
}
|
||
|
||
// Re-read the ticket immediately before any write. This prevents a stale
|
||
// model decision from overwriting a human change made during inference.
|
||
if (result.ChangeCategory || (result.Reply && canReply)) && !s.cfg.DryRun {
|
||
fresh, err := s.glpi.GetTicket(ctx, id)
|
||
if err != nil {
|
||
run.Reason = "prewrite_ticket_recheck_failed"
|
||
finish(err)
|
||
return err
|
||
}
|
||
if sourceVersion(fresh) != run.SourceVersion {
|
||
run.ExecutionChecks = append(run.ExecutionChecks, model.RuleCheck{Code: "execution_ticket_unchanged", Group: "execution", Label: "Ticket seit Analyse unverändert", Status: "fail", Blocking: true, Actual: "geändert", Expected: "unverändert"})
|
||
run.Outcome = "skipped"
|
||
run.Reason = "ticket_changed_before_write"
|
||
if result.ChangeCategory {
|
||
run.CategoryDecision = "category_ticket_changed_before_write"
|
||
}
|
||
if result.Reply && canReply {
|
||
run.ReplyDecision = "reply_ticket_changed_before_write"
|
||
}
|
||
run.PolicyReason = run.CategoryDecision + "; " + run.ReplyDecision
|
||
s.metrics.Skipped.Add(1)
|
||
finish(nil)
|
||
return nil
|
||
}
|
||
run.ExecutionChecks = append(run.ExecutionChecks, model.RuleCheck{Code: "execution_ticket_unchanged", Group: "execution", Label: "Ticket seit Analyse unverändert", Status: "pass", Actual: "unverändert", Expected: "unverändert"})
|
||
}
|
||
|
||
if result.ChangeCategory && !s.cfg.DryRun {
|
||
if err := s.glpi.SetCategory(ctx, id, result.CategoryID); err != nil {
|
||
run.ExecutionChecks = append(run.ExecutionChecks, model.RuleCheck{Code: "execution_category_write", Group: "execution", Label: "Kategorie konnte in GLPI geschrieben werden", Status: "fail", Blocking: true, Actual: err.Error(), Expected: "erfolgreich"})
|
||
run.Reason = "category_write_failed"
|
||
run.CategoryDecision = "category_write_failed"
|
||
run.PolicyReason = run.CategoryDecision + "; " + run.ReplyDecision
|
||
finish(err)
|
||
return err
|
||
}
|
||
run.ExecutionChecks = append(run.ExecutionChecks, model.RuleCheck{Code: "execution_category_write", Group: "execution", Label: "Kategorie konnte in GLPI geschrieben werden", Status: "pass", Actual: fmt.Sprintf("#%d", result.CategoryID), Expected: "erfolgreich"})
|
||
run.CategoryChanged = true
|
||
run.CategoryDecision = "category_written"
|
||
s.metrics.CategoryChanged.Add(1)
|
||
} else if result.ChangeCategory {
|
||
run.CategoryDecision = "category_accepted_dry_run"
|
||
}
|
||
run.PolicyReason = run.CategoryDecision + "; " + run.ReplyDecision
|
||
|
||
if result.Reply && canReply {
|
||
// If category was just changed by this process, date_mod will legitimately
|
||
// differ. Compare the decision-relevant ticket fields instead and require
|
||
// the category we expect before posting a reply.
|
||
if !s.cfg.DryRun {
|
||
fresh, err := s.glpi.GetTicket(ctx, id)
|
||
if err != nil {
|
||
run.Reason = "prereply_ticket_recheck_failed"
|
||
finish(err)
|
||
return err
|
||
}
|
||
expectedCategory := t.CategoryID
|
||
if result.ChangeCategory {
|
||
expectedCategory = result.CategoryID
|
||
}
|
||
if !sameDecisionSource(t, fresh, expectedCategory) {
|
||
run.ReplyProposed = false
|
||
run.Reason = "ticket_changed_before_reply"
|
||
run.ReplyDecision = "reply_ticket_changed_before_write"
|
||
run.PolicyReason = run.CategoryDecision + "; " + run.ReplyDecision
|
||
run.Outcome = "skipped"
|
||
s.metrics.Skipped.Add(1)
|
||
finish(nil)
|
||
return nil
|
||
}
|
||
}
|
||
followups, err = s.glpi.GetFollowups(ctx, id)
|
||
if err != nil {
|
||
run.Reason = "followup_recheck_failed"
|
||
finish(err)
|
||
return err
|
||
}
|
||
if len(followups) > 0 {
|
||
run.ExecutionChecks = append(run.ExecutionChecks, model.RuleCheck{Code: "execution_followup_recheck", Group: "execution", Label: "Unmittelbar vor Antwort ist weiterhin kein Followup vorhanden", Status: "fail", Blocking: true, Actual: fmt.Sprintf("%d Followups", len(followups)), Expected: "0 Followups"})
|
||
run.ReplyProposed = false
|
||
run.Reason = "followup_appeared_before_write"
|
||
run.ReplyDecision = "reply_followup_appeared_before_write"
|
||
} else if !s.cfg.DryRun {
|
||
run.ExecutionChecks = append(run.ExecutionChecks, model.RuleCheck{Code: "execution_followup_recheck", Group: "execution", Label: "Unmittelbar vor Antwort ist weiterhin kein Followup vorhanden", Status: "pass", Actual: "0 Followups", Expected: "0 Followups"})
|
||
if err := s.glpi.AddFollowup(ctx, id, result.ReplyText, result.ReplyIsHTML); err != nil {
|
||
run.ExecutionChecks = append(run.ExecutionChecks, model.RuleCheck{Code: "execution_reply_write", Group: "execution", Label: "Antwort konnte in GLPI geschrieben werden", Status: "fail", Blocking: true, Actual: err.Error(), Expected: "erfolgreich"})
|
||
run.Reason = "reply_write_failed"
|
||
run.ReplyDecision = "reply_write_failed"
|
||
run.PolicyReason = run.CategoryDecision + "; " + run.ReplyDecision
|
||
finish(err)
|
||
return err
|
||
}
|
||
run.ExecutionChecks = append(run.ExecutionChecks, model.RuleCheck{Code: "execution_reply_write", Group: "execution", Label: "Antwort konnte in GLPI geschrieben werden", Status: "pass", Actual: "erfolgreich", Expected: "erfolgreich"})
|
||
run.ReplyWritten = true
|
||
run.ReplyDecision = "reply_written"
|
||
s.metrics.Replies.Add(1)
|
||
} else {
|
||
run.ReplyDecision = "reply_accepted_dry_run"
|
||
}
|
||
run.PolicyReason = run.CategoryDecision + "; " + run.ReplyDecision
|
||
}
|
||
|
||
// Persist the final GLPI version after our own write so the next poll does
|
||
// not immediately process the same self-induced modification again.
|
||
if !s.cfg.DryRun && (run.CategoryChanged || run.ReplyWritten) {
|
||
if finalTicket, e := s.glpi.GetTicket(ctx, id); e == nil {
|
||
run.SourceVersion = sourceVersion(finalTicket)
|
||
}
|
||
}
|
||
run.PolicyReason = run.CategoryDecision + "; " + run.ReplyDecision
|
||
run.Outcome = "processed"
|
||
s.metrics.Processed.Add(1)
|
||
finish(nil)
|
||
return nil
|
||
}
|
||
|
||
// DiagnoseRun returns the persisted, historical decision record. The rule
|
||
// checks stored on the run are the authoritative explanation of the policy at
|
||
// execution time.
|
||
func (s *Service) DiagnoseRun(ctx context.Context, runID string) (model.RunRecord, error) {
|
||
_ = ctx
|
||
r, ok := s.state.FindRun(strings.TrimSpace(runID))
|
||
if !ok {
|
||
return model.RunRecord{}, fmt.Errorf("run %q not found", runID)
|
||
}
|
||
return r, nil
|
||
}
|
||
|
||
// DiagnoseKnowledge recalculates one arbitrary knowledge article against the
|
||
// current ticket/index. This is intentionally marked as a current re-evaluation
|
||
// when the GLPI ticket changed since the historical run.
|
||
func (s *Service) DiagnoseKnowledge(ctx context.Context, runID, knowledgeID string) (model.KnowledgeDiagnostic, error) {
|
||
run, ok := s.state.FindRun(strings.TrimSpace(runID))
|
||
if !ok {
|
||
return model.KnowledgeDiagnostic{}, fmt.Errorf("run %q not found", runID)
|
||
}
|
||
doc, ok := s.knowledge.ByID(strings.TrimSpace(knowledgeID))
|
||
if !ok {
|
||
return model.KnowledgeDiagnostic{}, fmt.Errorf("knowledge %q not found", knowledgeID)
|
||
}
|
||
t, err := s.glpi.GetTicket(ctx, run.TicketID)
|
||
if err != nil {
|
||
return model.KnowledgeDiagnostic{}, fmt.Errorf("load current ticket: %w", err)
|
||
}
|
||
cats, err := s.getCategories(ctx)
|
||
if err != nil {
|
||
return model.KnowledgeDiagnostic{}, fmt.Errorf("load categories: %w", err)
|
||
}
|
||
query := t.Name + "\n" + stripHTML(t.Content)
|
||
indexedHits, err := s.knowledge.Search(ctx, query, 0, cats)
|
||
if err != nil {
|
||
return model.KnowledgeDiagnostic{}, err
|
||
}
|
||
allHits := knowledge.FilterHitsBySources(indexedHits, s.cfg.KnowledgeAllowedSources, 0)
|
||
maxCandidates := s.cfg.KnowledgeTopK
|
||
if maxCandidates <= 0 {
|
||
maxCandidates = 6
|
||
}
|
||
llmHits, cutoff := selectKnowledgeCandidates(allHits, maxCandidates, s.cfg.KnowledgeRetrievalFloor, s.cfg.KnowledgeCandidateMaxGap)
|
||
llmSet := knowledgeHitIDSet(llmHits)
|
||
initialRank := 0
|
||
initialScore := 0.0
|
||
for i, h := range allHits {
|
||
if h.Doc.ID == doc.ID {
|
||
initialRank = i + 1
|
||
initialScore = h.Score
|
||
break
|
||
}
|
||
}
|
||
post := s.knowledge.RerankForCategory(allHits, run.AIRecommendedCategoryID)
|
||
var hit *model.KnowledgeHit
|
||
for i := range post {
|
||
if post[i].Doc.ID == doc.ID {
|
||
hit = &post[i]
|
||
break
|
||
}
|
||
}
|
||
if hit == nil {
|
||
return model.KnowledgeDiagnostic{}, fmt.Errorf("knowledge %q is not in active index", knowledgeID)
|
||
}
|
||
_, sent := llmSet[doc.ID]
|
||
reason := candidateSelectionReason(initialRank, initialScore, sent, cutoff, s.cfg.KnowledgeRetrievalFloor, maxCandidates)
|
||
decision := model.Decision{}
|
||
decision.Category.ID = run.AIRecommendedCategoryID
|
||
decision.Category.Confidence = run.AICategoryConfidence
|
||
decision.Reply.Allowed = run.AIReplyRecommended
|
||
decision.Reply.Confidence = run.AIReplyConfidence
|
||
decision.Reply.KnowledgeID = doc.ID
|
||
decision.Reason = run.AIReason
|
||
ctxData := model.ContextSnapshot{}
|
||
if s.context != nil && s.cfg.ContextEnabled {
|
||
ctxData = s.context.Collect(ctx, t)
|
||
}
|
||
res, _ := s.policy.Evaluate(t, decision, cats, []model.KnowledgeHit{*hit}, ctxData)
|
||
required := s.cfg.KnowledgeMinScore
|
||
if doc.MinScore > required {
|
||
required = doc.MinScore
|
||
}
|
||
checks := append([]model.RuleCheck(nil), res.ReplyChecks...)
|
||
checks = append([]model.RuleCheck{
|
||
{Code: "candidate_in_active_index", Group: "retrieval", Label: "Artikel ist im aktiven Knowledge-Index", Status: "pass", Actual: "ja", Expected: "ja"},
|
||
{Code: "candidate_retrieval_floor", Group: "retrieval", Label: "Retrieval-Score erreicht Floor", Status: passFail(initialScore >= s.cfg.KnowledgeRetrievalFloor), Blocking: initialScore < s.cfg.KnowledgeRetrievalFloor, Actual: percentText(initialScore), Expected: ">= " + percentText(s.cfg.KnowledgeRetrievalFloor)},
|
||
{Code: "candidate_dynamic_cutoff", Group: "retrieval", Label: "Artikel liegt innerhalb des dynamischen Top-K-Abstands", Status: passFail(initialScore >= cutoff), Blocking: initialScore < cutoff, Actual: percentText(initialScore), Expected: ">= " + percentText(cutoff), Detail: fmt.Sprintf("Bester Treffer minus %.1f Prozentpunkte, mindestens Retrieval-Floor.", s.cfg.KnowledgeCandidateMaxGap*100)},
|
||
{Code: "candidate_sent_to_ai", Group: "retrieval", Label: "Artikel wurde an die KI übergeben", Status: passFail(sent), Blocking: !sent, Actual: boolText(sent), Expected: "ja", Detail: reason},
|
||
}, checks...)
|
||
return model.KnowledgeDiagnostic{
|
||
RunID: run.RunID, TicketID: run.TicketID, KnowledgeID: doc.ID, Title: doc.Title, Source: doc.Source,
|
||
CurrentTicketChanged: sourceVersion(t) != run.SourceVersion, RetrievalRank: initialRank, RetrievalScore: initialScore,
|
||
SemanticScore: hit.SemanticScore, TitleScore: hit.TitleScore, LexicalScore: hit.LexicalScore, KeywordScore: hit.KeywordScore, CategoryScore: hit.CategoryScore,
|
||
CandidateCutoff: cutoff, SentToAI: sent, SelectionReason: reason, AISelected: run.AIKnowledgeID == doc.ID,
|
||
EvidenceScore: res.KnowledgeEvidenceScore, RequiredScore: required, BestChunkExcerpt: hit.BestChunkExcerpt, BestQueryExcerpt: hit.BestQueryExcerpt,
|
||
ExternalCategories: append([]string(nil), doc.ExternalCategories...), UnmappedCategories: append([]string(nil), doc.UnmappedExternalCategories...), Checks: checks, Document: doc,
|
||
}, nil
|
||
}
|
||
|
||
func candidateSelectionReason(rank int, score float64, sent bool, cutoff, floor float64, maxCandidates int) string {
|
||
if sent {
|
||
return "sent_to_ai"
|
||
}
|
||
if score < floor {
|
||
return "below_retrieval_floor"
|
||
}
|
||
if score < cutoff {
|
||
return "outside_candidate_gap"
|
||
}
|
||
if maxCandidates > 0 && rank > maxCandidates {
|
||
return "max_candidates_reached"
|
||
}
|
||
return "not_selected"
|
||
}
|
||
|
||
func auditKnowledgeCandidates(hits []model.KnowledgeHit, globalMin float64, limit int, sentToAI map[string]struct{}, cutoff, retrievalFloor float64, maxCandidates int) []model.KnowledgeCandidateAudit {
|
||
if limit <= 0 || limit > len(hits) {
|
||
limit = len(hits)
|
||
}
|
||
out := make([]model.KnowledgeCandidateAudit, 0, limit)
|
||
for idx, h := range hits[:limit] {
|
||
required := globalMin
|
||
if h.Doc.MinScore > required {
|
||
required = h.Doc.MinScore
|
||
}
|
||
_, wasSent := sentToAI[h.Doc.ID]
|
||
reason := candidateSelectionReason(idx+1, h.Score, wasSent, cutoff, retrievalFloor, maxCandidates)
|
||
out = append(out, model.KnowledgeCandidateAudit{
|
||
ID: h.Doc.ID, Title: h.Doc.Title, Source: h.Doc.Source, Score: h.Score,
|
||
SemanticScore: h.SemanticScore, TitleScore: h.TitleScore, LexicalScore: h.LexicalScore, KeywordScore: h.KeywordScore,
|
||
CategoryScore: h.CategoryScore, RequiredScore: required, AutoReply: h.Doc.AutoReply,
|
||
BestChunkExcerpt: h.BestChunkExcerpt, BestQueryExcerpt: h.BestQueryExcerpt,
|
||
QueryChunkCount: h.QueryChunkCount, DocumentChunkCount: h.DocumentChunkCount, SentToAI: wasSent, RetrievalRank: idx + 1, SelectionReason: reason,
|
||
})
|
||
}
|
||
return out
|
||
}
|
||
|
||
func selectKnowledgeCandidates(hits []model.KnowledgeHit, maxCandidates int, retrievalFloor, maxGap float64) ([]model.KnowledgeHit, float64) {
|
||
if len(hits) == 0 || maxCandidates <= 0 {
|
||
return nil, retrievalFloor
|
||
}
|
||
best := hits[0].Score
|
||
if best < retrievalFloor {
|
||
return nil, retrievalFloor
|
||
}
|
||
cutoff := best - maxGap
|
||
if cutoff < retrievalFloor {
|
||
cutoff = retrievalFloor
|
||
}
|
||
capacity := maxCandidates
|
||
if len(hits) < capacity {
|
||
capacity = len(hits)
|
||
}
|
||
out := make([]model.KnowledgeHit, 0, capacity)
|
||
for _, h := range hits {
|
||
if h.Score < cutoff || h.Score < retrievalFloor {
|
||
break
|
||
}
|
||
out = append(out, h)
|
||
if len(out) >= maxCandidates {
|
||
break
|
||
}
|
||
}
|
||
return out, cutoff
|
||
}
|
||
|
||
func knowledgeHitIDSet(hits []model.KnowledgeHit) map[string]struct{} {
|
||
out := make(map[string]struct{}, len(hits))
|
||
for _, h := range hits {
|
||
out[h.Doc.ID] = struct{}{}
|
||
}
|
||
return out
|
||
}
|
||
|
||
func auditContextDetails(c model.ContextSnapshot, limit int) []model.ContextAuditItem {
|
||
if limit <= 0 {
|
||
limit = 5
|
||
}
|
||
out := make([]model.ContextAuditItem, 0, limit*4)
|
||
for i, x := range c.Changes {
|
||
if i >= limit {
|
||
break
|
||
}
|
||
out = append(out, model.ContextAuditItem{Kind: "change", ID: x.ID, Name: x.Name, Relevance: x.Relevance, Detail: strings.TrimSpace(x.PlannedBegin + " – " + x.PlannedEnd)})
|
||
}
|
||
for i, x := range c.MajorIncidents {
|
||
if i >= limit {
|
||
break
|
||
}
|
||
out = append(out, model.ContextAuditItem{Kind: "incident", ID: x.ID, Name: x.Name, Relevance: x.Relevance, Status: fmt.Sprint(x.StatusID), Detail: auditExcerpt(x.Content, 320)})
|
||
}
|
||
for i, x := range c.ServiceIssues {
|
||
if i >= limit {
|
||
break
|
||
}
|
||
name := x.MonitorName
|
||
if name == "" {
|
||
name = x.IncidentTitle
|
||
}
|
||
out = append(out, model.ContextAuditItem{Kind: "uptime", ID: x.MonitorID, Name: name, Relevance: x.Relevance, Status: x.Status, Detail: auditExcerpt(x.Message, 320)})
|
||
}
|
||
for i, x := range c.UserDevices {
|
||
if i >= limit {
|
||
break
|
||
}
|
||
name := x.Name
|
||
if name == "" {
|
||
name = fmt.Sprintf("%s #%d", x.ItemType, x.ID)
|
||
}
|
||
parts := make([]string, 0, 3)
|
||
for _, v := range []string{x.Serial, x.InventoryNumber, x.Location} {
|
||
if strings.TrimSpace(v) != "" {
|
||
parts = append(parts, strings.TrimSpace(v))
|
||
}
|
||
}
|
||
detail := strings.Join(parts, " · ")
|
||
out = append(out, model.ContextAuditItem{Kind: "device", ID: x.ID, Name: name, Status: x.Status, Detail: detail})
|
||
}
|
||
return out
|
||
}
|
||
|
||
func auditExcerpt(v string, max int) string {
|
||
v = strings.Join(strings.Fields(v), " ")
|
||
if max <= 0 || len(v) <= max {
|
||
return v
|
||
}
|
||
return strings.TrimSpace(v[:max]) + "…"
|
||
}
|
||
|
||
func (s *Service) getCategories(ctx context.Context) ([]model.Category, error) {
|
||
s.catMu.RLock()
|
||
if len(s.categories) > 0 && time.Since(s.catAt) < 10*time.Minute {
|
||
out := append([]model.Category(nil), s.categories...)
|
||
s.catMu.RUnlock()
|
||
return s.enrichCategories(out), nil
|
||
}
|
||
s.catMu.RUnlock()
|
||
cats, err := s.glpi.GetCategories(ctx)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
s.catMu.Lock()
|
||
s.categories = append([]model.Category(nil), cats...)
|
||
s.catAt = time.Now()
|
||
s.catMu.Unlock()
|
||
return s.enrichCategories(cats), nil
|
||
}
|
||
|
||
func (s *Service) enrichCategories(cats []model.Category) []model.Category {
|
||
out := append([]model.Category(nil), cats...)
|
||
byID := make(map[int64]*model.Category, len(out))
|
||
for i := range out {
|
||
byID[out[i].ID] = &out[i]
|
||
out[i].Hints = append(out[i].Hints, semanticCategoryHints(out[i])...)
|
||
}
|
||
categorySources := sourceSet(s.cfg.KnowledgeCategorySources)
|
||
for _, doc := range s.knowledge.List() {
|
||
if _, allowed := categorySources[strings.ToLower(strings.TrimSpace(doc.Source))]; !allowed {
|
||
continue
|
||
}
|
||
for _, id := range doc.Categories {
|
||
if c := byID[id]; c != nil {
|
||
c.Hints = appendUnique(c.Hints, doc.Title)
|
||
for _, k := range doc.Keywords {
|
||
c.Hints = appendUnique(c.Hints, k)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
if s.cfg.LearningEnabled && s.learning != nil {
|
||
for i := range out {
|
||
out[i].Examples = s.learning.ExamplesFor(out[i].ID, s.cfg.LearningExamplesPerCategory)
|
||
}
|
||
}
|
||
return out
|
||
}
|
||
|
||
// Categories exposes the same enriched category catalogue that is supplied to
|
||
// Ollama. It is used by the authenticated dashboard for human feedback.
|
||
func (s *Service) Categories(ctx context.Context) ([]model.Category, error) {
|
||
return s.getCategories(ctx)
|
||
}
|
||
|
||
func (s *Service) RecordCategoryFeedback(ctx context.Context, runID string, categoryID int64) (model.LearningExample, error) {
|
||
if !s.cfg.LearningEnabled || s.learning == nil {
|
||
return model.LearningExample{}, fmt.Errorf("learning is disabled")
|
||
}
|
||
run, ok := s.state.FindRun(strings.TrimSpace(runID))
|
||
if !ok {
|
||
return model.LearningExample{}, fmt.Errorf("run not found")
|
||
}
|
||
cats, err := s.getCategories(ctx)
|
||
if err != nil {
|
||
return model.LearningExample{}, err
|
||
}
|
||
name := categoryName(cats, categoryID)
|
||
if categoryID <= 0 || name == "" {
|
||
return model.LearningExample{}, fmt.Errorf("unknown category id %d", categoryID)
|
||
}
|
||
t, err := s.glpi.GetTicket(ctx, run.TicketID)
|
||
if err != nil {
|
||
return model.LearningExample{}, err
|
||
}
|
||
if sourceVersion(t) != run.SourceVersion {
|
||
return model.LearningExample{}, fmt.Errorf("ticket changed since this run; process the current ticket state before teaching it")
|
||
}
|
||
ex := model.LearningExample{RunID: run.RunID, TicketID: t.ID, Subject: strings.TrimSpace(t.Name), Text: compactLearningText(stripHTML(t.Content), 1200), CategoryID: categoryID, CategoryName: name, AIRecommendedCategoryID: run.AIRecommendedCategoryID, AIConfidence: run.AICategoryConfidence, Correction: run.AIRecommendedCategoryID != categoryID, Source: "human-confirmed"}
|
||
return s.learning.Add(ex)
|
||
}
|
||
func (s *Service) LearningExamples() []model.LearningExample {
|
||
if s.learning == nil {
|
||
return nil
|
||
}
|
||
return s.learning.List()
|
||
}
|
||
func (s *Service) DeleteLearning(id string) error {
|
||
if s.learning == nil {
|
||
return fmt.Errorf("learning is disabled")
|
||
}
|
||
return s.learning.Delete(id)
|
||
}
|
||
func (s *Service) LearningCount() int {
|
||
if s.learning == nil {
|
||
return 0
|
||
}
|
||
return s.learning.Count()
|
||
}
|
||
|
||
func appendUnique(in []string, v string) []string {
|
||
v = strings.TrimSpace(v)
|
||
if v == "" {
|
||
return in
|
||
}
|
||
for _, x := range in {
|
||
if strings.EqualFold(strings.TrimSpace(x), v) {
|
||
return in
|
||
}
|
||
}
|
||
return append(in, v)
|
||
}
|
||
func compactLearningText(v string, max int) string {
|
||
v = strings.Join(strings.Fields(v), " ")
|
||
r := []rune(v)
|
||
if len(r) <= max {
|
||
return v
|
||
}
|
||
return string(r[:max]) + "…"
|
||
}
|
||
func semanticCategoryHints(c model.Category) []string {
|
||
name := strings.ToLower(c.Name + " " + c.CompleteName)
|
||
var h []string
|
||
add := func(vals ...string) {
|
||
for _, v := range vals {
|
||
h = appendUnique(h, v)
|
||
}
|
||
}
|
||
if strings.Contains(name, "active directory") || strings.Contains(name, "entra") || strings.Contains(name, "identity") || strings.Contains(name, "benutzerkonto") || strings.Contains(name, "account") {
|
||
add("Benutzerkonto", "Anmeldung / Login", "Konto gesperrt", "Passwort", "Domänenkonto", "Gruppen und Berechtigungen", "Authentifizierung")
|
||
}
|
||
if strings.Contains(name, "druck") || strings.Contains(name, "printer") {
|
||
add("Drucker", "Drucken nicht möglich", "Druckwarteschlange", "Netzwerkdrucker", "Toner", "Papierstau")
|
||
}
|
||
if strings.Contains(name, "vpn") {
|
||
add("VPN-Verbindung", "Remote Access", "Gateway", "GlobalProtect", "Tunnel", "Verbindungsaufbau")
|
||
}
|
||
if strings.Contains(name, "mail") || strings.Contains(name, "outlook") || strings.Contains(name, "exchange") {
|
||
add("E-Mail", "Outlook", "Postfach", "E-Mail Versand und Empfang", "Exchange")
|
||
}
|
||
if strings.Contains(name, "netz") || strings.Contains(name, "network") || strings.Contains(name, "wlan") || strings.Contains(name, "wifi") {
|
||
add("Netzwerk", "LAN", "WLAN", "Keine Verbindung", "DNS", "IP-Adresse")
|
||
}
|
||
if strings.Contains(name, "hardware") || strings.Contains(name, "client") || strings.Contains(name, "arbeitsplatz") {
|
||
add("Arbeitsplatzgerät", "Notebook", "PC", "Dockingstation", "Peripherie")
|
||
}
|
||
return h
|
||
}
|
||
|
||
func categoryName(categories []model.Category, id int64) string {
|
||
if id == 0 {
|
||
return "Nicht gesetzt"
|
||
}
|
||
for _, c := range categories {
|
||
if c.ID == id {
|
||
return categoryDisplayName(c)
|
||
}
|
||
}
|
||
return ""
|
||
}
|
||
|
||
func (s *Service) statusAllowed(id int64) bool {
|
||
for _, allowed := range s.cfg.GLPIAllowedStatusIDs {
|
||
if id == allowed {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
func sourceVersion(t model.Ticket) string {
|
||
// Do not rely on date_mod alone: two changes can happen within the same
|
||
// timestamp resolution and some API projections may omit it. Requesters and
|
||
// linked items are decision-relevant because they feed the context collector.
|
||
payload := fmt.Sprintf("%d\x00%s\x00%s\x00%s\x00%d\x00%d\x00%v\x00%v", t.ID, t.DateMod, t.Name, t.Content, t.StatusID, t.CategoryID, t.RequesterIDs, t.Items)
|
||
h := sha256.Sum256([]byte(payload))
|
||
return hex.EncodeToString(h[:])
|
||
}
|
||
|
||
func sameDecisionSource(original, fresh model.Ticket, expectedCategory int64) bool {
|
||
if fresh.Name != original.Name || fresh.Content != original.Content || fresh.StatusID != original.StatusID || fresh.CategoryID != expectedCategory {
|
||
return false
|
||
}
|
||
if fmt.Sprint(fresh.RequesterIDs) != fmt.Sprint(original.RequesterIDs) || fmt.Sprint(fresh.Items) != fmt.Sprint(original.Items) {
|
||
return false
|
||
}
|
||
return true
|
||
}
|
||
func newRunID() string { b := make([]byte, 8); _, _ = rand.Read(b); return hex.EncodeToString(b) }
|
||
func stripHTML(s string) string {
|
||
r := strings.NewReplacer("<br>", "\n", "<br/>", "\n", "<br />", "\n", "</p>", "\n")
|
||
s = r.Replace(s)
|
||
var b strings.Builder
|
||
inside := false
|
||
for _, ch := range s {
|
||
if ch == '<' {
|
||
inside = true
|
||
continue
|
||
}
|
||
if ch == '>' {
|
||
inside = false
|
||
continue
|
||
}
|
||
if !inside {
|
||
b.WriteRune(ch)
|
||
}
|
||
}
|
||
return strings.TrimSpace(b.String())
|
||
}
|
||
func shortlistCategories(t model.Ticket, cats []model.Category, limit int) []model.Category {
|
||
if limit <= 0 || len(cats) <= limit {
|
||
return cats
|
||
}
|
||
q := strings.Fields(strings.ToLower(t.Name + " " + stripHTML(t.Content)))
|
||
type scored struct {
|
||
c model.Category
|
||
s int
|
||
}
|
||
ss := make([]scored, 0, len(cats))
|
||
for _, c := range cats {
|
||
name := strings.ToLower(c.Name + " " + c.CompleteName + " " + strings.Join(c.Hints, " ") + " " + strings.Join(c.Examples, " "))
|
||
score := 0
|
||
for _, w := range q {
|
||
if len(w) >= 3 && strings.Contains(name, w) {
|
||
score++
|
||
}
|
||
}
|
||
if c.ID == t.CategoryID {
|
||
score += 100
|
||
}
|
||
ss = append(ss, scored{c, score})
|
||
}
|
||
sort.SliceStable(ss, func(i, j int) bool { return ss[i].s > ss[j].s })
|
||
out := make([]model.Category, 0, limit)
|
||
for i := 0; i < limit && i < len(ss); i++ {
|
||
out = append(out, ss[i].c)
|
||
}
|
||
return out
|
||
}
|