Janady13/missionmatch / cmd/opportunity-scope-normalize/main.go

Source identity

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

Download complete source text ↓289 lines · 12.5 KB
// Command opportunity-scope-normalize turns PUBLIC federal notice text into cited draft scope items,
// directive Section 5.2 and Phase 2 item 3.
//
// What it will and will not send off the machine is the whole point of the design:
//
//   - The ONLY text it composes is public SAM.gov notice metadata already cached in sam_opportunities.
//     Tenant uploaded documents, candidate data, scoring inputs, and slate contents never enter this path.
//   - Before any egress it REDACTS contact shapes (emails, phone numbers, identifiers) that a public notice
//     may legitimately carry but that scope extraction does not need, then hands the payload to the scrub
//     egress gate under the REQUIREMENT_NORMALIZE class. The gate is the authority: if it refuses, the
//     notice is skipped and nothing leaves.
//   - Every call that does leave is audited by backend, payload class, byte count, and content hash, never
//     by content.
//
// What comes back is treated as a proposal, never as fact. Model output is validated by
// opportunities.ValidateScopeCandidates, which rejects an item with no quote and, more importantly, an item
// whose quote cannot be found in the source text, because a fabricated citation reads exactly like a real
// one. Only survivors are written, always as review_status extracted, which no scoring path can see. A named
// human then reviews and approves.
//
// Usage:
//
//	opportunity-scope-normalize -dsn $MM_MIGRATOR_DATABASE_URL -megamind $MM_MEGAMIND_URL [-limit 10] [-dry-run]
package main

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

	"github.com/google/uuid"
	"github.com/jackc/pgx/v5/pgxpool"

	"github.com/missionworkforce/missionmatch/internal/modules/ai"
	"github.com/missionworkforce/missionmatch/internal/modules/opportunities"
	"github.com/missionworkforce/missionmatch/internal/platform/inference"
	"github.com/missionworkforce/missionmatch/internal/platform/scrub"
	"github.com/missionworkforce/missionmatch/internal/platform/secrets"
)

// normalizeSystem declares the schema and the grounding contract. The quote requirement is stated as the
// hard rule it is, because the validator enforces it afterwards either way and a model told the rule up
// front produces far fewer rejected items.
const normalizeSystem = `You extract contract scope from PUBLIC federal solicitation text.
Return ONLY compact JSON of the form {"items":[{"scope_type":"...","title":"...","description":"...","quote":"...","section_label":"...","confidence":0.0}]}.
scope_type MUST be one of: objective, task, deliverable, labor_category, staffing_quantity, skill,
certification, clearance, customer_experience, location, travel, schedule, security_control, reporting,
transition, key_personnel, pricing_instruction, evaluation_factor, compliance_clause, assumption,
dependency, risk, other.
THE QUOTE RULE: every item MUST carry a quote copied WORD FOR WORD from the text you were given. Never
paraphrase a quote and never invent one. An item whose quote is not present verbatim in the source is
discarded. If the text does not support an item, do not produce that item. Returning fewer, well grounded
items is correct.`

var (
	emailRe = regexp.MustCompile(`[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}`)
	phoneRe = regexp.MustCompile(`\(?\d{3}\)?[\s.\-]?\d{3}[\s.\-]?\d{4}`)
	uuidRe  = regexp.MustCompile(`[0-9a-fA-F]{8}\-[0-9a-fA-F]{4}\-[0-9a-fA-F]{4}\-[0-9a-fA-F]{4}\-[0-9a-fA-F]{12}`)
)

