Janady13/missionmatch / cmd/wire-extractions/main.go

Source identity

Field	Recorded value
Source type	GitHub source
Original record	Janady13/missionmatch / cmd/wire-extractions/main.go
Date	Not specified in source
Repository	Janady13/missionmatch
Revision	720d47fae476b0d38477ae4401697d372b1ea2bc
Branch	feature/opportunity-to-outcome
Path or archive member	cmd/wire-extractions/main.go
Original SHA-256	5807c64ac86f4d798cfc017a87fa694761540bb3d4ca704c6f3bc16c1a8b0439
Git blob SHA	c94653fd48f2d35e2c26d4f043479826237ff9f6
Reading-copy SHA-256	5807c64ac86f4d798cfc017a87fa694761540bb3d4ca704c6f3bc16c1a8b0439
Original source ↗
Recovered source transcript

Download complete source text ↓321 lines · 12.2 KB
// Command wire-extractions connects what the SAM.gov learning already read to the requirements the
// scoring engine actually reads.
//
// THE GAP THIS CLOSES. solicitation_extractions holds 2163 successful reads, each carrying a labor
// category, a required clearance, required skills and required years, at 0.825 average confidence, every
// row correctly linked to a notice. required_skills holds ZERO rows. So the machine had already read what
// these jobs need and nothing carried it across, which meant the Skills component, twenty five of the
// hundred scoring points, was being computed against an empty list for every candidate on every
// requirement. The reasoning that produced the extraction was sound and the wire was missing.
//
// THE JOIN IS THREE HOPS AND EACH ONE MATTERS:
//
//	requirements.opportunity_id  -> opportunities.id
//	opportunities.federal_ref    -> opportunity_records.id   (text holding a uuid)
//	solicitation_extractions.opportunity_record_id -> opportunity_records.id
//
// The middle hop is the one that is easy to get wrong. This schema carries two opportunity tables and a
// text column that holds a uuid, and assuming a requirement knows its own notice directly is the exact
// mistake that made key personnel commitments save successfully and appear nowhere earlier in this build.
//
// NOTHING IS OVERWRITTEN AND NOTHING IS INVENTED. A field a person has already filled in is left alone,
// because a machine reading a solicitation is evidence and a person typing into the product is a
// decision, and the decision wins. A skill the ontology does not recognise is REPORTED rather than
// created: growing the skill ontology is a deliberate act with its own command, and silently minting
// entries from free text is how an ontology becomes a junk drawer.
//
// Env: MM_MIGRATOR_DATABASE_URL. Flags: -dry-run to see what would change, -org to limit the run.
package main

import (
	"context"
	"encoding/json"
	"flag"
	"log"
	"os"
	"sort"
	"strings"
	"time"

	"github.com/jackc/pgx/v5/pgxpool"
)

// minConfidence is the floor an extraction must clear before its reading is written into a requirement.
//
// The corpus average is 0.825 and the distribution has a long tail of parse noise. Below this the model
// was guessing at a document it could not read properly, and a guessed labor category is worse than an
// empty one because it looks like knowledge.
const minConfidence = 0.60

type extraction struct {
	requirementID string
	orgID         string
	laborCategory string
	clearance     string
	years         int
	skills        []string
	confidence    float64
	haveLabor     bool
	haveClearance bool
	haveYears     bool
}

