@@ -0,0 +1,410 @@
|
||||
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
|
||||
}
|
||||
Reference in New Issue
Block a user