Janady13/missionmatch / internal/integration/curation_integration_test.go

1//go:build integration 2 3package integration 4 5import ( 6 "context" 7 "encoding/json" 8 "errors" 9 "testing" 10 "time" 11 12 "github.com/google/uuid" 13 "github.com/jackc/pgx/v5" 14 15 "github.com/missionworkforce/missionmatch/internal/modules/ai" 16 "github.com/missionworkforce/missionmatch/internal/modules/matching" 17 "github.com/missionworkforce/missionmatch/internal/modules/matching/engine" 18 "github.com/missionworkforce/missionmatch/internal/platform/audit" 19 "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" 20) 21 22// TestCurationTwoStepGateAndInertEdge is the WS8 Phase 4 end to end proof against real Postgres. It proves the 23// load bearing property of the whole workstream: a curated edge is staged INERT on approve and goes live ONLY 24// after a SECOND distinct verify step, and the engine math is never touched. The flow: 25// 26// (a) seed an org with recurring approved substitution evidence (a candidate skill B accepted in place of a 27// required skill A at a high band, approved twice) plus the governance aggregate that confirms the 28// pattern, then run the generator and assert a pending substitutes_for_edge proposal appears with the 29// right from (B), to (A), strength, and evidence. 30// (b) approve it and assert a skill_relationships row exists with verified_at NULL, then PROVE IT IS INERT: 31// a fresh DeterministicAliasResolver (the exact component that reads verified_at IS NOT NULL edges) does 32// NOT resolve B to A, so a score that would change if the edge were live is unchanged. 33// (c) verify it (the second human step) and assert verified_at is set, then prove the resolver now reads it: 34// a fresh resolver resolves B to A at the curated strength, so the edge is live. 35func TestCurationTwoStepGateAndInertEdge(t *testing.T) { 36 ctx := context.Background() 37 org := seedOrg(t) 38 39 // Two real global skills: A is required, B is the candidate substitute. Names are unique per run so the 40 // resolver's name path is unambiguous and the global unique on skills.name never collides across runs. 41 skillA := seedSkillByName(t, "ws8 required skill "+org.String()[:8]) 42 skillB := seedSkillByName(t, "ws8 candidate skill "+org.String()[:8]) 43 44 // Recurring approved substitution evidence: B matched A at a high band (0.93), non exact, approved, twice. 45 for i := 0; i < 2; i++ { 46 seedCurationMember(t, org, skillA, skillB, "0.93", "approved") 47 } 48 49 // Governance confidence: the contents free aggregate confirms the substitution pattern is reliably 50 // approved at the high band, so the double gate is satisfied. 51 seedSubstitutionAggregate(t, org, "high", "approved", 5, 5) 52 53 // Run the generator (app pool RLS scoped scan, migrator pool governance read and proposal write). 54 gen := ai.NewProposalGenerator( 55 appDB, 56 ai.NewMigratorGovernanceReader(migrator), 57 ai.NewMigratorProposalWriter(migrator), 58 audit.New(appDB), 59 ) 60 res, err := gen.GenerateProposalsForOrg(ctxOrg(org)) 61 if err != nil { 62 t.Fatalf("GenerateProposalsForOrg: %v", err) 63 } 64 if res.SubstitutionEdges != 1 { 65 t.Fatalf("generator produced %d substitution edges, want exactly 1", res.SubstitutionEdges) 66 } 67 68 // (a) A pending substitutes_for_edge proposal exists with from B, to A, the high band strength, and the 69 // recurring evidence counts. 70 prop := getOnlyPendingProposal(t, skillB, skillA) 71 if prop.Kind != "substitutes_for_edge" { 72 t.Fatalf("proposal kind = %q, want substitutes_for_edge", prop.Kind) 73 } 74 if prop.ProposedStrength == nil || prop.ProposedStrength.String() != "0.9" { 75 t.Fatalf("proposed strength = %v, want 0.90 for the high band", prop.ProposedStrength) 76 } 77 if prop.EvidenceSuccesses != 2 || prop.EvidenceObservations != 2 { 78 t.Fatalf("evidence = %d of %d, want 2 of 2", prop.EvidenceSuccesses, prop.EvidenceObservations) 79 } 80 if prop.ModelVersion != engine.ModelVersion { 81 t.Fatalf("proposal model_version = %q, want engine.ModelVersion %q", prop.ModelVersion, engine.ModelVersion) 82 } 83 84 // A rerun is idempotent: the pending proposal already exists, so no duplicate is created. 85 if _, err := gen.GenerateProposalsForOrg(ctxOrg(org)); err != nil { 86 t.Fatalf("generator rerun: %v", err) 87 } 88 if n := countProposals(t, skillB, skillA); n != 1 { 89 t.Fatalf("rerun created a duplicate proposal, count = %d, want 1", n) 90 } 91 92 apply := ai.NewCurationApplyService(migrator, audit.New(appDB), true) // enforce separation of duties 93 approver := uuid.New() 94 verifier := uuid.New() 95 96 // (b) APPROVE (the first human step). The proposal moves to approved and a staged INERT edge is inserted. 97 if _, err := apply.Approve(ctx, prop.ID, approver); err != nil { 98 t.Fatalf("Approve: %v", err) 99 } 100 edgeID, verifiedAt := substitutesEdge(t, skillB, skillA) 101 if edgeID == uuid.Nil { 102 t.Fatal("approve did not stage a skill_relationships edge") 103 } 104 if verifiedAt != nil { 105 t.Fatalf("staged edge verified_at = %v, want NULL (the edge must be inert until verify)", verifiedAt) 106 } 107 108 // PROVE INERT: a fresh resolver loads only verified_at IS NOT NULL edges, so it does NOT resolve B to A. 109 // This is the score that would change if the staged edge were live; it does not, so the edge is inert. 110 if matched := resolveBtoA(ctx, skillB, skillA); matched { 111 t.Fatal("INERT EDGE BROKEN: the staged verified_at NULL edge was read by the resolver and changed a match") 112 } 113 114 // The proposal approver and approved_at are recorded; the verifier is NOT yet set on the edge. 115 assertProposalApproved(t, prop.ID, approver) 116 117 // SEPARATION OF DUTIES: the approver cannot also verify when enforcement is on, so the two person control 118 // is real, not procedural. The edge stays staged and inert. 119 if _, err := apply.Verify(ctx, prop.ID, approver); !errors.Is(err, ai.ErrVerifierMustDifferFromApprover) { 120 t.Fatalf("self verify by the approver should be refused, got %v", err) 121 } 122 if _, stillNil := substitutesEdge(t, skillB, skillA); stillNil != nil { 123 t.Fatal("a refused self verify must leave the edge staged (verified_at NULL)") 124 } 125 126 // (c) VERIFY (the second, distinct human step). A DIFFERENT human verifies and the edge goes live. 127 if _, err := apply.Verify(ctx, prop.ID, verifier); err != nil { 128 t.Fatalf("Verify: %v", err) 129 } 130 _, verifiedAt2 := substitutesEdge(t, skillB, skillA) 131 if verifiedAt2 == nil { 132 t.Fatal("verify did not set verified_at on the staged edge") 133 } 134 assertEdgeVerifier(t, edgeID, verifier) 135 136 // PROVE LIVE: a fresh resolver now reads the verified edge and resolves B to A at the curated strength. 137 if matched := resolveBtoA(ctx, skillB, skillA); !matched { 138 t.Fatal("verified edge is not live: the resolver did not resolve B to A after verify") 139 } 140} 141 142// ---- seed and assertion helpers (migrator, BYPASSRLS, so they can write the global tables across tenants) ---- 143 144// seedSkillByName inserts a global skill with a unique name and returns its id. 145func seedSkillByName(t *testing.T, name string) uuid.UUID { 146 t.Helper() 147 var id uuid.UUID 148 if err := migrator.QueryRow(context.Background(), 149 "INSERT INTO skills (name) VALUES ($1) RETURNING id", name).Scan(&id); err != nil { 150 t.Fatalf("seed skill %q: %v", name, err) 151 } 152 return id 153} 154 155// seedCurationMember seeds one requirement, candidate, slate, decided slate member, and a candidate_evaluations 156// row whose matched_skills records the candidate skill B accepted for the required skill A at the given 157// similarity (non exact). It writes the real skill ids into the SkillMatch so the generator derives the pair by 158// id, exactly the production shape. 159func seedCurationMember(t *testing.T, org, skillA, skillB uuid.UUID, similarity, decision string) { 160 t.Helper() 161 ctx := context.Background() 162 q := dbgen.New(migrator) 163 164 req, err := q.CreateRequirement(ctx, dbgen.CreateRequirementParams{ 165 OrganizationID: org, Title: "WS8 Curation Role", RequiredLevel: "none", RequiredPoly: "none", 166 }) 167 if err != nil { 168 t.Fatalf("seed requirement: %v", err) 169 } 170 cand := seedCandidate(t, org, "none", "active", true) 171 172 slate, err := q.CreateSlate(ctx, dbgen.CreateSlateParams{ 173 OrganizationID: org, RequirementID: req.ID, Name: "WS8 Curation Slate", Status: "in_review", 174 }) 175 if err != nil { 176 t.Fatalf("seed slate: %v", err) 177 } 178 mv := engine.ModelVersion 179 if _, err := q.AddSlateMember(ctx, dbgen.AddSlateMemberParams{ 180 OrganizationID: org, SlateID: slate.ID, CandidateID: cand.ID, 181 TotalScore: dec0("80"), Band: "yellow", 182 ClearanceScore: dec0("100"), SkillsScore: dec0("80"), CertsScore: dec0("100"), 183 CustomerScore: dec0("100"), ExperienceScore: dec0("100"), LocationScore: dec0("100"), 184 Explanation: "seeded", Trace: []byte("[]"), GapSummary: []byte("[]"), ModelVersion: &mv, 185 }); err != nil { 186 t.Fatalf("seed slate member: %v", err) 187 } 188 if _, err := q.SetSlateMemberDecision(ctx, dbgen.SetSlateMemberDecisionParams{ 189 SlateID: slate.ID, CandidateID: cand.ID, Decision: decision, 190 }); err != nil { 191 t.Fatalf("seed decision: %v", err) 192 } 193 194 matched := []map[string]any{{ 195 "RequiredSkillID": skillA.String(), "RequiredSkillName": "req", 196 "CandidateSkillID": skillB.String(), "CandidateName": "cand", 197 "Similarity": similarity, "Exact": false, "Matched": true, 198 }} 199 mj, err := json.Marshal(matched) 200 if err != nil { 201 t.Fatalf("marshal matched skills: %v", err) 202 } 203 if err := q.UpsertCandidateEvaluation(ctx, dbgen.UpsertCandidateEvaluationParams{ 204 OrganizationID: org, CandidateID: cand.ID, RequirementID: req.ID, 205 TotalScore: dec0("80"), Band: "yellow", 206 ClearanceScore: dec0("100"), SkillsScore: dec0("80"), CertsScore: dec0("100"), 207 CustomerScore: dec0("100"), ExperienceScore: dec0("100"), LocationScore: dec0("100"), 208 Explanation: "seeded", MatchedSkills: mj, GapSummary: []byte("[]"), ModelVersion: engine.ModelVersion, 209 }); err != nil { 210 t.Fatalf("seed evaluation: %v", err) 211 } 212} 213 214// seedSubstitutionAggregate writes one live in tenant teaching_signal_aggregates row for the substitution 215// pattern, so the generator's governance confidence read finds the supporting evidence. 216func seedSubstitutionAggregate(t *testing.T, org uuid.UUID, band, outcome string, obs, succ int) { 217 t.Helper() 218 if _, err := migrator.Exec(context.Background(), 219 `INSERT INTO teaching_signal_aggregates 220 (tenant_id, dp_applied, pattern, band, outcome, model_version, observations, successes, window_start, window_end) 221 VALUES ($1, false, 'adjacent_skill_substitution_outcome', $2, $3, $4, $5, $6, '1970-01-01', now()::date)`, 222 org, band, outcome, engine.ModelVersion, obs, succ); err != nil { 223 t.Fatalf("seed substitution aggregate: %v", err) 224 } 225} 226 227// getOnlyPendingProposal reads the single pending proposal for the pair and fails if there is not exactly one. 228func getOnlyPendingProposal(t *testing.T, from, to uuid.UUID) dbgen.CurationProposal { 229 t.Helper() 230 rows, err := migrator.Query(context.Background(), 231 `SELECT id, kind, from_skill_id, to_skill_id, relation, proposed_strength, 232 evidence_observations, evidence_successes, model_version, status, approver, approved_at, created_at 233 FROM curation_proposals 234 WHERE from_skill_id = $1 AND to_skill_id = $2 AND status = 'pending'`, from, to) 235 if err != nil { 236 t.Fatalf("query pending proposal: %v", err) 237 } 238 defer rows.Close() 239 var out []dbgen.CurationProposal 240 for rows.Next() { 241 var p dbgen.CurationProposal 242 if err := rows.Scan(&p.ID, &p.Kind, &p.FromSkillID, &p.ToSkillID, &p.Relation, &p.ProposedStrength, 243 &p.EvidenceObservations, &p.EvidenceSuccesses, &p.ModelVersion, &p.Status, &p.Approver, &p.ApprovedAt, &p.CreatedAt); err != nil { 244 t.Fatalf("scan proposal: %v", err) 245 } 246 out = append(out, p) 247 } 248 if len(out) != 1 { 249 t.Fatalf("want exactly 1 pending proposal for the pair, got %d", len(out)) 250 } 251 return out[0] 252} 253 254// countProposals counts all proposals (any status) for the pair, to assert idempotency. 255func countProposals(t *testing.T, from, to uuid.UUID) int { 256 t.Helper() 257 var n int 258 if err := migrator.QueryRow(context.Background(), 259 "SELECT count(*) FROM curation_proposals WHERE from_skill_id = $1 AND to_skill_id = $2", from, to).Scan(&n); err != nil { 260 t.Fatalf("count proposals: %v", err) 261 } 262 return n 263} 264 265// substitutesEdge returns the staged or live skill_relationships edge id and its verified_at for the pair. 266func substitutesEdge(t *testing.T, from, to uuid.UUID) (uuid.UUID, *time.Time) { 267 t.Helper() 268 var id uuid.UUID 269 var verifiedAt *time.Time 270 err := migrator.QueryRow(context.Background(), 271 `SELECT id, verified_at FROM skill_relationships 272 WHERE from_skill_id = $1 AND to_skill_id = $2 AND relation = 'substitutes_for'`, from, to).Scan(&id, &verifiedAt) 273 if errors.Is(err, pgx.ErrNoRows) { 274 return uuid.Nil, nil 275 } 276 if err != nil { 277 t.Fatalf("query substitutes edge: %v", err) 278 } 279 return id, verifiedAt 280} 281 282// assertProposalApproved checks the proposal status, approver, and approved_at after the first human step. 283func assertProposalApproved(t *testing.T, id, approver uuid.UUID) { 284 t.Helper() 285 var status string 286 var gotApprover *uuid.UUID 287 var approvedAt *time.Time 288 if err := migrator.QueryRow(context.Background(), 289 "SELECT status, approver, approved_at FROM curation_proposals WHERE id = $1", id).Scan(&status, &gotApprover, &approvedAt); err != nil { 290 t.Fatalf("read approved proposal: %v", err) 291 } 292 if status != "approved" { 293 t.Fatalf("proposal status = %q, want approved", status) 294 } 295 if gotApprover == nil || *gotApprover != approver { 296 t.Fatalf("proposal approver = %v, want %v", gotApprover, approver) 297 } 298 if approvedAt == nil { 299 t.Fatal("proposal approved_at is NULL after approve") 300 } 301} 302 303// assertEdgeVerifier checks the verifier recorded on the live edge after the second human step. 304func assertEdgeVerifier(t *testing.T, edgeID, verifier uuid.UUID) { 305 t.Helper() 306 var gotVerifier *uuid.UUID 307 if err := migrator.QueryRow(context.Background(), 308 "SELECT verifier FROM skill_relationships WHERE id = $1", edgeID).Scan(&gotVerifier); err != nil { 309 t.Fatalf("read edge verifier: %v", err) 310 } 311 if gotVerifier == nil || *gotVerifier != verifier { 312 t.Fatalf("edge verifier = %v, want %v (distinct from the approver)", gotVerifier, verifier) 313 } 314} 315 316// resolveBtoA builds a FRESH DeterministicAliasResolver (so it re reads the edges from the DB, never a stale 317// cache) and asks whether candidate skill B resolves to required skill A. The resolver is the exact component 318// that loads verified_at IS NOT NULL substitutes_for edges, so this is the live test of the two step gate: it 319// returns false while the edge is staged inert and true once it is verified. 320func resolveBtoA(ctx context.Context, skillB, skillA uuid.UUID) bool { 321 r := matching.NewDeterministicAliasResolver(appDB) 322 _, ok := r.ResolveAlias(ctx, 323 engine.RequiredSkill{SkillID: skillA.String(), Name: "req", Mandatory: true}, 324 []engine.CandidateSkill{{SkillID: skillB.String(), Name: "cand"}}) 325 return ok 326} 327