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

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / internal/modules/staffing/service.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	internal/modules/staffing/service.go
Original SHA-256	ae472fb1cce39cac1806b17fd8156a8ede50371c18e9bfefbb7ac1384d368135
Git blob SHA	7c1827c458e89914b4bc9574d20dc6a7d938680a
Reading-copy SHA-256	ae472fb1cce39cac1806b17fd8156a8ede50371c18e9bfefbb7ac1384d368135
Original source ↗
Every recorded occurrence

These 2 records contain identical original source bytes. Their paths and revisions remain part of the archive.
Original record	Branch or archive	Revision / identifier	Original location
Janady13/missionmatch / internal/modules/staffing/service.go	feature/opportunity-to-outcome	720d47fae476b0d38477ae4401697d372b1ea2bc	Open original
Janady13/missionmatch / internal/modules/staffing/service.go	feature/aioe-upgrade	2178e7298d0e5d9af83afa12402fd587f0ba6ed8	Open original
Recovered source transcript

Download complete source text ↓510 lines · 19.8 KB
package staffing

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"hash/fnv"
	"sort"
	"strings"
	"time"

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5"
	"github.com/shopspring/decimal"

	"github.com/missionworkforce/missionmatch/internal/aioe"
	"github.com/missionworkforce/missionmatch/internal/modules/ai"
	"github.com/missionworkforce/missionmatch/internal/modules/matching"
	"github.com/missionworkforce/missionmatch/internal/modules/matching/engine"
	"github.com/missionworkforce/missionmatch/internal/platform/audit"
	"github.com/missionworkforce/missionmatch/internal/platform/database/dbgen"
	"github.com/missionworkforce/missionmatch/internal/platform/inference"
	"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)

const (
	w3LogicVersion     = "w3.v1"
	benchScanLimit     = 500 // matches the slate loop's candidate scan cap
	openRequirementCap = 500
)

// Service is the W3 application service. It runs every database operation inside a tenant transaction
// (RLS scoped), reuses the one engine.ScoreCandidate path through the matching package, narrates only
// behind the faithfulness gate, and audits every run and every decision.
type Service struct {
	db      *tenant.DB
	matcher engine.SkillMatcher
	aud     *audit.Writer
	infer   *inference.Client // may be nil: then every rationale is the deterministic template
	cfg     Config
	clock   func() time.Time
}

// NewService builds the service. db, aud, and infer may be nil (DB routes then return ErrNoDatabase; a nil
// audit skips audit in tests; a nil inference client means deterministic rationales).
func NewService(db *tenant.DB, matcher engine.SkillMatcher, aud *audit.Writer, infer *inference.Client, cfg Config) *Service {
	return &Service{db: db, matcher: matcher, aud: aud, infer: infer, cfg: cfg, clock: time.Now}
}

func modelVersion() string { return engine.ModelVersion + "+" + w3LogicVersion }

// scorerOnTx mirrors the matching service seam: load the tenant active AIOE weight profile once
// on the tx and score everything in the transaction under it, stamping the profile version. It also
// mirrors its failure posture: an active profile that exists but cannot be read fails the request
// rather than quietly scoring on defaults under the global.default label.
func (s *Service) scorerOnTx(ctx context.Context, tx pgx.Tx) (func(ctx context.Context, ec engine.Candidate, er engine.Requirement) engine.MatchResult, error) {
	w, label, err := aioe.LoadActiveWeightsOnTx(ctx, tx)
	if err != nil {
		return nil, err
	}
	return func(ctx context.Context, ec engine.Candidate, er engine.Requirement) engine.MatchResult {
		res := engine.ScoreCandidateWithWeights(ctx, ec, er, s.matcher, w)
		res.WeightProfileVersion = label
		return res
	}, nil
}

func (s *Service) orgID(ctx context.Context) (uuid.UUID, error) {
	raw, ok := tenant.OrgFromContext(ctx)
	if !ok {
		return uuid.Nil, tenant.ErrNoTenant
	}
	id, err := uuid.Parse(raw)
	if err != nil {
		return uuid.Nil, fmt.Errorf("staffing: invalid org id in context: %w", err)
	}
	return id, nil
}

func actorID(ctx context.Context) *uuid.UUID {
	if p, ok := tenant.UserFromContext(ctx); ok {
		if id, err := uuid.Parse(p.UserID); err == nil {
			return &id
		}
	}
	return nil
}