func main() {
	dryRun := flag.Bool("dry-run", false, "report what would change and write nothing")
	orgFilter := flag.String("org", "", "limit to one organization id")
	flag.Parse()

	dsn := firstNonEmpty(os.Getenv("MM_MIGRATOR_DATABASE_URL"), os.Getenv("MM_DATABASE_URL"))
	if dsn == "" {
		log.Fatal("wire-extractions: MM_MIGRATOR_DATABASE_URL is required")
	}
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
	defer cancel()

	pool, err := pgxpool.New(ctx, dsn)
	if err != nil {
		log.Fatalf("wire-extractions: pool: %v", err)
	}
	defer pool.Close()

	// The ontology, loaded once. Names and verified aliases together, which is the same pair the request
	// path resolver uses, so a term that resolves during a live match resolves here too.
	lookup, err := loadOntology(ctx, pool)
	if err != nil {
		log.Fatalf("wire-extractions: load ontology: %v", err)
	}
	log.Printf("wire-extractions: ontology holds %d resolvable terms", len(lookup))

	rows, err := readExtractions(ctx, pool, *orgFilter)
	if err != nil {
		log.Fatalf("wire-extractions: read: %v", err)
	}
	log.Printf("wire-extractions: %d requirement(s) have a reachable extraction, dry_run=%t", len(rows), *dryRun)
	if len(rows) == 0 {
		return
	}

	var wiredSkills, filledLabor, filledClearance, filledYears, lowConfidence int
	unresolved := map[string]int{}

	for _, e := range rows {
		if e.confidence > 0 && e.confidence < minConfidence {
			lowConfidence++
			continue
		}
		for _, raw := range e.skills {
			key := normalize(raw)
			if key == "" {
				continue
			}
			skillID, ok := lookup[key]
			if !ok {
				unresolved[key]++
				continue
			}
			if *dryRun {
				// Only count what a real run would actually INSERT. The write uses ON CONFLICT DO NOTHING,
				// so a pair that already exists changes nothing and must not be counted as change here
				// either, for the same reason the scalar preview asks the database instead of assuming.
				var exists bool
				if err := pool.QueryRow(ctx,
					`SELECT true FROM required_skills WHERE requirement_id=$1 AND skill_id=$2`,
					e.requirementID, skillID).Scan(&exists); err != nil {
					wiredSkills++
				}
				continue
			}
			// ON CONFLICT DO NOTHING rather than an existence check, so a rerun is safe and two runs
			// racing cannot produce a duplicate key error on a table whose primary key is exactly this pair.
			tag, err := pool.Exec(ctx, `
				INSERT INTO required_skills (organization_id, requirement_id, skill_id, mandatory)
				VALUES ($1, $2, $3, true)
				ON CONFLICT (requirement_id, skill_id) DO NOTHING`, e.orgID, e.requirementID, skillID)
			if err != nil {
				log.Printf("wire-extractions: requirement %s skill %q: %v", e.requirementID, raw, err)
				continue
			}
			wiredSkills += int(tag.RowsAffected())
		}

		// The scalar fields, each filled ONLY where the requirement is currently silent.
		if e.haveLabor {
			if n := updateIfEmpty(ctx, pool, *dryRun,
				`UPDATE requirements SET labor_category=$3, updated_at=now()
				  WHERE organization_id=$1 AND id=$2 AND coalesce(btrim(labor_category),'')=''`,
				e.orgID, e.requirementID, e.laborCategory); n > 0 {
				filledLabor += n
			}
		}
		if e.haveClearance {
			if n := updateIfEmpty(ctx, pool, *dryRun,
				`UPDATE requirements SET required_level=$3, updated_at=now()
				  WHERE organization_id=$1 AND id=$2 AND coalesce(required_level,'unknown')='unknown'`,
				e.orgID, e.requirementID, e.clearance); n > 0 {
				filledClearance += n
			}
		}
		if e.haveYears {
			if n := updateIfEmpty(ctx, pool, *dryRun,
				`UPDATE requirements SET required_years=$3, updated_at=now()
				  WHERE organization_id=$1 AND id=$2 AND coalesce(required_years,0)=0`,
				e.orgID, e.requirementID, e.years); n > 0 {
				filledYears += n
			}
		}
	}

	log.Printf("wire-extractions: DONE, skills wired=%d, labor category filled=%d, clearance filled=%d, years filled=%d, below confidence floor=%d",
		wiredSkills, filledLabor, filledClearance, filledYears, lowConfidence)

	// WHAT COULD NOT BE RESOLVED IS THE MOST USEFUL OUTPUT HERE. Each of these is a term a federal
	// solicitation asked for that the ontology cannot name, which is a gap in what the product knows
	// rather than a failure of this command. Ranked, so the ones worth teaching first are obvious.
	if len(unresolved) > 0 {
		type kv struct {
			term string
			n    int
		}
		list := make([]kv, 0, len(unresolved))
		total := 0
		for t, n := range unresolved {
			list = append(list, kv{t, n})
			total += n
		}
		sort.Slice(list, func(i, j int) bool {
			if list[i].n != list[j].n {
				return list[i].n > list[j].n
			}
			return list[i].term < list[j].term
		})
		log.Printf("wire-extractions: %d skill mention(s) across %d distinct terms did not resolve, top of the list:", total, len(list))
		for i, kv := range list {
			if i >= 20 {
				break
			}
			log.Printf("  %4d  %s", kv.n, kv.term)
		}
	}
}

