glpi-ai-agent/internal/agent/agent.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/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) error
	GetCategories(context.Context) ([]model.Category, error)
}
type AI interface {
	Ping(context.Context) error
	Analyse(context.Context, model.Ticket, []model.Category, []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
	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, s *state.Store, q *queue.Queue, m *metrics.Metrics, contextCollector ContextCollector) *Service {
	return &Service{cfg: cfg, glpi: g, ai: ai, knowledge: k, state: s, q: q, metrics: m, context: contextCollector, policy: NewPolicy(cfg.AutoCategory, cfg.AutoReply, cfg.CategoryConfidence, cfg.ReplyConfidence, cfg.KnowledgeMinScore, cfg.KnowledgeAllowedSources, cfg.KnowledgeAutoReplySources, cfg.CommunicationLanguage, cfg.CommunicationStyle, cfg.CommunicationSalutation, cfg.CommunicationClosing, cfg.CommunicationSignature, 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
	if s.state.Seen(t.ID, run.SourceVersion) {
		run.Outcome = "skipped"
		run.Reason = "already_processed"
		s.metrics.Skipped.Add(1)
		finish(nil)
		return nil
	}
	if !s.statusAllowed(t.StatusID) {
		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
	categories, err := s.getCategories(ctx)
	if err != nil {
		run.Reason = "categories_failed"
		finish(err)
		return err
	}
	promptCats := shortlistCategories(t, categories, s.cfg.CategoryPromptLimit)
	hits, err := s.knowledge.Search(ctx, t.Name+"\n"+stripHTML(t.Content), s.cfg.KnowledgeTopK)
	if err != nil {
		run.Reason = "knowledge_search_failed"
		finish(err)
		return err
	}
	if len(hits) > 0 {
		run.KnowledgeScore = hits[0].Score
	}
	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...)
		if contextData.Incomplete {
			s.metrics.ContextErrors.Add(1)
		}
	}
	decision, err := s.ai.Analyse(ctx, t, promptCats, hits, contextData)
	if err != nil {
		run.Reason = "ai_failed"
		finish(err)
		return err
	}
	result, err := s.policy.Evaluate(t, decision, categories, hits, contextData)
	if err != nil {
		run.Reason = "policy_rejected"
		finish(err)
		return err
	}
	run.CategoryProposed = result.CategoryID
	run.ReplyProposed = result.Reply
	run.KnowledgeID = result.KnowledgeID
	run.Reason = result.Reason
	if !canReply && result.Reply {
		run.ReplyProposed = false
		run.Reason = "existing_followup_no_reply"
	}

	// 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.Outcome = "skipped"
			run.Reason = "ticket_changed_before_write"
			s.metrics.Skipped.Add(1)
			finish(nil)
			return nil
		}
	}

	if result.ChangeCategory && !s.cfg.DryRun {
		if err := s.glpi.SetCategory(ctx, id, result.CategoryID); err != nil {
			run.Reason = "category_write_failed"
			finish(err)
			return err
		}
		run.CategoryChanged = true
		s.metrics.CategoryChanged.Add(1)
	} else if result.ChangeCategory {
		run.CategoryChanged = true
	}

	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.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.ReplyProposed = false
			run.Reason = "followup_appeared_before_write"
		} else if !s.cfg.DryRun {
			if err := s.glpi.AddFollowup(ctx, id, result.ReplyText); err != nil {
				run.Reason = "reply_write_failed"
				finish(err)
				return err
			}
			run.ReplyWritten = true
			s.metrics.Replies.Add(1)
		}
	}

	// 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.Outcome = "processed"
	s.metrics.Processed.Add(1)
	finish(nil)
	return nil
}
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 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 cats, nil
}
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)
		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
}