// auditOnTx records an AEDT event on the caller's tx and fails closed (a recommendation or decision that
// cannot be audited rolls back rather than being served unaudited). Demographics are never carried.
func (s *Service) auditOnTx(ctx context.Context, tx pgx.Tx, e audit.Event) error {
	if s.aud == nil {
		return nil
	}
	e.ActorUserID = actorID(ctx)
	return s.aud.OnTx(ctx, tx, e)
}

// servedScoreEvent is the screening stage AEDT record for a candidate surfaced in a recommendation, so the
// four fifths screening analysis stays complete (the same posture as screenMATCH and the W2 fix). It
// carries the candidate, the tool, the score, and the band, and NOTHING about demographics.
func servedScoreEvent(tool string, candidateID uuid.UUID, res engine.MatchResult) audit.Event {
	payload, _ := json.Marshal(map[string]any{"tool": tool, "score": res.Total.StringFixed(2), "band": res.Band.Code()})
	return audit.Event{Action: "aedt.score.served", EntityType: "candidate", EntityID: candidateID.String(), NewValues: payload}
}

// Decide records a human approve, reject, or dismiss on one recommendation. It locks the row, so two
// concurrent decisions serialize; a row that is not pending (already decided, or concurrently superseded)
// returns ErrConflict (409) rather than overwriting a decided or dead row. Rejecting only dismisses the
// suggestion; it never blacklists a candidate, and nothing here is automatic.
func (s *Service) Decide(ctx context.Context, recID uuid.UUID, req DecisionRequest) (RecommendationView, error) {
	if s.db == nil {
		return RecommendationView{}, ErrNoDatabase
	}
	decision := strings.TrimSpace(req.Decision)
	if decision != "approved" && decision != "rejected" && decision != "dismissed" {
		return RecommendationView{}, validationErr("decision", "must be one of approved, rejected, dismissed")
	}
	var notePtr *string
	if n := strings.TrimSpace(req.Note); n != "" {
		notePtr = &n
	}
	var out dbgen.StaffingRecommendation
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		q := dbgen.New(tx)
		cur, e := q.GetStaffingRecommendationForUpdate(ctx, recID) // row lock
		if errors.Is(e, pgx.ErrNoRows) {
			return ErrNotFound
		}
		if e != nil {
			return e
		}
		if cur.Status != "pending" {
			return ErrConflict // immutable once decided; superseded rows are dead
		}
		row, e := q.MarkRecommendationDecision(ctx, dbgen.MarkRecommendationDecisionParams{
			ID: recID, Status: decision, DecidedBy: actorID(ctx), DecisionNote: notePtr,
		})
		if errors.Is(e, pgx.ErrNoRows) {
			return ErrConflict // raced between the lock and the guarded update
		}
		if e != nil {
			return e
		}
		out = row
		// The note may carry candidate PII, so only the decision, the action, and a has_note flag are audited.
		payload, _ := json.Marshal(map[string]any{"decision": decision, "action": row.Action, "has_note": notePtr != nil})
		return s.auditOnTx(ctx, tx, audit.Event{Action: "aedt.staffing.decided", EntityType: "staffing_recommendation", EntityID: recID.String(), NewValues: payload})
	})
	if err != nil {
		return RecommendationView{}, err
	}
	return recommendationView(out), nil
}

// ListProactive returns the current pending proactive recommendations (the bench watch), capped and
// paginated. It reads the rows the scheduled recompute wrote; it does not generate.
func (s *Service) ListProactive(ctx context.Context, limit, offset int) (ProactiveResponse, error) {
	if s.db == nil {
		return ProactiveResponse{}, ErrNoDatabase
	}
	if limit <= 0 || limit > s.cfg.ReadPageMax {
		limit = s.cfg.ReadPageMax
	}
	if offset < 0 {
		offset = 0
	}
	var rows []dbgen.StaffingRecommendation
	err := s.db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error {
		var e error
		rows, e = dbgen.New(tx).ListProactiveRecommendations(ctx, dbgen.ListProactiveRecommendationsParams{Limit: int32(limit), Offset: int32(offset)})
		return e
	})
	if err != nil {
		return ProactiveResponse{}, err
	}
	return ProactiveResponse{Recommendations: recommendationViews(rows), Limit: limit, Offset: offset}, nil
}

// ---------- shared building blocks for the views ----------

// commitments captures who is committed where, read from approved slate members (the source of truth) and
// approved provisional W3 assigns.
type commitments struct {
	byCand     map[string]string // candidate id -> committed requirement id (slate preferred over provisional)
	countByReq map[string]int    // requirement id -> number of candidates committed to it
}

