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

1package ai 2 3// G9 recompete lifecycle and foresight graph. This file adds the federal lifecycle traversals as FIXED, 4// named, parameterized QueryIntents over the GLOBAL public market tables: opportunity_records (the unified 5// catalog whose record_kind partitions a record into forecast, active_solicitation, and contract_award), 6// usaspending_awards (the federal award cache), federal_mission_tags, and mission_areas. Where G1 reads the 7// competitor field and the incumbent of record, G9 reads the lifecycle a single procurement walks through 8// time: the forecast that announces it, the solicitation that opens it, and the predecessor award it 9// recompetes, plus the forward windows a capture team watches (the recompete clock and the forecast 10// runway). 11// 12// The graph NEVER computes a match score, a band, or any employment decision number. A prior award amount, 13// a forecast lead time, a recompete window are MARKET facts, structure the capture team reads, and they are 14// explicitly NOT match scores; the frozen engine in internal/modules/matching/engine stays the only scorer 15// and is untouched. The model never authors this SQL: every statement is a static, reviewed constant, 16// registered in the closed Registry, asserted by ValidateRegistry at boot, and re guarded by 17// GuardReadOnlyQuery on every execution. 18// 19// Honesty mandates from the GraphRAG review, encoded by construction: 20// - the procurement succession walk lives ENTIRELY in opportunity_records, the only table that carries the 21// full (agency, naics, psc, pop_state, set_aside) segment tuple plus the dedupe_key that links the 22// phases, so the succession never reaches for a column usaspending_awards lacks. It is a FIXED set of 23// LEFT JOINs (forecast to solicitation to predecessor award is at most three phases), NOT a recursive 24// CTE, exactly as the determinism mandate prefers when the data already carries the keys. 25// - the recompete clock UNIONs opportunity_records and usaspending_awards; the award arm keys ONLY on 26// awarding_agency (the only agency key the award table carries), each row carrying its source_table so an 27// award row and an opportunity row are never conflated. 28// - every incumbent edge is labeled prime-of-record (a constant label column), never the staffing 29// subcontractor, because the sub award labor reality is absent from the public award data; the prime 30// recipient or the catalog recorded incumbent is the only honest incumbent the data supports. 31// 32// Decimal, not float. Every dollar figure (estimated_value, award_amount) is Postgres numeric end to end and 33// is scanned into shopspring/decimal in Go, NEVER float64 and NEVER cast through ::float8: a surfaced market 34// figure is a decision adjacent number and the platform's decimal rule applies. 35 36import ( 37 "context" 38 "errors" 39 40 "github.com/jackc/pgx/v5" 41 "github.com/jackc/pgx/v5/pgtype" 42 "github.com/shopspring/decimal" 43 44 "github.com/missionworkforce/missionmatch/internal/platform/tenant" 45) 46 47// The registered names of the three G9 lifecycle intents. Each is a FIXED entry in the closed Registry in 48// tag.go, so ValidateRegistry asserts it at boot and ExecuteIntent runs it through the same guard, tenant 49// transaction, and audit posture as every other TAG intent. 50const ( 51 IntentProcurementSuccession = "procurement_succession" 52 IntentRecompeteClock = "recompete_clock" 53 IntentForecastRunway = "forecast_runway" 54) 55 56// procurementSuccessionSQL is the FIXED lifecycle succession. Given an active_solicitation opportunity_records 57// id ($1), it walks the dedupe_key succession within the SAME (agency, naics, psc, pop_state, set_aside) 58// segment to the predecessor contract_award and, when present, the announcing forecast, returning whether the 59// solicitation is new or a recompete, the prior award_amount (the predecessor contract_award estimated_value, 60// decimal) and the prior prime UEI labeled prime-of-record, with the cited record ids and dates. 61// 62// It is a FIXED depth set of LEFT JOINs, NOT a recursive CTE: a procurement passes through at most three 63// phases (forecast, solicitation, award), and every phase carries the same dedupe_key and the same full 64// segment tuple in opportunity_records, so a recursive walk would add nothing but a determinism risk. The 65// whole walk lives in opportunity_records, the only table that carries the full (agency, naics, psc, 66// pop_state, set_aside) tuple, so the succession never reaches for a column usaspending_awards lacks. 67// 68// Honesty and determinism, line by line: 69// - the anchor is the named solicitation; the predecessor award and the forecast are LEFT JOINed on the 70// shared dedupe_key AND the full five tuple, so a missing predecessor leaves is_recompete false (a new 71// start) rather than dropping the row, 72// - the predecessor award join excludes the anchor row itself (a.id <> o.id) so a solicitation never 73// matches itself, and picks the most recent predecessor by award_date with the id tiebreaker, a total 74// order for a byte stable single row, 75// - is_recompete is true exactly when a predecessor contract_award was found, is_new is its negation, 76// - the prior award amount is the predecessor estimated_value (numeric, scanned into decimal, never float), 77// and the prior prime UEI is labeled with the constant prime-of-record. 78const procurementSuccessionSQL = ` 79WITH anchor AS ( 80 SELECT 81 o.id, o.dedupe_key, o.agency, o.naics, o.psc, o.pop_state, o.set_aside, 82 o.title, o.posted_date, o.response_deadline 83 FROM opportunity_records o 84 WHERE o.id = $1 85 AND o.record_kind = 'active_solicitation' 86), predecessor AS ( 87 SELECT 88 a.id, a.incumbent_uei, a.incumbent_name, a.estimated_value, a.award_date 89 FROM opportunity_records a 90 JOIN anchor an 91 ON a.dedupe_key = an.dedupe_key 92 AND an.dedupe_key <> '' 93 AND a.agency = an.agency 94 AND a.naics = an.naics 95 AND a.psc = an.psc 96 AND a.pop_state = an.pop_state 97 AND a.set_aside = an.set_aside 98 AND a.id <> an.id 99 WHERE a.record_kind = 'contract_award' 100 ORDER BY a.award_date DESC NULLS LAST, a.id 101 LIMIT 1 102), forecast AS ( 103 SELECT f.id, f.estimated_award_date 104 FROM opportunity_records f 105 JOIN anchor an 106 ON f.dedupe_key = an.dedupe_key 107 AND an.dedupe_key <> '' 108 AND f.agency = an.agency 109 AND f.naics = an.naics 110 AND f.psc = an.psc 111 AND f.pop_state = an.pop_state 112 AND f.set_aside = an.set_aside 113 AND f.id <> an.id 114 WHERE f.record_kind = 'forecast' 115 ORDER BY f.estimated_award_date DESC NULLS LAST, f.id 116 LIMIT 1 117) 118SELECT 119 an.id::text AS solicitation_id, 120 an.title AS solicitation_title, 121 an.agency AS agency, 122 an.naics AS naics, 123 an.psc AS psc, 124 an.pop_state AS pop_state, 125 an.set_aside AS set_aside, 126 (p.id IS NOT NULL) AS is_recompete, 127 (p.id IS NULL) AS is_new, 128 COALESCE(p.id::text, '') AS predecessor_award_id, 129 COALESCE(p.incumbent_uei, '') AS incumbent_uei, 130 COALESCE(p.incumbent_name, '') AS incumbent_name, 131 CASE WHEN p.id IS NOT NULL THEN 'prime-of-record' ELSE '' END AS incumbent_role, 132 p.estimated_value AS prior_award_amount, 133 p.award_date AS prior_award_date, 134 COALESCE(f.id::text, '') AS forecast_id, 135 f.estimated_award_date AS forecast_estimated_award_date 136FROM anchor an 137LEFT JOIN predecessor p ON true 138LEFT JOIN forecast f ON true` 139 140// recompeteClockSQL is the FIXED recompete clock. Given an agency ($1) and a window in months ($2), it 141// returns the opportunity_records and the usaspending_awards whose pop_end_date falls inside the forward 142// window for that agency, the takeaway pipeline a capture team watches, each row carrying its source_table so 143// an award row and an opportunity row are never conflated, with the incumbent prime named and labeled 144// prime-of-record. The two arms are UNIONed and the result is ordered by pop_end_date with a record id 145// tiebreaker, a total order for byte stable output. 146// 147// Both arms key the agency the same honest way: opportunity_records on its agency column, awards on 148// awarding_agency (the only agency key the award table carries). estimated_value and award_amount are 149// numeric, scanned into decimal, never float. 150const recompeteClockSQL = ` 151SELECT 152 'opportunity_records' AS source_table, 153 o.id::text AS record_id, 154 o.title AS title, 155 o.agency AS agency, 156 o.naics AS naics, 157 o.incumbent_uei AS incumbent_uei, 158 'prime-of-record' AS incumbent_role, 159 o.estimated_value AS dollars, 160 o.value_source AS value_source, 161 o.pop_end_date AS pop_end_date 162FROM opportunity_records o 163WHERE o.agency = $1 164 AND o.pop_end_date IS NOT NULL 165 AND o.pop_end_date >= now()::date 166 AND o.pop_end_date <= (now()::date + ($2::int * 30)) 167UNION ALL 168SELECT 169 'usaspending_awards' AS source_table, 170 a.generated_internal_id AS record_id, 171 a.description AS title, 172 a.awarding_agency AS agency, 173 a.naics_code AS naics, 174 a.recipient_uei AS incumbent_uei, 175 'prime-of-record' AS incumbent_role, 176 a.award_amount AS dollars, 177 a.source AS value_source, 178 a.pop_end_date AS pop_end_date 179FROM usaspending_awards a 180WHERE a.awarding_agency = $1 181 AND a.pop_end_date IS NOT NULL 182 AND a.pop_end_date >= now()::date 183 AND a.pop_end_date <= (now()::date + ($2::int * 30)) 184ORDER BY pop_end_date, record_id` 185 186// forecastRunwaySQL is the FIXED forecast runway. Given a window in months ($1), it returns the 187// opportunity_records of record_kind forecast whose estimated_award_date falls inside the forward window, the 188// lead time view, joined to federal_mission_tags and mission_areas for the readable mission area and carrying 189// the place of performance state. estimated_value is numeric, scanned into decimal, never float. 190// 191// The mission area is a LEFT JOIN so an untagged forecast still appears (its mission area is the empty 192// string), and federal_mission_tags keys on the record_id as the text form of the opportunity uuid. The 193// result is ordered by estimated_award_date with the opportunity id tiebreaker, a total order for byte 194// stable output; the id appears in the ORDER BY even though one forecast can carry several mission tags, so a 195// multi tag forecast is still totally ordered. 196const forecastRunwaySQL = ` 197SELECT 198 o.id::text AS forecast_id, 199 o.title AS title, 200 o.agency AS agency, 201 o.naics AS naics, 202 o.psc AS psc, 203 o.pop_state AS pop_state, 204 coalesce(ma.code, '') AS mission_code, 205 coalesce(ma.name, '') AS mission_area, 206 o.estimated_value AS dollars, 207 o.value_source AS value_source, 208 o.estimated_award_date AS estimated_award_date 209FROM opportunity_records o 210LEFT JOIN federal_mission_tags fmt 211 ON fmt.record_table = 'opportunity_records' 212 AND fmt.record_id = o.id::text 213LEFT JOIN mission_areas ma 214 ON ma.code = fmt.mission_code 215WHERE o.record_kind = 'forecast' 216 AND o.estimated_award_date IS NOT NULL 217 AND o.estimated_award_date >= now()::date 218 AND o.estimated_award_date <= (now()::date + ($1::int * 30)) 219ORDER BY o.estimated_award_date, o.id, mission_code` 220 221// SuccessionRow is the lifecycle of one solicitation: the named solicitation, whether it is a new start or a 222// recompete, the predecessor award (if any) with its prior amount (decimal, never float) and prime UEI 223// labeled prime-of-record, and the announcing forecast (if any), with the cited record ids and dates. It is 224// pure market structure: there is no match score and no employment decision number on it. 225type SuccessionRow struct { 226 SolicitationID string `json:"solicitation_id"` 227 SolicitationTitle string `json:"solicitation_title"` 228 Agency string `json:"agency"` 229 NAICS string `json:"naics"` 230 PSC string `json:"psc"` 231 PopState string `json:"pop_state"` 232 SetAside string `json:"set_aside"` 233 IsRecompete bool `json:"is_recompete"` 234 IsNew bool `json:"is_new"` 235 PredecessorAwardID string `json:"predecessor_award_id,omitempty"` 236 IncumbentUEI string `json:"incumbent_uei,omitempty"` 237 IncumbentName string `json:"incumbent_name,omitempty"` 238 IncumbentRole string `json:"incumbent_role,omitempty"` 239 PriorAwardAmount *decimal.Decimal `json:"prior_award_amount,omitempty"` 240 PriorAwardDate string `json:"prior_award_date,omitempty"` 241 ForecastID string `json:"forecast_id,omitempty"` 242 ForecastEstAwardDate string `json:"forecast_estimated_award_date,omitempty"` 243} 244 245// ProcurementSuccession runs the registered procurement_succession intent for an active_solicitation 246// opportunity id and scans the result, carrying the prior award amount as decimal.Decimal (the numeric column 247// is scanned directly into decimal, never through float64). The walk is a FIXED set of LEFT JOINs over 248// opportunity_records, the only table that carries the full segment tuple, so no phantom award column is ever 249// touched. The generic ExecuteIntent path remains available to the narration layer. 250func ProcurementSuccession(ctx context.Context, db *tenant.DB, opportunityID string) ([]SuccessionRow, error) { 251 if db == nil { 252 return nil, errors.New("ai: no database") 253 } 254 var out []SuccessionRow 255 err := db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 256 if err := GuardReadOnlyQuery(procurementSuccessionSQL, ReadAllowlist); err != nil { 257 return err 258 } 259 rows, err := tx.Query(ctx, procurementSuccessionSQL, opportunityID) 260 if err != nil { 261 return err 262 } 263 defer rows.Close() 264 for rows.Next() { 265 var ( 266 r SuccessionRow 267 priorDate pgtype.Date 268 fcastDate pgtype.Date 269 ) 270 if err := rows.Scan(&r.SolicitationID, &r.SolicitationTitle, &r.Agency, &r.NAICS, &r.PSC, 271 &r.PopState, &r.SetAside, &r.IsRecompete, &r.IsNew, &r.PredecessorAwardID, 272 &r.IncumbentUEI, &r.IncumbentName, &r.IncumbentRole, &r.PriorAwardAmount, &priorDate, 273 &r.ForecastID, &fcastDate); err != nil { 274 return err 275 } 276 r.PriorAwardDate = isoDate(priorDate) 277 r.ForecastEstAwardDate = isoDate(fcastDate) 278 out = append(out, r) 279 } 280 return rows.Err() 281 }) 282 return out, err 283} 284 285// RecompeteClockRow is one recompete the capture team watches: the source table it came from (so an award is 286// never conflated with an opportunity), the record, the agency, the incumbent UEI labeled prime-of-record, 287// the dollars (decimal, never float), and the period of performance end that puts it in the window. It 288// carries no score. 289type RecompeteClockRow struct { 290 SourceTable string `json:"source_table"` 291 RecordID string `json:"record_id"` 292 Title string `json:"title"` 293 Agency string `json:"agency"` 294 NAICS string `json:"naics"` 295 IncumbentUEI string `json:"incumbent_uei"` 296 IncumbentRole string `json:"incumbent_role"` 297 Dollars *decimal.Decimal `json:"dollars,omitempty"` 298 ValueSource string `json:"value_source"` 299 PopEndDate string `json:"pop_end_date"` 300} 301 302// RecompeteClock runs the registered recompete_clock intent for an agency and a window in months, scanning the 303// dollars into decimal. The window is clamped to the bounded range by the registered ParamInt Min and Max on 304// the structured path; this typed helper clamps with ClampRecompeteWindowMonths so the scan is always 305// bounded. 306func RecompeteClock(ctx context.Context, db *tenant.DB, agency string, windowMonths int) ([]RecompeteClockRow, error) { 307 if db == nil { 308 return nil, errors.New("ai: no database") 309 } 310 var out []RecompeteClockRow 311 err := db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 312 if err := GuardReadOnlyQuery(recompeteClockSQL, ReadAllowlist); err != nil { 313 return err 314 } 315 rows, err := tx.Query(ctx, recompeteClockSQL, agency, ClampRecompeteWindowMonths(windowMonths)) 316 if err != nil { 317 return err 318 } 319 defer rows.Close() 320 for rows.Next() { 321 var ( 322 r RecompeteClockRow 323 popEnd pgtype.Date 324 ) 325 if err := rows.Scan(&r.SourceTable, &r.RecordID, &r.Title, &r.Agency, &r.NAICS, 326 &r.IncumbentUEI, &r.IncumbentRole, &r.Dollars, &r.ValueSource, &popEnd); err != nil { 327 return err 328 } 329 r.PopEndDate = isoDate(popEnd) 330 out = append(out, r) 331 } 332 return rows.Err() 333 }) 334 return out, err 335} 336 337// ForecastRunwayRow is one forecast landing in the forward window: the forecast, its agency and NAICS, its 338// mission area (from federal_mission_tags and mission_areas, the empty string when untagged), its place of 339// performance state, the dollars (decimal, never float), and the estimated award date that puts it on the 340// runway. It carries no score. 341type ForecastRunwayRow struct { 342 ForecastID string `json:"forecast_id"` 343 Title string `json:"title"` 344 Agency string `json:"agency"` 345 NAICS string `json:"naics"` 346 PSC string `json:"psc"` 347 PopState string `json:"pop_state"` 348 MissionCode string `json:"mission_code"` 349 MissionArea string `json:"mission_area"` 350 Dollars *decimal.Decimal `json:"dollars,omitempty"` 351 ValueSource string `json:"value_source"` 352 EstimatedAwardDate string `json:"estimated_award_date"` 353} 354 355// ForecastRunway runs the registered forecast_runway intent for a window in months, scanning the dollars into 356// decimal. The window is clamped to the bounded range so the scan is always bounded. 357func ForecastRunway(ctx context.Context, db *tenant.DB, windowMonths int) ([]ForecastRunwayRow, error) { 358 if db == nil { 359 return nil, errors.New("ai: no database") 360 } 361 var out []ForecastRunwayRow 362 err := db.RunInTenantTx(ctx, func(ctx context.Context, tx pgx.Tx) error { 363 if err := GuardReadOnlyQuery(forecastRunwaySQL, ReadAllowlist); err != nil { 364 return err 365 } 366 rows, err := tx.Query(ctx, forecastRunwaySQL, ClampRecompeteWindowMonths(windowMonths)) 367 if err != nil { 368 return err 369 } 370 defer rows.Close() 371 for rows.Next() { 372 var ( 373 r ForecastRunwayRow 374 estDate pgtype.Date 375 ) 376 if err := rows.Scan(&r.ForecastID, &r.Title, &r.Agency, &r.NAICS, &r.PSC, &r.PopState, 377 &r.MissionCode, &r.MissionArea, &r.Dollars, &r.ValueSource, &estDate); err != nil { 378 return err 379 } 380 r.EstimatedAwardDate = isoDate(estDate) 381 out = append(out, r) 382 } 383 return rows.Err() 384 }) 385 return out, err 386} 387