Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/missionmatch / internal/modules/ai/orchestrator.go |
| Date | Not specified in source |
| Repository | Janady13/missionmatch |
| Revision | 2178e7298d0e5d9af83afa12402fd587f0ba6ed8 |
| Branch | feature/aioe-upgrade |
| Path or archive member | internal/modules/ai/orchestrator.go |
| Original SHA-256 | 85cfcb2190dccfe07afe3701f330b595ab332dcf444c622445dd7c91604ba9cc |
| Git blob SHA | 7514981ca6220d054fcbca04402b3157b423c41e |
| Reading-copy SHA-256 | 85cfcb2190dccfe07afe3701f330b595ab332dcf444c622445dd7c91604ba9cc |
Recovered source transcript
Download complete source text ↓441 lines · 21.9 KB
package ai
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"github.com/jackc/pgx/v5"
"github.com/missionworkforce/missionmatch/internal/platform/audit"
"github.com/missionworkforce/missionmatch/internal/platform/inference"
"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)
// Route is the strategy the router picks for a question. The research recommends choosing the cheapest
// sufficient strategy per query (adaptive RAG), and routing aggregation questions to the database rather
// than to vector RAG (docs/AI_RESEARCH.md, routing + TAG).
type Route string
const (
RouteRefuse Route = "refuse" // unsafe input (eeo, injection)
RouteTAG Route = "tag" // the database must compute (count, sum, aggregate)
RouteRAG Route = "rag" // retrieve the tenant's candidates and narrate
RouteDirect Route = "direct" // answer from stable rules in context (CAG), no retrieval
RouteGraph Route = "graph" // walk a cited graph evidence path (G3 skill adjacency, G1 market backbone)
RoutePlan Route = "plan" // bounded plan and execute over a set of closed Registry intents (the planner)
)
// RationaleSource records whether the answer prose was confirmed by the faithfulness gate (ai_faithful) or
// fell back to the deterministic rendering of the evidence (deterministic).
const (
SourceAIFaithful = "ai_faithful"
SourceDeterministic = "deterministic"
)
// ErrNoModel marks that no inference client is configured (or it was not needed). Generation then falls
// back to the deterministic answer; it is never a hang and never a panic.
var ErrNoModel = errors.New("ai: no inference client configured")
// ErrUnknownIntent marks a TAG request for an intent that is not in the curated registry. The handler maps
// it to a 422, never a raw SQL error.
var ErrUnknownIntent = errors.New("ai: unknown query intent")
// Response is an answer plus the provenance the self correction gate and the audit need. The answer is
// ALWAYS grounded: it is the model's prose only when the faithfulness gate confirms it, otherwise the
// deterministic rendering of the same evidence. A refusal (EEO, unanswerable) carries no answer.
type Response struct {
Answer string `json:"answer"`
Route Route `json:"route"`
RationaleSource string `json:"rationale_source"` // ai_faithful | deterministic
Citations []string `json:"citations"` // the evidence the answer was grounded on
Faithful bool `json:"faithful"`
FaithScore float64 `json:"faithfulness_score"`
EvidenceCount int `json:"evidence_count"`
Refused bool `json:"refused"`
RefusalReason string `json:"refusal_reason,omitempty"`
FallbackReason string `json:"fallback_reason,omitempty"` // why the deterministic fallback was used; NEVER in the audit payload
MissionScope string `json:"mission_scope,omitempty"` // the mission area retrieval was scoped to (WS6), empty when unscoped
}
// Orchestrator routes a question across the augmented generation strategies, runs the faithfulness gate
// before any model prose ships, and audits every action. It holds the tenant database (RLS scoped) and the
// stateless inference client; it stores nothing itself.
type Orchestrator struct {
db *tenant.DB
client *inference.Client
aud *audit.Writer
mem *Memory
}
// NewOrchestrator builds the orchestrator. client may be nil (every answer is then deterministic, the
// graceful no plane posture); aud may be nil in tests.
func NewOrchestrator(db *tenant.DB, client *inference.Client, aud *audit.Writer) *Orchestrator {
return &Orchestrator{db: db, client: client, aud: aud}
}
// WithMemory wires the per tenant memory so RAG answers are memory augmented: the tenant's own recalled
// craft notes join the retrieved candidates as evidence. Memory is RLS scoped, so one tenant's memory can
// never surface in another's answer. Returns the orchestrator for chaining.
func (o *Orchestrator) WithMemory(m *Memory) *Orchestrator {
o.mem = m
return o
}
// Answer is the top level dispatcher: a structured intent runs the safe TAG path, or the structured GRAPH
// path when the named intent is one of the closed graph traversals; otherwise the rule based router picks the
// conservative GraphRAG path (a free form capture or skill question whose params can be bound unambiguously),
// CAG (stable rules), RAG (retrieve and narrate), or a refusal. KAG skill adjacency is a live capability
// reached directly (ai.SkillAdjacency). A free form aggregation question (RouteTAG with no intent) is served
// best effort by RAG, because the brain never authors free form SQL; an exact aggregation over tenant data
// uses the structured intent path. The model never authors SQL or params on any path: a graph intent and its
// arg slots are fixed, and a free form graph question whose required params cannot be extracted falls back to
// RAG rather than guessing.
func (o *Orchestrator) Answer(ctx context.Context, question, intentName string, args []any) (*Response, error) {
if strings.TrimSpace(intentName) != "" {
// A named GRAPH intent runs the structured graph path (the cited evidence walk, gated and audited as
// route graph); every other named intent runs the existing TAG path byte for byte as before.
if IsGraphIntent(intentName) {
return o.AnswerGraph(ctx, question, intentName, args)
}
return o.AnswerTAG(ctx, intentName, args)
}
// The conservative natural language graph router runs BEFORE the general Classify: a free form question
// that clearly names a capture or skill graph traversal AND carries the params to bind it unambiguously
// routes to the structured graph path; everything else (including a graph shaped question whose params
// could not be bound) falls through to the existing classification, default path invariance.
if gr := graphClassify(question); gr != nil {
return o.AnswerGraph(ctx, question, gr.intent, gr.args)
}
// THE PLANNER. A COMPOUND cross plane question (one that names more than one plane, for example a bench
// word AND a market word) that the single shot routes above could not bind to one intent routes to the
// planner, which selects and runs a SET of closed Registry intents and composes their evidence. This runs
// AFTER the single intent fast paths and BEFORE the Classify RAG fallback. The signal is conservative: a
// non compound question is byte for byte unchanged (isCompoundQuestion returns false), so a working single
// shot route is never regressed. AnswerPlan itself runs the EEO guards first and falls back to RAG when no
// stable plan binds, so an unbindable compound question is also unchanged.
if isCompoundQuestion(question) {
return o.AnswerPlan(ctx, question)
}
switch Classify(question) {
case RouteRefuse:
return o.refuse(ctx, RouteRefuse, "the question references prohibited demographic data")
case RouteRAG, RouteTAG:
return o.AnswerRAG(ctx, question, 8)
default:
return o.AnswerDirect(ctx, question)
}
}
// Classify is the lightweight router. A trained classifier is the documented upgrade; rule based routing
// over question shape is competitive at this scale and is transparent (docs/AI_RESEARCH.md, routing).
func Classify(question string) Route {
q := strings.ToLower(question)
// Refuse a question that names the eeo schema OR that asks about a protected attribute in plain words.
if ForbidEEO(q) != nil || MentionsProtectedAttribute(q) {
return RouteRefuse
}
// Knowledge first: a definitional or explanatory question the curated knowledge base can answer routes to
// the direct CAG path even when it also contains a retrieval trigger word. "How does the match score work"
// contains "match" (a RAG trigger) but is a definition question the knowledge base answers crisply, so it
// must not fall into retrieval. This only fires when a knowledge card actually matches, so a definitional
// shape with no card ("what are our hardest staffing gaps") still falls through to the data routes below.
if isDefinitionalQuestion(q) && len(SelectKnowledge(question, 1)) > 0 {
return RouteDirect
}
for _, kw := range []string{"how many", "count", "number of", "total", "sum", "average", "percentage", "ratio", "mix", "breakdown"} {
if strings.Contains(q, kw) {
return RouteTAG
}
}
for _, kw := range []string{"who", "which candidate", "find", "show me", "list", "top ", "best ", "match", "search"} {
if strings.Contains(q, kw) {
return RouteRAG
}
}
return RouteDirect
}
// isDefinitionalQuestion reports whether the question asks for an explanation or definition (what a thing is,
// what it means, how a mechanism works) rather than for tenant data. The patterns are anchored so a data
// question that starts "how many" or "how much" is NOT caught: "how do" and "how does" match, "how many" does
// not, so a count question still routes to TAG. Paired with a knowledge card hit in Classify, this is what
// sends a definition question to the taught knowledge path instead of retrieval.
func isDefinitionalQuestion(q string) bool {
q = strings.TrimSpace(q)
for _, p := range []string{
"what is ", "what are ", "what does ", "what makes ", "what counts",
"how does ", "how do ", "how is ", "how are ",
"explain", "define ", "definition of", "tell me about", "describe ", "meaning of",
"why does ", "why is ", "why do ", "what's ",
} {
if strings.HasPrefix(q, p) || strings.Contains(q, " "+p) {
return true
}
}
return false
}
// AnswerRAG retrieves the tenant's candidates for the question, narrates strictly from that evidence, and
// runs the faithfulness gate before returning. The deterministic engine remains the floor: this narrates
// over retrieved rows, it never scores. Retrieval needs the plane to embed the query, so a plane outage
// surfaces as a typed error (through the circuit breaker), never a hang.
func (o *Orchestrator) AnswerRAG(ctx context.Context, question string, limit int) (*Response, error) {
if ForbidEEO(question) != nil || MentionsProtectedAttribute(question) {
return o.refuse(ctx, RouteRAG, "the question references prohibited demographic data")
}
if o.client == nil {
return nil, ErrNoModel // retrieval cannot embed the query without the plane
}
// Mission scoping (WS6 Option A): if the question names a mission area from the curated mission_spine
// vocabulary, constrain retrieval to candidates whose full_name, headline, or location text matches it.
// An unnamed mission is "", which runs the unscoped retrieval byte for byte as before.
mission := MissionFromQuestion(question)
hits, err := HybridSearch(ctx, o.db, o.client, question, mission, limit)
if err != nil {
return nil, err
}
if len(hits) == 0 {
reason := "no candidates matched the question"
if mission != "" {
reason = "no candidates matched the question within the " + mission + " mission scope"
}
return o.refuse(ctx, RouteRAG, reason)
}
// Evidence carries the MATCHED TEXT, not just the name.
//
// HybridSearch already computes a ts_headline snippet precisely so the narration and the faithfulness
// gate can cite why a candidate matched, and discarding it left the model with nothing but names to
// reason from: every substantive question ("who holds an active TS/SCI clearance") was structurally
// unanswerable, because the answer is in the candidate text and the text was never passed.
//
// The raw fused score is deliberately NOT shown. It is an internal Reciprocal Rank Fusion artifact on
// an arbitrary scale, roughly 1/60 per arm, and it carries no meaning a reader or a model can use. In
// practice the model treated those floats as data and reasoned about them out loud, which is a
// fabrication surface rather than evidence. Rank position is the honest ordinal signal, so that is what
// the evidence states.
evidence := make([]string, len(hits))
names := make([]string, len(hits))
for i, h := range hits {
line := fmt.Sprintf("%d. %s", i+1, h.FullName)
if s := strings.TrimSpace(h.Snippet); s != "" {
line += ": " + s
}
evidence[i] = line
names[i] = h.FullName
}
// Memory augmented generation: the tenant's own recalled craft notes join the evidence (RLS scoped, so
// never another tenant's). Best effort, never fatal: a recall error or no client just omits the notes.
if o.mem != nil {
if notes, mErr := o.mem.Recall(ctx, question, 3); mErr == nil {
for _, n := range notes {
evidence = append(evidence, "memory note: "+n)
}
}
}
deterministic := "Top matching candidates: " + strings.Join(names, ", ") + "."
model, genErr := o.generate(ctx,
"Answer the question using ONLY the listed candidates as evidence. Cite candidates by name. If the evidence does not answer the question, say so.",
"QUESTION: "+question+"\n\nCANDIDATES:\n"+strings.Join(evidence, "\n"))
resp, aerr := o.finalize(ctx, RouteRAG, model, genErr, deterministic, evidence)
if aerr != nil {
return nil, aerr
}
resp.MissionScope = mission
return resp, nil
}
// AnswerTAG runs a curated, parameterized, read only intent and narrates the database's own result. This
// is the path for aggregation questions; the model never authors SQL. The data is the database's, so a
// plane outage degrades only the prose: the deterministic rendering of the rows is always returned.
func (o *Orchestrator) AnswerTAG(ctx context.Context, intentName string, args []any) (*Response, error) {
intent, ok := IntentByName(intentName)
if !ok {
return nil, fmt.Errorf("%w: %q", ErrUnknownIntent, intentName)
}
res, err := ExecuteIntent(ctx, o.db, intent, args)
if err != nil {
return nil, err
}
rendered, _ := json.Marshal(map[string]any{"columns": res.Columns, "rows": res.Rows})
evidence := []string{string(rendered)}
deterministic := renderTable(intent.Description, res.Columns, res.Rows)
model, genErr := o.generate(ctx,
"State the result plainly from the table data. Do not invent numbers; report exactly what the rows say.",
"INTENT: "+intent.Description+"\n\nRESULT:\n"+string(rendered))
return o.finalize(ctx, RouteTAG, model, genErr, deterministic, evidence)
}
// AnswerDirect answers from the stable public rules in context (CAG): no retrieval, no tenant data. The
// rules are contents free, so this never touches tenant rows; a plane outage returns the rules themselves.
func (o *Orchestrator) AnswerDirect(ctx context.Context, question string) (*Response, error) {
if ForbidEEO(question) != nil || MentionsProtectedAttribute(question) {
return o.refuse(ctx, RouteDirect, "the question references prohibited demographic data")
}
rules := PublicRules()
// Teach the answer from the curated knowledge base: select the cards relevant to the question so only the
// relevant knowledge enters the model context, and cite each card's source. The standing rules are always
// the base evidence, so the scoring and clearance facts are always available.
cards := SelectKnowledge(question, 4)
citations := []string{"Platform scoring and clearance rules"}
var taught strings.Builder
for _, c := range cards {
taught.WriteString("\n\n[")
taught.WriteString(c.Topic)
taught.WriteString("] ")
taught.WriteString(c.Content)
taught.WriteString(" (source: ")
taught.WriteString(c.Source)
taught.WriteString(")")
citations = append(citations, c.Topic+" (source: "+c.Source+")")
}
// The deterministic answer is the taught knowledge when a card matched, else the standing rules, so a plane
// outage still returns exactly what the brain was taught and never a guess.
deterministic := renderKnowledge(cards)
if deterministic == "" {
deterministic = rules
}
model, genErr := o.generate(ctx,
"Answer using ONLY the platform rules and the taught knowledge below as evidence, and cite the topics you used. If the evidence does not answer the question, say so plainly. Never invent a number, a weight, a rule, a definition, or a fact you were not given.",
"QUESTION: "+question+"\n\nRULES:\n"+rules+"\n\nTAUGHT KNOWLEDGE:"+taught.String())
return o.finalize(ctx, RouteDirect, model, genErr, deterministic, citations)
}
// generate calls the plane (the injection guard is wrapped on the gateway side). A nil client is not an
// error to the caller: it returns ErrNoModel so finalize takes the deterministic path.
func (o *Orchestrator) generate(ctx context.Context, system, prompt string) (string, error) {
if o.client == nil {
return "", ErrNoModel
}
out, err := o.client.Generate(ctx, inference.GenerateRequest{System: system, Prompt: prompt, MaxTokens: 400, Temperature: 0.1})
if err != nil {
return "", err
}
return strings.TrimSpace(out.Text), nil
}
// finalize is the self correction step. It returns the model prose ONLY when the faithfulness gate confirms
// it is supported by the evidence; otherwise it returns the deterministic rendering of the same evidence
// (a 200, rationale_source deterministic), recording why. The audit records the route, the rationale source,
// and the evidence count, never the answer text, the question, the fallback reason, or any candidate.
func (o *Orchestrator) finalize(ctx context.Context, route Route, modelAnswer string, genErr error, deterministic string, citations []string) (*Response, error) {
resp := &Response{Route: route, Citations: citations, EvidenceCount: len(citations)}
if genErr != nil || strings.TrimSpace(modelAnswer) == "" {
resp.Answer = deterministic
resp.RationaleSource = SourceDeterministic
resp.FallbackReason = fallbackReason(genErr)
if err := o.audit(ctx, "aedt.ai.answered", map[string]any{"route": route, "rationale_source": SourceDeterministic, "evidence": len(citations)}); err != nil {
return nil, err
}
return resp, nil
}
verdict, err := o.client.CheckFaithfulness(ctx, modelAnswer, citations)
if err == nil && verdict.Supported {
resp.Answer = modelAnswer
resp.RationaleSource = SourceAIFaithful
resp.Faithful = true
resp.FaithScore = verdict.Score
} else {
resp.Answer = deterministic
resp.RationaleSource = SourceDeterministic
if err != nil {
resp.FallbackReason = fallbackReason(err)
} else {
resp.FaithScore = verdict.Score
resp.FallbackReason = "could_not_confirm"
}
}
if err := o.audit(ctx, "aedt.ai.answered", map[string]any{"route": route, "rationale_source": resp.RationaleSource, "faithful": resp.Faithful, "evidence": len(citations)}); err != nil {
return nil, err
}
return resp, nil
}
func fallbackReason(err error) string {
switch {
case err == nil:
return "could_not_confirm"
case errors.Is(err, ErrNoModel):
return "no_model"
case errors.Is(err, inference.ErrCircuitOpen):
return "circuit_open"
case errors.Is(err, context.DeadlineExceeded):
return "plane_timeout"
default:
return "plane_error"
}
}
// renderTable builds a deterministic, readable answer from a TAG result, so the aggregation question is
// answered exactly from the database even when the model prose is withheld.
func renderTable(description string, columns []string, rows [][]any) string {
if len(rows) == 0 {
return description + ": no rows."
}
var b strings.Builder
b.WriteString(description + ": ")
parts := make([]string, 0, len(rows))
for _, row := range rows {
cells := make([]string, 0, len(row))
for i, v := range row {
col := ""
if i < len(columns) {
col = columns[i] + "="
}
cells = append(cells, fmt.Sprintf("%s%v", col, v))
}
parts = append(parts, strings.Join(cells, " "))
}
b.WriteString(strings.Join(parts, "; "))
b.WriteString(".")
return b.String()
}
// refuse builds a refusal and records it. The audit error is returned, so a refusal that cannot be
// recorded is an error rather than a silently unaudited refusal.
//
// Failing closed here is safe in the direction that matters. A refusal grants nothing: the caller gets no
// answer, no evidence, and no tenant data whether it is served as a 422 refusal or as a 503. So the only
// thing lost by failing closed is the polite explanation, and the thing gained is that an EEO probe, or a
// question from an unauthenticated tenant context, can never be repeated invisibly. That matches
// auth.auditSessionRejected, which already fails a rejected session closed for exactly this reason.
func (o *Orchestrator) refuse(ctx context.Context, route Route, reason string) (*Response, error) {
if err := o.audit(ctx, "aedt.ai.refused", map[string]any{"route": route, "reason": reason}); err != nil {
return nil, err
}
return &Response{Route: route, Refused: true, RefusalReason: reason}, nil
}
// ErrAuditFailed marks that an AEDT event for an AI action could not be written. The handler maps it to a
// 503, never to a 4xx: nothing is wrong with the caller's request, the platform simply cannot record the
// decision right now, and under the audit package contract an unrecordable decision is not served.
var ErrAuditFailed = errors.New("ai: the AEDT audit event could not be recorded")
// audit writes one append only AEDT event for an AI action. Payloads carry the route and verdict only,
// never the answer text, the question, the fallback reason, or any candidate, so the audit holds no PII.
//
// It FAILS CLOSED. Every AI route audits through this one helper, and it used to discard the write error,
// so an AI answer whose AEDT record could not be written was served anyway. That is the exact posture
// internal/platform/audit and matching.auditOnTx forbid: a decision that cannot be recorded rolls back
// rather than being served unaudited. An automated employment decision tool whose audit trail is missing
// the decisions it actually made is worse than one that is briefly unavailable, because the gap is
// invisible and unrecoverable after the fact.
//
// The one case that is NOT a failure is an orchestrator with no audit writer or no database at all. That
// is a deliberate configuration (the unit tests, and the no plane posture), not a failed write, so it
// stays a no op and answers still flow.
func (o *Orchestrator) audit(ctx context.Context, action string, payload map[string]any) error {
if o.aud == nil || o.db == nil {
return nil
}
body, _ := json.Marshal(payload)
if err := o.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
return o.aud.OnTx(ctx, tx, audit.Event{Action: action, EntityType: "ai", NewValues: body})
}); err != nil {
return fmt.Errorf("%w: %s: %v", ErrAuditFailed, action, err)
}
return nil
}