central_ai_llm_master/src/hub/hub.go

// Package hub wires the five steps: hear, understand, check, route, answer.
//
// The HTTP layer in main.go stays thin. This package holds the pipeline so it
// can be tested against a fake project without a network.
package hub

import (
	"bytes"
	"context"
	"encoding/json"
	"strings"
	"sync"
	"time"

	"hub/answer"
	"hub/apierr"
	"hub/config"
	"hub/destinations"
	"hub/hear"
	"hub/logbook"
	"hub/route"
	"hub/understand"
)

// Service is the wired Hub.
type Service struct {
	Config      *config.Config
	Rules       *understand.Rules
	Router      understand.Decider
	Dispatcher  *route.Dispatcher
	Approvals   *Approvals
	Log         *logbook.Log
	Speaker     route.Announcer
	Paperclip   route.JobFiler
	Started     time.Time
	HealthProbe func(ctx context.Context) map[string]string

	capMu      sync.Mutex
	capTopics  []string
	capPeople  map[string][]string
	capFetched time.Time
}

// capabilitySnapshot is the live project ability list, keyed by project for
// people so a name from one project never leaks into another.
type capabilitySnapshot struct {
	topics []string
	people map[string][]string
}

// CapabilityTTL is how long the Hub keeps a project capability list.
const CapabilityTTL = 30 * time.Second

// Ask runs the five steps for one POST /ask request. It never returns an error
// for a routing fault: a fault becomes a status in the reply.
func (s *Service) Ask(ctx context.Context, req hear.Request) answer.Response {
	var outcome route.Outcome

	// Step 2a: the never-route check. No LLM runs on a control action.
	if understand.IsControlAction(req.Text) {
		outcome = route.Refused()
	} else {
		caps := s.capabilities(ctx)
		decision, err := s.Router.Decide(ctx, understand.Input{
			Question: req.Text,
			Person:   req.PersonOrDefault(),
			Topics:   caps.topics,
			Rules:    s.Rules,
		})
		if err != nil {
			outcome = route.Outcome{
				Kind:   route.KindNoMatch,
				Status: route.StatusUnavailable,
				Answer: "I cannot work out where to send that.",
				Reason: err.Error(),
			}
		} else {
			req = applyPerson(req, decision, caps.people)
			outcome = s.Dispatcher.Dispatch(ctx, decision, req)
		}
	}

	resp := answer.Build(req, outcome)

	// Answer step: speak a house reply. House TTS is out of band; the Hub
	// starts it and does not wait for the speakers (docs/hub-api.md section 8).
	if !outcome.Spoken && answer.ShouldSpeak(resp) && s.Speaker != nil {
		text := resp.Speech.Text
		speak := s.Speaker
		go func() {
			_ = speak.Announce(context.Background(), text)
		}()
	}
	return resp
}

// capabilities reads the live capability topics and people from every project,
// with a short cache. A project that is down contributes nothing and does not
// block.
func (s *Service) capabilities(ctx context.Context) capabilitySnapshot {
	if s.Dispatcher == nil || len(s.Dispatcher.Projects) == 0 {
		return capabilitySnapshot{}
	}
	s.capMu.Lock()
	if time.Since(s.capFetched) < CapabilityTTL && s.capTopics != nil {
		snap := capabilitySnapshot{topics: s.capTopics, people: s.capPeople}
		s.capMu.Unlock()
		return snap
	}
	s.capMu.Unlock()

	type result struct {
		project string
		caps    destinations.Capabilities
	}
	results := make([]result, 0, len(s.Dispatcher.Projects))
	var mu sync.Mutex
	var wg sync.WaitGroup

	for name, client := range s.Dispatcher.Projects {
		wg.Add(1)
		go func(name string, c route.ProjectQuerier) {
			defer wg.Done()
			cctx, cancel := context.WithTimeout(ctx, 1*time.Second)
			defer cancel()
			caps, err := c.Capabilities(cctx)
			if err != nil {
				return
			}
			mu.Lock()
			results = append(results, result{project: name, caps: caps})
			mu.Unlock()
		}(name, client)
	}
	wg.Wait()

	var topics []string
	seen := map[string]bool{}
	people := map[string][]string{}
	for _, r := range results {
		for _, t := range r.caps.Topics {
			t = strings.ToLower(strings.TrimSpace(t))
			if t == "" || seen[t] {
				continue
			}
			seen[t] = true
			topics = append(topics, t)
		}
		for _, person := range r.caps.People {
			person = strings.ToLower(strings.TrimSpace(person))
			if person == "" {
				continue
			}
			people[r.project] = append(people[r.project], person)
		}
	}

	s.capMu.Lock()
	s.capTopics = topics
	s.capPeople = people
	s.capFetched = time.Now()
	s.capMu.Unlock()
	return capabilitySnapshot{topics: topics, people: people}
}

// applyPerson sets req.Person from a household name in the question when the
// caller did not name a person. It matches only the routed project's published
// people list, so a name from another project does not leak across. When no
// name matches, PersonOrDefault still falls back to the asker (docs/hub-api.md
// section 5).
func applyPerson(req hear.Request, decision understand.Decision, people map[string][]string) hear.Request {
	if req.Person != "" {
		return req
	}
	var projects []string
	switch decision.Action {
	case understand.ActionRoute:
		projects = []string{decision.Project}
	case understand.ActionFanout:
		projects = decision.Projects
	default:
		return req
	}
	for _, p := range projects {
		if name := understand.MatchPerson(req.Text, people[p]); name != "" {
			req.Person = name
			return req
		}
	}
	return req
}

