Janady13/missionmatch / internal/integration/teaching_producer_integration_test.go

1//go:build integration 2 3package integration 4 5import ( 6 "context" 7 "encoding/json" 8 "testing" 9 10 "github.com/google/uuid" 11 "github.com/jackc/pgx/v5" 12 "github.com/shopspring/decimal" 13 14 "github.com/missionworkforce/missionmatch/internal/modules/ai" 15 "github.com/missionworkforce/missionmatch/internal/modules/matching/engine" 16 "github.com/missionworkforce/missionmatch/internal/platform/audit" 17 "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" 18 "github.com/missionworkforce/missionmatch/internal/platform/inference" 19) 20 21// TestTeachingProducerAggregatesInTenantRows proves the WS8 Phase 1 producer end to end against real 22// Postgres: the read is RLS scoped (it sees only the caller org's slate decisions, never a second org's), 23// the audit event commits on the same tenant tx, and the per tenant aggregates land in the global 24// teaching_signal_aggregates table with tenant_id set and dp_applied false. It seeds TWO orgs with distinct 25// reviewer craft; running the producer for org A must fold only org A's evidence. 26func TestTeachingProducerAggregatesInTenantRows(t *testing.T) { 27 // Org A: a clearance gap override (required secret, clearance_score 0, approved) plus an adjacent skill 28 // substitution stored in the evaluation (a non exact match at cosine 0.91, a high band), approved. 29 orgA := seedOrg(t) 30 _ = seedTeachingMember(t, orgA, teachingSeed{ 31 requiredLevel: "secret", 32 clearanceScore: "0", 33 experience: "100", 34 decision: "approved", 35 matchedSkills: []teachSeedSkill{{similarity: "0.91", exact: false, matched: true}}, 36 }) 37 38 // Org B: a completely different shape (no clearance required, an experience shortfall, rejected). If the 39 // producer's read were not RLS scoped, org A's run would fold this too and the counts would be wrong. 40 orgB := seedOrg(t) 41 _ = seedTeachingMember(t, orgB, teachingSeed{ 42 requiredLevel: "none", 43 clearanceScore: "100", 44 experience: "20", 45 decision: "rejected", 46 matchedSkills: nil, 47 }) 48 49 // A teaching client with TeachingEnabled false: Teach validates the boundary but transmits nothing. 50 teacher, err := inference.New(inference.Config{BaseURL: "http://127.0.0.1:0", TeachingEnabled: false}) 51 if err != nil { 52 t.Fatalf("inference client: %v", err) 53 } 54 // The aggregate writer is backed by the MIGRATOR pool (the global table the app role cannot write). 55 writer := ai.NewMigratorAggregateWriter(migrator) 56 prod := ai.NewTeachingProducer(appDB, writer, audit.New(appDB), teacher) 57 58 res, err := prod.AggregateForOrg(ctxOrg(orgA)) 59 if err != nil { 60 t.Fatalf("AggregateForOrg(orgA): %v", err) 61 } 62 // RLS scoped read: org A has exactly one slate member, so exactly one decision was read, never org B's. 63 if res.DecisionsRead != 1 { 64 t.Fatalf("DecisionsRead = %d, want 1 (the read must be RLS scoped to org A only)", res.DecisionsRead) 65 } 66 67 // Org A's evidence folds two patterns: clearance_gap_override_outcome (unknown band, approved) and 68 // adjacent_skill_substitution_outcome (high band, approved). Both must be persisted for org A. 69 rows := teachingAggregatesForTenant(t, orgA) 70 if len(rows) != len(res.Signals) || len(rows) == 0 { 71 t.Fatalf("persisted %d aggregate rows for org A, produced %d signals", len(rows), len(res.Signals)) 72 } 73 sawClearance, sawSubstitution := false, false 74 for _, r := range rows { 75 if r.TenantID == nil || *r.TenantID != orgA { 76 t.Fatalf("aggregate tenant_id = %v, want org A %v (a live in tenant row)", r.TenantID, orgA) 77 } 78 if r.DpApplied { 79 t.Fatal("aggregate dp_applied = true, want false for a live in tenant row") 80 } 81 if r.ModelVersion != engine.ModelVersion { 82 t.Fatalf("aggregate model_version = %q, want engine.ModelVersion %q", r.ModelVersion, engine.ModelVersion) 83 } 84 switch r.Pattern { 85 case "clearance_gap_override_outcome": 86 sawClearance = true 87 if r.Band != "unknown" || r.Outcome != "approved" || r.Observations != 1 || r.Successes != 1 { 88 t.Fatalf("clearance aggregate = %+v, want unknown/approved/1/1", r) 89 } 90 case "adjacent_skill_substitution_outcome": 91 sawSubstitution = true 92 if r.Band != "high" || r.Outcome != "approved" || r.Observations != 1 || r.Successes != 1 { 93 t.Fatalf("substitution aggregate = %+v, want high/approved/1/1", r) 94 } 95 } 96 } 97 if !sawClearance || !sawSubstitution { 98 t.Fatalf("missing expected aggregates: clearance=%v substitution=%v", sawClearance, sawSubstitution) 99 } 100 101 // Org B has no aggregates, proving the producer never wrote for an org it was not run on. 102 if got := teachingAggregatesForTenant(t, orgB); len(got) != 0 { 103 t.Fatalf("org B has %d aggregate rows, want 0 (the producer ran only for org A)", len(got)) 104 } 105 106 // The teaching.signal.aggregated audit event committed on org A's tenant tx, carrying counts only. 107 var auditCount int 108 var newVals []byte 109 if err := appDB.RunInTenantTx(ctxOrg(orgA), func(ctx context.Context, tx pgx.Tx) error { 110 return tx.QueryRow(ctx, 111 `SELECT count(*), max(new_values::text) FROM audit_logs 112 WHERE organization_id=$1 AND action='teaching.signal.aggregated'`, 113 orgA).Scan(&auditCount, &newVals) 114 }); err != nil { 115 t.Fatalf("count teaching audit: %v", err) 116 } 117 if auditCount != 1 { 118 t.Fatalf("want 1 teaching.signal.aggregated audit for org A, got %d", auditCount) 119 } 120 var payload map[string]int 121 if err := json.Unmarshal(newVals, &payload); err != nil { 122 t.Fatalf("audit payload not the counts only shape: %v (%s)", err, newVals) 123 } 124 if payload["decisions_aggregated"] != 1 { 125 t.Fatalf("audit decisions_aggregated = %d, want 1", payload["decisions_aggregated"]) 126 } 127 128 // A rerun is idempotent (ON CONFLICT upserts in place), so the row count does not grow. 129 if _, err := prod.AggregateForOrg(ctxOrg(orgA)); err != nil { 130 t.Fatalf("AggregateForOrg(orgA) rerun: %v", err) 131 } 132 if got := teachingAggregatesForTenant(t, orgA); len(got) != len(rows) { 133 t.Fatalf("rerun changed the aggregate row count from %d to %d (upsert must be idempotent)", len(rows), len(got)) 134 } 135} 136 137// ---- seed helpers (write across tenants via the BYPASSRLS migrator) ---- 138 139type teachSeedSkill struct { 140 similarity string 141 exact bool 142 matched bool 143} 144 145type teachingSeed struct { 146 requiredLevel string 147 clearanceScore string 148 experience string 149 decision string 150 matchedSkills []teachSeedSkill 151} 152 153// seedTeachingMember seeds one requirement, one slate, one decided slate member, and (when matchedSkills is 154// set) one candidate_evaluations row carrying the match evidence the producer reads. It returns the 155// candidate id. Everything is written through the migrator so the seed can span orgs. 156func seedTeachingMember(t *testing.T, org uuid.UUID, s teachingSeed) uuid.UUID { 157 t.Helper() 158 ctx := context.Background() 159 q := dbgen.New(migrator) 160 161 req, err := q.CreateRequirement(ctx, dbgen.CreateRequirementParams{ 162 OrganizationID: org, Title: "Teaching Role", RequiredLevel: s.requiredLevel, RequiredPoly: "none", 163 }) 164 if err != nil { 165 t.Fatalf("seed requirement: %v", err) 166 } 167 cand := seedCandidate(t, org, "none", "active", true) 168 169 slate, err := q.CreateSlate(ctx, dbgen.CreateSlateParams{ 170 OrganizationID: org, RequirementID: req.ID, Name: "Teaching Slate", Status: "in_review", 171 }) 172 if err != nil { 173 t.Fatalf("seed slate: %v", err) 174 } 175 176 mv := engine.ModelVersion 177 if _, err := q.AddSlateMember(ctx, dbgen.AddSlateMemberParams{ 178 OrganizationID: org, SlateID: slate.ID, CandidateID: cand.ID, 179 TotalScore: dec0("80"), Band: "yellow", 180 ClearanceScore: dec0(s.clearanceScore), SkillsScore: dec0("100"), CertsScore: dec0("100"), 181 CustomerScore: dec0("100"), ExperienceScore: dec0(s.experience), LocationScore: dec0("100"), 182 Explanation: "seeded", Trace: []byte("[]"), GapSummary: []byte("[]"), ModelVersion: &mv, 183 }); err != nil { 184 t.Fatalf("seed slate member: %v", err) 185 } 186 if _, err := q.SetSlateMemberDecision(ctx, dbgen.SetSlateMemberDecisionParams{ 187 SlateID: slate.ID, CandidateID: cand.ID, Decision: s.decision, 188 }); err != nil { 189 t.Fatalf("seed decision: %v", err) 190 } 191 192 if len(s.matchedSkills) > 0 { 193 matched := make([]map[string]any, 0, len(s.matchedSkills)) 194 for _, sk := range s.matchedSkills { 195 matched = append(matched, map[string]any{ 196 "RequiredSkillID": "", "RequiredSkillName": "skill", "CandidateSkillID": "", 197 "CandidateName": "", "Similarity": sk.similarity, "Exact": sk.exact, "Matched": sk.matched, 198 }) 199 } 200 mj, err := json.Marshal(matched) 201 if err != nil { 202 t.Fatalf("marshal matched skills: %v", err) 203 } 204 if err := q.UpsertCandidateEvaluation(ctx, dbgen.UpsertCandidateEvaluationParams{ 205 OrganizationID: org, CandidateID: cand.ID, RequirementID: req.ID, 206 TotalScore: dec0("80"), Band: "yellow", 207 ClearanceScore: dec0(s.clearanceScore), SkillsScore: dec0("100"), CertsScore: dec0("100"), 208 CustomerScore: dec0("100"), ExperienceScore: dec0(s.experience), LocationScore: dec0("100"), 209 Explanation: "seeded", MatchedSkills: mj, GapSummary: []byte("[]"), ModelVersion: engine.ModelVersion, 210 }); err != nil { 211 t.Fatalf("seed evaluation: %v", err) 212 } 213 } 214 return cand.ID 215} 216 217// teachingAggregatesForTenant reads the global teaching_signal_aggregates rows for one tenant through the 218// migrator (the table is global, so the migrator sees all rows; we filter by tenant_id in the query). 219func teachingAggregatesForTenant(t *testing.T, tenantID uuid.UUID) []dbgen.TeachingSignalAggregate { 220 t.Helper() 221 rows, err := migrator.Query(context.Background(), 222 `SELECT id, tenant_id, dp_applied, pattern, band, outcome, model_version, observations, successes, 223 window_start, window_end, created_at 224 FROM teaching_signal_aggregates WHERE tenant_id=$1 225 ORDER BY pattern, band, outcome`, tenantID) 226 if err != nil { 227 t.Fatalf("query aggregates: %v", err) 228 } 229 defer rows.Close() 230 var out []dbgen.TeachingSignalAggregate 231 for rows.Next() { 232 var r dbgen.TeachingSignalAggregate 233 if err := rows.Scan(&r.ID, &r.TenantID, &r.DpApplied, &r.Pattern, &r.Band, &r.Outcome, 234 &r.ModelVersion, &r.Observations, &r.Successes, &r.WindowStart, &r.WindowEnd, &r.CreatedAt); err != nil { 235 t.Fatalf("scan aggregate: %v", err) 236 } 237 out = append(out, r) 238 } 239 if err := rows.Err(); err != nil { 240 t.Fatalf("rows err: %v", err) 241 } 242 return out 243} 244 245// dec0 builds a decimal for a seed value, failing the test on a malformed literal. 246func dec0(s string) decimal.Decimal { return decimal.RequireFromString(s) } 247