Source identity
| Field | Recorded value |
|---|---|
| Source type | GitHub source |
| Original record | Janady13/missionmatch / cmd/import-bench/main.go |
| Date | Not specified in source |
| Repository | Janady13/missionmatch |
| Revision | 720d47fae476b0d38477ae4401697d372b1ea2bc |
| Branch | feature/opportunity-to-outcome |
| Path or archive member | cmd/import-bench/main.go |
| Original SHA-256 | a9d73f3d57fc170871969fe954241a4c9878634e504b3f33f39be9c769471ef5 |
| Git blob SHA | 24b7f8c307c3cf7e93a87a035cef919b992f1d74 |
| Reading-copy SHA-256 | a9d73f3d57fc170871969fe954241a4c9878634e504b3f33f39be9c769471ef5 |
Recovered source transcript
Download complete source text ↓455 lines · 19.9 KB
// Command import-bench is the WS9 CUI safe bulk loader for a real cleared workforce bench. It reads a
// directory of resumes and a structured CSV or JSONL roster, maps each entry onto the existing
// sourcing.SourcedPerson, dedups and writes the candidate plus structured skills, certs, and agencies
// through the migrator role, composes a NAME FREE embed profile and embeds it BEFORE the write so the
// inference round trip never holds a row lock, and tags the candidate's missions through the application
// role under RLS after the candidate commits. It writes one bench_import_log compliance row per committed
// record and one append only audit summary (counts and model only, no PII) at the end. The frozen scoring
// engine is never touched.
//
// Seven WS9 invariants are enforced here. Invariant 1: when -embed is set, the REAL vector model probe
// (ai.VerifyVectorModel, the same gate the api runs at boot, NOT the bare Health liveness ping) runs at
// startup and the binary refuses to embed and exits non zero on a model mismatch. Invariant 2: two pools,
// a migrator pool for the candidate, skills, embed, and bench_import_log writes and a SEPARATE application
// pool for the candidate_mission_tags write, which runs inside RunInTenantTx so RLS validates the org on
// every tag insert. Invariant 6: the embed profile is composed and embedded BEFORE the migrator write
// transaction opens, so a slow plane never holds the candidate row lock across the embed timeout; on an
// embed failure the row commits without a vector and the embed-backfill only-missing fallback fills it
// later. Invariant 7: -legal-basis is REQUIRED, the binary refuses to run if it is empty, and the recorded
// source is bench_file for resumes or bench_csv for profiles, never the PDL string.
//
// Env (server side only): MM_DATABASE_URL (the app DSN, NOBYPASSRLS, required), MM_MIGRATOR_DATABASE_URL
// (the migrator DSN, BYPASSRLS; falls back to MM_DATABASE_URL), and for -embed MM_MEGAMIND_URL plus
// MM_MEGAMIND_EMBED_MODEL.
//
// Build and run FIPS validated: CGO_ENABLED=0 GOFIPS140=v1.0.0 go build, run with GODEBUG=fips140=on.
package main
import (
"context"
"crypto/fips140"
"crypto/pbkdf2"
"crypto/sha256"
"encoding/json"
"errors"
"flag"
"log"
"math"
"os"
"strings"
"time"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/jackc/pgx/v5/pgxpool"
pgvector "github.com/pgvector/pgvector-go"
"github.com/missionworkforce/missionmatch/internal/modules/ai"
"github.com/missionworkforce/missionmatch/internal/modules/federalsignal"
"github.com/missionworkforce/missionmatch/internal/platform/audit"
"github.com/missionworkforce/missionmatch/internal/platform/benchimport"
"github.com/missionworkforce/missionmatch/internal/platform/database"
"github.com/missionworkforce/missionmatch/internal/platform/database/dbgen"
"github.com/missionworkforce/missionmatch/internal/platform/extraction"
"github.com/missionworkforce/missionmatch/internal/platform/inference"
"github.com/missionworkforce/missionmatch/internal/platform/missionclass"
"github.com/missionworkforce/missionmatch/internal/platform/security"
"github.com/missionworkforce/missionmatch/internal/platform/sourcing"
"github.com/missionworkforce/missionmatch/internal/platform/tenant"
)
const embedDimension = 384
// errStopBatch is the sentinel a yield closure returns to stop the FileSource walk once the -size cap is
// reached. The FileSource treats a non nil yield error as a stop, so the run ends cleanly at the cap.
var errStopBatch = errors.New("import-bench: size cap reached")
func main() {
resumesDir := flag.String("resumes", "", "directory of resume files to import (PDF, DOCX, HTML, or text)")
profilesPath := flag.String("profiles", "", "path to a structured CSV or JSONL roster to import")
orgFlag := flag.String("org", "", "the organization id to import the bench into (uuid, required)")
legalBasis := flag.String("legal-basis", "", "the lawful basis recorded against every imported record (required, invariant 7)")
embed := flag.Bool("embed", false, "compose a name free profile and embed it before the write (gated on the real vector model probe)")
dryRun := flag.Bool("dry-run", false, "extract, resolve, and report dedup hits, then roll back, writing nothing")
size := flag.Int("size", 0, "cap the number of records processed (0 means no cap)")
flag.Parse()
// FIPS self test: derive one PBKDF2 key at the production cost under the active crypto module and fail
// closed if it errors, so a build that dropped the validated module is caught at boot, not mid run. This
// is the same in build FIPS gate the api and tag-missions binaries use.
if err := fipsSelfTest(); err != nil {
log.Fatalf("import-bench: FIPS self test (PBKDF2 HMAC SHA 256) failed: %v", err)
}
log.Printf("import-bench: fips_enabled=%v module=%s", fips140.Enabled(), fips140.Version())
// Invariant 7: refuse to run without an explicit operator provided legal basis. It is NEVER defaulted to
// the PDL legitimate interest string. The recorded source is bench_file or bench_csv, set per source
// below, never PDL.
basis := strings.TrimSpace(*legalBasis)
if basis == "" {
log.Fatal("import-bench: -legal-basis is required (invariant 7), pass the owner provided controller basis, for example \"client provided cleared bench, client is the data controller\"")
}
if strings.TrimSpace(*resumesDir) == "" && strings.TrimSpace(*profilesPath) == "" {
log.Fatal("import-bench: pass at least one of -resumes (a directory) or -profiles (a CSV or JSONL path)")
}
if strings.TrimSpace(*orgFlag) == "" {
log.Fatal("import-bench: -org (the target organization id) is required")
}
orgID, err := uuid.Parse(strings.TrimSpace(*orgFlag))
if err != nil {
log.Fatalf("import-bench: bad org id %q: %v", *orgFlag, err)
}
appDSN := strings.TrimSpace(os.Getenv("MM_DATABASE_URL"))
if appDSN == "" {
log.Fatal("import-bench: MM_DATABASE_URL (the app DSN) is required")
}
migDSN := firstNonEmpty(os.Getenv("MM_MIGRATOR_DATABASE_URL"), appDSN)
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Hour)
defer cancel()
// Invariant 2, two pools. The migrator pool carries the candidate, skills, embed, and bench_import_log
// writes (BYPASSRLS, because those writes set an explicit organization_id and no GUC). The SEPARATE app
// pool carries the candidate_mission_tags write inside RunInTenantTx as the NOBYPASSRLS application role,
// so RLS validates the org on every tag insert. database.NewPool registers the pgvector types on both,
// which the embedding write on the migrator pool needs.
migPool, err := database.NewPool(ctx, migDSN)
if err != nil {
log.Fatalf("import-bench: migrator pool: %v", err)
}
defer migPool.Close()
if err := database.AssertMigratorRole(ctx, migPool); err != nil {
log.Fatalf("import-bench: migrator pool role check: %v", err)
}
appPool, err := database.NewPool(ctx, appDSN)
if err != nil {
log.Fatalf("import-bench: app pool: %v", err)
}
defer appPool.Close()
// Fail closed unless this is the NOBYPASSRLS app role, so the tag write can never bypass tenant RLS.
if err := database.AssertRole(ctx, appPool, database.AppRoleName()); err != nil {
log.Fatalf("import-bench: app pool role check: %v", err)
}
appDB := tenant.New(appPool)
// Build the AliasIndex ONCE from the global alias pairs (read via the migrator role, the alias tables
// are global reference data) and share it across every record, mirroring cmd/tag-missions.
idx, err := federalsignal.BuildAliasIndex(ctx, federalsignal.NewRepo(migPool))
if err != nil {
log.Fatalf("import-bench: build alias index: %v", err)
}
// Invariant 1: when -embed is set, build the inference client and run the REAL vector model probe BEFORE
// processing any record. ai.VerifyVectorModel re embeds a sample of stored skill names and compares the
// cosine against the configured model, the exact gate the api runs at boot. It is NOT the Health ping,
// which has no model identity. On a proven mismatch the binary refuses to embed and exits non zero.
var embedClient *inference.Client
modelName := envDefault("MM_MEGAMIND_EMBED_MODEL", "qllama/bge-small-en-v1.5")
if *embed {
megaURL := strings.TrimSpace(os.Getenv("MM_MEGAMIND_URL"))
if megaURL == "" {
log.Fatal("import-bench: -embed needs MM_MEGAMIND_URL (the inference plane base url)")
}
embedClient, err = inference.New(inference.Config{
BaseURL: megaURL, EmbedModel: modelName, Dimension: embedDimension, Timeout: 60 * time.Second, MaxRetries: 2,
})
if err != nil {
log.Fatalf("import-bench: inference client: %v", err)
}
// The probe reads the global skills table; the app pool is the NOBYPASSRLS role, which sees the global
// skills rows and the catalog dimension checks, so it yields the same verdict the api boot gate does.
if err := ai.VerifyVectorModel(ctx, appDB, embedClient, embedDimension); err != nil {
log.Fatalf("import-bench: vector model probe FAILED, refusing to embed (invariant 1): %v", err)
}
log.Printf("import-bench: vector model probe PASS (stored vectors match %s at dim %d)", modelName, embedDimension)
}
ext := extraction.NewDeterministic() // certs and agencies from the profile text; skills come structured
r := &runner{
migPool: migPool,
appDB: appDB,
ext: ext,
idx: idx,
client: embedClient,
embed: *embed,
dryRun: *dryRun,
orgID: orgID,
basis: basis,
cap: *size,
}
// Choose the source. Resumes record source bench_file, profiles record bench_csv (invariant 7), never PDL.
if dir := strings.TrimSpace(*resumesDir); dir != "" {
src := benchimport.NewResumeSource(ext)
br, rerr := src.Resumes(ctx, dir, func(rec benchimport.BenchRecord, file string) error {
return r.process(ctx, rec, "bench_file", file)
})
if rerr != nil && !errors.Is(rerr, errStopBatch) {
log.Fatalf("import-bench: walk resumes: %v", rerr)
}
r.batch = addBatch(r.batch, br)
}
if path := strings.TrimSpace(*profilesPath); path != "" {
src := benchimport.NewStructuredSource()
br, perr := src.Profiles(ctx, path, func(rec benchimport.BenchRecord, batch string) error {
return r.process(ctx, rec, "bench_csv", batch)
})
if perr != nil && !errors.Is(perr, errStopBatch) {
log.Fatalf("import-bench: decode profiles: %v", perr)
}
r.batch = addBatch(r.batch, br)
}
// One append only audit summary at the end, attributed to the target org. A bench import is a TENANT
// operation into a specific org, so the summary runs through audit.Writer.OnTx inside RunInTenantTx on
// the APP pool, with the org from the -org flag set on the context via WithOrg. The summary row therefore
// carries the correct organization_id, not a NULL system org. It is counts and the model only, NO PII and
// no candidate id. A dry run still records a summary with dry_run true so an operator has a trail.
writeBatchAudit(ctx, appDB, r.orgID, modelName, *embed, *dryRun, r.batch, r.counts)
log.Printf("import-bench: DONE read=%d skipped=%d mapped=%d new=%d enriched=%d embed_skipped=%d dry_run=%v",
r.batch.Read, r.batch.Skipped, r.batch.Mapped, r.counts.New, r.counts.Enriched, r.counts.EmbedSkipped, *dryRun)
}
// runner holds the per run dependencies and accumulators so the FileSource yield closures stay thin. The
// legal basis and org are validated once in main and read here, never re derived.
type runner struct {
migPool *pgxpool.Pool
appDB *tenant.DB
ext extraction.Extractor
idx missionclass.AliasIndex
client *inference.Client
embed bool
dryRun bool
orgID uuid.UUID
basis string
cap int
batch benchimport.BatchResult
counts runCounts
}
// runCounts accumulates the per record outcomes. EmbedSkipped counts records that committed without a
// vector because the embed network call failed (the embed-backfill only-missing fallback fills them later).
type runCounts struct {
Processed int
New int
Enriched int
EmbedSkipped int
}
// process ingests one bench record honoring invariants 2 and 6. It composes the name free embed profile and
// calls inference.Client.Embed BEFORE opening the migrator write transaction (invariant 6), so the embed
// network round trip is never inside the lock window. It opens ONE migrator transaction and calls
// sourcing.IngestBenchRecord, which writes the candidate, skills, the precomputed vector, and one
// bench_import_log row atomically. In a dry run the transaction is rolled back so an operator previews dedup
// hits without writing. After the migrator transaction commits (never in a dry run), it tags the candidate's
// missions on the APP pool inside RunInTenantTx (invariant 2), so the tag write is never in the migrator tx.
func (r *runner) process(ctx context.Context, rec benchimport.BenchRecord, source, fileOrBatch string) error {
if r.cap > 0 && r.counts.Processed >= r.cap {
return errStopBatch
}
r.counts.Processed++
sp := rec.ToSourcedPerson()
// Invariant 6: compose and embed BEFORE the write transaction. A nil vector commits the row without an
// embedding. The embed call is the ONLY network round trip and it happens here, outside the tx below.
var vec *pgvector.Vector
if r.embed && r.client != nil {
profile := composeEmbedProfile(sp)
res, eerr := r.client.Embed(ctx, []string{profile})
switch {
case eerr != nil:
// Commit the row WITHOUT a vector and rely on the documented embed-backfill only-missing fallback.
log.Printf("import-bench: embed failed for one record, committing without a vector (run embed-backfill only-missing to fill it): %v", eerr)
r.counts.EmbedSkipped++
case len(res.Embeddings) == 1 && len(res.Embeddings[0]) == embedDimension:
v := append([]float32(nil), res.Embeddings[0]...)
normalize(v)
nv := pgvector.NewVector(v)
vec = &nv
default:
// A width or count surprise the Embed width check did not raise: treat as a skipped embed, never
// write a malformed vector, and lean on the only-missing fallback.
log.Printf("import-bench: embed returned an unexpected shape for one record, committing without a vector")
r.counts.EmbedSkipped++
}
}
// One migrator transaction: candidate plus skills plus the precomputed embed plus one bench_import_log
// row, atomic. In a dry run the transaction is rolled back so nothing is written.
tx, err := r.migPool.Begin(ctx)
if err != nil {
return err
}
committed := false
defer func() {
if !committed {
_ = tx.Rollback(ctx)
}
}()
q := dbgen.New(tx)
out, err := sourcing.IngestBenchRecord(ctx, q, r.orgID, sp, r.ext, vec, source, fileOrBatch, r.basis)
if err != nil {
log.Printf("import-bench: ingest %q: %v", redactRef(rec), err)
return nil // skip one bad record, never abort the batch
}
if r.dryRun {
// Preview only: report the dedup verdict, then roll back via the deferred Rollback. No candidate, no
// log, and no tag is written.
if out.New {
log.Printf("import-bench: [dry run] would INSERT a new candidate from %s", source)
r.counts.New++
} else {
log.Printf("import-bench: [dry run] would ENRICH an existing candidate (dedup hit) from %s", source)
r.counts.Enriched++
}
return nil
}
if err := tx.Commit(ctx); err != nil {
log.Printf("import-bench: commit one record: %v", err)
return nil
}
committed = true
if out.New {
r.counts.New++
} else {
r.counts.Enriched++
}
// Invariant 2: tag the candidate's missions on the APP pool inside RunInTenantTx AFTER the candidate
// commits. The org is set on the context via tenant.WithOrg and read from the GUC by the tag write,
// never threaded as a caller argument, so the RLS WITH CHECK validates the org on every tag insert. This
// write is NEVER in the migrator transaction above.
orgCtx := tenant.WithOrg(ctx, r.orgID.String())
headline := composeEmbedProfile(sp)
if terr := sourcing.TagCandidateMissions(orgCtx, r.appDB, out.CandidateID, headline, sp.Skills, r.idx); terr != nil {
// The candidate is committed; a tag write failure is logged and the run continues, the tag-missions
// backfill can re tag later. It does not roll back the committed candidate.
log.Printf("import-bench: tag missions for one candidate failed (candidate committed, re run tag-missions to retry): %v", terr)
}
return nil
}
// composeEmbedProfile builds the NAME FREE embed profile for a sourced person: the headline (JobTitle or
// Role plus the first six skills) followed by the full skill list. It mirrors the embed-backfill candidate
// profile composer and the invariant 5 test composer exactly. By construction it reads ONLY structured
// tokens (JobTitle, Role, Skills); the resume mapping never routes a name, an email, or a phone into any
// of those fields, so no PII can reach the inference plane.
func composeEmbedProfile(sp sourcing.SourcedPerson) string {
skills := sp.Skills
if len(skills) > 6 {
skills = skills[:6]
}
headline := sp.JobTitle
if headline == "" {
headline = sp.Role
}
if len(skills) > 0 {
if headline != "" {
headline += " | "
}
headline += strings.Join(skills, ", ")
}
headline = strings.TrimSpace(headline)
return strings.Join([]string{
headline,
"skills: " + strings.Join(sp.Skills, ", "),
}, " | ")
}
// writeBatchAudit writes ONE append only audit summary for the run, attributed to the target org. A bench
// import is a tenant operation, so the summary runs through audit.Writer.OnTx inside RunInTenantTx on the
// app pool: the org is set on the context via WithOrg and read from the validated GUC, so the row carries
// the correct organization_id rather than a NULL system org. The payload is counts and the model only, NO
// PII and no candidate id. A dry run still records a summary with dry_run true so an operator has a trail of
// the preview.
func writeBatchAudit(ctx context.Context, appDB *tenant.DB, orgID uuid.UUID, model string, embed, dryRun bool, batch benchimport.BatchResult, counts runCounts) {
payload, _ := json.Marshal(map[string]any{
"read": batch.Read,
"skipped": batch.Skipped,
"mapped": batch.Mapped,
"new": counts.New,
"enriched": counts.Enriched,
"embed_skipped": counts.EmbedSkipped,
"embed": embed,
"dry_run": dryRun,
"model_version": model,
})
w := audit.New(appDB)
orgCtx := tenant.WithOrg(ctx, orgID.String())
err := appDB.RunInTenantTx(orgCtx, func(ctx context.Context, tx pgx.Tx) error {
return w.OnTx(ctx, tx, audit.Event{
Action: "bench.import.batch",
EntityType: "bench_import",
NewValues: payload,
})
})
if err != nil {
log.Printf("import-bench: write batch audit summary: %v", err)
}
}
// addBatch sums two BatchResult counters so a run over both a resume directory and a profiles file reports
// one combined read, skipped, and mapped total.
func addBatch(a, b benchimport.BatchResult) benchimport.BatchResult {
return benchimport.BatchResult{Read: a.Read + b.Read, Skipped: a.Skipped + b.Skipped, Mapped: a.Mapped + b.Mapped}
}
// redactRef returns a non PII reference for a log line about a failed record: the job title only, never the
// name, email, or phone, so an error log never leaks a person.
func redactRef(rec benchimport.BenchRecord) string {
if rec.JobTitle == "" {
return "untitled"
}
return rec.JobTitle
}
// normalize L2 normalizes a vector in place so cosine over the resume_embedding column is meaningful,
// matching the embeddings package contract and the embed-backfill writer.
func normalize(v []float32) {
var sum float64
for _, x := range v {
sum += float64(x) * float64(x)
}
if sum == 0 {
return
}
inv := float32(1.0 / math.Sqrt(sum))
for i := range v {
v[i] *= inv
}
}
// fipsSelfTest derives one PBKDF2 key at the production iteration count under the active crypto module. It
// is the in build FIPS gate, identical to the api and tag-missions binaries, so this loader binary is held
// to the same validated crypto posture.
func fipsSelfTest() error {
_, err := pbkdf2.Key(sha256.New, "missionmatch fips self test", []byte("self test salt"), security.PBKDF2Iterations, 32)
return err
}
func firstNonEmpty(vals ...string) string {
for _, v := range vals {
if s := strings.TrimSpace(v); s != "" {
return s
}
}
return ""
}
func envDefault(k, def string) string {
if v := strings.TrimSpace(os.Getenv(k)); v != "" {
return v
}
return def
}