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

1package ai 2 3// teaching.go is the WS8 in tenant teaching producer (Phase 1, LIVE). It reads one tenant's reviewer craft 4// (the slate decisions and the frozen match evidence) and folds it into the closed scrub.TeachingSignal 5// vocabulary of contents free counts, so MEGAMIND can learn the CRAFT of the work (abstract patterns) and 6// never a client fact. It is LIVE in the sense that it runs in production and persists the per tenant 7// aggregates, but transmission to the inference plane stays gated off by TeachingEnabled until multiple 8// live tenants exist, so nothing leaves the box yet. 9// 10// The producer never touches the engine math. It reads what the frozen engine already wrote (the decision, 11// the sub scores, the gap summary, the matched skills) and counts it. It never changes a weight, a band 12// cut, the semantic threshold, or any served score. 13// 14// Three correctness boundaries hold by construction: 15// 16// 1. The READ is RLS scoped. It runs inside RunInTenantTx on the NOBYPASSRLS app pool, so Postgres Row 17// Level Security restricts the join to the caller's own org. One tenant's producer can never read 18// another tenant's decisions. 19// 2. The aggregate WRITE goes through the MIGRATOR pool seam, never the app role. teaching_signal_aggregates 20// is a global, migrator owned table the app role can only SELECT, so the live in tenant rows are written 21// through an injected Writer backed by the migrator pool. 22// 3. The audit event runs through audit.Writer.OnTx INSIDE the same tenant tx that read the decisions, so 23// it commits with the read and carries the correct org. It carries counts only, no PII, no candidate id, 24// and no raw score. 25// 26// Everything score affecting is deferred to the two human step curation pipeline (Phase 4); this producer 27// only aggregates evidence. 28 29import ( 30 "context" 31 "encoding/json" 32 "sort" 33 "time" 34 35 "github.com/google/uuid" 36 "github.com/jackc/pgx/v5" 37 "github.com/jackc/pgx/v5/pgtype" 38 "github.com/shopspring/decimal" 39 40 "github.com/missionworkforce/missionmatch/internal/modules/matching/engine" 41 "github.com/missionworkforce/missionmatch/internal/platform/audit" 42 "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" 43 "github.com/missionworkforce/missionmatch/internal/platform/inference" 44 "github.com/missionworkforce/missionmatch/internal/platform/scrub" 45 "github.com/missionworkforce/missionmatch/internal/platform/tenant" 46) 47 48// init registers the live engine model version with the scrub boundary so a signal stamped with 49// engine.ModelVersion is accepted, and a drifted version is refused the same way an unknown Pattern is. The 50// version is pinned from the engine package (never a literal), the only module here allowed to import it. 51func init() { 52 scrub.RegisterModelVersion(engine.ModelVersion) 53} 54 55// teachingThresholds keeps the few derived classification rules in one place so the folding logic is a pure 56// function of the stored evidence and the rules are visible in one spot for review. 57const ( 58 // clearanceRequiredNone and clearanceRequiredUnknown are the two required_level values that mean no 59 // clearance is required (matching engine.scoreClearance, which passes everyone below Confidential). Any 60 // other required_level means a clearance was demanded, so a zero clearance_score on an approved member is 61 // a reviewer override of a clearance gap. 62 clearanceRequiredNone = "none" 63 clearanceRequiredUnknown = "unknown" 64) 65 66// DecisionRow is the contents free craft evidence for one slate member: the reviewer decision, the stored 67// sub scores that reveal a clearance gap or an experience shortfall, the frozen gap summary, the required 68// clearance level, and the matched skills evidence from the candidate's persisted evaluation. It is the 69// shape ListSlateDecisionEvidenceForOrgRow already returns; the producer reads only these closed fields and 70// never a name, an email, or a resume. 71type DecisionRow = dbgen.ListSlateDecisionEvidenceForOrgRow 72 73// DecisionReader is the RLS scoped read seam. The live implementation runs inside RunInTenantTx on the app 74// pool; a test injects a fake returning fixture rows so the folding logic unit tests with no database. 75type DecisionReader interface { 76 ListDecisionEvidence(ctx context.Context) ([]DecisionRow, error) 77} 78 79// AggregateWriter is the migrator pool write seam for the global teaching_signal_aggregates table. The live 80// implementation is backed by the migrator pool (the app role has SELECT only on this global table); a test 81// injects a fake that records the upserts so the producer can be exercised with no database. 82type AggregateWriter interface { 83 UpsertTeachingSignalAggregate(ctx context.Context, arg dbgen.UpsertTeachingSignalAggregateParams) error 84} 85 86// AppReader is the app pool RunInTenantTx seam. *tenant.DB satisfies it. It is an interface so the audit on 87// the same tenant tx can be exercised, while the folding logic itself unit tests through the DecisionReader. 88type AppReader interface { 89 RunInTenantTx(ctx context.Context, fn func(ctx context.Context, tx pgx.Tx) error) error 90} 91 92// *tenant.DB is the live AppReader, a compile time assertion that the app pool satisfies the read seam. 93var _ AppReader = (*tenant.DB)(nil) 94 95// readerFromTx builds a DecisionReader bound to a tenant transaction, so the RLS scoped read runs on the 96// same connection whose app.current_org_id GUC RunInTenantTx set. 97type readerFromTx struct{ tx pgx.Tx } 98 99// readerFromTx is the live DecisionReader: a compile time assertion that the tenant tx backed reader honors 100// the seam the producer reads through. 101var _ DecisionReader = readerFromTx{} 102 103func (r readerFromTx) ListDecisionEvidence(ctx context.Context) ([]DecisionRow, error) { 104 return dbgen.New(r.tx).ListSlateDecisionEvidenceForOrg(ctx) 105} 106 107// migratorAggregateWriter backs AggregateWriter with the migrator pool. It is a thin adapter so the 108// producer depends on the interface, not the concrete dbgen DBTX, and a test can substitute a fake. 109type migratorAggregateWriter struct{ db dbgen.DBTX } 110 111func (w migratorAggregateWriter) UpsertTeachingSignalAggregate(ctx context.Context, arg dbgen.UpsertTeachingSignalAggregateParams) error { 112 return dbgen.New(w.db).UpsertTeachingSignalAggregate(ctx, arg) 113} 114 115// NewMigratorAggregateWriter wraps a migrator scoped dbgen.DBTX (the migrator pool) as the global aggregate 116// writer. The caller passes the migrator pool here, never the app pool, because the app role cannot write 117// the global teaching_signal_aggregates table. 118func NewMigratorAggregateWriter(migrator dbgen.DBTX) AggregateWriter { 119 return migratorAggregateWriter{db: migrator} 120} 121 122// TeachingProducer folds one tenant's reviewer craft into scrub.TeachingSignal counts and persists the live 123// in tenant aggregates. It is LIVE but transmission stays gated off by the inference client's TeachingEnabled 124// flag, so calling Teach validates the boundary end to end (scrub.Marshal) without emitting anything off box. 125type TeachingProducer struct { 126 app AppReader // app pool RunInTenantTx, the RLS scoped read boundary 127 writer AggregateWriter // migrator pool, the global aggregate write seam (never the app role) 128 audit *audit.Writer // audit.Writer.OnTx on the tenant tx 129 teacher *inference.Client // Teach, gated off by TeachingEnabled 130 now func() time.Time // injectable clock so the window dates are deterministic in tests 131} 132 133// NewTeachingProducer wires the live producer. app is the app pool (RLS scoped reads), writer is backed by 134// the MIGRATOR pool (global aggregate writes), auditW records the per tenant event on the tenant tx, and 135// teacher is the inference client (Teach stays gated off by TeachingEnabled). 136func NewTeachingProducer(app AppReader, writer AggregateWriter, auditW *audit.Writer, teacher *inference.Client) *TeachingProducer { 137 return &TeachingProducer{app: app, writer: writer, audit: auditW, teacher: teacher, now: time.Now} 138} 139 140// Result reports what one AggregateForOrg run folded: how many slate decisions it read and the signals it 141// produced. It carries counts only, never a candidate id or a score, so it is safe to log. 142type Result struct { 143 DecisionsRead int 144 Signals []scrub.TeachingSignal 145} 146 147// AggregateForOrg reads the org's slate decisions and folds them into per tenant craft aggregates. The org 148// is taken from ctx (the validated tenant), never an argument. 149// 150// Order of operations: 151// 1. Inside RunInTenantTx on the app pool (RLS scoped), read the decision evidence and emit the 152// teaching.signal.aggregated audit on the SAME tenant tx, so the audit commits with the read. 153// 2. After the read commits, fold the rows into scrub.TeachingSignal counts (a pure function). 154// 3. For each signal, call inference.Client.Teach (boundary checked, transmission gated off). 155// 4. Persist each signal into teaching_signal_aggregates through the MIGRATOR pool writer, tenant_id set, 156// dp_applied false (the live in tenant rows). The app role cannot write this global table. 157func (p *TeachingProducer) AggregateForOrg(ctx context.Context) (Result, error) { 158 orgID, ok := tenant.OrgFromContext(ctx) 159 if !ok { 160 return Result{}, tenant.ErrNoTenant 161 } 162 163 // 1. RLS scoped read plus the audit on the SAME tenant tx (invariant 5: the per tenant event commits 164 // with the read it records and carries the correct org). 165 var rows []DecisionRow 166 err := p.app.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 167 var rerr error 168 rows, rerr = readerFromTx{tx: tx}.ListDecisionEvidence(ctx) 169 if rerr != nil { 170 return rerr 171 } 172 // Counts only in the audit payload: how many decisions this run aggregated. No candidate id, no 173 // score, no name. The org is read from ctx by OnTx, never passed here. 174 newVals, merr := json.Marshal(map[string]int{"decisions_aggregated": len(rows)}) 175 if merr != nil { 176 return merr 177 } 178 return p.audit.OnTx(ctx, tx, audit.Event{ 179 Action: "teaching.signal.aggregated", 180 EntityType: "teaching_signal", 181 NewValues: newVals, 182 }) 183 }) 184 if err != nil { 185 return Result{}, err 186 } 187 188 // 2. Fold the evidence into the closed craft vocabulary (a pure function of the rows plus the window). 189 signals := fold(rows, engine.ModelVersion) 190 191 // 3 and 4. Exercise the scrub boundary through Teach (gated off, emits nothing) and persist the live in 192 // tenant aggregates through the migrator pool writer. Factored out so the teach and persist loop 193 // unit tests with fakes and no database, while the tenant tx read and audit above are covered by 194 // the integration test. 195 if perr := p.persist(ctx, orgID, signals); perr != nil { 196 return Result{}, perr 197 } 198 199 return Result{DecisionsRead: len(rows), Signals: signals}, nil 200} 201 202// persist runs the per signal teach and write loop: for each folded signal it calls inference.Client.Teach 203// (which validates the scrub boundary and, with TeachingEnabled false, transmits nothing) and then upserts 204// the live in tenant aggregate through the migrator pool writer with tenant_id set and dp_applied false. It 205// touches no app pool and no tenant tx, so it is safe to unit test with a fake writer and a gated off teacher. 206func (p *TeachingProducer) persist(ctx context.Context, orgID string, signals []scrub.TeachingSignal) error { 207 tenantUUID, perr := parseOrgUUID(orgID) 208 if perr != nil { 209 return perr 210 } 211 start, end := p.window() 212 for _, sig := range signals { 213 if p.teacher != nil { 214 if terr := p.teacher.Teach(ctx, sig); terr != nil { 215 // Teach validates the boundary (scrub.Marshal). A failure means the producer built a signal 216 // the boundary refuses, which is a bug to surface, not to swallow. Transmission is gated off. 217 return terr 218 } 219 } 220 if werr := p.writer.UpsertTeachingSignalAggregate(ctx, dbgen.UpsertTeachingSignalAggregateParams{ 221 TenantID: &tenantUUID, // live in tenant row: tenant_id is set 222 DpApplied: false, // not the cross tenant post DP path 223 Pattern: sig.Pattern, 224 Band: string(sig.Band), 225 Outcome: string(sig.Outcome), 226 ModelVersion: sig.ModelVersion, 227 Observations: int32(sig.Observations), 228 Successes: int32(sig.Successes), 229 WindowStart: pgDate(start), 230 WindowEnd: pgDate(end), 231 }); werr != nil { 232 return werr 233 } 234 } 235 return nil 236} 237 238// foldKey identifies one aggregate bucket: a craft pattern at a similarity band for a decision outcome. The 239// producer counts observations and successes per key, then turns each key into one scrub.TeachingSignal. 240type foldKey struct { 241 pattern string 242 band scrub.SimilarityBand 243 outcome scrub.Outcome 244} 245 246type foldCount struct { 247 observations int 248 successes int 249} 250 251// fold is the pure folding logic, a function of the stored decision evidence and the model version. It has 252// no DB, no clock, and no I/O, so it unit tests with fixture rows. 253// 254// It folds into the four closed scrub.Patterns. Which patterns are derivable from the stored evidence today: 255// 256// - adjacent_skill_substitution_outcome: DERIVABLE. candidate_evaluations.matched_skills carries each 257// required to candidate SkillMatch with its Similarity and an Exact flag (engine SkillMatch, no json 258// tags so the keys are the Go field names). A match with Matched true and Exact false is an adjacent or 259// semantic substitution. The Band comes from the cosine bucket of its Similarity, the Outcome from the 260// slate decision. A member with no stored evaluation (LEFT join NULL) contributes nothing here, which is 261// correct, the producer never fabricates a substitution it cannot see. 262// 263// - clearance_gap_override_outcome: DERIVABLE. A clearance was required when required_level is neither 264// none nor unknown (matching engine.scoreClearance). A zero clearance_score on such a member is the 265// engine recording an unmet clearance, so a member with a required clearance and clearance_score zero is 266// a clearance gap; the reviewer decision generalizes whether they overrode it. The Band is unknown, 267// because clearance carries no similarity. 268// 269// - experience_shortfall_outcome: DERIVABLE. The engine writes experience_score below 100 when the 270// candidate has fewer years than required (engine.scoreExperience). An experience shortfall is therefore 271// a member with experience_score below 100; the decision generalizes how reviewers treat it. Note the 272// engine does NOT add an experience entry to gap_summary (it appends only clearance, skill, and cert 273// gaps), so the score is the authoritative signal, not the gap summary text. The Band is unknown. 274// 275// - reranker_label_agreement: ZERO TODAY. No reranker label and no reranker agreement is persisted with a 276// slate decision or a candidate evaluation. The inference reranker (inference.Client.Rerank) is not 277// stored alongside the frozen decision, so there is no stored evidence to fold. The producer derives 278// nothing for this pattern and emits no signal for it, rather than fabricate a count. When a reranker 279// label is persisted in a later phase, this fold gains a real branch. 280func fold(rows []DecisionRow, modelVersion string) []scrub.TeachingSignal { 281 counts := map[foldKey]*foldCount{} 282 283 add := func(pattern string, band scrub.SimilarityBand, outcome scrub.Outcome, success bool) { 284 k := foldKey{pattern: pattern, band: band, outcome: outcome} 285 c := counts[k] 286 if c == nil { 287 c = &foldCount{} 288 counts[k] = c 289 } 290 c.observations++ 291 if success { 292 c.successes++ 293 } 294 } 295 296 for _, row := range rows { 297 outcome, ok := outcomeFromDecision(row.Decision) 298 if !ok { 299 continue // an out of vocabulary decision is skipped rather than folded into a wrong bucket 300 } 301 // A success is the reviewer approving the candidate despite (or with) the observed evidence; it is 302 // the positive class a craft signal generalizes over. 303 approved := outcome == scrub.OutcomeApproved 304 305 // adjacent_skill_substitution_outcome: one observation per non exact matched skill, banded by its 306 // similarity. A member with no stored evaluation contributes none. 307 for _, sm := range parseMatchedSkills(row.MatchedSkills) { 308 if sm.Matched && !sm.Exact { 309 add("adjacent_skill_substitution_outcome", bandFromSimilarity(sm.Similarity), outcome, approved) 310 } 311 } 312 313 // clearance_gap_override_outcome: a required clearance the engine scored zero is a clearance gap. 314 if clearanceRequired(row.RequiredLevel) && row.ClearanceScore.IsZero() { 315 add("clearance_gap_override_outcome", scrub.BandUnknown, outcome, approved) 316 } 317 318 // experience_shortfall_outcome: an experience_score below 100 is a years shortfall. 319 if row.ExperienceScore.LessThan(decimal.NewFromInt(100)) { 320 add("experience_shortfall_outcome", scrub.BandUnknown, outcome, approved) 321 } 322 323 // reranker_label_agreement: no stored evidence today, intentionally folds nothing (see the doc above). 324 } 325 326 return signalsFromCounts(counts, modelVersion) 327} 328 329// signalsFromCounts turns the folded buckets into scrub.TeachingSignals in a deterministic order (by 330// pattern, then band, then outcome), so the producer output and the persisted rows are stable across runs. 331func signalsFromCounts(counts map[foldKey]*foldCount, modelVersion string) []scrub.TeachingSignal { 332 keys := make([]foldKey, 0, len(counts)) 333 for k := range counts { 334 keys = append(keys, k) 335 } 336 sort.Slice(keys, func(i, j int) bool { 337 if keys[i].pattern != keys[j].pattern { 338 return keys[i].pattern < keys[j].pattern 339 } 340 if keys[i].band != keys[j].band { 341 return keys[i].band < keys[j].band 342 } 343 return keys[i].outcome < keys[j].outcome 344 }) 345 out := make([]scrub.TeachingSignal, 0, len(keys)) 346 for _, k := range keys { 347 c := counts[k] 348 out = append(out, scrub.TeachingSignal{ 349 Pattern: k.pattern, 350 Band: k.band, 351 Outcome: k.outcome, 352 Observations: c.observations, 353 Successes: c.successes, 354 ModelVersion: modelVersion, 355 }) 356 } 357 return out 358} 359 360// matchedSkill mirrors the closed subset of engine.SkillMatch the producer needs. The engine type has no 361// json tags, so the persisted jsonb keys are the Go field names; these tags pin those names so the producer 362// is independent of the engine package's json encoding and reads only the evidence fields it folds. 363type matchedSkill struct { 364 Similarity decimal.Decimal `json:"Similarity"` 365 Exact bool `json:"Exact"` 366 Matched bool `json:"Matched"` 367} 368 369// parseMatchedSkills decodes the candidate_evaluations.matched_skills jsonb. A NULL column (a slate member 370// never screened through screenMATCH) or an empty array yields no skills, so the producer folds nothing for 371// such a member rather than guess. 372func parseMatchedSkills(raw []byte) []matchedSkill { 373 if len(raw) == 0 { 374 return nil 375 } 376 var skills []matchedSkill 377 if err := json.Unmarshal(raw, &skills); err != nil { 378 return nil // unreadable evidence is treated as absent, never fabricated 379 } 380 return skills 381} 382 383// outcomeFromDecision maps the slate_members.decision vocabulary to the scrub.Outcome vocabulary. They are 384// the same three labels by construction (the scrub Outcome constants were taken verbatim from 385// slate_members.decision), so this is a checked passthrough that refuses an unexpected label. 386func outcomeFromDecision(decision string) (scrub.Outcome, bool) { 387 switch decision { 388 case "approved": 389 return scrub.OutcomeApproved, true 390 case "rejected": 391 return scrub.OutcomeRejected, true 392 case "pending": 393 return scrub.OutcomePending, true 394 default: 395 return "", false 396 } 397} 398 399// bandFromSimilarity buckets a cosine similarity into the coarse scrub.SimilarityBand, the public 400// quantization the teaching boundary speaks in (never the raw score). The cut points match the scrub band 401// documentation: exact at 1.00, high 0.90 to 0.99, medium 0.80 to 0.89, low below 0.80. A negative or 402// missing similarity is unknown. 403func bandFromSimilarity(sim decimal.Decimal) scrub.SimilarityBand { 404 switch { 405 case sim.GreaterThanOrEqual(decimal.NewFromInt(1)): 406 return scrub.BandExact 407 case sim.GreaterThanOrEqual(decimal.RequireFromString("0.90")): 408 return scrub.BandHigh 409 case sim.GreaterThanOrEqual(decimal.RequireFromString("0.80")): 410 return scrub.BandMedium 411 case sim.GreaterThanOrEqual(decimal.Zero): 412 return scrub.BandLow 413 default: 414 return scrub.BandUnknown 415 } 416} 417 418// clearanceRequired reports whether the requirement demanded a clearance, matching engine.scoreClearance 419// which passes everyone when the required level is below Confidential (the none and unknown labels). 420func clearanceRequired(requiredLevel string) bool { 421 return requiredLevel != clearanceRequiredNone && requiredLevel != clearanceRequiredUnknown 422} 423 424// window returns the observation window the producer stamps on each aggregate. v1 aggregates the full 425// stored history into one open ended window ending today, so a rerun refreshes the same window row in place 426// (the upsert key includes window_start). The window dates carry no client fact, only the run date. 427func (p *TeachingProducer) window() (start, end time.Time) { 428 now := p.now().UTC() 429 // A fixed epoch start keeps window_start stable across reruns so the upsert key is stable; the end moves 430 // to the run date. This is the all history window for v1, not a rolling window. 431 start = time.Date(1970, 1, 1, 0, 0, 0, 0, time.UTC) 432 end = time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, time.UTC) 433 return start, end 434} 435 436// pgDate wraps a date for the date columns. The producer always sets a valid date, never NULL. 437func pgDate(t time.Time) pgtype.Date { 438 return pgtype.Date{Time: t, Valid: true} 439} 440 441// parseOrgUUID parses the validated org id from context into a uuid for the tenant_id column. The org 442// originates from RunInTenantTx's validated context, so a parse failure is a programming error, not client 443// input. 444func parseOrgUUID(orgID string) (uuid.UUID, error) { 445 return uuid.Parse(orgID) 446} 447