Janady13/missionmatch / internal/modules/staffing/service.go

1package staffing 2 3import ( 4 "context" 5 "encoding/json" 6 "errors" 7 "fmt" 8 "hash/fnv" 9 "sort" 10 "strings" 11 "time" 12 13 "github.com/google/uuid" 14 "github.com/jackc/pgx/v5" 15 "github.com/shopspring/decimal" 16 17 "github.com/missionworkforce/missionmatch/internal/modules/ai" 18 "github.com/missionworkforce/missionmatch/internal/modules/matching" 19 "github.com/missionworkforce/missionmatch/internal/modules/matching/engine" 20 "github.com/missionworkforce/missionmatch/internal/platform/audit" 21 "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" 22 "github.com/missionworkforce/missionmatch/internal/platform/inference" 23 "github.com/missionworkforce/missionmatch/internal/platform/tenant" 24) 25 26const ( 27 w3LogicVersion = "w3.v1" 28 benchScanLimit = 500 // matches the slate loop's candidate scan cap 29 openRequirementCap = 500 30) 31 32// Service is the W3 application service. It runs every database operation inside a tenant transaction 33// (RLS scoped), reuses the one engine.ScoreCandidate path through the matching package, narrates only 34// behind the faithfulness gate, and audits every run and every decision. 35type Service struct { 36 db *tenant.DB 37 matcher engine.SkillMatcher 38 aud *audit.Writer 39 infer *inference.Client // may be nil: then every rationale is the deterministic template 40 cfg Config 41 clock func() time.Time 42} 43 44// NewService builds the service. db, aud, and infer may be nil (DB routes then return ErrNoDatabase; a nil 45// audit skips audit in tests; a nil inference client means deterministic rationales). 46func NewService(db *tenant.DB, matcher engine.SkillMatcher, aud *audit.Writer, infer *inference.Client, cfg Config) *Service { 47 return &Service{db: db, matcher: matcher, aud: aud, infer: infer, cfg: cfg, clock: time.Now} 48} 49 50func modelVersion() string { return engine.ModelVersion + "+" + w3LogicVersion } 51 52func (s *Service) orgID(ctx context.Context) (uuid.UUID, error) { 53 raw, ok := tenant.OrgFromContext(ctx) 54 if !ok { 55 return uuid.Nil, tenant.ErrNoTenant 56 } 57 id, err := uuid.Parse(raw) 58 if err != nil { 59 return uuid.Nil, fmt.Errorf("staffing: invalid org id in context: %w", err) 60 } 61 return id, nil 62} 63 64func actorID(ctx context.Context) *uuid.UUID { 65 if p, ok := tenant.UserFromContext(ctx); ok { 66 if id, err := uuid.Parse(p.UserID); err == nil { 67 return &id 68 } 69 } 70 return nil 71} 72 73// auditOnTx records an AEDT event on the caller's tx and fails closed (a recommendation or decision that 74// cannot be audited rolls back rather than being served unaudited). Demographics are never carried. 75func (s *Service) auditOnTx(ctx context.Context, tx pgx.Tx, e audit.Event) error { 76 if s.aud == nil { 77 return nil 78 } 79 e.ActorUserID = actorID(ctx) 80 return s.aud.OnTx(ctx, tx, e) 81} 82 83// servedScoreEvent is the screening stage AEDT record for a candidate surfaced in a recommendation, so the 84// four fifths screening analysis stays complete (the same posture as screenMATCH and the W2 fix). It 85// carries the candidate, the tool, the score, and the band, and NOTHING about demographics. 86func servedScoreEvent(tool string, candidateID uuid.UUID, res engine.MatchResult) audit.Event { 87 payload, _ := json.Marshal(map[string]any{"tool": tool, "score": res.Total.StringFixed(2), "band": res.Band.Code()}) 88 return audit.Event{Action: "aedt.score.served", EntityType: "candidate", EntityID: candidateID.String(), NewValues: payload} 89} 90 91// Decide records a human approve, reject, or dismiss on one recommendation. It locks the row, so two 92// concurrent decisions serialize; a row that is not pending (already decided, or concurrently superseded) 93// returns ErrConflict (409) rather than overwriting a decided or dead row. Rejecting only dismisses the 94// suggestion; it never blacklists a candidate, and nothing here is automatic. 95func (s *Service) Decide(ctx context.Context, recID uuid.UUID, req DecisionRequest) (RecommendationView, error) { 96 if s.db == nil { 97 return RecommendationView{}, ErrNoDatabase 98 } 99 decision := strings.TrimSpace(req.Decision) 100 if decision != "approved" && decision != "rejected" && decision != "dismissed" { 101 return RecommendationView{}, validationErr("decision", "must be one of approved, rejected, dismissed") 102 } 103 var notePtr *string 104 if n := strings.TrimSpace(req.Note); n != "" { 105 notePtr = &n 106 } 107 var out dbgen.StaffingRecommendation 108 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 109 q := dbgen.New(tx) 110 cur, e := q.GetStaffingRecommendationForUpdate(ctx, recID) // row lock 111 if errors.Is(e, pgx.ErrNoRows) { 112 return ErrNotFound 113 } 114 if e != nil { 115 return e 116 } 117 if cur.Status != "pending" { 118 return ErrConflict // immutable once decided; superseded rows are dead 119 } 120 row, e := q.MarkRecommendationDecision(ctx, dbgen.MarkRecommendationDecisionParams{ 121 ID: recID, Status: decision, DecidedBy: actorID(ctx), DecisionNote: notePtr, 122 }) 123 if errors.Is(e, pgx.ErrNoRows) { 124 return ErrConflict // raced between the lock and the guarded update 125 } 126 if e != nil { 127 return e 128 } 129 out = row 130 // The note may carry candidate PII, so only the decision, the action, and a has_note flag are audited. 131 payload, _ := json.Marshal(map[string]any{"decision": decision, "action": row.Action, "has_note": notePtr != nil}) 132 return s.auditOnTx(ctx, tx, audit.Event{Action: "aedt.staffing.decided", EntityType: "staffing_recommendation", EntityID: recID.String(), NewValues: payload}) 133 }) 134 if err != nil { 135 return RecommendationView{}, err 136 } 137 return recommendationView(out), nil 138} 139 140// ListProactive returns the current pending proactive recommendations (the bench watch), capped and 141// paginated. It reads the rows the scheduled recompute wrote; it does not generate. 142func (s *Service) ListProactive(ctx context.Context, limit, offset int) (ProactiveResponse, error) { 143 if s.db == nil { 144 return ProactiveResponse{}, ErrNoDatabase 145 } 146 if limit <= 0 || limit > s.cfg.ReadPageMax { 147 limit = s.cfg.ReadPageMax 148 } 149 if offset < 0 { 150 offset = 0 151 } 152 var rows []dbgen.StaffingRecommendation 153 err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 154 var e error 155 rows, e = dbgen.New(tx).ListProactiveRecommendations(ctx, dbgen.ListProactiveRecommendationsParams{Limit: int32(limit), Offset: int32(offset)}) 156 return e 157 }) 158 if err != nil { 159 return ProactiveResponse{}, err 160 } 161 return ProactiveResponse{Recommendations: recommendationViews(rows), Limit: limit, Offset: offset}, nil 162} 163 164// ---------- shared building blocks for the views ---------- 165 166// commitments captures who is committed where, read from approved slate members (the source of truth) and 167// approved provisional W3 assigns. 168type commitments struct { 169 byCand map[string]string // candidate id -> committed requirement id (slate preferred over provisional) 170 countByReq map[string]int // requirement id -> number of candidates committed to it 171} 172 173func buildCommitments(rows []dbgen.ListCommitmentsRow) commitments { 174 c := commitments{byCand: map[string]string{}, countByReq: map[string]int{}} 175 src := map[string]string{} 176 for _, r := range rows { 177 cand := r.CandidateID.String() 178 if existing, ok := c.byCand[cand]; ok { 179 if src[cand] == "slate" || r.Source != "slate" { 180 _ = existing 181 continue // keep the slate commitment, ignore a later provisional one 182 } 183 } 184 c.byCand[cand] = r.RequirementID.String() 185 src[cand] = r.Source 186 } 187 for _, req := range c.byCand { 188 c.countByReq[req]++ 189 } 190 return c 191} 192 193// buildPriorities maps each open requirement id to its priority multiplier. An unknown opportunity value 194// is imputed with the median of the known values (a documented neutral fallback, never zero, which would 195// deprioritize every gap with un ingested dollars); if no value is known at all, every requirement gets 196// the neutral minimum. The applied value source is returned for the run parameters. 197func buildPriorities(cfg Config, rows []dbgen.ListOpenRequirementsWithValueRow) (map[string]decimal.Decimal, string) { 198 var known []decimal.Decimal 199 for _, r := range rows { 200 if r.OpportunityValue != nil { 201 known = append(known, *r.OpportunityValue) 202 } 203 } 204 median := medianDecimal(known) 205 usedFallback := false 206 pr := make(map[string]decimal.Decimal, len(rows)) 207 for _, r := range rows { 208 v := median 209 if r.OpportunityValue != nil { 210 v = *r.OpportunityValue 211 } else if len(known) > 0 { 212 usedFallback = true 213 } 214 pr[r.ID.String()] = PriorityMultiplier(v, cfg.ValueReference, cfg.PriorityMin, cfg.PriorityMax) 215 } 216 switch { 217 case len(known) == 0: 218 return pr, "neutral" 219 case usedFallback: 220 return pr, "median_fallback" 221 default: 222 return pr, "opportunity_value" 223 } 224} 225 226func medianDecimal(vals []decimal.Decimal) decimal.Decimal { 227 if len(vals) == 0 { 228 return decimal.Zero 229 } 230 s := append([]decimal.Decimal(nil), vals...) 231 sort.Slice(s, func(i, j int) bool { return s[i].LessThan(s[j]) }) 232 n := len(s) 233 if n%2 == 1 { 234 return s[n/2] 235 } 236 return s[n/2-1].Add(s[n/2]).Div(decimal.NewFromInt(2)) 237} 238 239func completeness(ec engine.Candidate) decimal.Decimal { 240 present := 0 241 if len(ec.Skills) > 0 { 242 present++ 243 } 244 if len(ec.Certifications) > 0 { 245 present++ 246 } 247 if len(ec.CustomerExperience) > 0 { 248 present++ 249 } 250 if ec.Location != "" { 251 present++ 252 } 253 return decimal.NewFromInt(int64(present)).Div(decimal.NewFromInt(4)) 254} 255 256// scoreBench scores every supplied candidate against one requirement through the one engine path and 257// builds the classifier inputs (commitment, provenance, completeness, and, when withReskill, the adjacency 258// path for a near miss). It returns the bench plus the frozen results keyed by candidate id (so the caller 259// persists the exact evaluation each recommendation references). 260func (s *Service) scoreBench(ctx context.Context, tx pgx.Tx, er engine.Requirement, gapPriorityUnused decimal.Decimal, cands []dbgen.Candidate, commit commitments, priorities map[string]decimal.Decimal, withReskill bool, cfg Config, asOf time.Time) ([]ScoredCandidate, map[string]engine.MatchResult, error) { 261 q := dbgen.New(tx) 262 adjCache := map[string][]ai.AdjacentSkill{} 263 bench := make([]ScoredCandidate, 0, len(cands)) 264 resByCand := make(map[string]engine.MatchResult, len(cands)) 265 for _, c := range cands { 266 ec, err := matching.LoadEngineCandidate(ctx, q, c, asOf) 267 if err != nil { 268 return nil, nil, err 269 } 270 res := engine.ScoreCandidate(ctx, ec, er, s.matcher) 271 cid := c.ID.String() 272 resByCand[cid] = res 273 sc := ScoredCandidate{ 274 CandidateID: cid, 275 Available: ec.Available, 276 Total: res.Total, 277 Band: res.Band, 278 ClearanceSatisfied: res.Clearance.Equal(decimal.NewFromInt(100)), 279 Provenance: c.ClearanceProvenance, 280 Completeness: completeness(ec), 281 } 282 if committedReq, ok := commit.byCand[cid]; ok { 283 sc.CommittedRequirementID = committedReq 284 if p, ok := priorities[committedReq]; ok { 285 sc.CommittedPriority = p 286 } else { 287 sc.CommittedPriority = cfg.PriorityMin 288 } 289 } 290 if withReskill && res.Total.LessThan(cfg.StaffingFloor) && sc.ClearanceSatisfied && ec.Available { 291 sc.Reskill = s.findReskill(ctx, tx, res, ec, cfg, adjCache) 292 } 293 bench = append(bench, sc) 294 } 295 return bench, resByCand, nil 296} 297 298// findReskill checks whether a near miss holds a skill adjacent to a missing required skill, using the live 299// KAG skill adjacency (which is looser than the engine's own semantic match threshold, so this is the 300// "could be trained to fit" zone). Adjacency unavailability is not an error: it just means no reskill path. 301func (s *Service) findReskill(ctx context.Context, tx pgx.Tx, res engine.MatchResult, ec engine.Candidate, cfg Config, cache map[string][]ai.AdjacentSkill) *ReskillFact { 302 held := make(map[string]bool, len(ec.Skills)) 303 for _, sk := range ec.Skills { 304 held[sk.Name] = true 305 } 306 for _, m := range res.MatchedSkills { 307 if m.Matched { 308 continue 309 } 310 name := m.RequiredSkillName 311 adj, ok := cache[name] 312 if !ok { 313 a, err := ai.SkillAdjacencyOnTx(ctx, tx, name, 15) 314 if err != nil { 315 continue 316 } 317 adj = a 318 cache[name] = a 319 } 320 for _, a := range adj { // ordered nearest first 321 d := decimal.NewFromFloat(a.Distance) 322 if d.LessThanOrEqual(cfg.AdjacencyThreshold) && held[a.Name] { 323 return &ReskillFact{HeldSkill: a.Name, MissingSkill: name, Distance: d.Round(4)} 324 } 325 } 326 } 327 return nil 328} 329 330// enrichedRec is a classified recommendation with its narrated rationale, ready to persist. 331type enrichedRec struct { 332 rec Recommendation 333 rationale string 334 source RationaleSource 335} 336 337// enrich narrates each recommendation (faithfulness gated, deterministic fallback) and records the 338// fallback reason in the evidence. It runs OUTSIDE any database transaction so a slow plane never holds a 339// transaction open. 340func (s *Service) enrich(ctx context.Context, recs []Recommendation) []enrichedRec { 341 out := make([]enrichedRec, 0, len(recs)) 342 for _, r := range recs { 343 rationale, source, reason := s.narrate(ctx, r.Action, r.Evidence) 344 if reason != "" { 345 r.Evidence.FaithFallbackReason = reason 346 } 347 out = append(out, enrichedRec{rec: r, rationale: rationale, source: source}) 348 } 349 return out 350} 351 352// persistRun writes one run and its recommendations atomically: it supersedes the prior pending rows of 353// this scope, creates the run, upserts the frozen candidate_evaluations each non source recommendation 354// references (the engine evaluation), inserts the recommendations, audits each served score, and audits the 355// run. It fails closed on any audit error. 356// pairKey identifies one frozen engine result by its (candidate, requirement) pair, so a candidate that 357// appears against several requirements (the proactive view) maps to the correct evaluation each time. 358func pairKey(candidateID, requirementID string) string { return candidateID + "|" + requirementID } 359 360func (s *Service) persistRun(ctx context.Context, tx pgx.Tx, orgID uuid.UUID, view View, scopeRef *uuid.UUID, params []byte, tool string, recs []enrichedRec, resByPair map[string]engine.MatchResult, supersede func(context.Context, *dbgen.Queries) error) (dbgen.RecommendationRun, []dbgen.StaffingRecommendation, error) { 361 q := dbgen.New(tx) 362 if supersede != nil { 363 if err := supersede(ctx, q); err != nil { 364 return dbgen.RecommendationRun{}, nil, err 365 } 366 } 367 run, err := q.CreateRecommendationRun(ctx, dbgen.CreateRecommendationRunParams{ 368 OrganizationID: orgID, View: string(view), ScopeRef: scopeRef, ModelVersion: modelVersion(), Parameters: params, CreatedBy: actorID(ctx), 369 }) 370 if err != nil { 371 return dbgen.RecommendationRun{}, nil, fmt.Errorf("create run: %w", err) 372 } 373 out := make([]dbgen.StaffingRecommendation, 0, len(recs)) 374 for _, er := range recs { 375 reqID, err := uuid.Parse(er.rec.RequirementID) 376 if err != nil { 377 return dbgen.RecommendationRun{}, nil, err 378 } 379 var candPtr *uuid.UUID 380 if er.rec.CandidateID != "" { 381 cid, err := uuid.Parse(er.rec.CandidateID) 382 if err != nil { 383 return dbgen.RecommendationRun{}, nil, err 384 } 385 candPtr = &cid 386 // Persist the frozen evaluation this recommendation references (the FK target) and audit the 387 // served score for the surfaced candidate. 388 res, ok := resByPair[pairKey(er.rec.CandidateID, er.rec.RequirementID)] 389 if !ok { 390 return dbgen.RecommendationRun{}, nil, fmt.Errorf("staffing: missing engine result for candidate %s on requirement %s", er.rec.CandidateID, er.rec.RequirementID) 391 } 392 evalParams, err := matching.ToCandidateEvaluationParams(orgID, cid, reqID, res, actorID(ctx)) 393 if err != nil { 394 return dbgen.RecommendationRun{}, nil, err 395 } 396 if err := q.UpsertCandidateEvaluation(ctx, evalParams); err != nil { 397 return dbgen.RecommendationRun{}, nil, fmt.Errorf("upsert evaluation: %w", err) 398 } 399 if err := s.auditOnTx(ctx, tx, servedScoreEvent(tool, cid, res)); err != nil { 400 return dbgen.RecommendationRun{}, nil, err 401 } 402 } 403 evidence, err := json.Marshal(er.rec.Evidence) 404 if err != nil { 405 return dbgen.RecommendationRun{}, nil, err 406 } 407 row, err := q.CreateStaffingRecommendation(ctx, dbgen.CreateStaffingRecommendationParams{ 408 OrganizationID: orgID, RunID: run.ID, RequirementID: reqID, CandidateID: candPtr, 409 Action: string(er.rec.Action), Rank: int32(er.rec.Rank), Confidence: er.rec.Confidence, 410 ValueScore: er.rec.ValueScore, Rationale: er.rationale, RationaleSource: string(er.source), Evidence: evidence, 411 }) 412 if err != nil { 413 return dbgen.RecommendationRun{}, nil, fmt.Errorf("create recommendation: %w", err) 414 } 415 out = append(out, row) 416 } 417 counts := map[string]int{} 418 for _, er := range recs { 419 counts[string(er.rec.Action)]++ 420 } 421 payload, _ := json.Marshal(map[string]any{"run_id": run.ID.String(), "view": string(view), "actions": counts, "model_version": run.ModelVersion}) 422 if err := s.auditOnTx(ctx, tx, audit.Event{Action: "aedt.staffing.recommended", EntityType: "recommendation_run", EntityID: run.ID.String(), NewValues: payload}); err != nil { 423 return dbgen.RecommendationRun{}, nil, err 424 } 425 return run, out, nil 426} 427 428func runParams(cfg Config, valueSource string) []byte { 429 b, _ := json.Marshal(map[string]any{ 430 "staffing_floor": cfg.StaffingFloor.String(), 431 "adjacency_threshold": cfg.AdjacencyThreshold.String(), 432 "value_reference": cfg.ValueReference.String(), 433 "value_scale_precision": cfg.ValueScalePrecision, 434 "value_source": valueSource, 435 }) 436 return b 437} 438 439// parseOverrides validates the optional per run floor and adjacency overrides, returning a ValidationError 440// (422) with field messages when either is out of range. 441func parseOverrides(floor, adjacency *string) (*decimal.Decimal, *decimal.Decimal, *ValidationError) { 442 fields := map[string]string{} 443 var fp, ap *decimal.Decimal 444 if floor != nil { 445 f, err := decimal.NewFromString(strings.TrimSpace(*floor)) 446 if err != nil || f.LessThan(decimal.Zero) || f.GreaterThan(decimal.NewFromInt(100)) { 447 fields["staffing_floor"] = "must be a decimal in [0,100]" 448 } else { 449 fp = &f 450 } 451 } 452 if adjacency != nil { 453 a, err := decimal.NewFromString(strings.TrimSpace(*adjacency)) 454 if err != nil || a.LessThan(decimal.Zero) || a.GreaterThan(decimal.NewFromInt(2)) { 455 fields["adjacency_threshold"] = "must be a decimal in [0,2]" 456 } else { 457 ap = &a 458 } 459 } 460 if len(fields) > 0 { 461 return nil, nil, &ValidationError{Fields: fields} 462 } 463 return fp, ap, nil 464} 465 466// w3AdvisoryNamespace namespaces the W3 transaction advisory locks so a key cannot collide with another 467// subsystem's advisory lock. 468const w3AdvisoryNamespace uint64 = 0x573341 // "W3A" 469 470// advisoryKey derives a stable lock key from the view, the org, and a scope (a requirement id for a gap, a 471// candidate id for a proactive run, empty for a portfolio). Two writes that supersede and reinsert the same 472// (view, org, scope) serialize; writes to different scopes proceed in parallel. 473func advisoryKey(view View, org uuid.UUID, scope string) int64 { 474 h := fnv.New64a() 475 _, _ = h.Write([]byte(string(view))) 476 _, _ = h.Write([]byte{0}) 477 _, _ = h.Write(org[:]) 478 _, _ = h.Write([]byte{0}) 479 _, _ = h.Write([]byte(scope)) 480 return int64(h.Sum64() ^ w3AdvisoryNamespace) 481} 482 483// acquireAdvisoryLock takes a transaction scoped advisory lock; the second writer on the same key blocks 484// until the first commits, then supersedes and rewrites. The lock auto releases at transaction end. 485func acquireAdvisoryLock(ctx context.Context, tx pgx.Tx, key int64) error { 486 _, err := tx.Exec(ctx, "SELECT pg_advisory_xact_lock($1)", key) 487 return err 488} 489