func buildCommitments(rows []dbgen.ListCommitmentsRow) commitments {
	c := commitments{byCand: map[string]string{}, countByReq: map[string]int{}}
	src := map[string]string{}
	for _, r := range rows {
		cand := r.CandidateID.String()
		if existing, ok := c.byCand[cand]; ok {
			if src[cand] == "slate" || r.Source != "slate" {
				_ = existing
				continue // keep the slate commitment, ignore a later provisional one
			}
		}
		c.byCand[cand] = r.RequirementID.String()
		src[cand] = r.Source
	}
	for _, req := range c.byCand {
		c.countByReq[req]++
	}
	return c
}

// buildPriorities maps each open requirement id to its priority multiplier. An unknown opportunity value
// is imputed with the median of the known values (a documented neutral fallback, never zero, which would
// deprioritize every gap with un ingested dollars); if no value is known at all, every requirement gets
// the neutral minimum. The applied value source is returned for the run parameters.
func buildPriorities(cfg Config, rows []dbgen.ListOpenRequirementsWithValueRow) (map[string]decimal.Decimal, string) {
	var known []decimal.Decimal
	for _, r := range rows {
		if r.OpportunityValue != nil {
			known = append(known, *r.OpportunityValue)
		}
	}
	median := medianDecimal(known)
	usedFallback := false
	pr := make(map[string]decimal.Decimal, len(rows))
	for _, r := range rows {
		v := median
		if r.OpportunityValue != nil {
			v = *r.OpportunityValue
		} else if len(known) > 0 {
			usedFallback = true
		}
		pr[r.ID.String()] = PriorityMultiplier(v, cfg.ValueReference, cfg.PriorityMin, cfg.PriorityMax)
	}
	switch {
	case len(known) == 0:
		return pr, "neutral"
	case usedFallback:
		return pr, "median_fallback"
	default:
		return pr, "opportunity_value"
	}
}

func medianDecimal(vals []decimal.Decimal) decimal.Decimal {
	if len(vals) == 0 {
		return decimal.Zero
	}
	s := append([]decimal.Decimal(nil), vals...)
	sort.Slice(s, func(i, j int) bool { return s[i].LessThan(s[j]) })
	n := len(s)
	if n%2 == 1 {
		return s[n/2]
	}
	return s[n/2-1].Add(s[n/2]).Div(decimal.NewFromInt(2))
}

func completeness(ec engine.Candidate) decimal.Decimal {
	present := 0
	if len(ec.Skills) > 0 {
		present++
	}
	if len(ec.Certifications) > 0 {
		present++
	}
	if len(ec.CustomerExperience) > 0 {
		present++
	}
	if ec.Location != "" {
		present++
	}
	return decimal.NewFromInt(int64(present)).Div(decimal.NewFromInt(4))
}

// scoreBench scores every supplied candidate against one requirement through the one engine path and
// builds the classifier inputs (commitment, provenance, completeness, and, when withReskill, the adjacency
// path for a near miss). It returns the bench plus the frozen results keyed by candidate id (so the caller
// persists the exact evaluation each recommendation references).
func (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) {
	q := dbgen.New(tx)
	adjCache := map[string][]ai.AdjacentSkill{}
	bench := make([]ScoredCandidate, 0, len(cands))
	score, serr := s.scorerOnTx(ctx, tx)
	if serr != nil {
		return nil, nil, serr
	}
	resByCand := make(map[string]engine.MatchResult, len(cands))
	for _, c := range cands {
		ec, err := matching.LoadEngineCandidate(ctx, q, c, asOf)
		if err != nil {
			return nil, nil, err
		}
		res := score(ctx, ec, er)
		cid := c.ID.String()
		resByCand[cid] = res
		sc := ScoredCandidate{
			CandidateID:        cid,
			Available:          ec.Available,
			Total:              res.Total,
			Band:               res.Band,
			ClearanceSatisfied: res.Clearance.Equal(decimal.NewFromInt(100)),
			Provenance:         c.ClearanceProvenance,
			Completeness:       completeness(ec),
		}
		if committedReq, ok := commit.byCand[cid]; ok {
			sc.CommittedRequirementID = committedReq
			if p, ok := priorities[committedReq]; ok {
				sc.CommittedPriority = p
			} else {
				sc.CommittedPriority = cfg.PriorityMin
			}
		}
		if withReskill && res.Total.LessThan(cfg.StaffingFloor) && sc.ClearanceSatisfied && ec.Available {
			sc.Reskill = s.findReskill(ctx, tx, res, ec, cfg, adjCache)
		}
		bench = append(bench, sc)
	}
	return bench, resByCand, nil
}

