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

1package ai 2 3// G1 market and relationship graph. This file adds the federal capture intelligence traversals as FIXED, 4// named, parameterized QueryIntents over the GLOBAL public market tables: usaspending_awards (the federal 5// award cache), opportunity_records (the unified opportunity catalog), and sam_opportunities (the SAM 6// solicitation feed with its dotted agency hierarchy). Where the G3 skill graph walks a curated adjacency 7// recursively, the G1 market graph is mostly AGGREGATE reads: the competitor field, the incumbent of 8// record, the recompete pipeline, and the agency buying pattern are all GROUP BY rollups, not recursion, 9// because the data already carries the keys the question needs (the agency hierarchy is a dotted 10// full_parent_path string, so a string split and a GROUP BY answers the buyer question without a recursive 11// CTE, exactly as the determinism mandate prefers). 12// 13// The graph NEVER computes a match score, a band, or any employment decision number. A dollar share, a 14// competitor rank, an estimated value band are MARKET aggregates, structure the capture team reads, and 15// they are explicitly NOT match scores; the frozen engine in internal/modules/matching/engine stays the 16// only scorer and is untouched. The model never authors this SQL: every statement is a static, reviewed 17// constant, registered in the closed Registry, asserted by ValidateRegistry at boot, and re guarded by 18// GuardReadOnlyQuery on every execution. 19// 20// Honesty mandates from the GraphRAG review, encoded in the SQL by construction: 21// - usaspending_awards carries NO psc, NO set_aside, and NO pop_state column, so the awards self join and 22// the competitor field key ONLY on (awarding_agency, naics_code). The finer (agency, naics, psc, 23// set_aside, pop_state) slices come from opportunity_records, which does carry that full tuple, and 24// those edges are labeled opportunity_records sourced so a reader never mistakes an award for a slice it 25// cannot support. 26// - every incumbent edge is labeled prime-of-record (a constant label column), never the staffing 27// subcontractor, because the sub award labor reality is absent from the public award data; the prime 28// recipient is the only honest incumbent the data supports. 29// - the agency hierarchy rollup splits the sam_opportunities full_parent_path dotted string and groups, it 30// does NOT use a recursive CTE, because the hierarchy is already a materialized path and recursion would 31// add nothing but a determinism risk. 32// 33// Decimal, not float. Every dollar figure (award_amount, estimated_value) and every share is Postgres 34// numeric end to end and is scanned into shopspring/decimal in Go, NEVER float64 and NEVER cast through 35// ::float8: a surfaced market figure is a decision adjacent number and the platform's decimal rule applies. 36// The recency weight is a numeric step function (more recent award years weigh more), kept inside the 37// numeric type so the weighted share stays exact. 38 39import ( 40 "context" 41 "errors" 42 43 "github.com/jackc/pgx/v5" 44 "github.com/jackc/pgx/v5/pgtype" 45 "github.com/shopspring/decimal" 46 47 "github.com/missionworkforce/missionmatch/internal/platform/tenant" 48) 49 50// The registered names of the four G1 market intents. Each is a FIXED entry in the closed Registry in 51// tag.go, so ValidateRegistry asserts it at boot and ExecuteIntent runs it through the same guard, tenant 52// transaction, and audit posture as every other TAG intent. 53const ( 54 IntentSegmentCompetitorField = "segment_competitor_field" 55 IntentIncumbentOfRecord = "incumbent_of_record" 56 IntentRecompetePipeline = "recompete_pipeline" 57 IntentAgencyBuyingPattern = "agency_buying_pattern" 58) 59 60// primeOfRecordLabel is the constant incumbent edge label. Every incumbent the G1 graph surfaces is the 61// PRIME recipient of record, never the staffing subcontractor: the public award data carries the prime 62// recipient only, so labeling the edge prime-of-record is the honest statement of what the data supports. 63// It is an identifier constant, so its hyphen is allowed; it never appears in surfaced prose. 64const primeOfRecordLabel = "prime-of-record" 65 66// DefaultRecompeteWindowMonths is the default recompete window when a caller binds a non positive one: the 67// 12 to 24 month horizon the capture pipeline watches. 68const DefaultRecompeteWindowMonths = 24 69 70// maxRecompeteWindowMonths bounds a caller bound window so an oversized one cannot turn the pipeline into an 71// unbounded scan. A five year horizon is the meaningful ceiling for a recompete watch. 72const maxRecompeteWindowMonths = 60 73 74// DefaultCompetitorLookbackYears is the default recency lookback for the competitor field when a caller 75// binds a non positive one: a five year award history is the standard capture baseline. 76const DefaultCompetitorLookbackYears = 5 77 78// maxCompetitorLookbackYears bounds the lookback so an oversized one cannot widen the scan unbounded. 79const maxCompetitorLookbackYears = 15 80 81// segmentCompetitorFieldSQL is the FIXED competitor field aggregate. Given an agency ($1), a NAICS ($2), 82// and a recency lookback in years ($3), it returns, per recipient UEI, the recency weighted dollar share of 83// awards in that (agency, naics) segment, ordered share DESC with a UEI tiebreaker, so a dominant share 84// reads wired and a fragmented field reads open. The honesty mandate is enforced by construction: 85// 86// - the segment key is ONLY (awarding_agency, naics_code), the only keys usaspending_awards carries, never 87// a psc or set_aside or pop_state the table does not have, 88// - is_galapagos labels the tenant's own past performance row, so the capture team sees where it already 89// stands in its own field, 90// - the recency weight is a numeric step function on the award year (the more recent the period of 91// performance end, the heavier the weight), kept in the numeric type so the weighted dollars and the 92// share stay exact decimal, 93// - the share is the recency weighted dollars over the segment total recency weighted dollars, computed in 94// numeric, never float; a window function over the segment supplies the denominator, 95// - the evidence is the contributing award count per recipient, so the narration can cite the specific 96// awards behind a share. 97// 98// Determinism: the ORDER BY ends with the recipient_uei tiebreaker, a total order over the grouped rows, so 99// the same inputs yield byte identical rows. 100const segmentCompetitorFieldSQL = ` 101WITH segment AS ( 102 SELECT 103 a.recipient_uei, 104 a.recipient_name, 105 a.is_galapagos, 106 a.award_amount, 107 CASE 108 WHEN a.pop_end_date >= (now()::date - ($3::int * 365)) THEN 1.00::numeric 109 WHEN a.pop_end_date >= (now()::date - ((2 * $3)::int * 365)) THEN 0.50::numeric 110 ELSE 0.25::numeric 111 END AS recency_weight 112 FROM usaspending_awards a 113 WHERE a.awarding_agency = $1 114 AND a.naics_code = $2 115 AND a.award_amount IS NOT NULL 116), weighted AS ( 117 SELECT 118 recipient_uei, 119 max(recipient_name) AS recipient_name, 120 bool_or(is_galapagos) AS is_galapagos, 121 count(*) AS award_count, 122 sum(award_amount * recency_weight)::numeric AS weighted_dollars 123 FROM segment 124 GROUP BY recipient_uei 125) 126SELECT 127 w.recipient_uei, 128 w.recipient_name, 129 w.is_galapagos, 130 w.award_count, 131 w.weighted_dollars, 132 CASE 133 WHEN sum(w.weighted_dollars) OVER () = 0 THEN 0::numeric 134 ELSE (w.weighted_dollars / sum(w.weighted_dollars) OVER ())::numeric 135 END AS dollar_share 136FROM weighted w 137ORDER BY dollar_share DESC, w.recipient_uei` 138 139// incumbentOfRecordSQL is the FIXED incumbent walk. Given an opportunity_records id ($1), it walks to the 140// predecessor award in usaspending_awards keyed on the SAME (agency, naics) the award carries plus the 141// opportunity's dedupe_key matched against the award's PIID and id, returning the prime recipient UEI 142// labeled prime-of-record with the matched award as evidence. It is a JOIN, not a recursion: the 143// predecessor is one hop away on a shared key, so a recursive CTE would add nothing. 144// 145// The honesty mandate is enforced: the award side keys only on (awarding_agency = o.agency, naics_code = 146// o.naics), the only keys the award carries, and the incumbent edge is the constant prime-of-record label, 147// never the staffing sub. The dedupe_key match is the cross source link the opportunity catalog already 148// computes (the normalized solicitation or award id), tried against the award's PIID and its generated id, 149// so a recompete opportunity finds its own predecessor award. The most recent matching award (by period of 150// performance end) is the incumbent of record; the ORDER BY ends with the award generated_internal_id 151// tiebreaker for a byte stable single row pick. 152const incumbentOfRecordSQL = ` 153SELECT 154 o.id AS opportunity_id, 155 o.title AS opportunity_title, 156 o.agency AS agency, 157 o.naics AS naics, 158 a.recipient_uei AS incumbent_uei, 159 a.recipient_name AS incumbent_name, 160 'prime-of-record' AS incumbent_role, 161 a.generated_internal_id AS award_id, 162 a.award_amount AS award_amount, 163 a.pop_end_date AS award_pop_end_date 164FROM opportunity_records o 165JOIN usaspending_awards a 166 ON a.awarding_agency = o.agency 167 AND a.naics_code = o.naics 168 AND o.dedupe_key <> '' 169 AND (a.award_id = o.dedupe_key OR a.generated_internal_id = o.dedupe_key) 170WHERE o.id = $1 171ORDER BY a.pop_end_date DESC NULLS LAST, a.generated_internal_id` 172 173// recompetePipelineSQL is the FIXED recompete pipeline. Given an agency ($1) and a window in months ($2), 174// it returns the opportunity_records and the usaspending_awards (priced or not) whose pop_end_date falls 175// inside the forward window, the takeaway pipeline a capture team watches, each row carrying its source so an award 176// row and an opportunity row are never conflated. The two arms are UNIONed and the result is ordered by 177// pop_end_date with a record id tiebreaker, a total order for byte stable output. 178// 179// Both arms key the agency the same honest way: opportunity_records on its agency column, awards on 180// awarding_agency. The award arm labels the incumbent prime-of-record (the only incumbent the data 181// supports); the opportunity arm carries its own incumbent_uei when the catalog recorded one. estimated_value 182// and award_amount are numeric, scanned into decimal, never float. 183const recompetePipelineSQL = ` 184SELECT 185 'opportunity_records' AS source_table, 186 o.id::text AS record_id, 187 o.title AS title, 188 o.agency AS agency, 189 o.naics AS naics, 190 o.incumbent_uei AS incumbent_uei, 191 o.estimated_value AS dollars, 192 o.value_source AS value_source, 193 o.pop_end_date AS pop_end_date 194FROM opportunity_records o 195WHERE o.agency = $1 196 AND o.pop_end_date IS NOT NULL 197 AND o.pop_end_date >= now()::date 198 AND o.pop_end_date <= (now()::date + ($2::int * 30)) 199UNION 200SELECT 201 'usaspending_awards' AS source_table, 202 a.generated_internal_id AS record_id, 203 a.description AS title, 204 a.awarding_agency AS agency, 205 a.naics_code AS naics, 206 a.recipient_uei AS incumbent_uei, 207 a.award_amount AS dollars, 208 a.source AS value_source, 209 a.pop_end_date AS pop_end_date 210FROM usaspending_awards a 211WHERE a.awarding_agency = $1 212 AND a.pop_end_date IS NOT NULL 213 AND a.pop_end_date >= now()::date 214 AND a.pop_end_date <= (now()::date + ($2::int * 30)) 215ORDER BY pop_end_date, record_id` 216 217// agencyBuyingPatternSQL is the FIXED buyer pattern. Given an agency or department prefix ($1) and an 218// optional NAICS filter ($2, empty string means every NAICS), it returns the NAICS and PSC distribution, the 219// set_aside mix, and the estimated_value band for that buyer, the where to position view. The agency match 220// uses the sam_opportunities full_parent_path DOTTED string: the buyer is matched when the first segment of 221// the dotted path (the department, split on the first dot) equals the prefix OR the prefix is a leading 222// segment of the path, so a department prefix rolls up all its sub agencies WITHOUT a recursive CTE. 223// 224// The distribution comes from opportunity_records (which carries the full naics, psc, set_aside, 225// estimated_value tuple) joined to the agencies the path rollup identified, so the slice is honestly 226// opportunity_records sourced. estimated_value is numeric; the band thresholds are numeric literals, so the 227// banding stays exact decimal. The result groups by (naics, psc, set_aside, value_band) and orders by the 228// opportunity count DESC with a total order tiebreaker on the grouped keys. 229const agencyBuyingPatternSQL = ` 230WITH agency_match AS ( 231 SELECT DISTINCT s.department, s.sub_agency 232 FROM sam_opportunities s 233 WHERE split_part(s.full_parent_path, '.', 1) = $1 234 OR s.full_parent_path = $1 235 OR s.full_parent_path LIKE ($1 || '.%') 236), banded AS ( 237 SELECT 238 o.naics, 239 o.psc, 240 o.set_aside, 241 o.estimated_value, 242 CASE 243 WHEN o.estimated_value IS NULL THEN 'unpriced' 244 WHEN o.estimated_value < 1000000::numeric THEN 'under_1m' 245 WHEN o.estimated_value < 10000000::numeric THEN '1m_to_10m' 246 WHEN o.estimated_value < 100000000::numeric THEN '10m_to_100m' 247 ELSE 'over_100m' 248 END AS value_band 249 FROM opportunity_records o 250 WHERE ($2 = '' OR o.naics = $2) 251 AND ( 252 o.agency = $1 253 OR EXISTS ( 254 SELECT 1 FROM agency_match am 255 WHERE am.department = o.agency OR am.sub_agency = o.agency 256 ) 257 ) 258) 259SELECT 260 b.naics, 261 b.psc, 262 b.set_aside, 263 b.value_band, 264 count(*) AS opportunity_count, 265 sum(coalesce(b.estimated_value, 0))::numeric AS banded_value 266FROM banded b 267GROUP BY b.naics, b.psc, b.set_aside, b.value_band 268ORDER BY opportunity_count DESC, b.naics, b.psc, b.set_aside, b.value_band` 269 270// ClampRecompeteWindowMonths normalizes a requested recompete window to the bounded range: a non positive or 271// missing window becomes the default 24 months, and a window above the ceiling is capped, so the pipeline 272// scan is always bounded. 273func ClampRecompeteWindowMonths(months int) int { 274 if months <= 0 { 275 return DefaultRecompeteWindowMonths 276 } 277 if months > maxRecompeteWindowMonths { 278 return maxRecompeteWindowMonths 279 } 280 return months 281} 282 283// ClampCompetitorLookbackYears normalizes a requested competitor lookback to the bounded range. 284func ClampCompetitorLookbackYears(years int) int { 285 if years <= 0 { 286 return DefaultCompetitorLookbackYears 287 } 288 if years > maxCompetitorLookbackYears { 289 return maxCompetitorLookbackYears 290 } 291 return years 292} 293 294// CompetitorRow is one recipient in a competitor field: the recipient, whether it is the tenant's own past 295// performance (is_galapagos), the count of contributing awards (the evidence), the recency weighted dollars, 296// and the recency weighted dollar share, both decimal, never float. It is pure market structure: there is no 297// match score and no employment decision number on it. 298type CompetitorRow struct { 299 RecipientUEI string `json:"recipient_uei"` 300 RecipientName string `json:"recipient_name"` 301 IsGalapagos bool `json:"is_galapagos"` 302 AwardCount int `json:"award_count"` 303 WeightedDollars decimal.Decimal `json:"weighted_dollars"` 304 DollarShare decimal.Decimal `json:"dollar_share"` 305} 306 307// SegmentCompetitorField runs the registered segment_competitor_field intent and scans the result into typed 308// rows, carrying the weighted dollars and the share as decimal.Decimal (the numeric columns are scanned 309// directly into decimal, never through float64). The agency and naics key the (agency, naics) segment the 310// only way usaspending_awards supports; the lookback is clamped to the bounded range. The generic 311// ExecuteIntent path remains available to the narration layer. 312func SegmentCompetitorField(ctx context.Context, db *tenant.DB, agency, naics string, lookbackYears int) ([]CompetitorRow, error) { 313 if db == nil { 314 return nil, errors.New("ai: no database") 315 } 316 var out []CompetitorRow 317 err := db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 318 if err := GuardReadOnlyQuery(segmentCompetitorFieldSQL, ReadAllowlist); err != nil { 319 return err 320 } 321 rows, err := tx.Query(ctx, segmentCompetitorFieldSQL, agency, naics, ClampCompetitorLookbackYears(lookbackYears)) 322 if err != nil { 323 return err 324 } 325 defer rows.Close() 326 for rows.Next() { 327 var r CompetitorRow 328 if err := rows.Scan(&r.RecipientUEI, &r.RecipientName, &r.IsGalapagos, &r.AwardCount, &r.WeightedDollars, &r.DollarShare); err != nil { 329 return err 330 } 331 out = append(out, r) 332 } 333 return rows.Err() 334 }) 335 return out, err 336} 337 338// IncumbentEdge is the predecessor award an opportunity recompetes against: the opportunity, the matched 339// award, and the prime recipient UEI labeled prime-of-record (the only incumbent the public award data 340// supports). award_amount is decimal, never float. It carries no score and no employment decision number. 341type IncumbentEdge struct { 342 OpportunityID string `json:"opportunity_id"` 343 OpportunityTitle string `json:"opportunity_title"` 344 Agency string `json:"agency"` 345 NAICS string `json:"naics"` 346 IncumbentUEI string `json:"incumbent_uei"` 347 IncumbentName string `json:"incumbent_name"` 348 IncumbentRole string `json:"incumbent_role"` 349 AwardID string `json:"award_id"` 350 AwardAmount *decimal.Decimal `json:"award_amount,omitempty"` 351 AwardPopEndDate string `json:"award_pop_end_date,omitempty"` 352} 353 354// IncumbentOfRecord runs the registered incumbent_of_record intent and scans the matched predecessor awards 355// into typed edges. Every returned edge is labeled prime-of-record by construction (the SQL emits the 356// constant label), so a caller never mistakes the prime recipient for the staffing subcontractor the data 357// does not carry. award_amount is scanned into decimal. 358func IncumbentOfRecord(ctx context.Context, db *tenant.DB, opportunityID string) ([]IncumbentEdge, error) { 359 if db == nil { 360 return nil, errors.New("ai: no database") 361 } 362 var out []IncumbentEdge 363 err := db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 364 if err := GuardReadOnlyQuery(incumbentOfRecordSQL, ReadAllowlist); err != nil { 365 return err 366 } 367 rows, err := tx.Query(ctx, incumbentOfRecordSQL, opportunityID) 368 if err != nil { 369 return err 370 } 371 defer rows.Close() 372 for rows.Next() { 373 var ( 374 e IncumbentEdge 375 popEnd pgtype.Date 376 ) 377 if err := rows.Scan(&e.OpportunityID, &e.OpportunityTitle, &e.Agency, &e.NAICS, 378 &e.IncumbentUEI, &e.IncumbentName, &e.IncumbentRole, &e.AwardID, &e.AwardAmount, &popEnd); err != nil { 379 return err 380 } 381 e.AwardPopEndDate = isoDate(popEnd) 382 out = append(out, e) 383 } 384 return rows.Err() 385 }) 386 return out, err 387} 388 389// PipelineRow is one recompete the capture team watches: the source table it came from (so an award is never 390// conflated with an opportunity), the record, the agency, the incumbent UEI, the dollars (decimal, never 391// float), and the period of performance end that puts it in the window. It carries no score. 392type PipelineRow struct { 393 SourceTable string `json:"source_table"` 394 RecordID string `json:"record_id"` 395 Title string `json:"title"` 396 Agency string `json:"agency"` 397 NAICS string `json:"naics"` 398 IncumbentUEI string `json:"incumbent_uei"` 399 Dollars *decimal.Decimal `json:"dollars,omitempty"` 400 ValueSource string `json:"value_source"` 401 PopEndDate string `json:"pop_end_date"` 402} 403 404// RecompetePipeline runs the registered recompete_pipeline intent for an agency and a window in months, 405// scanning the dollars into decimal. The window is clamped to the bounded range so the scan is always 406// bounded. 407func RecompetePipeline(ctx context.Context, db *tenant.DB, agency string, windowMonths int) ([]PipelineRow, error) { 408 if db == nil { 409 return nil, errors.New("ai: no database") 410 } 411 var out []PipelineRow 412 err := db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 413 if err := GuardReadOnlyQuery(recompetePipelineSQL, ReadAllowlist); err != nil { 414 return err 415 } 416 rows, err := tx.Query(ctx, recompetePipelineSQL, agency, ClampRecompeteWindowMonths(windowMonths)) 417 if err != nil { 418 return err 419 } 420 defer rows.Close() 421 for rows.Next() { 422 var ( 423 r PipelineRow 424 popEnd pgtype.Date 425 ) 426 if err := rows.Scan(&r.SourceTable, &r.RecordID, &r.Title, &r.Agency, &r.NAICS, 427 &r.IncumbentUEI, &r.Dollars, &r.ValueSource, &popEnd); err != nil { 428 return err 429 } 430 r.PopEndDate = isoDate(popEnd) 431 out = append(out, r) 432 } 433 return rows.Err() 434 }) 435 return out, err 436} 437 438// BuyingPatternRow is one slice of an agency's buying pattern: the NAICS, PSC, set_aside, and value band of 439// the buyer's opportunities, the count in that slice, and the banded value (decimal, never float). It is 440// opportunity_records sourced, the only table that carries the full slice tuple. It carries no score. 441type BuyingPatternRow struct { 442 NAICS string `json:"naics"` 443 PSC string `json:"psc"` 444 SetAside string `json:"set_aside"` 445 ValueBand string `json:"value_band"` 446 OpportunityCount int `json:"opportunity_count"` 447 BandedValue decimal.Decimal `json:"banded_value"` 448} 449 450// AgencyBuyingPattern runs the registered agency_buying_pattern intent for an agency or department prefix and 451// an optional NAICS filter, scanning the banded value into decimal. The agency rollup is the full_parent_path 452// string split, NOT a recursive CTE. 453func AgencyBuyingPattern(ctx context.Context, db *tenant.DB, agencyPrefix, naics string) ([]BuyingPatternRow, error) { 454 if db == nil { 455 return nil, errors.New("ai: no database") 456 } 457 var out []BuyingPatternRow 458 err := db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 459 if err := GuardReadOnlyQuery(agencyBuyingPatternSQL, ReadAllowlist); err != nil { 460 return err 461 } 462 rows, err := tx.Query(ctx, agencyBuyingPatternSQL, agencyPrefix, naics) 463 if err != nil { 464 return err 465 } 466 defer rows.Close() 467 for rows.Next() { 468 var r BuyingPatternRow 469 if err := rows.Scan(&r.NAICS, &r.PSC, &r.SetAside, &r.ValueBand, &r.OpportunityCount, &r.BandedValue); err != nil { 470 return err 471 } 472 out = append(out, r) 473 } 474 return rows.Err() 475 }) 476 return out, err 477} 478 479// isoDate renders a nullable Postgres date as a stable ISO 8601 calendar date string, so an evidence date 480// marshals deterministically and a null date renders as the empty string. The dollar columns above scan 481// straight into *decimal.Decimal (the same path the awards repo uses), so no numeric ever passes through 482// float64. 483func isoDate(d pgtype.Date) string { 484 if !d.Valid { 485 return "" 486 } 487 return d.Time.Format("2006-01-02") 488} 489