func main() {
	dsn := flag.String("dsn", "", "migrator DSN, defaults to MM_MIGRATOR_DATABASE_URL")
	mega := flag.String("megamind", "", "MEGAMIND gateway base URL, defaults to MM_MEGAMIND_URL")
	model := flag.String("model", envOr("MM_MEGAMIND_GENERATE_MODEL", "llama3.1:8b"), "generation model")
	limit := flag.Int("limit", 5, "how many tracked opportunities to normalize in this run")
	dryRun := flag.Bool("dry-run", false, "compose, gate, and validate, but write nothing")
	tokenFlag := flag.String("token", "", "gateway bearer token (defaults to the secret store)")
	flag.Parse()
	if *dsn == "" {
		*dsn = os.Getenv("MM_MIGRATOR_DATABASE_URL")
	}
	if *mega == "" {
		*mega = os.Getenv("MM_MEGAMIND_URL")
	}

	if *dsn == "" || *mega == "" {
		log.Fatal("opportunity-scope-normalize: -dsn and -megamind are required")
	}
	ctx := context.Background()
	pool, err := pgxpool.New(ctx, *dsn)
	if err != nil {
		log.Fatalf("pool: %v", err)
	}
	defer pool.Close()

	token := *tokenFlag
	if token == "" {
		if st, serr := secrets.Open(os.Getenv("MM_SECRETS_FILE")); serr == nil {
			token, _ = st.Get("MM_MEGAMIND_TOKEN")
		}
	}
	client, err := inference.New(inference.Config{
		BaseURL: *mega, GenerateModel: *model, Dimension: 384,
		Timeout: 120 * time.Second, MaxRetries: 2, Token: token,
	})
	if err != nil {
		log.Fatalf("inference: %v", err)
	}
	if err := client.Health(ctx); err != nil {
		log.Fatalf("MEGAMIND not reachable at %s: %v", *mega, err)
	}

	// Only opportunities that trace to a public SAM notice are eligible. A customer or partner sourced
	// opportunity carries private scope and must never reach this path.
	// s.provenance is SELECTED, never assumed. Migration 0081 makes that column born with the row and
	// immutable, so what comes back is what the database says about where this text came from, and the gate
	// below decides from that rather than from anything this program believes.
	rows, err := pool.Query(ctx, `
		SELECT t.organization_id::text, t.id::text, s.notice_id, s.provenance,
		       COALESCE(s.title,''), COALESCE(s.department,''), COALESCE(s.sub_agency,''),
		       COALESCE(s.naics_code,''), COALESCE(s.psc_code,''), COALESCE(s.set_aside,''),
		       COALESCE(s.notice_type,''), COALESCE(s.pop_city,''), COALESCE(s.pop_state,'')
		FROM tracked_opportunities t
		JOIN sam_opportunities s ON s.notice_id = t.sam_notice_id
		WHERE t.source_type = 'sam_gov' AND t.sam_notice_id <> ''
		  AND NOT EXISTS (SELECT 1 FROM opportunity_scope_items i
		                  WHERE i.organization_id = t.organization_id AND i.opportunity_id = t.id)
		ORDER BY t.created_at DESC
		LIMIT $1`, *limit)
	if err != nil {
		log.Fatalf("select opportunities: %v", err)
	}
	// provs carries the provenance of EVERY row this payload was composed from. Today a notice is composed
	// from exactly one public cache row, so the set has one element; when a composer later draws on several
	// rows it appends each one's stored value and the gate refuses if any of them may not leave.
	type job struct {
		org, opp, notice, text string
		provs                  []scrub.Provenance
	}
	var jobs []job
	for rows.Next() {
		var org, opp, notice, prov, title, dept, sub, naics, psc, setAside, kind, city, state string
		if err := rows.Scan(&org, &opp, &notice, &prov, &title, &dept, &sub, &naics, &psc, &setAside, &kind, &city, &state); err != nil {
			rows.Close()
			log.Fatalf("scan: %v", err)
		}
		jobs = append(jobs, job{
			org: org, opp: opp, notice: notice,
			text:  composeNoticeText(title, dept, sub, naics, psc, setAside, kind, city, state),
			provs: []scrub.Provenance{scrub.Provenance(prov)},
		})
	}
	rows.Close()
	if len(jobs) == 0 {
		log.Print("opportunity-scope-normalize: nothing to normalize (no tracked SAM opportunity without scope)")
		return
	}
	log.Printf("opportunity-scope-normalize: %d opportunities, dry_run=%t", len(jobs), *dryRun)

	totalAccepted, totalRejected := 0, 0
	for _, j := range jobs {
		payload := redactContacts(j.text)

		// THE GATE. Composed public text only, contact shapes already redacted, and the gate still has the
		// final say. A refusal skips the notice; nothing is sent and nothing is written.
		// j.provs came from the DATABASE, not from this program. If a future composer draws on a tenant
		// document, the stored provenance travels with it and this call refuses without anyone remembering to.
		// THE GATE AND THE AUDIT ROW ARE ONE CALL, and it FAILS CLOSED. Until 29 July 2026 this line called
		// AllowExternal and wrote the returned metadata to the log below and nowhere else, so the one class
		// of call that leaves this machine was the one class with no persisted audit trail.
		var orgUUID *uuid.UUID
		if id, perr := uuid.Parse(j.org); perr == nil {
			orgUUID = &id
		}
		meta, err := ai.AllowExternalAudited(ctx, pool, orgUUID, "megamind-gateway", j.provs, scrub.ClassRequirementNormalize, []byte(payload))
		if err != nil {
			log.Printf("  notice %s: egress REFUSED, nothing sent, skipping: %v", j.notice, err)
			continue
		}
		auditJSON, _ := json.Marshal(meta)
		log.Printf("  notice %s: egress allowed and audited %s", j.notice, string(auditJSON))
		contentHash, _ := meta["content_sha256"].(string)

		resp, gerr := client.Generate(ctx, inference.GenerateRequest{
			System: normalizeSystem, Prompt: "SOURCE TEXT:\n" + payload, Temperature: 0, MaxTokens: 900,
		})
		if gerr != nil {
			log.Printf("  notice %s: generation failed: %v", j.notice, gerr)
			continue
		}
		// WHICH backend actually served it. The gateway falls back to the local runtime when the selected
		// backend fails, so without this the audit cannot distinguish a real external call from a failure
		// that quietly became a local one.
		log.Printf("  notice %s: served by model %q, %d tokens", j.notice, resp.Model, resp.Tokens)
		if oerr := ai.RecordExternalOutcome(ctx, pool, orgUUID, contentHash, resp.Model, resp.Tokens); oerr != nil {
			log.Printf("  notice %s: WARNING outcome audit row not written: %v", j.notice, oerr)
		}
		accepted, rejected, verr := opportunities.ValidateScopeCandidates([]byte(resp.Text), payload)
		if verr != nil {
			log.Printf("  notice %s: model payload rejected: %v", j.notice, verr)
			continue
		}
		for _, r := range rejected {
			log.Printf("    REJECTED %q: %s", clip(r.Title, 48), r.Reason)
		}
		totalRejected += len(rejected)
		if *dryRun {
			for _, a := range accepted {
				log.Printf("    would write [%s] %s (cited)", a.ScopeType, clip(a.Title, 60))
			}
			totalAccepted += len(accepted)
			continue
		}
		written, werr := writeScope(ctx, pool, j.org, j.opp, accepted)
		if werr != nil {
			log.Printf("  notice %s: write failed: %v", j.notice, werr)
			continue
		}
		totalAccepted += written
		log.Printf("  notice %s: wrote %d cited scope items, rejected %d", j.notice, written, len(rejected))
	}
	log.Printf("opportunity-scope-normalize: DONE accepted=%d rejected=%d", totalAccepted, totalRejected)
}

