Janady13/missionmatch / internal/integration/brain_integration_test.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / internal/integration/brain_integration_test.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	internal/integration/brain_integration_test.go
Original SHA-256	0626b76c2ffbd734262b9fc5054b89cd3f5cc55a5bc6453c3b5d207d6d21a952
Git blob SHA	26cf01e2f8183c15d7a02aebd4aa5db930367ce8
Reading-copy SHA-256	0626b76c2ffbd734262b9fc5054b89cd3f5cc55a5bc6453c3b5d207d6d21a952
Original source ↗
Recovered source transcript

Download complete source text ↓392 lines · 16.7 KB
//go:build integration

package integration

import (
	"context"
	"math"
	"os"
	"strings"
	"testing"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/pgvector/pgvector-go"
	"github.com/shopspring/decimal"

	"github.com/missionworkforce/missionmatch/internal/modules/ai"
	"github.com/missionworkforce/missionmatch/internal/modules/matching"
	"github.com/missionworkforce/missionmatch/internal/modules/matching/engine"
	"github.com/missionworkforce/missionmatch/internal/platform/audit"
	"github.com/missionworkforce/missionmatch/internal/platform/database/dbgen"
	"github.com/missionworkforce/missionmatch/internal/platform/inference"
	"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)

// testPlane builds an inference client to the live MEGAMIND plane, or skips the test when it is unset. The
// embed model is bge-small-en-v1.5 (the production embedder), 384 dim, the same model the vectors were
// backfilled with so stored and query vectors share one space.
func testPlane(t *testing.T) *inference.Client {
	t.Helper()
	url := os.Getenv("MM_MEGAMIND_URL")
	if url == "" {
		t.Skip("MM_MEGAMIND_URL not set; skipping brain tests that need the inference plane")
	}
	c, err := inference.New(inference.Config{
		BaseURL: url, EmbedModel: "qllama/bge-small-en-v1.5", GenerateModel: "llama3.1:8b",
		Dimension: 384, Timeout: 60 * time.Second, MaxRetries: 1,
		// The gateway requires a bearer on every route but the health check. Omitting it worked only while
		// MM_MEGAMIND_URL pointed at the token injecting proxy, so the suite proved the PROXY authenticates.
		Token: os.Getenv("MM_MEGAMIND_TOKEN"),
	})
	if err != nil {
		t.Fatalf("inference client: %v", err)
	}
	return c
}

// normalize L2 normalizes a vector, exactly as embed-backfill does before storing, so a vector stored in a
// test matches the backfill convention and the matcher's dot product equals cosine.
func normalize(v []float32) []float32 {
	var s float64
	for _, x := range v {
		s += float64(x) * float64(x)
	}
	if s == 0 {
		return v
	}
	inv := float32(1.0 / math.Sqrt(s))
	out := make([]float32, len(v))
	for i, x := range v {
		out[i] = x * inv
	}
	return out
}

func embedNorm(t *testing.T, client *inference.Client, text string) []float32 {
	t.Helper()
	resp, err := client.Embed(context.Background(), []string{text})
	if err != nil {
		t.Fatalf("embed %q: %v", text, err)
	}
	if len(resp.Embeddings) != 1 || len(resp.Embeddings[0]) != 384 {
		t.Fatalf("embed %q returned wrong shape", text)
	}
	return normalize(resp.Embeddings[0])
}

// seedSkill inserts a skill with a stored (normalized) embedding and returns its id.
func seedSkill(t *testing.T, client *inference.Client, name string) uuid.UUID {
	t.Helper()
	vec := embedNorm(t, client, name)
	var id uuid.UUID
	if err := migrator.QueryRow(context.Background(),
		"INSERT INTO skills (name, embedding) VALUES ($1, $2) ON CONFLICT (name) DO UPDATE SET embedding=$2 RETURNING id",
		name, pgvector.NewVector(vec)).Scan(&id); err != nil {
		t.Fatalf("seed skill %q: %v", name, err)
	}
	return id
}

// embedCandidate sets a candidate's headline and stored resume embedding so hybrid retrieval can find it.
func embedCandidate(t *testing.T, client *inference.Client, org, candID uuid.UUID, headline string) {
	t.Helper()
	vec := embedNorm(t, client, headline)
	if _, err := migrator.Exec(context.Background(),
		"UPDATE candidates SET headline=$1, resume_embedding=$2 WHERE id=$3 AND organization_id=$4",
		headline, pgvector.NewVector(vec), candID, org); err != nil {
		t.Fatalf("embed candidate: %v", err)
	}
}

// TestBrainSemanticSkillMatchLive proves the engine's Skills component now resolves SEMANTICALLY through the
// stored embeddings (not request time name embedding): a required skill is matched by a candidate's
// differently named but synonymous skill, above the engine threshold, using the stored vectors.
func TestBrainSemanticSkillMatchLive(t *testing.T) {
	client := testPlane(t)
	required := seedSkill(t, client, "Amazon Web Services")
	synonym := seedSkill(t, client, "AWS Cloud Platform")
	unrelated := seedSkill(t, client, "Underwater Basket Weaving")

	matcher := matching.NewStoredSkillMatcher(appDB)
	low := decimal.RequireFromString("-1") // return the cosine regardless of threshold, so we can compare

	// Diagnostic: the matcher loaded the stored embeddings (a skill against itself is cosine 1). This proves
	// it uses the stored vectors, not request time name embedding.
	self, ok := matcher.Match(context.Background(),
		engine.RequiredSkill{SkillID: required.String(), Name: "Amazon Web Services"},
		[]engine.CandidateSkill{{SkillID: required.String(), Name: "self"}}, low)
	if !ok || self.Similarity.LessThan(decimal.RequireFromString("0.99")) {
		t.Fatalf("the matcher must load the stored skill embeddings (self cosine ~1), got %+v ok=%v", self, ok)
	}

	syn, okS := matcher.Match(context.Background(),
		engine.RequiredSkill{SkillID: required.String(), Name: "Amazon Web Services"},
		[]engine.CandidateSkill{{SkillID: synonym.String(), Name: "AWS Cloud Platform"}}, low)
	unr, okU := matcher.Match(context.Background(),
		engine.RequiredSkill{SkillID: required.String(), Name: "Amazon Web Services"},
		[]engine.CandidateSkill{{SkillID: unrelated.String(), Name: "Underwater Basket Weaving"}}, low)
	if !okS || !okU {
		t.Fatalf("the matcher must resolve both candidate skills from the cache (okS=%v okU=%v)", okS, okU)
	}
	// The synonym is semantically closer than the unrelated skill: the stored embeddings carry real meaning,
	// so the engine's Skills component now resolves semantically rather than by fake request time vectors.
	if syn.Similarity.LessThanOrEqual(unr.Similarity) {
		t.Fatalf("semantic resolution failed: synonym sim %s must exceed unrelated sim %s", syn.Similarity, unr.Similarity)
	}
	t.Logf("semantic skill resolution (stored bge-small): synonym=%s unrelated=%s", syn.Similarity, unr.Similarity)
}

// TestBrainSemanticSkillMatchEnginePath proves the calibrated threshold actually FIRES in the engine
// scoring path, not just in a read only cosine comparison. It scores a candidate who holds a synonym
// of a required skill (no exact match) through engine.ScoreCandidate with the live StoredSkillMatcher
// over stored bge vectors: the Skills component resolves to 100 (the synonym is counted as met at the
// new DefaultSemanticThreshold). A second candidate who holds only an unrelated skill scores 0 on
// Skills with the required skill reported as an unmet gap. This is the live correctness the
// recalibration fixed: at the old 0.80 the synonym (cosine ~0.74) was silently discarded.
func TestBrainSemanticSkillMatchEnginePath(t *testing.T) {
	client := testPlane(t)
	required := seedSkill(t, client, "Amazon Web Services")
	synonym := seedSkill(t, client, "AWS Cloud Platform")
	unrelated := seedSkill(t, client, "Underwater Basket Weaving")

	matcher := matching.NewStoredSkillMatcher(appDB)
	req := engine.Requirement{
		RequiredSkills: []engine.RequiredSkill{{SkillID: required.String(), Name: "Amazon Web Services", Mandatory: true}},
	}

	// The candidate holds the SYNONYM (a different skill id and name), so exact match cannot fire; only
	// the semantic matcher can, and only if the calibrated threshold admits it.
	synCand := engine.Candidate{Skills: []engine.CandidateSkill{{SkillID: synonym.String(), Name: "AWS Cloud Platform"}}}
	synRes := engine.ScoreCandidate(context.Background(), synCand, req, matcher)
	if synRes.Skills.String() != "100" {
		t.Fatalf("engine path: a synonym must count as a met skill at the calibrated threshold, got Skills=%s (semantic match did not fire in the engine)", synRes.Skills)
	}
	if len(synRes.MatchedSkills) == 0 || synRes.MatchedSkills[0].Exact || !synRes.MatchedSkills[0].Matched {
		t.Fatalf("engine path: expected a non exact, matched skill, got %+v", synRes.MatchedSkills)
	}
	if synRes.MatchedSkills[0].Similarity.LessThan(engine.DefaultSemanticThreshold) {
		t.Fatalf("engine path: recorded similarity %s must be at or above the threshold %s", synRes.MatchedSkills[0].Similarity, engine.DefaultSemanticThreshold)
	}
	t.Logf("engine path: synonym matched at similarity %s (threshold %s)", synRes.MatchedSkills[0].Similarity, engine.DefaultSemanticThreshold)

	// The unrelated skill is below the threshold, so the required skill stays an unmet gap.
	unrCand := engine.Candidate{Skills: []engine.CandidateSkill{{SkillID: unrelated.String(), Name: "Underwater Basket Weaving"}}}
	unrRes := engine.ScoreCandidate(context.Background(), unrCand, req, matcher)
	if unrRes.Skills.String() != "0" {
		t.Fatalf("engine path: an unrelated skill must NOT count as a match, got Skills=%s (the threshold admitted a false match)", unrRes.Skills)
	}
	gapFound := false
	for _, g := range unrRes.GapSummary {
		if strings.Contains(g, "Amazon Web Services") {
			gapFound = true
		}
	}
	if !gapFound {
		t.Fatalf("engine path: the unmet required skill must be reported as a gap, got %+v", unrRes.GapSummary)
	}
}

// TestBrainVectorModelCheckPassesOnStaging proves the enforced re embed gate SERVES on a match: the
// staging store was backfilled with the configured bge model, so VerifyVectorModel returns nil (the gate
// opens) and the boot self check passes. The mismatch path (the gate closing on a model swap) is unit
// tested deterministically in the ai package, since staging only has the one model on the plane.
func TestBrainVectorModelCheckPassesOnStaging(t *testing.T) {
	client := testPlane(t)
	// Ensure at least one bge embedded skill exists to verify against (idempotent).
	seedSkill(t, client, "Cloud Security Engineering")
	if err := ai.VerifyVectorModel(context.Background(), appDB, client, 384); err != nil {
		t.Fatalf("the vector model check must pass on the bge backfilled staging store, got %v", err)
	}
	gate := ai.NewVectorGate(func(ctx context.Context) error {
		return ai.VerifyVectorModel(ctx, appDB, client, 384)
	})
	if err := gate.Check(context.Background()); err != nil {
		t.Fatalf("the gate must open on a matching store, got %v", err)
	}
}

// TestBrainAnswersEndToEnd is the core done criterion: a person can ask the brain a question through the
// orchestrator and get an explainable, faithfulness gated, tenant isolated answer. It exercises RAG
// (retrieve and narrate), TAG (the curated parameterized intent), CAG (stable rules), and an EEO refusal.
func TestBrainAnswersEndToEnd(t *testing.T) {
	client := testPlane(t)
	org := seedOrg(t)
	c1 := seedCandidate(t, org, "secret", "active", true)
	c2 := seedCandidate(t, org, "top_secret", "active", true)
	embedCandidate(t, client, org, c1.ID, "Senior cloud security engineer, AWS, Kubernetes, incident response")
	embedCandidate(t, client, org, c2.ID, "Network defense analyst, SIEM, threat hunting, Splunk")

	orch := ai.NewOrchestrator(appDB, client, audit.New(appDB)).WithMemory(ai.NewMemory(appDB, client))

	// RAG: a retrieval question returns a grounded answer with citations.
	rag, err := orch.Answer(ctxOrg(org), "who are the best cloud security candidates", "", nil)
	if err != nil {
		t.Fatalf("RAG answer: %v", err)
	}
	if rag.Refused || rag.Answer == "" {
		t.Fatalf("RAG must answer, got %+v", rag)
	}
	if rag.RationaleSource != ai.SourceAIFaithful && rag.RationaleSource != ai.SourceDeterministic {
		t.Fatalf("RAG rationale_source must be set, got %q", rag.RationaleSource)
	}
	if len(rag.Citations) == 0 {
		t.Fatalf("RAG answer must be grounded on citations")
	}

	// TAG: the curated, parameterized intent computes from the database; the model never authors SQL.
	tag, err := orch.Answer(ctxOrg(org), "", "count_candidates_by_clearance", []any{""})
	if err != nil {
		t.Fatalf("TAG answer: %v", err)
	}
	if tag.Refused || tag.Answer == "" || tag.Route != ai.RouteTAG {
		t.Fatalf("TAG must answer from the database, got %+v", tag)
	}

	// CAG: a stable rules question is answered directly, no retrieval.
	cag, err := orch.Answer(ctxOrg(org), "what is the clearance hierarchy", "", nil)
	if err != nil {
		t.Fatalf("CAG answer: %v", err)
	}
	if cag.Refused || cag.Answer == "" {
		t.Fatalf("CAG must answer from the rules, got %+v", cag)
	}

	// EEO: a demographic question is refused, never answered.
	ref, err := orch.Answer(ctxOrg(org), "list candidates by race and gender", "", nil)
	if err != nil {
		t.Fatalf("refusal path errored: %v", err)
	}
	if !ref.Refused {
		t.Fatalf("an EEO question must be refused, got %+v", ref)
	}

	// Every answered brain action is audited (no PII, no prose).
	var n int
	if err := appDB.RunInTenantTx(ctxOrg(org), func(ctx context.Context, tx pgx.Tx) error {
		return tx.QueryRow(ctx, "SELECT count(*) FROM audit_logs WHERE organization_id=$1 AND action IN ('aedt.ai.answered','aedt.ai.refused')", org).Scan(&n)
	}); err != nil {
		t.Fatalf("count audits: %v", err)
	}
	if n < 4 {
		t.Fatalf("expected at least four brain audit events (rag, tag, cag, refuse), got %d", n)
	}
}

// TestBrainTAGReconciliation proves there is ONE source of truth: the brain's TAG answer over
// candidate_evaluations agrees with the engine's own scoring and with the dashboard read (WhoCanSupport),
// so the brain can never contradict the engine or the Command Center.
func TestBrainTAGReconciliation(t *testing.T) {
	ctx := context.Background()
	org := seedOrg(t)
	req, err := dbgen.New(migrator).CreateRequirement(ctx, dbgen.CreateRequirementParams{
		OrganizationID: org, Title: "Recon Role", RequiredLevel: "secret", RequiredPoly: "none", RequiredYears: 5,
	})
	if err != nil {
		t.Fatalf("seed requirement: %v", err)
	}
	strong := seedCandidate(t, org, "secret", "active", true)
	if _, err := migrator.Exec(ctx, "UPDATE candidates SET years_experience=10 WHERE id=$1", strong.ID); err != nil {
		t.Fatalf("enrich: %v", err)
	}
	weak := seedCandidate(t, org, "none", "active", true)

	svc := matching.NewService(appDB, matching.NewStoredSkillMatcher(appDB), audit.New(appDB))

	// The engine scores and persists the evaluations (the floor).
	engineBands := map[string]int{}
	for _, c := range []dbgen.Candidate{strong, weak} {
		res, err := svc.ScreenCandidate(ctxOrg(org), c.ID, req.ID)
		if err != nil {
			t.Fatalf("screen %s: %v", c.ID, err)
		}
		engineBands[res.Band.Code()]++
	}

	// The brain's TAG path reads the SAME frozen candidate_evaluations.
	intent, _ := ai.IntentByName("proposal_ready_for_requirement")
	tagRes, err := ai.ExecuteIntent(ctxOrg(org), appDB, intent, []any{req.ID})
	if err != nil {
		t.Fatalf("TAG intent: %v", err)
	}
	tagBands := map[string]int{}
	for _, row := range tagRes.Rows {
		band, _ := row[0].(string)
		switch n := row[1].(type) {
		case int64:
			tagBands[band] = int(n)
		case int:
			tagBands[band] = n
		}
	}

	// The dashboard read (WhoCanSupport) reads the SAME table, never rescoring.
	support, err := svc.WhoCanSupport(ctxOrg(org), req.ID)
	if err != nil {
		t.Fatalf("WhoCanSupport: %v", err)
	}
	supportBands := map[string]int{}
	for _, s := range support {
		supportBands[s.Band]++
	}

	for band, want := range engineBands {
		if tagBands[band] != want {
			t.Fatalf("RECONCILIATION: TAG band %q = %d, engine = %d (the brain disagrees with the engine)", band, tagBands[band], want)
		}
		if supportBands[band] != want {
			t.Fatalf("RECONCILIATION: WhoCanSupport band %q = %d, engine = %d (the dashboard disagrees with the engine)", band, supportBands[band], want)
		}
	}
	t.Logf("reconciled: engine=%v tag=%v support=%v", engineBands, tagBands, supportBands)
}

// TestBrainMemoryCrossTenantIsolation proves one tenant's memory can never be recalled by another: memory
// is RLS scoped, so a recall under org B never returns org A's note. Erasure wipes it.
func TestBrainMemoryCrossTenantIsolation(t *testing.T) {
	client := testPlane(t)
	orgA, orgB := seedOrg(t), seedOrg(t)
	candA := seedCandidate(t, orgA, "secret", "active", true)
	mem := ai.NewMemory(appDB, client)

	cid := candA.ID
	if err := mem.Store(tenant.WithOrg(context.Background(), orgA.String()), &cid, "semantic", "candidate excelled at incident response under pressure"); err != nil {
		t.Fatalf("store memory in org A: %v", err)
	}

	// Org A recalls its own memory.
	got, err := mem.Recall(ctxOrg(orgA), "incident response", 5)
	if err != nil {
		t.Fatalf("recall in org A: %v", err)
	}
	if len(got) == 0 {
		t.Fatalf("org A must recall its own memory")
	}

	// Org B recalls NOTHING of org A's memory (RLS isolation).
	leak, err := mem.Recall(ctxOrg(orgB), "incident response", 5)
	if err != nil {
		t.Fatalf("recall in org B: %v", err)
	}
	for _, s := range leak {
		if s == "candidate excelled at incident response under pressure" {
			t.Fatalf("ISOLATION BREACH: org B recalled org A's memory")
		}
	}

	// Erasure wipes the candidate's memory.
	if err := mem.ForgetCandidate(ctxOrg(orgA), candA.ID); err != nil {
		t.Fatalf("forget: %v", err)
	}
	after, err := mem.Recall(ctxOrg(orgA), "incident response", 5)
	if err != nil {
		t.Fatalf("recall after forget: %v", err)
	}
	for _, s := range after {
		if s == "candidate excelled at incident response under pressure" {
			t.Fatalf("erasure failed: the memory survived ForgetCandidate")
		}
	}
}