// findReskill checks whether a near miss holds a skill adjacent to a missing required skill, using the live
// KAG skill adjacency (which is looser than the engine's own semantic match threshold, so this is the
// "could be trained to fit" zone). Adjacency unavailability is not an error: it just means no reskill path.
func (s *Service) findReskill(ctx context.Context, tx pgx.Tx, res engine.MatchResult, ec engine.Candidate, cfg Config, cache map[string][]ai.AdjacentSkill) *ReskillFact {
	held := make(map[string]bool, len(ec.Skills))
	for _, sk := range ec.Skills {
		held[sk.Name] = true
	}
	for _, m := range res.MatchedSkills {
		if m.Matched {
			continue
		}
		name := m.RequiredSkillName
		adj, ok := cache[name]
		if !ok {
			a, err := ai.SkillAdjacencyOnTx(ctx, tx, name, 15)
			if err != nil {
				continue
			}
			adj = a
			cache[name] = a
		}
		for _, a := range adj { // ordered nearest first
			d := decimal.NewFromFloat(a.Distance)
			if d.LessThanOrEqual(cfg.AdjacencyThreshold) && held[a.Name] {
				return &ReskillFact{HeldSkill: a.Name, MissingSkill: name, Distance: d.Round(4)}
			}
		}
	}
	return nil
}

// enrichedRec is a classified recommendation with its narrated rationale, ready to persist.
type enrichedRec struct {
	rec       Recommendation
	rationale string
	source    RationaleSource
}

// enrich narrates each recommendation (faithfulness gated, deterministic fallback) and records the
// fallback reason in the evidence. It runs OUTSIDE any database transaction so a slow plane never holds a
// transaction open.
func (s *Service) enrich(ctx context.Context, recs []Recommendation) []enrichedRec {
	out := make([]enrichedRec, 0, len(recs))
	for _, r := range recs {
		rationale, source, reason := s.narrate(ctx, r.Action, r.Evidence)
		if reason != "" {
			r.Evidence.FaithFallbackReason = reason
		}
		out = append(out, enrichedRec{rec: r, rationale: rationale, source: source})
	}
	return out
}

// persistRun writes one run and its recommendations atomically: it supersedes the prior pending rows of
// this scope, creates the run, upserts the frozen candidate_evaluations each non source recommendation
// references (the engine evaluation), inserts the recommendations, audits each served score, and audits the
// run. It fails closed on any audit error.
// pairKey identifies one frozen engine result by its (candidate, requirement) pair, so a candidate that
// appears against several requirements (the proactive view) maps to the correct evaluation each time.
func pairKey(candidateID, requirementID string) string { return candidateID + "|" + requirementID }

func (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) {
	q := dbgen.New(tx)
	if supersede != nil {
		if err := supersede(ctx, q); err != nil {
			return dbgen.RecommendationRun{}, nil, err
		}
	}
	run, err := q.CreateRecommendationRun(ctx, dbgen.CreateRecommendationRunParams{
		OrganizationID: orgID, View: string(view), ScopeRef: scopeRef, ModelVersion: modelVersion(), Parameters: params, CreatedBy: actorID(ctx),
	})
	if err != nil {
		return dbgen.RecommendationRun{}, nil, fmt.Errorf("create run: %w", err)
	}
	out := make([]dbgen.StaffingRecommendation, 0, len(recs))
	for _, er := range recs {
		reqID, err := uuid.Parse(er.rec.RequirementID)
		if err != nil {
			return dbgen.RecommendationRun{}, nil, err
		}
		var candPtr *uuid.UUID
		if er.rec.CandidateID != "" {
			cid, err := uuid.Parse(er.rec.CandidateID)
			if err != nil {
				return dbgen.RecommendationRun{}, nil, err
			}
			candPtr = &cid
			// Persist the frozen evaluation this recommendation references (the FK target) and audit the
			// served score for the surfaced candidate.
			res, ok := resByPair[pairKey(er.rec.CandidateID, er.rec.RequirementID)]
			if !ok {
				return dbgen.RecommendationRun{}, nil, fmt.Errorf("staffing: missing engine result for candidate %s on requirement %s", er.rec.CandidateID, er.rec.RequirementID)
			}
			evalParams, err := matching.ToCandidateEvaluationParams(orgID, cid, reqID, res, actorID(ctx))
			if err != nil {
				return dbgen.RecommendationRun{}, nil, err
			}
			if err := q.UpsertCandidateEvaluation(ctx, evalParams); err != nil {
				return dbgen.RecommendationRun{}, nil, fmt.Errorf("upsert evaluation: %w", err)
			}
			if err := s.auditOnTx(ctx, tx, servedScoreEvent(tool, cid, res)); err != nil {
				return dbgen.RecommendationRun{}, nil, err
			}
		}
		evidence, err := json.Marshal(er.rec.Evidence)
		if err != nil {
			return dbgen.RecommendationRun{}, nil, err
		}
		row, err := q.CreateStaffingRecommendation(ctx, dbgen.CreateStaffingRecommendationParams{
			OrganizationID: orgID, RunID: run.ID, RequirementID: reqID, CandidateID: candPtr,
			Action: string(er.rec.Action), Rank: int32(er.rec.Rank), Confidence: er.rec.Confidence,
			ValueScore: er.rec.ValueScore, Rationale: er.rationale, RationaleSource: string(er.source), Evidence: evidence,
		})
		if err != nil {
			return dbgen.RecommendationRun{}, nil, fmt.Errorf("create recommendation: %w", err)
		}
		out = append(out, row)
	}
	counts := map[string]int{}
	for _, er := range recs {
		counts[string(er.rec.Action)]++
	}
	payload, _ := json.Marshal(map[string]any{"run_id": run.ID.String(), "view": string(view), "actions": counts, "model_version": run.ModelVersion})
	if err := s.auditOnTx(ctx, tx, audit.Event{Action: "aedt.staffing.recommended", EntityType: "recommendation_run", EntityID: run.ID.String(), NewValues: payload}); err != nil {
		return dbgen.RecommendationRun{}, nil, err
	}
	return run, out, nil
}

