package staffing import ( "context" "sort" "strings" "time" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/missionworkforce/missionmatch/internal/modules/matching" "github.com/missionworkforce/missionmatch/internal/modules/matching/engine" "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" ) // RecomputeProactive refreshes the proactive bench watch. For each available candidate it scores the open // requirements, keeps the best fits at or above the staffing floor, and writes a fresh proactive run for // that candidate, superseding their prior pending proactive rows. It is the scheduled job behind the host // timer (the same pattern as mm-opp-sync), idempotent, and it writes deterministic rationales (no model // call in the batch, so the job is fast and never floods the inference plane). It returns the number of // candidates that produced at least one fit. // // The freshness of the watch depends on the ingestion timers: a newly posted opportunity only unlocks a // fit once the opportunity and requirement rows exist. The build report notes this dependency. func (s *Service) RecomputeProactive(ctx context.Context) (int, error) { if s.db == nil { return 0, ErrNoDatabase } orgID, err := s.orgID(ctx) if err != nil { return 0, err } cfg := s.cfg asOf := s.clock() type job struct { candID uuid.UUID recs []Recommendation resByPair map[string]engine.MatchResult } var jobs []job var valueSource string err = s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { q := dbgen.New(tx) openRows, e := q.ListOpenRequirementsWithValue(ctx, dbgen.ListOpenRequirementsWithValueParams{Limit: openRequirementCap, Offset: 0}) if e != nil { return e } _, valueSource = buildPriorities(cfg, openRows) // proactive does not weight by dollars, but record the source type reqE struct { id string er engine.Requirement } reqs := make([]reqE, 0, len(openRows)) for _, rr := range openRows { er, e := matching.LoadEngineRequirement(ctx, q, rr.ID) if e != nil { return e } reqs = append(reqs, reqE{id: rr.ID.String(), er: er}) } cands, e := q.ListCandidates(ctx, dbgen.ListCandidatesParams{Limit: benchScanLimit, Offset: 0}) if e != nil { return e } for _, c := range cands { if !c.Available { continue // the bench watch is for candidates available to staff } ec, e := matching.LoadEngineCandidate(ctx, q, c, asOf) if e != nil { return e } if !ec.Available { continue } type fit struct { reqID string res engine.MatchResult } var fits []fit for _, r := range reqs { res := engine.ScoreCandidate(ctx, ec, r.er, s.matcher) if res.Total.GreaterThanOrEqual(cfg.StaffingFloor) { fits = append(fits, fit{reqID: r.id, res: res}) } } if len(fits) == 0 { continue } sort.SliceStable(fits, func(i, j int) bool { if d := fits[i].res.Total.Cmp(fits[j].res.Total); d != 0 { return d > 0 } return fits[i].reqID < fits[j].reqID }) if len(fits) > cfg.ProactiveBestFitLimit { fits = fits[:cfg.ProactiveBestFitLimit] } cid := c.ID.String() prov := c.ClearanceProvenance comp := completeness(ec) recs := make([]Recommendation, 0, len(fits)) rbp := make(map[string]engine.MatchResult, len(fits)) for i, f := range fits { conf := Confidence(cfg, ScoredCandidate{Total: f.res.Total, Provenance: prov, Completeness: comp}) recs = append(recs, Recommendation{ RequirementID: f.reqID, CandidateID: cid, Action: ActionAssign, Rank: i + 1, Confidence: conf, Evidence: Evidence{ EngineTotal: f.res.Total.StringFixed(2), Band: f.res.Band.Code(), Trigger: &TriggerChange{Kind: "recompute", DetectedAt: asOf.Format(time.RFC3339)}, }, }) rbp[pairKey(cid, f.reqID)] = f.res } jobs = append(jobs, job{candID: c.ID, recs: recs, resByPair: rbp}) } return nil }) if err != nil { return 0, err } processed := 0 for _, j := range jobs { enriched := enrichDeterministic(j.recs) scope := j.candID candLock := j.candID e := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { // Serialize the proactive recompute for THIS candidate so overlapping cron invocations cannot // leave two live pending sets for one candidate. if le := acquireAdvisoryLock(ctx, tx, advisoryKey(ViewProactive, orgID, candLock.String())); le != nil { return le } _, _, perr := s.persistRun(ctx, tx, orgID, ViewProactive, &scope, runParams(cfg, valueSource), "staffing.proactive", enriched, j.resByPair, func(ctx context.Context, q *dbgen.Queries) error { cp := candLock return q.SupersedePendingProactive(ctx, &cp) }) return perr }) if e != nil { return processed, e } processed++ } return processed, nil } // enrichDeterministic narrates each recommendation with the deterministic template only, no model call. func enrichDeterministic(recs []Recommendation) []enrichedRec { out := make([]enrichedRec, 0, len(recs)) for _, r := range recs { rationale := strings.Join(evidenceStatements(r.Action, r.Evidence), " ") if rationale == "" { rationale = "Recommended by the deterministic engine." } r.Evidence.FaithFallbackReason = "proactive_batch" out = append(out, enrichedRec{rec: r, rationale: rationale, source: RationaleDeterministic}) } return out }