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

1package ai 2 3// curation.go is the WS8 Phase 4 deterministic curation proposal generator (LIVE). It turns one tenant's 4// recurring, human reviewed reviewer craft into curation_proposals that a human approves, then a SECOND human 5// verifies, before anything affects a future score. It never changes the engine math, a weight, a band cut, 6// the 0.62 semantic threshold, or any already served score. Its ONLY downstream effect is, after TWO human 7// steps, an additive verified skill_relationships edge the resolver reads, which changes FUTURE scores only. 8// 9// The generator answers a question the contents free teaching_signal_aggregates rows cannot, because those 10// rows are by design contents free (a pattern, a band, an outcome, and counts, NEVER a skill id). They tell 11// us the CROSS CUTTING GOVERNANCE CONFIDENCE: is the adjacent_skill_substitution_outcome pattern reliably 12// approved at a high band. They do not name a from_skill or a to_skill. So the generator gets the SPECIFIC 13// recurring approved pairs from a WITHIN TENANT, RLS scoped scan of the tenant's own candidate_evaluations 14// (its own human reviewed evaluation data, which is allowed in tenant and is reviewed by a human before it 15// ever becomes a global edge), and proposes a pair ONLY when both hold: the tenant's own evaluations show the 16// pair was approved repeatedly above a curated threshold, AND the contents free aggregate confirms the 17// pattern is reliably approved at that band. Under the one tenant posture this is correct and safe: the scan 18// is RLS scoped, the proposal is a SUGGESTION, and two humans must approve then verify before a score moves. 19// 20// The candidate_evaluations.matched_skills jsonb carries the full engine SkillMatch, which has BOTH ids and 21// names (RequiredSkillID, RequiredSkillName, CandidateSkillID, CandidateName, Similarity, Exact, Matched). So 22// a recurring approved non exact pair is identified by id directly when the matcher populated ids, and by 23// resolving the skill name to a global skills.id when only names are present. The proposal's from_skill_id is 24// the CANDIDATE skill (the substitute the reviewer accepted) and to_skill_id is the REQUIRED skill, so a 25// later verified substitutes_for edge reads exactly the way the resolver expects (a candidate skill 26// substitutes for the required skill the candidate did not hold exactly). 27// 28// Three correctness boundaries hold by construction, the same three the producer holds: 29// 30// 1. The evaluation SCAN is RLS scoped. It runs inside RunInTenantTx on the NOBYPASSRLS app pool, so one 31// tenant's generator can only ever read its own evaluations. 32// 2. The proposal WRITE, the idempotency check, and the governance confidence read go through the MIGRATOR 33// pool seam, never the app role. curation_proposals and teaching_signal_aggregates are global, migrator 34// owned tables the app role can only SELECT. 35// 3. The proposal created event runs through audit.Writer.System with a NULL org, because a proposal is a 36// global object, never a per tenant one. It carries skill ids and counts only, never PII. 37 38import ( 39 "context" 40 "encoding/json" 41 "errors" 42 "sort" 43 44 "github.com/google/uuid" 45 "github.com/jackc/pgx/v5" 46 "github.com/shopspring/decimal" 47 48 "github.com/missionworkforce/missionmatch/internal/modules/matching/engine" 49 "github.com/missionworkforce/missionmatch/internal/platform/audit" 50 "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" 51 "github.com/missionworkforce/missionmatch/internal/platform/scrub" 52 "github.com/missionworkforce/missionmatch/internal/platform/tenant" 53) 54 55// Curation thresholds keep the few classification rules in one place so the proposal selection is a pure 56// function of the evidence and the rules are visible for review. 57const ( 58 // minSubstitutionApprovals is the curated floor on how many times a specific candidate to required pair 59 // must have been APPROVED by a reviewer in this tenant's own evaluations before the generator will 60 // propose a substitutes_for edge. A one off approval is not craft, it is an anecdote, so it never 61 // proposes. Two or more recurring approvals is the floor. 62 minSubstitutionApprovals = 2 63 64 // substitutionRelation is the relation a substitutes_for_edge proposal asks for, matching the resolver 65 // vocabulary the engine consults (substitutes_for and broader_than). 66 substitutionRelation = "substitutes_for" 67 68 // minOverrideApprovals is the curated floor on how many times a clearance gap override was approved in 69 // this tenant before the generator raises a reciprocity_review for human governance. A clearance override 70 // is never turned into a scoring edge; it is surfaced for a human to review the policy, evidence only. 71 minOverrideApprovals = 2 72 73 // kindSubstitutesForEdge, kindReciprocityReview, and kindRecalibrationEvidence are the three proposal 74 // kinds, matching the curation_proposals kind CHECK. 75 kindSubstitutesForEdge = "substitutes_for_edge" 76 kindReciprocityReview = "reciprocity_review" 77 kindRecalibrationEvidence = "recalibration_evidence" 78) 79 80// strengthByBand maps a similarity band to the curated proposed strength of the substitutes_for edge. The 81// strength is a partial credit weight in the 0.62 to 1.0 range the resolver honors; it is NOT the engine 82// math, only the strength the resolver multiplies into a partial skill match once a human verifies the edge. 83// A high band pair earns a stronger curated strength than a medium band one; a low or unknown band never 84// proposes a substitution at all (see selectSubstitutionProposals), so they are absent here. 85var strengthByBand = map[scrub.SimilarityBand]string{ 86 scrub.BandHigh: "0.90", 87 scrub.BandMedium: "0.80", 88} 89 90// pairKey identifies one specific recurring substitution: a candidate skill that a reviewer accepted in place 91// of a required skill the candidate did not hold exactly. Both sides are resolved skill ids. 92type pairKey struct { 93 fromSkillID string // the CANDIDATE skill (the accepted substitute) 94 toSkillID string // the REQUIRED skill it stood in for 95} 96 97// pairEvidence accumulates the approval and total counts for one pair and the strongest band the reviewers 98// accepted it at. The band drives the proposed strength; the counts drive the threshold and the evidence 99// recorded on the proposal. 100type pairEvidence struct { 101 approvals int 102 total int 103 band scrub.SimilarityBand 104} 105 106// evalRow is the contents free craft evidence the generator scans: the reviewer decision and the persisted 107// matched skills (which carry the skill ids and names plus the similarity and the exact flag). It is the same 108// ListSlateDecisionEvidenceForOrgRow shape the producer reads, so both run off one RLS scoped query. 109type evalRow = dbgen.ListSlateDecisionEvidenceForOrgRow 110 111// EvaluationScanner is the RLS scoped read seam for the tenant's own evaluation evidence and the skill name 112// resolution. The live implementation runs inside RunInTenantTx on the app pool; a test injects a fake so the 113// derivation and selection logic unit tests with no database. 114type EvaluationScanner interface { 115 // ScanEvaluations returns the org's slate decisions joined to the persisted match evidence, RLS scoped. 116 ScanEvaluations(ctx context.Context) ([]evalRow, error) 117 // ResolveSkillID resolves a skill name to its global skills.id. It returns ok false when no skill matches, 118 // so a matched_skills entry that carries a name the ontology does not know yields no pair rather than a 119 // fabricated id. The resolution runs on the same app tx (skills is global, app SELECT). 120 ResolveSkillID(ctx context.Context, name string) (uuid.UUID, bool, error) 121} 122 123// GovernanceReader reads the cross cutting governance confidence (the tenant's teaching_signal_aggregates 124// rows). The live implementation is backed by the MIGRATOR pool (the global table the app role can only 125// SELECT); a test injects a fake so selection unit tests with no database. 126type GovernanceReader interface { 127 GovernanceConfidence(ctx context.Context, tenantID uuid.UUID, modelVersion string) ([]dbgen.TeachingSignalAggregate, error) 128} 129 130// systemAuditor is the global (NULL org) audit seam the generator records the proposal created event through. 131// *audit.Writer satisfies it. It is an interface so the write loop unit tests with a no op auditor and no 132// database, while the live path records through audit.Writer.System exactly as the invariant requires. 133type systemAuditor interface { 134 System(ctx context.Context, e audit.Event) error 135} 136 137// *audit.Writer is the live systemAuditor. 138var _ systemAuditor = (*audit.Writer)(nil) 139 140// ProposalWriter is the migrator pool write seam for curation_proposals. The live implementation is backed by 141// the MIGRATOR pool; a test injects a fake that records the inserts and answers the idempotency check. 142type ProposalWriter interface { 143 // ActiveProposalExists reports whether a pending or approved proposal already exists for the triple, so the 144 // generator never creates a duplicate suggestion for a pair a human is already deciding or has approved. 145 ActiveProposalExists(ctx context.Context, fromSkillID, toSkillID uuid.UUID, relation string) (bool, error) 146 // InsertProposal creates one pending proposal and returns its id. 147 InsertProposal(ctx context.Context, arg dbgen.InsertCurationProposalParams) (uuid.UUID, error) 148} 149 150// scannerFromTx binds an EvaluationScanner to a tenant transaction, so the RLS scoped read runs on the same 151// connection whose app.current_org_id GUC RunInTenantTx set. 152type scannerFromTx struct{ tx pgx.Tx } 153 154var _ EvaluationScanner = scannerFromTx{} 155 156func (s scannerFromTx) ScanEvaluations(ctx context.Context) ([]evalRow, error) { 157 return dbgen.New(s.tx).ListSlateDecisionEvidenceForOrg(ctx) 158} 159 160func (s scannerFromTx) ResolveSkillID(ctx context.Context, name string) (uuid.UUID, bool, error) { 161 id, err := dbgen.New(s.tx).ResolveSkillIDByName(ctx, name) 162 if errors.Is(err, pgx.ErrNoRows) { 163 return uuid.Nil, false, nil 164 } 165 if err != nil { 166 return uuid.Nil, false, err 167 } 168 return id, true, nil 169} 170 171// migratorGovernanceReader backs GovernanceReader with the migrator pool. 172type migratorGovernanceReader struct{ db dbgen.DBTX } 173 174func (g migratorGovernanceReader) GovernanceConfidence(ctx context.Context, tenantID uuid.UUID, modelVersion string) ([]dbgen.TeachingSignalAggregate, error) { 175 return dbgen.New(g.db).ListTeachingAggregatesForTenant(ctx, dbgen.ListTeachingAggregatesForTenantParams{ 176 TenantID: &tenantID, ModelVersion: modelVersion, 177 }) 178} 179 180// NewMigratorGovernanceReader wraps a migrator scoped dbgen.DBTX as the governance confidence reader. The 181// caller passes the migrator pool here, never the app pool, because teaching_signal_aggregates is a global 182// table the app role can only SELECT. 183func NewMigratorGovernanceReader(migrator dbgen.DBTX) GovernanceReader { 184 return migratorGovernanceReader{db: migrator} 185} 186 187// migratorProposalWriter backs ProposalWriter with the migrator pool. 188type migratorProposalWriter struct{ db dbgen.DBTX } 189 190func (w migratorProposalWriter) ActiveProposalExists(ctx context.Context, fromSkillID, toSkillID uuid.UUID, relation string) (bool, error) { 191 _, err := dbgen.New(w.db).GetActiveProposalForPair(ctx, dbgen.GetActiveProposalForPairParams{ 192 FromSkillID: &fromSkillID, ToSkillID: &toSkillID, Relation: &relation, 193 }) 194 if errors.Is(err, pgx.ErrNoRows) { 195 return false, nil 196 } 197 if err != nil { 198 return false, err 199 } 200 return true, nil 201} 202 203func (w migratorProposalWriter) InsertProposal(ctx context.Context, arg dbgen.InsertCurationProposalParams) (uuid.UUID, error) { 204 row, err := dbgen.New(w.db).InsertCurationProposal(ctx, arg) 205 if err != nil { 206 return uuid.Nil, err 207 } 208 return row.ID, nil 209} 210 211// NewMigratorProposalWriter wraps a migrator scoped dbgen.DBTX as the proposal writer. The caller passes the 212// migrator pool here, never the app pool, because curation_proposals is a global table the app role can only 213// SELECT. 214func NewMigratorProposalWriter(migrator dbgen.DBTX) ProposalWriter { 215 return migratorProposalWriter{db: migrator} 216} 217 218// ProposalGenerator turns one tenant's recurring approved reviewer craft into pending curation_proposals. The 219// evaluation scan is RLS scoped on the app pool; the governance read and the proposal write go through the 220// migrator pool; the proposal created event is a global System audit. 221type ProposalGenerator struct { 222 app AppReader // app pool RunInTenantTx, the RLS scoped evaluation scan boundary 223 governance GovernanceReader // migrator pool, the contents free governance confidence read 224 writer ProposalWriter // migrator pool, the proposal write and idempotency seam 225 audit systemAuditor // audit.Writer.System for the global proposal created event 226} 227 228// NewProposalGenerator wires the live generator. app is the app pool (RLS scoped evaluation scans), governance 229// and writer are backed by the MIGRATOR pool (the global tables), and auditW records the global proposal 230// created event through System. 231func NewProposalGenerator(app AppReader, governance GovernanceReader, writer ProposalWriter, auditW systemAuditor) *ProposalGenerator { 232 return &ProposalGenerator{app: app, governance: governance, writer: writer, audit: auditW} 233} 234 235// ProposalResult reports what one GenerateProposalsForOrg run produced: how many evaluations it scanned and 236// the proposals it created (by kind). It carries counts and skill ids only, never PII, so it is safe to log. 237type ProposalResult struct { 238 EvaluationsScanned int 239 Created int 240 SubstitutionEdges int 241 ReciprocityReviews int 242 RecalibrationRows int 243} 244 245// GenerateProposalsForOrg scans the org's own evaluations for recurring approved craft and emits pending 246// curation_proposals. The org is taken from ctx (the validated tenant), never an argument. 247// 248// Order of operations: 249// 1. Inside RunInTenantTx on the app pool (RLS scoped), scan the org's slate decisions joined to the 250// persisted match evidence and resolve each pair's skill ids (by id when the matcher populated them, by 251// name through the global skills table otherwise). This is the ONLY app pool step. 252// 2. Read the contents free governance confidence (the tenant's teaching_signal_aggregates) through the 253// migrator pool. 254// 3. Select the pairs that clear the curated threshold AND are supported by the governance confidence, then 255// insert each as a pending substitutes_for_edge proposal through the migrator pool, idempotently (never a 256// duplicate for a pair already pending or approved). Audit teaching.proposal.created through System. 257// 4. Emit reciprocity_review proposals for recurring approved clearance gap overrides, and one 258// recalibration_evidence proposal carrying the aggregate counts for WS4 (evidence only, never a threshold 259// change), the same idempotent and audited way. 260func (p *ProposalGenerator) GenerateProposalsForOrg(ctx context.Context) (ProposalResult, error) { 261 orgID, ok := tenant.OrgFromContext(ctx) 262 if !ok { 263 return ProposalResult{}, tenant.ErrNoTenant 264 } 265 tenantUUID, err := parseOrgUUID(orgID) 266 if err != nil { 267 return ProposalResult{}, err 268 } 269 270 // 1. RLS scoped evaluation scan plus the skill id resolution, all on the app tx. 271 var rows []evalRow 272 var pairs map[pairKey]*pairEvidence 273 var overrideApprovals int 274 if rerr := p.app.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 275 s := scannerFromTx{tx: tx} 276 var serr error 277 rows, serr = s.ScanEvaluations(ctx) 278 if serr != nil { 279 return serr 280 } 281 pairs, serr = derivePairs(ctx, rows, s) 282 if serr != nil { 283 return serr 284 } 285 overrideApprovals = countClearanceOverrideApprovals(rows) 286 return nil 287 }); rerr != nil { 288 return ProposalResult{}, rerr 289 } 290 291 // 2. The contents free governance confidence (migrator pool, the global table). 292 aggregates, gerr := p.governance.GovernanceConfidence(ctx, tenantUUID, engine.ModelVersion) 293 if gerr != nil { 294 return ProposalResult{}, gerr 295 } 296 297 // 3 and 4. Pure selection over the derived pairs and the governance confidence, then the idempotent, 298 // audited write of each proposal through the migrator pool. 299 plans := selectProposals(pairs, overrideApprovals, aggregates) 300 res := ProposalResult{EvaluationsScanned: len(rows)} 301 for _, plan := range plans { 302 created, werr := p.writeProposal(ctx, plan) 303 if werr != nil { 304 return res, werr 305 } 306 if !created { 307 continue 308 } 309 res.Created++ 310 switch plan.kind { 311 case kindSubstitutesForEdge: 312 res.SubstitutionEdges++ 313 case kindReciprocityReview: 314 res.ReciprocityReviews++ 315 case kindRecalibrationEvidence: 316 res.RecalibrationRows++ 317 } 318 } 319 return res, nil 320} 321 322// proposalPlan is one candidate proposal the pure selection produced, before the idempotency check and the 323// write. A substitutes_for_edge plan carries the resolved from and to skill ids and the band derived strength; 324// the reciprocity_review and recalibration_evidence plans carry counts only, no skill ids. 325type proposalPlan struct { 326 kind string 327 fromSkillID *uuid.UUID 328 toSkillID *uuid.UUID 329 relation string 330 strength *decimal.Decimal 331 observations int 332 successes int 333} 334 335// writeProposal performs the idempotency check and the insert for one plan through the migrator pool, then 336// emits the global teaching.proposal.created audit. It returns created false (with no error) when an active 337// proposal already exists for the pair, so a rerun never duplicates a suggestion. The audit carries skill ids 338// and counts only, NEVER PII, and runs through System with a NULL org because a proposal is a global object. 339func (p *ProposalGenerator) writeProposal(ctx context.Context, plan proposalPlan) (bool, error) { 340 // Idempotency only applies to a skill pair (a substitutes_for_edge). The evidence only kinds 341 // (reciprocity_review, recalibration_evidence) carry no pair, so there is nothing to dedupe on a pair key; 342 // the per run window scoping in the caller keeps them from multiplying. 343 if plan.kind == kindSubstitutesForEdge && plan.fromSkillID != nil && plan.toSkillID != nil { 344 exists, err := p.writer.ActiveProposalExists(ctx, *plan.fromSkillID, *plan.toSkillID, plan.relation) 345 if err != nil { 346 return false, err 347 } 348 if exists { 349 return false, nil 350 } 351 } 352 353 var relation *string 354 if plan.relation != "" { 355 r := plan.relation 356 relation = &r 357 } 358 id, err := p.writer.InsertProposal(ctx, dbgen.InsertCurationProposalParams{ 359 Kind: plan.kind, 360 FromSkillID: plan.fromSkillID, 361 ToSkillID: plan.toSkillID, 362 Relation: relation, 363 ProposedStrength: plan.strength, 364 EvidenceObservations: int32(plan.observations), 365 EvidenceSuccesses: int32(plan.successes), 366 ModelVersion: engine.ModelVersion, 367 }) 368 if err != nil { 369 return false, err 370 } 371 372 payload, merr := json.Marshal(map[string]any{ 373 "kind": plan.kind, 374 "relation": plan.relation, 375 "evidence_observations": plan.observations, 376 "evidence_successes": plan.successes, 377 }) 378 if merr != nil { 379 return false, merr 380 } 381 if aerr := p.audit.System(ctx, audit.Event{ 382 Action: "teaching.proposal.created", 383 EntityType: "curation_proposal", 384 EntityID: id.String(), 385 NewValues: payload, 386 }); aerr != nil { 387 return false, aerr 388 } 389 return true, nil 390} 391 392// matchedSkillFull mirrors the full closed subset of engine.SkillMatch the generator needs to identify a 393// specific pair. The engine type has no json tags, so the persisted jsonb keys are the Go field names; these 394// tags pin those names. The generator reads the ids first and falls back to the names, never any other field. 395type matchedSkillFull struct { 396 RequiredSkillID string `json:"RequiredSkillID"` 397 RequiredSkillName string `json:"RequiredSkillName"` 398 CandidateSkillID string `json:"CandidateSkillID"` 399 CandidateName string `json:"CandidateName"` 400 Similarity decimal.Decimal `json:"Similarity"` 401 Exact bool `json:"Exact"` 402 Matched bool `json:"Matched"` 403} 404 405// derivePairs scans the evaluation rows for recurring APPROVED non exact (required skill, candidate skill) 406// pairs and counts, per pair, how many times it was approved and how many times it was seen at all, tracking 407// the strongest band the reviewers accepted it at. It resolves each side to a global skills.id: by id when 408// the matcher populated the SkillMatch ids, and by resolving the name through the skills table otherwise. A 409// pair that cannot be resolved to two ids is skipped (never a fabricated id). An exact match is never a 410// substitution, so it is skipped here. derivePairs runs on the app tx because the name resolution reads the 411// global skills table on the same connection; the counting itself is pure. 412func derivePairs(ctx context.Context, rows []evalRow, scanner EvaluationScanner) (map[pairKey]*pairEvidence, error) { 413 out := map[pairKey]*pairEvidence{} 414 // resolveCache memoizes name to id resolution across rows so one skill name resolves once per run. 415 resolveCache := map[string]*uuid.UUID{} 416 resolve := func(id, name string) (string, error) { 417 if id != "" { 418 return id, nil 419 } 420 if name == "" { 421 return "", nil 422 } 423 if cached, ok := resolveCache[name]; ok { 424 if cached == nil { 425 return "", nil 426 } 427 return cached.String(), nil 428 } 429 rid, found, err := scanner.ResolveSkillID(ctx, name) 430 if err != nil { 431 return "", err 432 } 433 if !found { 434 resolveCache[name] = nil 435 return "", nil 436 } 437 resolveCache[name] = &rid 438 return rid.String(), nil 439 } 440 441 for _, row := range rows { 442 outcome, ok := outcomeFromDecision(row.Decision) 443 if !ok { 444 continue 445 } 446 approved := outcome == scrub.OutcomeApproved 447 for _, sm := range parseMatchedSkillsFull(row.MatchedSkills) { 448 if !sm.Matched || sm.Exact { 449 continue // only a non exact accepted substitution is a candidate pair 450 } 451 fromID, err := resolve(sm.CandidateSkillID, sm.CandidateName) 452 if err != nil { 453 return nil, err 454 } 455 toID, err := resolve(sm.RequiredSkillID, sm.RequiredSkillName) 456 if err != nil { 457 return nil, err 458 } 459 if fromID == "" || toID == "" || fromID == toID { 460 continue // unresolved or a self pair is never a substitution 461 } 462 k := pairKey{fromSkillID: fromID, toSkillID: toID} 463 ev := out[k] 464 if ev == nil { 465 ev = &pairEvidence{band: scrub.BandUnknown} 466 out[k] = ev 467 } 468 ev.total++ 469 if approved { 470 ev.approvals++ 471 if band := bandFromSimilarity(sm.Similarity); bandRank(band) > bandRank(ev.band) { 472 ev.band = band // track the strongest band an approval was accepted at 473 } 474 } 475 } 476 } 477 return out, nil 478} 479 480// countClearanceOverrideApprovals counts how many slate decisions in this tenant approved a candidate whose 481// engine clearance_score was zero against a required clearance, which is a reviewer overriding a clearance 482// gap. It is contents free (a count, no candidate id), and it feeds the reciprocity_review proposal, which is 483// evidence for a human governance review, never a scoring edge. 484func countClearanceOverrideApprovals(rows []evalRow) int { 485 n := 0 486 for _, row := range rows { 487 outcome, ok := outcomeFromDecision(row.Decision) 488 if !ok || outcome != scrub.OutcomeApproved { 489 continue 490 } 491 if clearanceRequired(row.RequiredLevel) && row.ClearanceScore.IsZero() { 492 n++ 493 } 494 } 495 return n 496} 497 498// selectProposals is the pure selection over the derived pairs, the clearance override count, and the 499// governance confidence. It is a function of the evidence and the curated thresholds, so it unit tests with 500// fixtures and no DB. 501// 502// For a substitutes_for_edge it requires THREE things to all hold: 503// - the pair was approved at least minSubstitutionApprovals times in this tenant's own evaluations, 504// - the strongest accepted band is high or medium (a low or unknown band similarity is too weak to curate), 505// - the contents free governance confidence confirms the adjacent_skill_substitution_outcome pattern is 506// reliably approved at that band (the aggregate has more approved than not at the band). 507// 508// That double gate is the design correction: the specific pair comes from the tenant's own reviewed 509// evaluations, and the cross cutting reliability comes from the contents free aggregate. Neither alone 510// proposes an edge. 511func selectProposals(pairs map[pairKey]*pairEvidence, overrideApprovals int, aggregates []dbgen.TeachingSignalAggregate) []proposalPlan { 512 var plans []proposalPlan 513 514 // Index the governance confidence by band for the substitution pattern, so the lookup is O(1) per pair. 515 subConfidence := substitutionConfidenceByBand(aggregates) 516 517 // Deterministic order: sort the pair keys so the proposals are stable across runs. 518 keys := make([]pairKey, 0, len(pairs)) 519 for k := range pairs { 520 keys = append(keys, k) 521 } 522 sort.Slice(keys, func(i, j int) bool { 523 if keys[i].fromSkillID != keys[j].fromSkillID { 524 return keys[i].fromSkillID < keys[j].fromSkillID 525 } 526 return keys[i].toSkillID < keys[j].toSkillID 527 }) 528 529 for _, k := range keys { 530 ev := pairs[k] 531 if ev.approvals < minSubstitutionApprovals { 532 continue 533 } 534 strengthStr, bandable := strengthByBand[ev.band] 535 if !bandable { 536 continue // a low or unknown band is too weak to curate as a substitution 537 } 538 if !subConfidence[ev.band] { 539 continue // the contents free aggregate does not support the pattern at this band 540 } 541 from, err1 := uuid.Parse(k.fromSkillID) 542 to, err2 := uuid.Parse(k.toSkillID) 543 if err1 != nil || err2 != nil { 544 continue // a malformed id never proposes 545 } 546 strength := decimal.RequireFromString(strengthStr) 547 plans = append(plans, proposalPlan{ 548 kind: kindSubstitutesForEdge, 549 fromSkillID: &from, 550 toSkillID: &to, 551 relation: substitutionRelation, 552 strength: &strength, 553 observations: ev.total, 554 successes: ev.approvals, 555 }) 556 } 557 558 // A reciprocity_review for recurring approved clearance gap overrides. Evidence only, never a scoring 559 // edge: it asks a human to review whether the tenant's clearance policy and the reviewers' overrides are 560 // reciprocal. It carries no skill ids, only the counts. 561 if overrideApprovals >= minOverrideApprovals { 562 plans = append(plans, proposalPlan{ 563 kind: kindReciprocityReview, 564 relation: "", 565 observations: overrideApprovals, 566 successes: overrideApprovals, 567 }) 568 } 569 570 // One recalibration_evidence proposal carrying the reranker agreement aggregate when present, for WS4. 571 // Evidence only: it never changes a threshold automatically, it gives a human the counts to recalibrate. 572 if obs, succ, ok := recalibrationEvidence(aggregates); ok { 573 plans = append(plans, proposalPlan{ 574 kind: kindRecalibrationEvidence, 575 relation: "", 576 observations: obs, 577 successes: succ, 578 }) 579 } 580 581 return plans 582} 583 584// substitutionConfidenceByBand reduces the governance aggregates to a per band yes or no: is the 585// adjacent_skill_substitution_outcome pattern reliably approved at the band (more approved observations than 586// not). It is the contents free reliability check that gates a specific pair from the tenant's own 587// evaluations. 588func substitutionConfidenceByBand(aggregates []dbgen.TeachingSignalAggregate) map[scrub.SimilarityBand]bool { 589 approved := map[scrub.SimilarityBand]int{} 590 other := map[scrub.SimilarityBand]int{} 591 for _, a := range aggregates { 592 if a.Pattern != "adjacent_skill_substitution_outcome" { 593 continue 594 } 595 band := scrub.SimilarityBand(a.Band) 596 if a.Outcome == string(scrub.OutcomeApproved) { 597 approved[band] += int(a.Successes) 598 } else { 599 other[band] += int(a.Observations) 600 } 601 } 602 out := map[scrub.SimilarityBand]bool{} 603 for band, n := range approved { 604 out[band] = n > 0 && n >= other[band] 605 } 606 return out 607} 608 609// recalibrationEvidence sums the reranker_label_agreement aggregate counts across bands, when any exist, for a 610// single recalibration_evidence proposal. It returns ok false when there is no reranker agreement aggregate, 611// so no empty evidence proposal is created. The reranker pattern folds nothing today (no stored reranker 612// label), so this is dormant until that evidence exists, exactly like the producer's fold. 613func recalibrationEvidence(aggregates []dbgen.TeachingSignalAggregate) (observations, successes int, ok bool) { 614 for _, a := range aggregates { 615 if a.Pattern != "reranker_label_agreement" { 616 continue 617 } 618 observations += int(a.Observations) 619 successes += int(a.Successes) 620 ok = true 621 } 622 return observations, successes, ok 623} 624 625// parseMatchedSkillsFull decodes the candidate_evaluations.matched_skills jsonb into the full closed subset 626// the generator needs (the ids and names plus the similarity and the flags). A NULL or empty column yields no 627// skills, and unreadable bytes are treated as absent, never fabricated, the same posture as the producer. 628func parseMatchedSkillsFull(raw []byte) []matchedSkillFull { 629 if len(raw) == 0 { 630 return nil 631 } 632 var skills []matchedSkillFull 633 if err := json.Unmarshal(raw, &skills); err != nil { 634 return nil 635 } 636 return skills 637} 638 639// bandRank orders the similarity bands so derivePairs can track the strongest band an approval was accepted 640// at. Higher is stronger. Unknown and low rank lowest; exact ranks highest but never reaches here for a non 641// exact substitution. 642func bandRank(b scrub.SimilarityBand) int { 643 switch b { 644 case scrub.BandExact: 645 return 4 646 case scrub.BandHigh: 647 return 3 648 case scrub.BandMedium: 649 return 2 650 case scrub.BandLow: 651 return 1 652 default: 653 return 0 654 } 655} 656