Janady13/missionmatch / cmd/teach-aggregate/main.go

1// Command teach-aggregate runs the WS8 teaching loop cadence pass. For each organization it runs the in 2// tenant teaching producer (folds that tenant's own reviewer craft into the closed scrub.TeachingSignal 3// vocabulary of contents free counts and persists the per tenant teaching_signal_aggregates rows) and then 4// the deterministic curation proposal generator (turns recurring approved craft into pending 5// curation_proposals a human approves, then a SECOND human verifies, before anything affects a future score). 6// Both passes are idempotent and safe to rerun: every upsert refreshes in place and every proposal insert is 7// deduped on the active pair. It is a one shot job and a cron candidate like tag-missions and mm-opp-sync. 8// 9// THE HONEST SCORE INVARIANT. This pass never changes a weight, a band cut, the 0.62 semantic threshold, the 10// engine ModelVersion math, or any already served and audited score. The producer only COUNTS what the frozen 11// engine already wrote, and the generator only emits a SUGGESTION. The only score affecting change anywhere is 12// an additive curated resolver edge that two humans must approve then verify, which changes FUTURE scores only, 13// recorded as a model input change (ontology.edge.added), never an engine change. Mission tags and craft 14// aggregates are reference data, NEVER a scoring input, and the pass never reaches the eeo schema. 15// 16// TWO POOL PATTERN (the tag-missions precedent). The READS that fold a tenant's craft and scan its evaluations 17// run through the NOBYPASSRLS application role inside each org's RunInTenantTx, so Postgres Row Level Security 18// is exercised on every row and one tenant can never read another's. Every WRITE to a global migrator owned 19// table (teaching_signal_aggregates, curation_proposals) goes through the BYPASSRLS migrator pool, never the 20// app role, which has SELECT only on those tables. 21// 22// THE CROSS TENANT DIFFERENTIAL PRIVACY PATH STAYS OFF. The cross tenant DP aggregation runs ONLY when 23// MM_FEDERATED_DP_ENABLED is true AND the contributing live non test tenant count is at least MinTenants. The 24// contributing tenant set is defined by an EXPLICIT production org allow list, the env var 25// MM_FEDERATED_PRODUCTION_ORGS (comma separated org uuids, default EMPTY). With the default empty allow list 26// the contributing count is zero, below MinTenants, so the cross tenant path is skipped and logged, which is 27// the correct posture today: a real five tenant cohort does not exist and the flag is off. NOTE the Mission A 28// tenant_directory is the future source of this allow list (the durable list of live non test tenants), and the 29// Mironov safe noise sampler (federated.activeAggregator.Noise, an activation blocker today) is a HARD blocker 30// before the flag is ever set. 31// 32// Env (server side only): MM_DATABASE_URL (the app DSN, NOBYPASSRLS, required), MM_MIGRATOR_DATABASE_URL (the 33// migrator DSN, BYPASSRLS, used for the global aggregate and proposal writes and to enumerate org ids; falls 34// back to MM_DATABASE_URL), MM_FEDERATED_DP_ENABLED / MM_FEDERATED_DP_EPSILON / MM_FEDERATED_DP_DELTA (the DP 35// config, default off), MM_FEDERATED_PRODUCTION_ORGS (the contributing tenant allow list, default empty). 36// 37// Build and run FIPS validated: CGO_ENABLED=0 GOFIPS140=v1.0.0 go build, run with GODEBUG=fips140=on. 38package main 39 40import ( 41 "context" 42 "crypto/fips140" 43 "crypto/pbkdf2" 44 "crypto/sha256" 45 "flag" 46 "log" 47 "os" 48 "strings" 49 "time" 50 51 "github.com/google/uuid" 52 "github.com/jackc/pgx/v5/pgxpool" 53 54 "github.com/missionworkforce/missionmatch/internal/modules/ai" 55 "github.com/missionworkforce/missionmatch/internal/modules/matching/engine" 56 "github.com/missionworkforce/missionmatch/internal/platform/audit" 57 "github.com/missionworkforce/missionmatch/internal/platform/config" 58 "github.com/missionworkforce/missionmatch/internal/platform/database" 59 "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" 60 "github.com/missionworkforce/missionmatch/internal/platform/inference" 61 "github.com/missionworkforce/missionmatch/internal/platform/security" 62 "github.com/missionworkforce/missionmatch/internal/platform/tenant" 63) 64 65func main() { 66 orgFlag := flag.String("org", "", "limit the pass to a single organization id (default: all orgs)") 67 flag.Parse() 68 69 // FIPS self test: derive one PBKDF2 key at the production cost under the active crypto module and fail 70 // closed if it errors, so a build that dropped the validated module is caught at boot, not mid run. This is 71 // the same in build FIPS gate the api, missionclass-sync, and tag-missions binaries use. 72 if err := fipsSelfTest(); err != nil { 73 log.Fatalf("teach-aggregate: FIPS self test (PBKDF2 HMAC SHA 256) failed: %v", err) 74 } 75 log.Printf("teach-aggregate: fips_enabled=%v module=%s", fips140.Enabled(), fips140.Version()) 76 77 cfg, err := config.Load() 78 if err != nil { 79 log.Fatalf("teach-aggregate: config: %v", err) 80 } 81 82 appDSN := strings.TrimSpace(os.Getenv("MM_DATABASE_URL")) 83 if appDSN == "" { 84 log.Fatal("teach-aggregate: MM_DATABASE_URL (the app DSN) is required") 85 } 86 migDSN := firstNonEmpty(os.Getenv("MM_MIGRATOR_DATABASE_URL"), appDSN) 87 88 ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) 89 defer cancel() 90 91 // App pool: the per tenant craft folds and evaluation scans run here, under RLS, never the migrator bypass. 92 appPool, err := database.NewPool(ctx, appDSN) 93 if err != nil { 94 log.Fatalf("teach-aggregate: app pool: %v", err) 95 } 96 defer appPool.Close() 97 // Fail closed unless this is the NOBYPASSRLS app role, so the pass can never bypass tenant RLS. 98 if err := database.AssertRole(ctx, appPool, "missionmatch_app"); err != nil { 99 log.Fatalf("teach-aggregate: app pool role check: %v", err) 100 } 101 appDB := tenant.New(appPool) 102 103 // Migrator pool: the global aggregate and proposal writes plus the org enumeration. No tenant scoped read 104 // runs here. teaching_signal_aggregates and curation_proposals are global migrator owned tables the app 105 // role can only SELECT. 106 migPool, err := pgxpool.New(ctx, migDSN) 107 if err != nil { 108 log.Fatalf("teach-aggregate: migrator pool: %v", err) 109 } 110 defer migPool.Close() 111 if err := database.AssertMigratorRole(ctx, migPool); err != nil { 112 log.Fatalf("teach-aggregate: %v", err) 113 } 114 115 // The teaching client carries TeachingEnabled from config (default false). With it false, the producer's 116 // per signal Teach call validates the scrub boundary end to end but transmits nothing off the box, the 117 // correct in tenant LIVE posture. A nil client is also fine (the producer skips Teach), but build a real one 118 // when a gateway URL is configured so the boundary is exercised exactly as it will be in production. 119 var teacher *inference.Client 120 if cfg.Inference.URL != "" { 121 t, terr := inference.New(inference.Config{ 122 BaseURL: cfg.Inference.URL, EmbedModel: cfg.Inference.EmbedModel, RerankModel: cfg.Inference.RerankModel, 123 GenerateModel: cfg.Inference.GenerateModel, Dimension: cfg.Inference.Dimension, 124 Timeout: time.Duration(cfg.Inference.TimeoutSeconds) * time.Second, MaxRetries: cfg.Inference.MaxRetries, 125 BreakerTrips: cfg.Inference.BreakerTrips, BreakerCool: time.Duration(cfg.Inference.BreakerCoolSec) * time.Second, 126 TeachingEnabled: cfg.Inference.TeachingEnabled, 127 }) 128 if terr != nil { 129 log.Fatalf("teach-aggregate: inference client: %v", terr) 130 } 131 teacher = t 132 } 133 134 // The producer's audit is on the APP pool, because the per tenant teaching.signal.aggregated event commits 135 // on the same tenant tx that read the decisions (invariant 5). The generator's audit is the SAME writer used 136 // through System for the global teaching.proposal.created event (NULL org). 137 auditW := audit.New(appDB) 138 139 producer := ai.NewTeachingProducer(appDB, ai.NewMigratorAggregateWriter(migPool), auditW, teacher) 140 generator := ai.NewProposalGenerator(appDB, ai.NewMigratorGovernanceReader(migPool), ai.NewMigratorProposalWriter(migPool), auditW) 141 142 orgIDs, err := listOrgs(ctx, migPool, *orgFlag) 143 if err != nil { 144 log.Fatalf("teach-aggregate: list orgs: %v", err) 145 } 146 if len(orgIDs) == 0 { 147 log.Print("teach-aggregate: no organizations to process") 148 return 149 } 150 151 // The model_version is pinned from engine.ModelVersion, never a literal, so the whole pass stamps the live 152 // scoring contract and a drifted version is refused at the scrub boundary (the producer registers it at 153 // init). Logged once so the run record names the contract its aggregates sit on. 154 log.Printf("teach-aggregate: model_version=%s (pinned from engine.ModelVersion)", engine.ModelVersion) 155 156 var totalSignals, totalProposals int 157 for _, orgID := range orgIDs { 158 orgCtx := tenant.WithOrg(ctx, orgID.String()) 159 160 // 1. The in tenant producer: fold this tenant's reviewer craft into per tenant aggregates. RLS scoped 161 // read on the app pool, global write on the migrator pool, audit on the same tenant tx. 162 pRes, perr := producer.AggregateForOrg(orgCtx) 163 if perr != nil { 164 log.Fatalf("teach-aggregate: org %s aggregate: %v", orgID, perr) 165 } 166 167 // 2. The deterministic curation generator: turn recurring approved craft into pending proposals. RLS 168 // scoped evaluation scan on the app pool, governance read and proposal write on the migrator pool, 169 // global proposal created audit through System. Idempotent: a rerun never duplicates an active pair. 170 gRes, gerr := generator.GenerateProposalsForOrg(orgCtx) 171 if gerr != nil { 172 log.Fatalf("teach-aggregate: org %s generate proposals: %v", orgID, gerr) 173 } 174 175 log.Printf("teach-aggregate: org %s folded %d decisions into %d signals, scanned %d evaluations, created %d proposals (%d substitution, %d reciprocity, %d recalibration)", 176 orgID, pRes.DecisionsRead, len(pRes.Signals), gRes.EvaluationsScanned, gRes.Created, 177 gRes.SubstitutionEdges, gRes.ReciprocityReviews, gRes.RecalibrationRows) 178 totalSignals += len(pRes.Signals) 179 totalProposals += gRes.Created 180 } 181 182 // 3. The CROSS TENANT differential privacy aggregation, gated OFF by posture. It runs ONLY when the flag is 183 // on AND a real contributing cohort of at least MinTenants live non test tenants exists. With the default 184 // empty allow list the contributing count is zero, so this is skipped and logged today, which is correct. 185 runCrossTenantDP(cfg.Federated) 186 187 log.Printf("teach-aggregate: DONE, %d org(s), persisted %d teaching signals and created %d curation proposals", 188 len(orgIDs), totalSignals, totalProposals) 189} 190 191// runCrossTenantDP runs the cross tenant differential privacy aggregation when, and ONLY when, two conditions 192// both hold: MM_FEDERATED_DP_ENABLED is true, and the contributing tenant count (the live non test orgs in the 193// explicit MM_FEDERATED_PRODUCTION_ORGS allow list) is at least MinTenants. Today the allow list defaults to 194// empty, so the contributing count is zero, below MinTenants, and the path is skipped and logged. This is the 195// correct posture: a real five tenant cohort does not exist and the flag is off. When a real cohort eventually 196// exists, the Mironov safe noise sampler (federated Noise, an activation blocker today) must be implemented 197// FIRST, and the contributing tenant set should be sourced from the Mission A tenant_directory rather than this 198// env allow list. The function deliberately does NO emission yet beyond logging the gate decision; the active 199// aggregator construction itself refuses while the flag is off (federated.NewActiveAggregator returns 200// ErrDisabled), the hard construction guard this function mirrors at the cadence layer. 201func runCrossTenantDP(fed config.Federated) { 202 contributing := contributingTenants() 203 minTenants := ai.DefaultFederatedConfig().MinTenants 204 205 if !fed.DPEnabled { 206 log.Printf("teach-aggregate: cross tenant DP SKIPPED, MM_FEDERATED_DP_ENABLED is false (the in tenant posture, no federated emission)") 207 return 208 } 209 if len(contributing) < minTenants { 210 log.Printf("teach-aggregate: cross tenant DP SKIPPED, %d contributing tenant(s) in MM_FEDERATED_PRODUCTION_ORGS is below MinTenants %d (no real cohort exists yet)", 211 len(contributing), minTenants) 212 return 213 } 214 // Unreachable in tenant LIVE today (the allow list is empty so the count is always below MinTenants). When a 215 // real cohort exists this is where the gated aggregation would run, AFTER the Mironov safe sampler ships. It 216 // is a hard fail until then so an operator who flips the flag prematurely gets a legible refusal, never a 217 // silent naive Gaussian release. 218 log.Fatalf("teach-aggregate: cross tenant DP requested with %d contributing tenants but the Mironov safe noise sampler is an activation blocker, refusing to release (implement federated.Noise first)", 219 len(contributing)) 220} 221 222// contributingTenants parses MM_FEDERATED_PRODUCTION_ORGS into the set of live non test contributing org ids. 223// It is EMPTY by default, so the contributing count is zero and the cross tenant path stays off. Only a well 224// formed uuid counts; a malformed entry is skipped rather than silently widening the cohort. The roughly 1088 225// test orgs are excluded by construction, because they are simply never in this explicit production allow list. 226func contributingTenants() []uuid.UUID { 227 raw := strings.TrimSpace(os.Getenv("MM_FEDERATED_PRODUCTION_ORGS")) 228 if raw == "" { 229 return nil 230 } 231 var out []uuid.UUID 232 for _, part := range strings.Split(raw, ",") { 233 s := strings.TrimSpace(part) 234 if s == "" { 235 continue 236 } 237 id, err := uuid.Parse(s) 238 if err != nil { 239 log.Printf("teach-aggregate: ignoring malformed org id %q in MM_FEDERATED_PRODUCTION_ORGS", s) 240 continue 241 } 242 out = append(out, id) 243 } 244 return out 245} 246 247// listOrgs returns the organization ids to process. It reads them through the migrator role (which sees every 248// org) so one run covers the cluster; the per org folds and scans then run under the app role and RLS. A -org 249// value restricts the run to a single org for a targeted pass. 250func listOrgs(ctx context.Context, migPool *pgxpool.Pool, only string) ([]uuid.UUID, error) { 251 if only != "" { 252 id, err := uuid.Parse(strings.TrimSpace(only)) 253 if err != nil { 254 return nil, err 255 } 256 return []uuid.UUID{id}, nil 257 } 258 return dbgen.New(migPool).ListAllOrganizationIDs(ctx) 259} 260 261// fipsSelfTest derives one PBKDF2 key at the production iteration count under the active crypto module. It is 262// the in build FIPS gate, identical to the api, missionclass-sync, and tag-missions binaries, so this cadence 263// binary is held to the same validated crypto posture. 264func fipsSelfTest() error { 265 _, err := pbkdf2.Key(sha256.New, "missionmatch fips self test", []byte("self test salt"), security.PBKDF2Iterations, 32) 266 return err 267} 268 269func firstNonEmpty(vals ...string) string { 270 for _, v := range vals { 271 if s := strings.TrimSpace(v); s != "" { 272 return s 273 } 274 } 275 return "" 276} 277