// LogAsk writes one log line for a completed request (docs/hub-api.md section
// 10). It holds no secret.
func (s *Service) LogAsk(req hear.Request, resp answer.Response, latency time.Duration) {
	if s.Log == nil {
		return
	}
	project := ""
	if resp.Route != nil {
		project = resp.Route.Project
	}
	routeText := string(resp.Shape)
	if resp.Route != nil {
		routeText = resp.Route.Kind
		if project != "" {
			routeText += ":" + project
		}
	}
	_ = s.Log.Write(logbook.Entry{
		Time:      time.Now().UTC().Format(time.RFC3339),
		RequestID: req.RequestID,
		Source:    string(req.Source),
		Asker:     req.Asker,
		Text:      req.Text,
		Route:     routeText,
		Status:    resp.Status,
		LatencyMS: latency.Milliseconds(),
		Shape:     string(resp.Shape),
	})
}

// ApproveRequest is the POST /approve body.
type ApproveRequest struct {
	ApprovalID string `json:"approval_id"`
	Decision   string `json:"decision"`
	Asker      string `json:"asker"`
	Device     string `json:"device"`
}

// ApproveResult is the result block.
type ApproveResult struct {
	Kind   string `json:"kind"`
	Status string `json:"status"`
	Detail string `json:"detail,omitempty"`
}

// ApproveResponse is the POST /approve reply.
type ApproveResponse struct {
	Status     string        `json:"status"`
	ApprovalID string        `json:"approval_id"`
	Decision   string        `json:"decision"`
	Result     ApproveResult `json:"result"`
	Answer     string        `json:"answer"`
}

// ErrBadApprove is returned for a malformed /approve body.
func ParseApprove(body []byte) (ApproveRequest, *apierr.Error) {
	var r ApproveRequest
	if err := decodeJSON(body, &r); err != nil {
		return ApproveRequest{}, apierr.New(400, "bad_request", "The request body is not valid JSON.")
	}
	r.ApprovalID = strings.TrimSpace(r.ApprovalID)
	r.Decision = strings.TrimSpace(r.Decision)
	r.Asker = strings.TrimSpace(r.Asker)
	if r.ApprovalID == "" {
		return ApproveRequest{}, apierr.New(422, "unprocessable", "Field 'approval_id' is required.")
	}
	if r.Asker != "sam" {
		return ApproveRequest{}, apierr.New(422, "unsupported_asker", "The Hub supports one asker, Sam.")
	}
	switch r.Decision {
	case "approve", "deny":
	default:
		return ApproveRequest{}, apierr.New(400, "bad_request", "Field 'decision' must be approve or deny.")
	}
	return r, nil
}

// Approve records one decision and runs the approved action. In v1 the only
// Hub-owned approval is a Paperclip job (docs/hub-api.md section 6).
func (s *Service) Approve(ctx context.Context, req ApproveRequest) (ApproveResponse, *apierr.Error) {
	ap, err := s.Approvals.Decide(req.ApprovalID)
	if err != nil {
		switch err {
		case ErrNoApproval:
			return ApproveResponse{}, apierr.New(404, "not_found", "That approval does not exist.")
		case ErrApprovalExpired:
			return ApproveResponse{}, apierr.New(409, "approval_expired", "That approval expired. Ask again.")
		case ErrAlreadyDecided:
			return ApproveResponse{}, apierr.New(409, "already_decided", "That approval was already decided.")
		default:
			return ApproveResponse{}, apierr.New(500, "internal", "The Hub had an internal problem.")
		}
	}

	resp := ApproveResponse{
		Status:     "ok",
		ApprovalID: ap.ID,
		Decision:   req.Decision,
		Result:     ApproveResult{Kind: ap.Action},
	}

	if req.Decision == "deny" {
		resp.Result.Status = "denied"
		resp.Answer = "Fine, I did not file it."
		return resp, nil
	}

	switch ap.Action {
	case "paperclip_job":
		if s.Paperclip == nil {
			resp.Result.Status = "failed"
			resp.Result.Detail = "Paperclip is not connected."
			resp.Answer = "I could not file the job."
			return resp, nil
		}
		_, err := s.Paperclip.FileJob(ctx, ap.Summary, "Filed by the Hub after an admin approval.")
		if err != nil {
			resp.Result.Status = "failed"
			resp.Result.Detail = "The job did not file."
			resp.Answer = "I could not file the job."
			return resp, nil
		}
		resp.Result.Status = "filed"
		resp.Result.Detail = "Issue filed."
		resp.Answer = "Filed the job."
		return resp, nil
	default:
		resp.Result.Status = "failed"
		resp.Result.Detail = "Unknown action."
		resp.Answer = "I do not know how to run that."
		return resp, nil
	}
}

// decodeJSON decodes one JSON object. Unknown fields are ignored so clients
// may add fields (docs/hub-api.md section 4).
func decodeJSON(body []byte, v any) error {
	return json.NewDecoder(bytes.NewReader(body)).Decode(v)
}