jbergner
94dbd4ccab
All checks were successful
release-tag / release-image (push) Successful in 2m32s
479 lines
16 KiB
Go
479 lines
16 KiB
Go
package web
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"log/slog"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/local/glpi-neural-brain/internal/sourceagent"
|
|
)
|
|
|
|
func (s *Server) sourceStoreAvailable(w http.ResponseWriter) bool {
|
|
if s.SourceAgents != nil {
|
|
return true
|
|
}
|
|
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "source agent service is not initialized"})
|
|
return false
|
|
}
|
|
func (s *Server) adminAuthorized(w http.ResponseWriter, r *http.Request) bool {
|
|
if s.authorized(r) {
|
|
return true
|
|
}
|
|
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"})
|
|
return false
|
|
}
|
|
|
|
func (s *Server) handleListSourceAgents(w http.ResponseWriter, r *http.Request) {
|
|
if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) {
|
|
return
|
|
}
|
|
ctx, cancel := contextTimeout(r, 10*time.Second)
|
|
defer cancel()
|
|
agents, err := s.SourceAgents.ListAgents(ctx)
|
|
if err != nil {
|
|
writeJSON(w, 500, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
tasks, err := s.SourceAgents.ListTasks(ctx, "")
|
|
if err != nil {
|
|
writeJSON(w, 500, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
stats, err := s.SourceAgents.Stats(ctx)
|
|
if err != nil {
|
|
writeJSON(w, 500, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
// Keep collection fields JSON-stable. A nil Go slice serializes as null,
|
|
// which previously crashed the Source Agents UI when there were no tasks.
|
|
if agents == nil {
|
|
agents = []sourceagent.Agent{}
|
|
}
|
|
if tasks == nil {
|
|
tasks = []sourceagent.Task{}
|
|
}
|
|
controllerPolicy, _ := s.SourceAgents.ControllerPolicy(ctx)
|
|
profiles, _ := s.SourceAgents.ListControllerProfiles(ctx)
|
|
controller := s.SourceAgents.ControllerStats(ctx)
|
|
controller["policy"] = controllerPolicy
|
|
controller["profiles"] = profiles
|
|
writeJSON(w, 200, map[string]any{"agents": agents, "tasks": tasks, "inbox": stats, "compute": s.SourceAgents.ComputeStats(), "controller": controller, "brain_url": s.PublicURL})
|
|
}
|
|
func (s *Server) handleCreateSourceAgent(w http.ResponseWriter, r *http.Request) {
|
|
if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) {
|
|
return
|
|
}
|
|
var in struct {
|
|
ID string `json:"id"`
|
|
Name string `json:"name"`
|
|
}
|
|
if err := decode(r, &in); err != nil {
|
|
writeJSON(w, 400, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
ctx, cancel := contextTimeout(r, 10*time.Second)
|
|
defer cancel()
|
|
a, token, err := s.SourceAgents.CreateAgent(ctx, in.ID, in.Name)
|
|
if err != nil {
|
|
writeJSON(w, 400, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusCreated, map[string]any{"agent": a, "token": token, "brain_url": s.PublicURL, "token_notice": "Der Token wird nur in dieser Antwort im Klartext ausgegeben."})
|
|
}
|
|
func (s *Server) handlePatchSourceAgent(w http.ResponseWriter, r *http.Request) {
|
|
if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) {
|
|
return
|
|
}
|
|
var in struct {
|
|
Enabled *bool `json:"enabled"`
|
|
}
|
|
if err := decode(r, &in); err != nil {
|
|
writeJSON(w, 400, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
if in.Enabled == nil {
|
|
writeJSON(w, 400, map[string]string{"error": "enabled is required"})
|
|
return
|
|
}
|
|
ctx, cancel := contextTimeout(r, 10*time.Second)
|
|
defer cancel()
|
|
if err := s.SourceAgents.SetAgentEnabled(ctx, r.PathValue("id"), *in.Enabled); err != nil {
|
|
writeJSON(w, 404, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
writeJSON(w, 200, map[string]any{"ok": true})
|
|
}
|
|
func (s *Server) handleDeleteSourceAgent(w http.ResponseWriter, r *http.Request) {
|
|
if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) {
|
|
return
|
|
}
|
|
ctx, cancel := contextTimeout(r, 10*time.Second)
|
|
defer cancel()
|
|
if err := s.SourceAgents.DeleteAgent(ctx, r.PathValue("id")); err != nil {
|
|
writeJSON(w, 500, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
writeJSON(w, 200, map[string]any{"ok": true})
|
|
}
|
|
func (s *Server) handleRotateSourceAgentToken(w http.ResponseWriter, r *http.Request) {
|
|
if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) {
|
|
return
|
|
}
|
|
ctx, cancel := contextTimeout(r, 10*time.Second)
|
|
defer cancel()
|
|
token, err := s.SourceAgents.RotateToken(ctx, r.PathValue("id"))
|
|
if err != nil {
|
|
writeJSON(w, 404, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
writeJSON(w, 200, map[string]any{"token": token, "brain_url": s.PublicURL, "token_notice": "Der neue Token wird nur in dieser Antwort im Klartext ausgegeben."})
|
|
}
|
|
func (s *Server) handleListSourceTasks(w http.ResponseWriter, r *http.Request) {
|
|
if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) {
|
|
return
|
|
}
|
|
ctx, cancel := contextTimeout(r, 10*time.Second)
|
|
defer cancel()
|
|
tasks, err := s.SourceAgents.ListTasks(ctx, r.PathValue("id"))
|
|
if err != nil {
|
|
writeJSON(w, 500, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
writeJSON(w, 200, map[string]any{"tasks": tasks})
|
|
}
|
|
func (s *Server) handleCreateSourceTask(w http.ResponseWriter, r *http.Request) {
|
|
if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) {
|
|
return
|
|
}
|
|
var t sourceagent.Task
|
|
if err := decode(r, &t); err != nil {
|
|
writeJSON(w, 400, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
t.AgentID = r.PathValue("id")
|
|
if strings.TrimSpace(t.ID) == "" {
|
|
t.ID = "task-" + strconv.FormatInt(time.Now().UnixNano(), 36)
|
|
}
|
|
ctx, cancel := contextTimeout(r, 10*time.Second)
|
|
defer cancel()
|
|
out, err := s.SourceAgents.UpsertTask(ctx, t)
|
|
if err != nil {
|
|
writeJSON(w, 400, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusCreated, out)
|
|
}
|
|
func (s *Server) handleUpdateSourceTask(w http.ResponseWriter, r *http.Request) {
|
|
if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) {
|
|
return
|
|
}
|
|
var t sourceagent.Task
|
|
if err := decode(r, &t); err != nil {
|
|
writeJSON(w, 400, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
t.ID = r.PathValue("id")
|
|
if t.AgentID == "" {
|
|
writeJSON(w, 400, map[string]string{"error": "agent_id required"})
|
|
return
|
|
}
|
|
ctx, cancel := contextTimeout(r, 10*time.Second)
|
|
defer cancel()
|
|
out, err := s.SourceAgents.UpsertTask(ctx, t)
|
|
if err != nil {
|
|
writeJSON(w, 400, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
writeJSON(w, 200, out)
|
|
}
|
|
func (s *Server) handleDeleteSourceTask(w http.ResponseWriter, r *http.Request) {
|
|
if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) {
|
|
return
|
|
}
|
|
ctx, cancel := contextTimeout(r, 10*time.Second)
|
|
defer cancel()
|
|
if err := s.SourceAgents.DeleteTask(ctx, r.PathValue("id")); err != nil {
|
|
writeJSON(w, 500, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
writeJSON(w, 200, map[string]any{"ok": true})
|
|
}
|
|
func (s *Server) handleSourceInbox(w http.ResponseWriter, r *http.Request) {
|
|
if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) {
|
|
return
|
|
}
|
|
limit := 100
|
|
if n, err := strconv.Atoi(r.URL.Query().Get("limit")); err == nil && n > 0 && n <= 1000 {
|
|
limit = n
|
|
}
|
|
ctx, cancel := contextTimeout(r, 10*time.Second)
|
|
defer cancel()
|
|
docs, err := s.SourceAgents.ListInbox(ctx, strings.TrimSpace(r.URL.Query().Get("status")), limit)
|
|
if err != nil {
|
|
writeJSON(w, 500, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
writeJSON(w, 200, map[string]any{"documents": docs})
|
|
}
|
|
func (s *Server) handleSourceInboxStatus(w http.ResponseWriter, r *http.Request) {
|
|
if !s.sourceStoreAvailable(w) || !s.adminAuthorized(w, r) {
|
|
return
|
|
}
|
|
ctx, cancel := contextTimeout(r, 10*time.Second)
|
|
defer cancel()
|
|
st, err := s.SourceAgents.Stats(ctx)
|
|
if err != nil {
|
|
writeJSON(w, 500, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
writeJSON(w, 200, st)
|
|
}
|
|
|
|
func bearerToken(r *http.Request) string {
|
|
v := strings.TrimSpace(r.Header.Get("Authorization"))
|
|
if strings.HasPrefix(v, "Bearer ") {
|
|
return strings.TrimSpace(strings.TrimPrefix(v, "Bearer "))
|
|
}
|
|
return ""
|
|
}
|
|
func (s *Server) authenticateSourceAgent(r *http.Request) (sourceagent.Agent, error) {
|
|
if s.SourceAgents == nil {
|
|
return sourceagent.Agent{}, errors.New("source agent service unavailable")
|
|
}
|
|
ctx, cancel := contextTimeout(r, 10*time.Second)
|
|
defer cancel()
|
|
a, err := s.SourceAgents.Authenticate(ctx, bearerToken(r))
|
|
if err != nil {
|
|
return a, err
|
|
}
|
|
if want := strings.TrimSpace(r.Header.Get("X-Brain-Agent-ID")); want != "" && want != a.ID {
|
|
return sourceagent.Agent{}, errors.New("agent id does not match token")
|
|
}
|
|
return a, nil
|
|
}
|
|
func (s *Server) handleAgentConfig(w http.ResponseWriter, r *http.Request) {
|
|
a, err := s.authenticateSourceAgent(r)
|
|
if err != nil {
|
|
writeJSON(w, 401, map[string]string{"error": "unauthorized"})
|
|
return
|
|
}
|
|
ctx, cancel := contextTimeout(r, 10*time.Second)
|
|
defer cancel()
|
|
cfg, err := s.SourceAgents.RemoteConfig(ctx, a)
|
|
if err != nil {
|
|
writeJSON(w, 500, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
if s.Engine != nil {
|
|
settings := s.Engine.RuntimeSettings()
|
|
cfg.Performance = sourceagent.AgentPerformance{SpeedMode: settings.SpeedMode, CPUWorkers: settings.SpeedCPUWorkers}
|
|
}
|
|
if err := s.SourceAgents.TouchAgent(ctx, a.ID); err != nil {
|
|
writeJSON(w, 500, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
writeJSON(w, 200, cfg)
|
|
}
|
|
|
|
func (s *Server) handleAgentPerformance(w http.ResponseWriter, r *http.Request) {
|
|
if _, err := s.authenticateSourceAgent(r); err != nil {
|
|
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"})
|
|
return
|
|
}
|
|
if s.Engine == nil {
|
|
writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "brain runtime unavailable"})
|
|
return
|
|
}
|
|
settings := s.Engine.RuntimeSettings()
|
|
writeJSON(w, http.StatusOK, sourceagent.AgentPerformance{SpeedMode: settings.SpeedMode, CPUWorkers: settings.SpeedCPUWorkers})
|
|
}
|
|
|
|
func (s *Server) handleAgentHeartbeat(w http.ResponseWriter, r *http.Request) {
|
|
a, err := s.authenticateSourceAgent(r)
|
|
if err != nil {
|
|
writeJSON(w, 401, map[string]string{"error": "unauthorized"})
|
|
return
|
|
}
|
|
var h sourceagent.Heartbeat
|
|
if err := decode(r, &h); err != nil {
|
|
writeJSON(w, 400, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
if h.AgentID != "" && h.AgentID != a.ID {
|
|
writeJSON(w, 403, map[string]string{"error": "agent_id mismatch"})
|
|
return
|
|
}
|
|
ctx, cancel := contextTimeout(r, 10*time.Second)
|
|
defer cancel()
|
|
if err := s.SourceAgents.Heartbeat(ctx, a.ID, h); err != nil {
|
|
writeJSON(w, 500, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
writeJSON(w, 200, map[string]any{"ok": true, "server_time": time.Now().UTC()})
|
|
}
|
|
func (s *Server) handleAgentIngest(w http.ResponseWriter, r *http.Request) {
|
|
a, err := s.authenticateSourceAgent(r)
|
|
if err != nil {
|
|
writeJSON(w, 401, map[string]string{"error": "unauthorized"})
|
|
return
|
|
}
|
|
var batch sourceagent.IngestBatch
|
|
dec := json.NewDecoder(io.LimitReader(r.Body, 32<<20))
|
|
if err := dec.Decode(&batch); err != nil {
|
|
writeJSON(w, 400, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
if batch.AgentID != "" && batch.AgentID != a.ID {
|
|
writeJSON(w, 403, map[string]string{"error": "agent_id mismatch"})
|
|
return
|
|
}
|
|
if batch.SchemaVersion != 0 && batch.SchemaVersion != sourceagent.SchemaVersion {
|
|
writeJSON(w, 400, map[string]string{"error": "unsupported schema_version"})
|
|
return
|
|
}
|
|
if strings.TrimSpace(batch.TaskID) == "" || len(batch.Documents) == 0 || len(batch.Documents) > 500 {
|
|
writeJSON(w, 400, map[string]string{"error": "task_id and 1..500 documents are required"})
|
|
return
|
|
}
|
|
ctx, cancel := contextTimeout(r, 60*time.Second)
|
|
defer cancel()
|
|
result, err := s.SourceAgents.Ingest(ctx, a.ID, batch.TaskID, batch.Documents)
|
|
if err != nil {
|
|
writeJSON(w, 500, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
if result.Accepted > 0 && s.Engine != nil {
|
|
s.Engine.WakeSourceInbox()
|
|
}
|
|
if err := s.SourceAgents.Heartbeat(ctx, a.ID, sourceagent.Heartbeat{AgentID: a.ID, Status: "ingest", Documents: result.Accepted}); err != nil {
|
|
// The ingest transaction already committed. Do not force a duplicate client
|
|
// retry only because the convenience heartbeat failed; surface it in logs.
|
|
slog.Warn("source agent post-ingest heartbeat failed", "agent_id", a.ID, "error", err)
|
|
}
|
|
writeJSON(w, http.StatusAccepted, result)
|
|
}
|
|
|
|
func (s *Server) handleAgentComputeClaim(w http.ResponseWriter, r *http.Request) {
|
|
a, err := s.authenticateSourceAgent(r)
|
|
if err != nil {
|
|
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"})
|
|
return
|
|
}
|
|
capable := false
|
|
for _, capability := range a.Capabilities {
|
|
if capability == sourceagent.ComputeKindVectorGraph {
|
|
capable = true
|
|
break
|
|
}
|
|
}
|
|
if !capable {
|
|
writeJSON(w, http.StatusForbidden, map[string]string{"error": "agent has not advertised vector_graph compute capability"})
|
|
return
|
|
}
|
|
job, ok := s.SourceAgents.ClaimVectorGraphJob(a.ID, 3*time.Minute)
|
|
if !ok {
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return
|
|
}
|
|
w.Header().Set("Content-Type", "application/octet-stream")
|
|
w.Header().Set("Cache-Control", "no-store")
|
|
w.Header().Set("X-Brain-Compute-Job-ID", job.Header.JobID)
|
|
w.Header().Set("X-Brain-Compute-Kind", sourceagent.ComputeKindVectorGraph)
|
|
w.WriteHeader(http.StatusOK)
|
|
if err := sourceagent.WriteVectorGraphJob(w, job); err != nil {
|
|
slog.Warn("source agent compute payload write failed", "agent_id", a.ID, "job_id", job.Header.JobID, "error", err)
|
|
}
|
|
}
|
|
|
|
func (s *Server) handleAgentComputeResult(w http.ResponseWriter, r *http.Request) {
|
|
a, err := s.authenticateSourceAgent(r)
|
|
if err != nil {
|
|
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"})
|
|
return
|
|
}
|
|
var result sourceagent.VectorGraphComputeResult
|
|
dec := json.NewDecoder(io.LimitReader(r.Body, 64<<20))
|
|
if err := dec.Decode(&result); err != nil {
|
|
writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
jobID := strings.TrimSpace(r.PathValue("id"))
|
|
if result.JobID == "" {
|
|
result.JobID = jobID
|
|
}
|
|
if result.JobID != jobID {
|
|
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "job_id mismatch"})
|
|
return
|
|
}
|
|
if result.SchemaVersion != 0 && result.SchemaVersion != sourceagent.SchemaVersion {
|
|
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "unsupported schema_version"})
|
|
return
|
|
}
|
|
if err := s.SourceAgents.CompleteVectorGraphJob(a.ID, result); err != nil {
|
|
writeJSON(w, http.StatusConflict, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusAccepted, map[string]any{"ok": true, "job_id": jobID})
|
|
}
|
|
|
|
func (s *Server) handleAgentArticleQualityComputeClaim(w http.ResponseWriter, r *http.Request) {
|
|
a, err := s.authenticateSourceAgent(r)
|
|
if err != nil {
|
|
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"})
|
|
return
|
|
}
|
|
capable := false
|
|
for _, capability := range a.Capabilities {
|
|
if capability == sourceagent.ComputeKindArticleQuality {
|
|
capable = true
|
|
break
|
|
}
|
|
}
|
|
if !capable {
|
|
writeJSON(w, http.StatusForbidden, map[string]string{"error": "agent has not advertised article_quality compute capability"})
|
|
return
|
|
}
|
|
job, ok := s.SourceAgents.ClaimArticleQualityJob(a.ID, 3*time.Minute)
|
|
if !ok {
|
|
w.WriteHeader(http.StatusNoContent)
|
|
return
|
|
}
|
|
w.Header().Set("Cache-Control", "no-store")
|
|
writeJSON(w, http.StatusOK, job)
|
|
}
|
|
|
|
func (s *Server) handleAgentArticleQualityComputeResult(w http.ResponseWriter, r *http.Request) {
|
|
a, err := s.authenticateSourceAgent(r)
|
|
if err != nil {
|
|
writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"})
|
|
return
|
|
}
|
|
var result sourceagent.ArticleQualityComputeResult
|
|
if err := json.NewDecoder(io.LimitReader(r.Body, 8<<20)).Decode(&result); err != nil {
|
|
writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
jobID := strings.TrimSpace(r.PathValue("id"))
|
|
if result.JobID == "" {
|
|
result.JobID = jobID
|
|
}
|
|
if result.JobID != jobID {
|
|
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "job_id mismatch"})
|
|
return
|
|
}
|
|
if result.SchemaVersion != 0 && result.SchemaVersion != sourceagent.SchemaVersion {
|
|
writeJSON(w, http.StatusBadRequest, map[string]string{"error": "unsupported schema_version"})
|
|
return
|
|
}
|
|
if err := s.SourceAgents.CompleteArticleQualityJob(a.ID, result); err != nil {
|
|
writeJSON(w, http.StatusConflict, map[string]string{"error": err.Error()})
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusAccepted, map[string]any{"ok": true, "job_id": jobID})
|
|
}
|