// writeScope inserts the survivors as review_status extracted, which no scoring path can see. The citation
// travels with the row: the schema CHECK refuses an extracted item that carries neither a document nor a
// quote hash, so an uncited write cannot happen even if this code were wrong.
func writeScope(ctx context.Context, pool *pgxpool.Pool, org, opp string, items []opportunities.ValidatedScope) (int, error) {
	n := 0
	for i, it := range items {
		if _, err := pool.Exec(ctx, `
			INSERT INTO opportunity_scope_items (
			  organization_id, opportunity_id, scope_type, title, description,
			  source_section_label, source_quote_hash, extraction_method, extraction_confidence,
			  review_status, display_order)
			VALUES ($1,$2,$3,$4,$5,$6,$7,'model_normalize',$8,'extracted',$9)`,
			org, opp, it.ScopeType, it.Title, it.Description,
			it.SectionLabel, it.QuoteHash, it.Confidence, i); err != nil {
			return n, err
		}
		n++
	}
	return n, nil
}

// composeNoticeText builds the public source text. Every value here comes from the cached PUBLIC SAM notice
// catalog, which carries no organization_id and no tenant content.
func composeNoticeText(title, dept, sub, naics, psc, setAside, kind, city, state string) string {
	parts := []string{}
	add := func(label, v string) {
		if strings.TrimSpace(v) != "" {
			parts = append(parts, label+v)
		}
	}
	add("", title)
	add("Department: ", dept)
	add("Sub agency: ", sub)
	add("Notice type: ", kind)
	add("NAICS: ", naics)
	add("PSC: ", psc)
	add("Set aside: ", setAside)
	if city != "" || state != "" {
		add("Place of performance: ", strings.TrimSpace(city+" "+state))
	}
	return strings.Join(parts, ". ") + "."
}

// redactContacts removes the contact shapes a public notice may carry. Scope extraction has no use for a
// contracting officer's email or phone, so the cheapest way to keep them off an external service is to not
// send them. The scrub gate independently refuses the payload if anything slips through.
func redactContacts(s string) string {
	s = emailRe.ReplaceAllString(s, "[contact redacted]")
	s = phoneRe.ReplaceAllString(s, "[phone redacted]")
	s = uuidRe.ReplaceAllString(s, "[id redacted]")
	return s
}

func envOr(k, def string) string {
	if v := os.Getenv(k); v != "" {
		return v
	}
	return def
}

func clip(s string, n int) string {
	if len(s) > n {
		return s[:n] + "..."
	}
	return s
}

var _ = fmt.Sprintf