Janady13/missionmatch / internal/modules/ai/orchestrator.go

1package ai 2 3import ( 4 "context" 5 "encoding/json" 6 "errors" 7 "fmt" 8 "strings" 9 10 "github.com/jackc/pgx/v5" 11 12 "github.com/missionworkforce/missionmatch/internal/platform/audit" 13 "github.com/missionworkforce/missionmatch/internal/platform/inference" 14 "github.com/missionworkforce/missionmatch/internal/platform/tenant" 15) 16 17// Route is the strategy the router picks for a question. The research recommends choosing the cheapest 18// sufficient strategy per query (adaptive RAG), and routing aggregation questions to the database rather 19// than to vector RAG (docs/AI_RESEARCH.md, routing + TAG). 20type Route string 21 22const ( 23 RouteRefuse Route = "refuse" // unsafe input (eeo, injection) 24 RouteTAG Route = "tag" // the database must compute (count, sum, aggregate) 25 RouteRAG Route = "rag" // retrieve the tenant's candidates and narrate 26 RouteDirect Route = "direct" // answer from stable rules in context (CAG), no retrieval 27 RouteGraph Route = "graph" // walk a cited graph evidence path (G3 skill adjacency, G1 market backbone) 28 RoutePlan Route = "plan" // bounded plan and execute over a set of closed Registry intents (the planner) 29) 30 31// RationaleSource records whether the answer prose was confirmed by the faithfulness gate (ai_faithful) or 32// fell back to the deterministic rendering of the evidence (deterministic). 33const ( 34 SourceAIFaithful = "ai_faithful" 35 SourceDeterministic = "deterministic" 36) 37 38// ErrNoModel marks that no inference client is configured (or it was not needed). Generation then falls 39// back to the deterministic answer; it is never a hang and never a panic. 40var ErrNoModel = errors.New("ai: no inference client configured") 41 42// ErrUnknownIntent marks a TAG request for an intent that is not in the curated registry. The handler maps 43// it to a 422, never a raw SQL error. 44var ErrUnknownIntent = errors.New("ai: unknown query intent") 45 46// Response is an answer plus the provenance the self correction gate and the audit need. The answer is 47// ALWAYS grounded: it is the model's prose only when the faithfulness gate confirms it, otherwise the 48// deterministic rendering of the same evidence. A refusal (EEO, unanswerable) carries no answer. 49type Response struct { 50 Answer string `json:"answer"` 51 Route Route `json:"route"` 52 RationaleSource string `json:"rationale_source"` // ai_faithful | deterministic 53 Citations []string `json:"citations"` // the evidence the answer was grounded on 54 Faithful bool `json:"faithful"` 55 FaithScore float64 `json:"faithfulness_score"` 56 EvidenceCount int `json:"evidence_count"` 57 Refused bool `json:"refused"` 58 RefusalReason string `json:"refusal_reason,omitempty"` 59 FallbackReason string `json:"fallback_reason,omitempty"` // why the deterministic fallback was used; NEVER in the audit payload 60 MissionScope string `json:"mission_scope,omitempty"` // the mission area retrieval was scoped to (WS6), empty when unscoped 61} 62 63// Orchestrator routes a question across the augmented generation strategies, runs the faithfulness gate 64// before any model prose ships, and audits every action. It holds the tenant database (RLS scoped) and the 65// stateless inference client; it stores nothing itself. 66type Orchestrator struct { 67 db *tenant.DB 68 client *inference.Client 69 aud *audit.Writer 70 mem *Memory 71} 72 73// NewOrchestrator builds the orchestrator. client may be nil (every answer is then deterministic, the 74// graceful no plane posture); aud may be nil in tests. 75func NewOrchestrator(db *tenant.DB, client *inference.Client, aud *audit.Writer) *Orchestrator { 76 return &Orchestrator{db: db, client: client, aud: aud} 77} 78 79// WithMemory wires the per tenant memory so RAG answers are memory augmented: the tenant's own recalled 80// craft notes join the retrieved candidates as evidence. Memory is RLS scoped, so one tenant's memory can 81// never surface in another's answer. Returns the orchestrator for chaining. 82func (o *Orchestrator) WithMemory(m *Memory) *Orchestrator { 83 o.mem = m 84 return o 85} 86 87// Answer is the top level dispatcher: a structured intent runs the safe TAG path, or the structured GRAPH 88// path when the named intent is one of the closed graph traversals; otherwise the rule based router picks the 89// conservative GraphRAG path (a free form capture or skill question whose params can be bound unambiguously), 90// CAG (stable rules), RAG (retrieve and narrate), or a refusal. KAG skill adjacency is a live capability 91// reached directly (ai.SkillAdjacency). A free form aggregation question (RouteTAG with no intent) is served 92// best effort by RAG, because the brain never authors free form SQL; an exact aggregation over tenant data 93// uses the structured intent path. The model never authors SQL or params on any path: a graph intent and its 94// arg slots are fixed, and a free form graph question whose required params cannot be extracted falls back to 95// RAG rather than guessing. 96func (o *Orchestrator) Answer(ctx context.Context, question, intentName string, args []any) (*Response, error) { 97 if strings.TrimSpace(intentName) != "" { 98 // A named GRAPH intent runs the structured graph path (the cited evidence walk, gated and audited as 99 // route graph); every other named intent runs the existing TAG path byte for byte as before. 100 if IsGraphIntent(intentName) { 101 return o.AnswerGraph(ctx, question, intentName, args) 102 } 103 return o.AnswerTAG(ctx, intentName, args) 104 } 105 // The conservative natural language graph router runs BEFORE the general Classify: a free form question 106 // that clearly names a capture or skill graph traversal AND carries the params to bind it unambiguously 107 // routes to the structured graph path; everything else (including a graph shaped question whose params 108 // could not be bound) falls through to the existing classification, default path invariance. 109 if gr := graphClassify(question); gr != nil { 110 return o.AnswerGraph(ctx, question, gr.intent, gr.args) 111 } 112 // THE PLANNER. A COMPOUND cross plane question (one that names more than one plane, for example a bench 113 // word AND a market word) that the single shot routes above could not bind to one intent routes to the 114 // planner, which selects and runs a SET of closed Registry intents and composes their evidence. This runs 115 // AFTER the single intent fast paths and BEFORE the Classify RAG fallback. The signal is conservative: a 116 // non compound question is byte for byte unchanged (isCompoundQuestion returns false), so a working single 117 // shot route is never regressed. AnswerPlan itself runs the EEO guards first and falls back to RAG when no 118 // stable plan binds, so an unbindable compound question is also unchanged. 119 if isCompoundQuestion(question) { 120 return o.AnswerPlan(ctx, question) 121 } 122 switch Classify(question) { 123 case RouteRefuse: 124 return o.refuse(ctx, RouteRefuse, "the question references prohibited demographic data"), nil 125 case RouteRAG, RouteTAG: 126 return o.AnswerRAG(ctx, question, 8) 127 default: 128 return o.AnswerDirect(ctx, question) 129 } 130} 131 132// Classify is the lightweight router. A trained classifier is the documented upgrade; rule based routing 133// over question shape is competitive at this scale and is transparent (docs/AI_RESEARCH.md, routing). 134func Classify(question string) Route { 135 q := strings.ToLower(question) 136 // Refuse a question that names the eeo schema OR that asks about a protected attribute in plain words. 137 if ForbidEEO(q) != nil || MentionsProtectedAttribute(q) { 138 return RouteRefuse 139 } 140 for _, kw := range []string{"how many", "count", "number of", "total", "sum", "average", "percentage", "ratio", "mix", "breakdown"} { 141 if strings.Contains(q, kw) { 142 return RouteTAG 143 } 144 } 145 for _, kw := range []string{"who", "which candidate", "find", "show me", "list", "top ", "best ", "match", "search"} { 146 if strings.Contains(q, kw) { 147 return RouteRAG 148 } 149 } 150 return RouteDirect 151} 152 153// AnswerRAG retrieves the tenant's candidates for the question, narrates strictly from that evidence, and 154// runs the faithfulness gate before returning. The deterministic engine remains the floor: this narrates 155// over retrieved rows, it never scores. Retrieval needs the plane to embed the query, so a plane outage 156// surfaces as a typed error (through the circuit breaker), never a hang. 157func (o *Orchestrator) AnswerRAG(ctx context.Context, question string, limit int) (*Response, error) { 158 if ForbidEEO(question) != nil || MentionsProtectedAttribute(question) { 159 return o.refuse(ctx, RouteRAG, "the question references prohibited demographic data"), nil 160 } 161 if o.client == nil { 162 return nil, ErrNoModel // retrieval cannot embed the query without the plane 163 } 164 // Mission scoping (WS6 Option A): if the question names a mission area from the curated mission_spine 165 // vocabulary, constrain retrieval to candidates whose full_name, headline, or location text matches it. 166 // An unnamed mission is "", which runs the unscoped retrieval byte for byte as before. 167 mission := MissionFromQuestion(question) 168 hits, err := HybridSearch(ctx, o.db, o.client, question, mission, limit) 169 if err != nil { 170 return nil, err 171 } 172 if len(hits) == 0 { 173 reason := "no candidates matched the question" 174 if mission != "" { 175 reason = "no candidates matched the question within the " + mission + " mission scope" 176 } 177 return o.refuse(ctx, RouteRAG, reason), nil 178 } 179 evidence := make([]string, len(hits)) 180 names := make([]string, len(hits)) 181 for i, h := range hits { 182 evidence[i] = fmt.Sprintf("%s (relevance %.4f)", h.FullName, h.Score) 183 names[i] = h.FullName 184 } 185 // Memory augmented generation: the tenant's own recalled craft notes join the evidence (RLS scoped, so 186 // never another tenant's). Best effort, never fatal: a recall error or no client just omits the notes. 187 if o.mem != nil { 188 if notes, mErr := o.mem.Recall(ctx, question, 3); mErr == nil { 189 for _, n := range notes { 190 evidence = append(evidence, "memory note: "+n) 191 } 192 } 193 } 194 deterministic := "Top matching candidates: " + strings.Join(names, ", ") + "." 195 model, genErr := o.generate(ctx, 196 "Answer the question using ONLY the listed candidates as evidence. Cite candidates by name. If the evidence does not answer the question, say so.", 197 "QUESTION: "+question+"\n\nCANDIDATES:\n"+strings.Join(evidence, "\n")) 198 resp := o.finalize(ctx, RouteRAG, model, genErr, deterministic, evidence) 199 resp.MissionScope = mission 200 return resp, nil 201} 202 203// AnswerTAG runs a curated, parameterized, read only intent and narrates the database's own result. This 204// is the path for aggregation questions; the model never authors SQL. The data is the database's, so a 205// plane outage degrades only the prose: the deterministic rendering of the rows is always returned. 206func (o *Orchestrator) AnswerTAG(ctx context.Context, intentName string, args []any) (*Response, error) { 207 intent, ok := IntentByName(intentName) 208 if !ok { 209 return nil, fmt.Errorf("%w: %q", ErrUnknownIntent, intentName) 210 } 211 res, err := ExecuteIntent(ctx, o.db, intent, args) 212 if err != nil { 213 return nil, err 214 } 215 rendered, _ := json.Marshal(map[string]any{"columns": res.Columns, "rows": res.Rows}) 216 evidence := []string{string(rendered)} 217 deterministic := renderTable(intent.Description, res.Columns, res.Rows) 218 model, genErr := o.generate(ctx, 219 "State the result plainly from the table data. Do not invent numbers; report exactly what the rows say.", 220 "INTENT: "+intent.Description+"\n\nRESULT:\n"+string(rendered)) 221 return o.finalize(ctx, RouteTAG, model, genErr, deterministic, evidence), nil 222} 223 224// AnswerDirect answers from the stable public rules in context (CAG): no retrieval, no tenant data. The 225// rules are contents free, so this never touches tenant rows; a plane outage returns the rules themselves. 226func (o *Orchestrator) AnswerDirect(ctx context.Context, question string) (*Response, error) { 227 if ForbidEEO(question) != nil || MentionsProtectedAttribute(question) { 228 return o.refuse(ctx, RouteDirect, "the question references prohibited demographic data"), nil 229 } 230 rules := PublicRules() 231 evidence := []string{rules} 232 model, genErr := o.generate(ctx, 233 "Answer using ONLY these platform rules as evidence. If the rules do not answer the question, say so. Do not invent a number or a weight.", 234 "QUESTION: "+question+"\n\nRULES:\n"+rules) 235 return o.finalize(ctx, RouteDirect, model, genErr, rules, evidence), nil 236} 237 238// generate calls the plane (the injection guard is wrapped on the gateway side). A nil client is not an 239// error to the caller: it returns ErrNoModel so finalize takes the deterministic path. 240func (o *Orchestrator) generate(ctx context.Context, system, prompt string) (string, error) { 241 if o.client == nil { 242 return "", ErrNoModel 243 } 244 out, err := o.client.Generate(ctx, inference.GenerateRequest{System: system, Prompt: prompt, MaxTokens: 400, Temperature: 0.1}) 245 if err != nil { 246 return "", err 247 } 248 return strings.TrimSpace(out.Text), nil 249} 250 251// finalize is the self correction step. It returns the model prose ONLY when the faithfulness gate confirms 252// it is supported by the evidence; otherwise it returns the deterministic rendering of the same evidence 253// (a 200, rationale_source deterministic), recording why. The audit records the route, the rationale source, 254// and the evidence count, never the answer text, the question, the fallback reason, or any candidate. 255func (o *Orchestrator) finalize(ctx context.Context, route Route, modelAnswer string, genErr error, deterministic string, citations []string) *Response { 256 resp := &Response{Route: route, Citations: citations, EvidenceCount: len(citations)} 257 if genErr != nil || strings.TrimSpace(modelAnswer) == "" { 258 resp.Answer = deterministic 259 resp.RationaleSource = SourceDeterministic 260 resp.FallbackReason = fallbackReason(genErr) 261 o.audit(ctx, "aedt.ai.answered", map[string]any{"route": route, "rationale_source": SourceDeterministic, "evidence": len(citations)}) 262 return resp 263 } 264 verdict, err := o.client.CheckFaithfulness(ctx, modelAnswer, citations) 265 if err == nil && verdict.Supported { 266 resp.Answer = modelAnswer 267 resp.RationaleSource = SourceAIFaithful 268 resp.Faithful = true 269 resp.FaithScore = verdict.Score 270 } else { 271 resp.Answer = deterministic 272 resp.RationaleSource = SourceDeterministic 273 if err != nil { 274 resp.FallbackReason = fallbackReason(err) 275 } else { 276 resp.FaithScore = verdict.Score 277 resp.FallbackReason = "could_not_confirm" 278 } 279 } 280 o.audit(ctx, "aedt.ai.answered", map[string]any{"route": route, "rationale_source": resp.RationaleSource, "faithful": resp.Faithful, "evidence": len(citations)}) 281 return resp 282} 283 284func fallbackReason(err error) string { 285 switch { 286 case err == nil: 287 return "could_not_confirm" 288 case errors.Is(err, ErrNoModel): 289 return "no_model" 290 case errors.Is(err, inference.ErrCircuitOpen): 291 return "circuit_open" 292 case errors.Is(err, context.DeadlineExceeded): 293 return "plane_timeout" 294 default: 295 return "plane_error" 296 } 297} 298 299// renderTable builds a deterministic, readable answer from a TAG result, so the aggregation question is 300// answered exactly from the database even when the model prose is withheld. 301func renderTable(description string, columns []string, rows [][]any) string { 302 if len(rows) == 0 { 303 return description + ": no rows." 304 } 305 var b strings.Builder 306 b.WriteString(description + ": ") 307 parts := make([]string, 0, len(rows)) 308 for _, row := range rows { 309 cells := make([]string, 0, len(row)) 310 for i, v := range row { 311 col := "" 312 if i < len(columns) { 313 col = columns[i] + "=" 314 } 315 cells = append(cells, fmt.Sprintf("%s%v", col, v)) 316 } 317 parts = append(parts, strings.Join(cells, " ")) 318 } 319 b.WriteString(strings.Join(parts, "; ")) 320 b.WriteString(".") 321 return b.String() 322} 323 324func (o *Orchestrator) refuse(ctx context.Context, route Route, reason string) *Response { 325 o.audit(ctx, "aedt.ai.refused", map[string]any{"route": route, "reason": reason}) 326 return &Response{Route: route, Refused: true, RefusalReason: reason} 327} 328 329// audit writes one append only AEDT event for an AI action. Payloads carry the route and verdict only, 330// never the answer text, the question, the fallback reason, or any candidate, so the audit holds no PII. 331func (o *Orchestrator) audit(ctx context.Context, action string, payload map[string]any) { 332 if o.aud == nil || o.db == nil { 333 return 334 } 335 body, _ := json.Marshal(payload) 336 _ = o.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 337 return o.aud.OnTx(ctx, tx, audit.Event{Action: action, EntityType: "ai", NewValues: body}) 338 }) 339} 340