// updateIfEmpty writes the value only where the column is currently silent, and counts honestly.
//
// THE FIRST VERSION OF THIS RETURNED 1 IN DRY RUN WITHOUT ASKING THE DATABASE ANYTHING, so a dry run
// claimed it would fill five labor categories and three clearances that were already filled. The real run
// then correctly reported zero and the two numbers disagreed. A dry run that overstates is worse than no
// dry run, because it is consulted precisely when somebody is deciding whether to trust the tool. It now
// runs the SAME predicate as a count, so the preview and the write can only ever agree.
func updateIfEmpty(ctx context.Context, pool *pgxpool.Pool, dryRun bool, sql string, args ...any) int {
	if dryRun {
		// UPDATE table SET ... WHERE <pred> becomes SELECT count(*) FROM table WHERE <pred>, with the
		// value placeholder still bound so the argument positions line up.
		where := sql[strings.Index(sql, " WHERE "):]
		table := strings.Fields(strings.TrimPrefix(strings.TrimSpace(sql), "UPDATE "))[0]
		// Every caller passes (organization, requirement, value) and the predicate uses only the first
		// two; the third is what SET would write. Passing all three to a count that binds two is an
		// argument count error, which is how the first attempt at this failed loudly rather than quietly,
		// and loudly was the right way for it to fail.
		var n int
		if err := pool.QueryRow(ctx, "SELECT count(*) FROM "+table+where, args[:2]...).Scan(&n); err != nil {
			log.Printf("wire-extractions: dry run count: %v", err)
			return 0
		}
		return n
	}
	tag, err := pool.Exec(ctx, sql, args...)
	if err != nil {
		log.Printf("wire-extractions: update: %v", err)
		return 0
	}
	return int(tag.RowsAffected())
}

// readExtractions walks the three hop join and returns one row per requirement that has an extraction.
func readExtractions(ctx context.Context, pool *pgxpool.Pool, org string) ([]extraction, error) {
	q := `
		SELECT r.id::text, r.organization_id::text,
		       coalesce(se.labor_category,''), coalesce(se.required_clearance,''),
		       coalesce(se.required_years, 0), coalesce(se.required_skills,'[]'::jsonb)::text,
		       coalesce(se.confidence, 0)
		  FROM requirements r
		  JOIN opportunities o  ON o.organization_id = r.organization_id AND o.id = r.opportunity_id
		  JOIN solicitation_extractions se ON se.opportunity_record_id::text = o.federal_ref
		 WHERE r.deleted_at IS NULL
		   AND se.extract_status = 'ok'`
	args := []any{}
	if strings.TrimSpace(org) != "" {
		q += ` AND r.organization_id = $1`
		args = append(args, strings.TrimSpace(org))
	}
	// Highest confidence first, so where a notice somehow carries two extractions the better one is the
	// one that fills the empty fields.
	q += ` ORDER BY se.confidence DESC NULLS LAST`

	rows, err := pool.Query(ctx, q, args...)
	if err != nil {
		return nil, err
	}
	defer rows.Close()

	var out []extraction
	for rows.Next() {
		var e extraction
		var skillsJSON string
		if err := rows.Scan(&e.requirementID, &e.orgID, &e.laborCategory, &e.clearance,
			&e.years, &skillsJSON, &e.confidence); err != nil {
			return nil, err
		}
		_ = json.Unmarshal([]byte(skillsJSON), &e.skills)
		e.haveLabor = strings.TrimSpace(e.laborCategory) != ""
		// "none" is a real answer from the extractor and it is NOT a clearance requirement, so it must not
		// be written into a column whose vocabulary treats every value as a bar to clear.
		e.haveClearance = e.clearance != "" && e.clearance != "none" && e.clearance != "unknown"
		e.haveYears = e.years > 0
		out = append(out, e)
	}
	return out, rows.Err()
}

// loadOntology maps every resolvable term to a skill id: canonical names and verified aliases together.
func loadOntology(ctx context.Context, pool *pgxpool.Pool) (map[string]string, error) {
	out := map[string]string{}
	rows, err := pool.Query(ctx, `
		SELECT lower(btrim(name)), id::text FROM skills
		UNION ALL
		SELECT lower(btrim(alias_text)), skill_id::text FROM skill_aliases WHERE verified_at IS NOT NULL`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	for rows.Next() {
		var term, id string
		if err := rows.Scan(&term, &id); err != nil {
			return nil, err
		}
		if term = normalize(term); term != "" {
			// First writer wins: a canonical name is read before aliases and should not be displaced.
			if _, seen := out[term]; !seen {
				out[term] = id
			}
		}
	}
	return out, rows.Err()
}

// normalize is the one place a term is folded for comparison, so the ontology and the extraction are
// always folded the same way. Case and surrounding punctuation only; nothing here rewrites meaning.
func normalize(s string) string {
	s = strings.ToLower(strings.TrimSpace(s))
	s = strings.Trim(s, ".,;:()[]\"'")
	return strings.Join(strings.Fields(s), " ")
}

func firstNonEmpty(vals ...string) string {
	for _, v := range vals {
		if s := strings.TrimSpace(v); s != "" {
			return s
		}
	}
	return ""
}