func runParams(cfg Config, valueSource string) []byte {
	b, _ := json.Marshal(map[string]any{
		"staffing_floor":        cfg.StaffingFloor.String(),
		"adjacency_threshold":   cfg.AdjacencyThreshold.String(),
		"value_reference":       cfg.ValueReference.String(),
		"value_scale_precision": cfg.ValueScalePrecision,
		"value_source":          valueSource,
	})
	return b
}

// parseOverrides validates the optional per run floor and adjacency overrides, returning a ValidationError
// (422) with field messages when either is out of range.
func parseOverrides(floor, adjacency *string) (*decimal.Decimal, *decimal.Decimal, *ValidationError) {
	fields := map[string]string{}
	var fp, ap *decimal.Decimal
	if floor != nil {
		f, err := decimal.NewFromString(strings.TrimSpace(*floor))
		if err != nil || f.LessThan(decimal.Zero) || f.GreaterThan(decimal.NewFromInt(100)) {
			fields["staffing_floor"] = "must be a decimal in [0,100]"
		} else {
			fp = &f
		}
	}
	if adjacency != nil {
		a, err := decimal.NewFromString(strings.TrimSpace(*adjacency))
		if err != nil || a.LessThan(decimal.Zero) || a.GreaterThan(decimal.NewFromInt(2)) {
			fields["adjacency_threshold"] = "must be a decimal in [0,2]"
		} else {
			ap = &a
		}
	}
	if len(fields) > 0 {
		return nil, nil, &ValidationError{Fields: fields}
	}
	return fp, ap, nil
}

// w3AdvisoryNamespace namespaces the W3 transaction advisory locks so a key cannot collide with another
// subsystem's advisory lock.
const w3AdvisoryNamespace uint64 = 0x573341 // "W3A"

// advisoryKey derives a stable lock key from the view, the org, and a scope (a requirement id for a gap, a
// candidate id for a proactive run, empty for a portfolio). Two writes that supersede and reinsert the same
// (view, org, scope) serialize; writes to different scopes proceed in parallel.
func advisoryKey(view View, org uuid.UUID, scope string) int64 {
	h := fnv.New64a()
	_, _ = h.Write([]byte(string(view)))
	_, _ = h.Write([]byte{0})
	_, _ = h.Write(org[:])
	_, _ = h.Write([]byte{0})
	_, _ = h.Write([]byte(scope))
	return int64(h.Sum64() ^ w3AdvisoryNamespace)
}

// acquireAdvisoryLock takes a transaction scoped advisory lock; the second writer on the same key blocks
// until the first commits, then supersedes and rewrites. The lock auto releases at transaction end.
func acquireAdvisoryLock(ctx context.Context, tx pgx.Tx, key int64) error {
	_, err := tx.Exec(ctx, "SELECT pg_advisory_xact_lock($1)", key)
	return err
}