Janady13/missionmatch / docs/plans/2026-06-29-ws1-ontology-sync.md

1# WS1 Ontology Sync Implementation Plan 2 3> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task by task. Steps use checkbox (`- [ ]`) syntax for tracking. 4 5**Goal:** Build `cmd/ontology-sync`, a Go command that ingests the GovCon vocabulary from free federal sources into a new global ontology store, so the matcher has hundreds of real cleared workforce terms to match against instead of 13. 6 7**Architecture:** Mirror the proven `cmd/opp-sync` adapter pattern. A new `internal/modules/ontology` package defines a `VocabRecord`, an `OntologySource` interface, and an `Ingester` that retains each raw payload to object storage, upserts the normalized record into a new global `ontology_terms` table (migrator writes, app role reads), and writes one audit event per source. Adapters under `internal/modules/ontology/adapters` implement each source. The IC mission spine is a curated table seeded by the migration, not an adapter. This plan delivers the foundation plus the keyless adapters; O*NET ships dormant behind a missing key. 8 9**Tech Stack:** Go 1.26, pgx/v5 (`pgxpool`), goose migrations, `github.com/xuri/excelize/v2` (pure Go XLSX reader, new dependency), the existing `internal/platform/storage` Storer and `internal/platform/audit` Writer, the existing `internal/platform/extraction` deterministic extractor. CGO_ENABLED=0, GOFIPS140=v1.0.0, no Python. 10 11**Scope note:** This is WS1 of the AI Grounding Program (see `docs/AI_GROUNDING_PROGRAM.md`). WS2 (alias and relationship layer), WS3 (clearance modeling), and Phases 2 and 3 each get their own plan. WS1 creates the staging and reference tables (`ontology_terms`, `ontology_crosswalk`, `mission_spine`); the matching facing promotion of these terms into the global `skills` table happens in WS2. 12 13--- 14 15## File Structure 16 17| File | Responsibility | New or Modify | 18| --- | --- | --- | 19| `db/migrations/0021_ontology.sql` | the three global ontology tables plus the seeded mission spine | Create | 20| `internal/modules/ontology/record.go` | `VocabRecord`, `TermKind` closed set | Create | 21| `internal/modules/ontology/source.go` | `OntologySource` interface, `ErrThrottled` | Create | 22| `internal/modules/ontology/ingest.go` | `Ingester` (retain, upsert, audit), `RecordWriter`, `Auditor`, `IngestResult` | Create | 23| `internal/modules/ontology/writer.go` | `DBWriter` upsert into `ontology_terms` via `pool.Exec` | Create | 24| `internal/modules/ontology/ingest_test.go` | ingester unit tests with `storage.NewMemory` and fakes | Create | 25| `internal/modules/ontology/adapters/psc.go` | PSC manual XLSX adapter + `parsePSC` | Create | 26| `internal/modules/ontology/adapters/psc_test.go` | `parsePSC` unit test on an in memory XLSX | Create | 27| `internal/modules/ontology/adapters/corpus.go` | mines our retained SAM and award text via the extractor | Create | 28| `internal/modules/ontology/adapters/corpus_test.go` | corpus adapter unit test | Create | 29| `internal/modules/ontology/adapters/nice.go` | NICE Framework components JSON adapter | Create | 30| `internal/modules/ontology/adapters/calc.go` | GSA CALC labor category sweep adapter | Create | 31| `internal/modules/ontology/adapters/onet.go` | O*NET adapter, dormant without `ONET_API_KEY` | Create | 32| `cmd/ontology-sync/main.go` | the CLI: wire sources, loop ingest, log | Create | 33| `Makefile` | add the `ontology-sync` FIPS build target | Modify | 34| `go.mod` / `go.sum` | add `github.com/xuri/excelize/v2` | Modify | 35 36--- 37 38## Task 1: Migration 0021, the global ontology tables 39 40**Files:** 41- Create: `db/migrations/0021_ontology.sql` 42 43- [ ] **Step 1: Write the migration** 44 45```sql 46-- +goose Up 47-- WS1 ontology staging and reference tables (AI Grounding Program). Like skills, sam_opportunities, 48-- and opportunity_records these are GLOBAL PUBLIC reference data, NOT tenant data: the federal 49-- vocabulary is identical for every org, so these tables carry NO organization_id and sit OUTSIDE Row 50-- Level Security. The runtime app role gets SELECT only; cmd/ontology-sync writes as the migrator role. 51-- ontology_terms is the raw pulled vocabulary with provenance; ontology_crosswalk links codes across 52-- frameworks (NICE to DCWF, SOC to skill); mission_spine is the hand curated IC vocabulary, seeded here. 53 54CREATE TABLE ontology_terms ( 55 id uuid PRIMARY KEY DEFAULT gen_random_uuid(), 56 source_name text NOT NULL, -- provenance label, for example PSC Manual 57 source_record_id text NOT NULL, -- the source's own id (upsert key with source_name) 58 term_kind text NOT NULL CHECK (term_kind IN ('skill','labor_category','work_role','ksa','psc_code','mission_area')), 59 term text NOT NULL DEFAULT '', -- the canonical text 60 code text NOT NULL DEFAULT '', -- the source code (PSC code, SOC code, NICE element id) 61 category text NOT NULL DEFAULT '', 62 description text NOT NULL DEFAULT '', 63 raw_object_key text NOT NULL DEFAULT '', -- object storage key for the retained raw payload 64 raw_sha256 text NOT NULL DEFAULT '', -- integrity stamp of the retained payload 65 fetched_at timestamptz NOT NULL, 66 synced_at timestamptz NOT NULL DEFAULT now(), 67 UNIQUE (source_name, source_record_id) 68); 69CREATE INDEX idx_ontology_terms_kind ON ontology_terms (term_kind); 70CREATE INDEX idx_ontology_terms_code ON ontology_terms (code); 71 72CREATE TABLE ontology_crosswalk ( 73 id uuid PRIMARY KEY DEFAULT gen_random_uuid(), 74 from_system text NOT NULL, -- for example NICE 75 from_code text NOT NULL, 76 to_system text NOT NULL, -- for example DCWF 77 to_code text NOT NULL, 78 relation text NOT NULL DEFAULT 'maps_to', 79 source_name text NOT NULL DEFAULT '', 80 synced_at timestamptz NOT NULL DEFAULT now(), 81 UNIQUE (from_system, from_code, to_system, to_code) 82); 83 84CREATE TABLE mission_spine ( 85 id uuid PRIMARY KEY DEFAULT gen_random_uuid(), 86 version text NOT NULL, 87 term text NOT NULL, 88 canonical_definition text NOT NULL DEFAULT '', 89 authority text NOT NULL DEFAULT '', -- ODNI, CNSSI 4009, JP 3-14 90 source_doc text NOT NULL DEFAULT '', 91 source_url text NOT NULL DEFAULT '', 92 effective_date date, 93 UNIQUE (version, term) 94); 95 96-- Starter curated spine (v1). Expanded by hand; each term cites its governing authority. 97INSERT INTO mission_spine (version, term, canonical_definition, authority, source_doc) VALUES 98 ('v1','SIGINT','Intelligence derived from signals, communications, electronic, and foreign instrumentation.','ODNI','Intelligence disciplines'), 99 ('v1','GEOINT','Intelligence from the exploitation and analysis of imagery and geospatial information.','ODNI','Intelligence disciplines'), 100 ('v1','HUMINT','Intelligence collected from human sources.','ODNI','Intelligence disciplines'), 101 ('v1','MASINT','Measurement and signature intelligence.','ODNI','Intelligence disciplines'), 102 ('v1','OSINT','Intelligence produced from publicly available information.','ODNI','IC OSINT Strategy'), 103 ('v1','Satellite C2','Command and control of satellite systems.','JP 3-14','Space Operations'), 104 ('v1','Cross Domain Solution','A controlled interface providing manual or automated access between security domains.','CNSSI 4009','National Information Assurance Glossary'), 105 ('v1','COMSEC','Communications security: measures to deny unauthorized access to telecommunications.','CNSSI 4009','National Information Assurance Glossary'), 106 ('v1','TEMPEST','Protection against compromising emanations.','CNSSI 4009','National Information Assurance Glossary'), 107 ('v1','Electronic Warfare','Military action using electromagnetic and directed energy.','JP 3-13.1','Electronic Warfare'); 108 109GRANT SELECT ON ontology_terms TO missionmatch_app; 110GRANT SELECT ON ontology_crosswalk TO missionmatch_app; 111GRANT SELECT ON mission_spine TO missionmatch_app; 112 113-- +goose Down 114DROP TABLE IF EXISTS mission_spine; 115DROP TABLE IF EXISTS ontology_crosswalk; 116DROP TABLE IF EXISTS ontology_terms; 117``` 118 119- [ ] **Step 2: Apply and roll back on a scratch database to prove both directions** 120 121Run (local scratch or M4 staging migrator DSN): 122```bash 123goose -dir db/migrations postgres "$MM_MIGRATOR_DATABASE_URL" up 124goose -dir db/migrations postgres "$MM_MIGRATOR_DATABASE_URL" status # 0021 applied, head v21 125goose -dir db/migrations postgres "$MM_MIGRATOR_DATABASE_URL" down # drops the three tables 126goose -dir db/migrations postgres "$MM_MIGRATOR_DATABASE_URL" up # reapply, leave applied 127``` 128Expected: up creates three tables and seeds 10 mission_spine rows; down drops cleanly; status shows head v21. 129 130- [ ] **Step 3: Commit** 131 132```bash 133git add db/migrations/0021_ontology.sql 134git commit -m "feat(ontology): add WS1 global ontology tables and seeded mission spine" 135``` 136 137--- 138 139## Task 2: VocabRecord and the OntologySource interface 140 141**Files:** 142- Create: `internal/modules/ontology/record.go` 143- Create: `internal/modules/ontology/source.go` 144- Test: `internal/modules/ontology/record_test.go` 145 146- [ ] **Step 1: Write the failing test** 147 148```go 149package ontology 150 151import "testing" 152 153func TestTermKindValid(t *testing.T) { 154 for _, k := range []TermKind{KindSkill, KindLaborCategory, KindWorkRole, KindKSA, KindPSC, KindMissionArea} { 155 if !k.Valid() { 156 t.Fatalf("expected %q valid", k) 157 } 158 } 159 if TermKind("bogus").Valid() { 160 t.Fatal("expected bogus invalid") 161 } 162} 163``` 164 165- [ ] **Step 2: Run to verify it fails** 166 167Run: `go test ./internal/modules/ontology/ -run TestTermKindValid` 168Expected: FAIL, undefined: TermKind. 169 170- [ ] **Step 3: Write record.go** 171 172```go 173// Package ontology ingests the GovCon vocabulary from free federal sources into the global ontology 174// store. It mirrors the opportunity_intelligence ingestion shape: a streaming source contract, a single 175// ingester that retains each raw payload and upserts the normalized record, and one audit event per run. 176package ontology 177 178import "encoding/json" 179 180// TermKind is the closed set of vocabulary record kinds. A closed set keeps the global table queryable 181// and lets the WS2 promotion step reason about provenance. 182type TermKind string 183 184const ( 185 KindSkill TermKind = "skill" 186 KindLaborCategory TermKind = "labor_category" 187 KindWorkRole TermKind = "work_role" 188 KindKSA TermKind = "ksa" 189 KindPSC TermKind = "psc_code" 190 KindMissionArea TermKind = "mission_area" 191) 192 193// Valid reports whether k is one of the allowed kinds (it must match the CHECK in migration 0021). 194func (k TermKind) Valid() bool { 195 switch k { 196 case KindSkill, KindLaborCategory, KindWorkRole, KindKSA, KindPSC, KindMissionArea: 197 return true 198 } 199 return false 200} 201 202// VocabRecord is one normalized vocabulary item from any ontology source. SourceName and SourceRecordID 203// form the upsert key. Term is the canonical text, Code is the source's own identifier (a PSC code, a SOC 204// code, a NICE element id), Category groups it, and Raw is the retained source payload. 205type VocabRecord struct { 206 SourceName string 207 SourceRecordID string 208 Kind TermKind 209 Term string 210 Code string 211 Category string 212 Description string 213 Raw json.RawMessage 214} 215``` 216 217- [ ] **Step 4: Write source.go** 218 219```go 220package ontology 221 222import ( 223 "context" 224 "errors" 225) 226 227// ErrThrottled signals a source hit its provider rate limit (a 429). The ingester treats it as a soft 228// stop: it keeps and reports whatever was ingested before the limit. 229var ErrThrottled = errors.New("ontology: source rate limited") 230 231// OntologySource is the provider adapter contract, one per free federal vocabulary feed. Each handles its 232// own paging internally and calls yield once per normalized record. A yield error stops the pull and is 233// returned, so a storage or database failure aborts the run fail closed. 234type OntologySource interface { 235 // Name is the provenance label stamped on every record this source yields (for example PSC Manual). 236 Name() string 237 // Fetch pulls the source's scope and calls yield once per normalized record. On a provider rate limit 238 // it returns ErrThrottled so the ingester can stop that source gracefully. 239 Fetch(ctx context.Context, yield func(VocabRecord) error) error 240} 241``` 242 243- [ ] **Step 5: Run to verify it passes** 244 245Run: `go test ./internal/modules/ontology/ -run TestTermKindValid` 246Expected: PASS. 247 248- [ ] **Step 6: Commit** 249 250```bash 251git add internal/modules/ontology/record.go internal/modules/ontology/source.go internal/modules/ontology/record_test.go 252git commit -m "feat(ontology): VocabRecord and OntologySource contract" 253``` 254 255--- 256 257## Task 3: The Ingester, RecordWriter, Auditor, and DBWriter 258 259**Files:** 260- Create: `internal/modules/ontology/ingest.go` 261- Create: `internal/modules/ontology/writer.go` 262- Test: `internal/modules/ontology/ingest_test.go` 263 264- [ ] **Step 1: Write the failing test** (uses `storage.NewMemory` and fakes, no database) 265 266```go 267package ontology 268 269import ( 270 "context" 271 "encoding/json" 272 "testing" 273 274 "github.com/missionworkforce/missionmatch/internal/platform/audit" 275 "github.com/missionworkforce/missionmatch/internal/platform/storage" 276) 277 278type fakeSource struct{ recs []VocabRecord } 279 280func (f *fakeSource) Name() string { return "Fake" } 281func (f *fakeSource) Fetch(_ context.Context, yield func(VocabRecord) error) error { 282 for _, r := range f.recs { 283 if err := yield(r); err != nil { 284 return err 285 } 286 } 287 return nil 288} 289 290type fakeWriter struct{ got []VocabRecord } 291 292func (w *fakeWriter) Upsert(_ context.Context, rec VocabRecord, key, sha string) error { 293 if key == "" || sha == "" { 294 return assertErr("expected non empty raw key and sha") 295 } 296 w.got = append(w.got, rec) 297 return nil 298} 299 300type fakeAuditor struct{ events int } 301 302func (a *fakeAuditor) System(_ context.Context, _ audit.Event) error { a.events++; return nil } 303 304func assertErr(s string) error { return &simpleErr{s} } 305 306type simpleErr struct{ s string } 307 308func (e *simpleErr) Error() string { return e.s } 309 310func TestIngesterRetainsUpsertsAudits(t *testing.T) { 311 src := &fakeSource{recs: []VocabRecord{ 312 {SourceName: "Fake", SourceRecordID: "R375", Kind: KindPSC, Term: "Maintenance of Aircraft", Code: "R375", Raw: json.RawMessage(`{"code":"R375"}`)}, 313 {SourceRecordID: "AC12", Kind: KindPSC, Term: "National Defense R and D", Code: "AC12"}, 314 }} 315 w := &fakeWriter{} 316 a := &fakeAuditor{} 317 ing := NewIngester(storage.NewMemory(), w, a) 318 319 res, err := ing.Ingest(context.Background(), src) 320 if err != nil { 321 t.Fatalf("ingest: %v", err) 322 } 323 if res.Fetched != 2 || res.Upserted != 2 { 324 t.Fatalf("counts: %+v", res) 325 } 326 if len(w.got) != 2 { 327 t.Fatalf("writer got %d", len(w.got)) 328 } 329 if w.got[1].SourceName != "Fake" { 330 t.Fatal("empty SourceName should default to the source Name") 331 } 332 if a.events != 1 { 333 t.Fatalf("expected one audit event, got %d", a.events) 334 } 335} 336 337func TestIngesterRejectsInvalidKind(t *testing.T) { 338 src := &fakeSource{recs: []VocabRecord{{SourceRecordID: "X", Kind: "bogus"}}} 339 ing := NewIngester(storage.NewMemory(), &fakeWriter{}, &fakeAuditor{}) 340 if _, err := ing.Ingest(context.Background(), src); err == nil { 341 t.Fatal("expected an error on an invalid kind") 342 } 343} 344``` 345 346- [ ] **Step 2: Run to verify it fails** 347 348Run: `go test ./internal/modules/ontology/ -run TestIngester` 349Expected: FAIL, undefined: NewIngester. 350 351- [ ] **Step 3: Write ingest.go** 352 353```go 354package ontology 355 356import ( 357 "context" 358 "encoding/json" 359 "errors" 360 "fmt" 361 362 "github.com/missionworkforce/missionmatch/internal/platform/audit" 363 "github.com/missionworkforce/missionmatch/internal/platform/storage" 364) 365 366// RecordWriter persists a normalized record. The pool backed implementation (NewDBWriter) upserts into 367// ontology_terms; tests use a fake. rawObjectKey and rawSHA256 are the storage pointer and integrity stamp. 368type RecordWriter interface { 369 Upsert(ctx context.Context, rec VocabRecord, rawObjectKey, rawSHA256 string) error 370} 371 372// Auditor records the platform ingest event. The concrete audit.Writer satisfies it through System. 373type Auditor interface { 374 System(ctx context.Context, e audit.Event) error 375} 376 377// IngestResult is the real, counted outcome of one source pull. Every count comes from records actually 378// retained and written, never a fabricated number. 379type IngestResult struct { 380 Source string `json:"source"` 381 Fetched int `json:"fetched"` 382 Upserted int `json:"upserted"` 383 ByKind map[string]int `json:"by_kind"` 384} 385 386// Ingester pulls a source, retains each raw payload in object storage, upserts the normalized record, and 387// writes one audit event for the run. It is the single ingestion path shared by every ontology source. 388type Ingester struct { 389 store storage.Storer 390 writer RecordWriter 391 auditor Auditor 392} 393 394// NewIngester builds an ingester. All three dependencies are required. 395func NewIngester(store storage.Storer, writer RecordWriter, auditor Auditor) *Ingester { 396 return &Ingester{store: store, writer: writer, auditor: auditor} 397} 398 399// Ingest pulls every record from src, retaining and upserting each, then writes the ingest audit event. A 400// storage or database failure aborts the pull fail closed. A provider rate limit is a soft stop. The audit 401// write is fail closed, and because every upsert is idempotent a rerun is safe. 402func (i *Ingester) Ingest(ctx context.Context, src OntologySource) (IngestResult, error) { 403 res := IngestResult{Source: src.Name(), ByKind: map[string]int{}} 404 err := src.Fetch(ctx, func(rec VocabRecord) error { 405 if rec.SourceName == "" { 406 rec.SourceName = src.Name() 407 } 408 if !rec.Kind.Valid() { 409 return fmt.Errorf("ontology: record %q from %s has invalid kind %q", rec.SourceRecordID, res.Source, rec.Kind) 410 } 411 if rec.SourceRecordID == "" { 412 return fmt.Errorf("ontology: record from %s has an empty source record id", res.Source) 413 } 414 key, sha, serr := i.retainRaw(ctx, rec) 415 if serr != nil { 416 return serr 417 } 418 if werr := i.writer.Upsert(ctx, rec, key, sha); werr != nil { 419 return werr 420 } 421 res.Fetched++ 422 res.Upserted++ 423 res.ByKind[string(rec.Kind)]++ 424 return nil 425 }) 426 if err != nil && !errors.Is(err, ErrThrottled) { 427 return res, err 428 } 429 if aerr := i.audit(ctx, res); aerr != nil { 430 return res, aerr 431 } 432 return res, nil 433} 434 435func (i *Ingester) retainRaw(ctx context.Context, rec VocabRecord) (string, string, error) { 436 payload := rec.Raw 437 if len(payload) == 0 { 438 payload = json.RawMessage("{}") 439 } 440 key := rawKey(rec.SourceName, rec.SourceRecordID) 441 sha := storage.SHA256Hex(payload) 442 if err := i.store.Put(ctx, key, payload); err != nil { 443 return "", "", fmt.Errorf("ontology: retain raw: %w", err) 444 } 445 return key, sha, nil 446} 447 448func (i *Ingester) audit(ctx context.Context, res IngestResult) error { 449 body, err := json.Marshal(res) 450 if err != nil { 451 return fmt.Errorf("ontology: marshal ingest audit: %w", err) 452 } 453 return i.auditor.System(ctx, audit.Event{ 454 Action: "ontology.ingest", 455 EntityType: "ontology_source", 456 EntityID: res.Source, 457 NewValues: body, 458 }) 459} 460 461func rawKey(source, id string) string { 462 return "ontology_terms/" + slug(source) + "/" + slug(id) + ".json" 463} 464 465func slug(s string) string { 466 out := make([]rune, 0, len(s)) 467 for _, r := range s { 468 switch { 469 case r >= 'a' && r <= 'z', r >= '0' && r <= '9', r == '.', r == '_': 470 out = append(out, r) 471 case r >= 'A' && r <= 'Z': 472 out = append(out, r+('a'-'A')) 473 default: 474 out = append(out, '_') 475 } 476 } 477 if len(out) == 0 { 478 return "unknown" 479 } 480 return string(out) 481} 482``` 483 484- [ ] **Step 4: Write writer.go** 485 486```go 487package ontology 488 489import ( 490 "context" 491 492 "github.com/jackc/pgx/v5/pgxpool" 493) 494 495// DBWriter upserts a VocabRecord into the global ontology_terms table. It runs as the migrator role (the 496// only role granted write on the global reference tables). The upsert key is (source_name, source_record_id), 497// so a rerun refreshes a term in place and the command is idempotent. 498type DBWriter struct{ pool *pgxpool.Pool } 499 500// NewDBWriter builds a DBWriter over a migrator pool. 501func NewDBWriter(pool *pgxpool.Pool) *DBWriter { return &DBWriter{pool: pool} } 502 503// Upsert writes one record, refreshing it on a (source_name, source_record_id) conflict. 504func (w *DBWriter) Upsert(ctx context.Context, rec VocabRecord, rawObjectKey, rawSHA256 string) error { 505 _, err := w.pool.Exec(ctx, ` 506INSERT INTO ontology_terms 507 (source_name, source_record_id, term_kind, term, code, category, description, raw_object_key, raw_sha256, fetched_at) 508VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9, now()) 509ON CONFLICT (source_name, source_record_id) DO UPDATE SET 510 term_kind=EXCLUDED.term_kind, term=EXCLUDED.term, code=EXCLUDED.code, category=EXCLUDED.category, 511 description=EXCLUDED.description, raw_object_key=EXCLUDED.raw_object_key, raw_sha256=EXCLUDED.raw_sha256, 512 synced_at=now()`, 513 rec.SourceName, rec.SourceRecordID, string(rec.Kind), rec.Term, rec.Code, rec.Category, rec.Description, 514 rawObjectKey, rawSHA256) 515 return err 516} 517``` 518 519- [ ] **Step 5: Run to verify it passes** 520 521Run: `go test ./internal/modules/ontology/ -run TestIngester` 522Expected: PASS (both tests). 523 524- [ ] **Step 6: Commit** 525 526```bash 527git add internal/modules/ontology/ingest.go internal/modules/ontology/writer.go internal/modules/ontology/ingest_test.go 528git commit -m "feat(ontology): ingester, db writer, and audited retain path" 529``` 530 531--- 532 533## Task 4: The PSC adapter (XLSX), the first real source 534 535**Files:** 536- Create: `internal/modules/ontology/adapters/psc.go` 537- Test: `internal/modules/ontology/adapters/psc_test.go` 538- Modify: `go.mod`, `go.sum` (add excelize) 539 540- [ ] **Step 1: Add the excelize dependency** 541 542Run: `go get github.com/xuri/excelize/v2@latest` 543Expected: go.mod and go.sum updated; excelize is pure Go (no cgo), so the FIPS build is unaffected. 544 545- [ ] **Step 2: Write the failing test** (build a tiny XLSX in memory, then parse it) 546 547```go 548package adapters 549 550import ( 551 "bytes" 552 "testing" 553 554 "github.com/xuri/excelize/v2" 555 556 "github.com/missionworkforce/missionmatch/internal/modules/ontology" 557) 558 559func buildPSCXLSX(t *testing.T) []byte { 560 t.Helper() 561 f := excelize.NewFile() 562 sh := f.GetSheetName(0) 563 _ = f.SetCellValue(sh, "A1", "PSC CODE") 564 _ = f.SetCellValue(sh, "B1", "PRODUCT AND SERVICE CODE NAME") 565 _ = f.SetCellValue(sh, "A2", "R425") 566 _ = f.SetCellValue(sh, "B2", "Engineering and Technical Services") 567 _ = f.SetCellValue(sh, "A3", "AC12") 568 _ = f.SetCellValue(sh, "B3", "National Defense R and D Services") 569 var buf bytes.Buffer 570 if err := f.Write(&buf); err != nil { 571 t.Fatalf("write xlsx: %v", err) 572 } 573 return buf.Bytes() 574} 575 576func TestParsePSC(t *testing.T) { 577 recs, err := parsePSC(buildPSCXLSX(t)) 578 if err != nil { 579 t.Fatalf("parse: %v", err) 580 } 581 if len(recs) != 2 { 582 t.Fatalf("want 2 records, got %d", len(recs)) 583 } 584 if recs[0].Code != "R425" || recs[0].Kind != ontology.KindPSC { 585 t.Fatalf("bad first record: %+v", recs[0]) 586 } 587 if recs[1].Term != "National Defense R and D Services" { 588 t.Fatalf("bad second term: %q", recs[1].Term) 589 } 590 if recs[0].SourceRecordID != "R425" { 591 t.Fatalf("source record id should be the code: %q", recs[0].SourceRecordID) 592 } 593} 594``` 595 596- [ ] **Step 3: Run to verify it fails** 597 598Run: `go test ./internal/modules/ontology/adapters/ -run TestParsePSC` 599Expected: FAIL, undefined: parsePSC. 600 601- [ ] **Step 4: Write psc.go** 602 603```go 604package adapters 605 606import ( 607 "bytes" 608 "context" 609 "encoding/json" 610 "fmt" 611 "io" 612 "net/http" 613 "strings" 614 "time" 615 616 "github.com/xuri/excelize/v2" 617 618 "github.com/missionworkforce/missionmatch/internal/modules/ontology" 619) 620 621// PSCSource ingests the federal Product and Service Code manual, the cleanest fully enumerable procurement 622// taxonomy. It reads the published XLSX (a URL or a local file) and yields one KindPSC record per code. 623type PSCSource struct { 624 url string 625 client *http.Client 626} 627 628// NewPSCSource builds the adapter. url is the PSC manual XLSX location (acquisition.gov publishes it). 629func NewPSCSource(url string, client *http.Client) *PSCSource { 630 if client == nil { 631 client = &http.Client{Timeout: 60 * time.Second} 632 } 633 return &PSCSource{url: url, client: client} 634} 635 636// Name is the provenance label. 637func (s *PSCSource) Name() string { return "PSC Manual" } 638 639// Fetch downloads the XLSX and yields one record per code. 640func (s *PSCSource) Fetch(ctx context.Context, yield func(ontology.VocabRecord) error) error { 641 data, err := s.download(ctx) 642 if err != nil { 643 return err 644 } 645 recs, err := parsePSC(data) 646 if err != nil { 647 return err 648 } 649 for _, r := range recs { 650 if err := yield(r); err != nil { 651 return err 652 } 653 } 654 return nil 655} 656 657func (s *PSCSource) download(ctx context.Context) ([]byte, error) { 658 req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.url, nil) 659 if err != nil { 660 return nil, fmt.Errorf("psc: new request: %w", err) 661 } 662 resp, err := s.client.Do(req) 663 if err != nil { 664 return nil, fmt.Errorf("psc: get: %w", err) 665 } 666 defer resp.Body.Close() 667 if resp.StatusCode != http.StatusOK { 668 return nil, fmt.Errorf("psc: provider returned %s", resp.Status) 669 } 670 return io.ReadAll(resp.Body) 671} 672 673// parsePSC reads the PSC manual XLSX bytes and returns one VocabRecord per code. It finds the code and name 674// columns by header text, so a column reorder in a later manual revision does not break it. 675func parsePSC(data []byte) ([]ontology.VocabRecord, error) { 676 f, err := excelize.OpenReader(bytes.NewReader(data)) 677 if err != nil { 678 return nil, fmt.Errorf("psc: open xlsx: %w", err) 679 } 680 defer f.Close() 681 sheet := f.GetSheetName(0) 682 rows, err := f.GetRows(sheet) 683 if err != nil { 684 return nil, fmt.Errorf("psc: read rows: %w", err) 685 } 686 if len(rows) < 2 { 687 return nil, fmt.Errorf("psc: sheet has no data rows") 688 } 689 codeCol, nameCol := -1, -1 690 for i, h := range rows[0] { 691 switch { 692 case strings.Contains(strings.ToUpper(h), "PSC") && strings.Contains(strings.ToUpper(h), "CODE"): 693 codeCol = i 694 case strings.Contains(strings.ToUpper(h), "NAME"): 695 nameCol = i 696 } 697 } 698 if codeCol < 0 || nameCol < 0 { 699 return nil, fmt.Errorf("psc: could not find the code and name columns in the header %v", rows[0]) 700 } 701 out := make([]ontology.VocabRecord, 0, len(rows)-1) 702 for _, row := range rows[1:] { 703 if codeCol >= len(row) || nameCol >= len(row) { 704 continue 705 } 706 code := strings.TrimSpace(row[codeCol]) 707 name := strings.TrimSpace(row[nameCol]) 708 if code == "" || name == "" { 709 continue 710 } 711 raw, _ := json.Marshal(map[string]string{"code": code, "name": name}) 712 out = append(out, ontology.VocabRecord{ 713 SourceName: "PSC Manual", 714 SourceRecordID: code, 715 Kind: ontology.KindPSC, 716 Term: name, 717 Code: code, 718 Category: "procurement", 719 Raw: raw, 720 }) 721 } 722 return out, nil 723} 724``` 725 726- [ ] **Step 5: Run to verify it passes** 727 728Run: `go test ./internal/modules/ontology/adapters/ -run TestParsePSC` 729Expected: PASS. 730 731- [ ] **Step 6: Commit** 732 733```bash 734git add internal/modules/ontology/adapters/psc.go internal/modules/ontology/adapters/psc_test.go go.mod go.sum 735git commit -m "feat(ontology): PSC manual XLSX adapter" 736``` 737 738--- 739 740## Task 5: The CLI, the Makefile target, and the FIPS build 741 742**Files:** 743- Create: `cmd/ontology-sync/main.go` 744- Modify: `Makefile` 745 746- [ ] **Step 1: Write cmd/ontology-sync/main.go** 747 748```go 749// Command ontology-sync ingests the GovCon vocabulary from free federal sources into the global 750// ontology_terms table, mirroring cmd/opp-sync. It runs as the migrator role and is safe to rerun: every 751// upsert is idempotent. The O*NET adapter is included only when ONET_API_KEY is set; the others need no key. 752// 753// Env: MM_MIGRATOR_DATABASE_URL (or MM_DATABASE_URL), MM_ONTOLOGY_RAW_DIR (object storage root, default 754// ./data/ontology-raw), MM_PSC_XLSX_URL (the PSC manual XLSX URL), ONET_API_KEY (optional, dormant adapter). 755package main 756 757import ( 758 "context" 759 "log" 760 "net/http" 761 "os" 762 "strings" 763 "time" 764 765 "github.com/jackc/pgx/v5/pgxpool" 766 767 "github.com/missionworkforce/missionmatch/internal/modules/ontology" 768 "github.com/missionworkforce/missionmatch/internal/modules/ontology/adapters" 769 "github.com/missionworkforce/missionmatch/internal/platform/audit" 770 "github.com/missionworkforce/missionmatch/internal/platform/storage" 771 "github.com/missionworkforce/missionmatch/internal/platform/tenant" 772) 773 774func main() { 775 dsn := firstNonEmpty(os.Getenv("MM_MIGRATOR_DATABASE_URL"), os.Getenv("MM_DATABASE_URL")) 776 if dsn == "" { 777 log.Fatal("ontology-sync: set MM_MIGRATOR_DATABASE_URL (the migrator DSN)") 778 } 779 rawDir := firstNonEmpty(os.Getenv("MM_ONTOLOGY_RAW_DIR"), "./data/ontology-raw") 780 pscURL := strings.TrimSpace(os.Getenv("MM_PSC_XLSX_URL")) 781 782 ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute) 783 defer cancel() 784 785 pool, err := pgxpool.New(ctx, dsn) 786 if err != nil { 787 log.Fatalf("ontology-sync: connect: %v", err) 788 } 789 defer pool.Close() 790 791 store, err := storage.NewLocal(rawDir) 792 if err != nil { 793 log.Fatalf("ontology-sync: storage: %v", err) 794 } 795 796 ing := ontology.NewIngester(store, ontology.NewDBWriter(pool), audit.New(tenant.New(pool))) 797 798 var sources []ontology.OntologySource 799 if pscURL != "" { 800 sources = append(sources, adapters.NewPSCSource(pscURL, &http.Client{Timeout: 60 * time.Second})) 801 } else { 802 log.Print("ontology-sync: MM_PSC_XLSX_URL not set, skipping the PSC adapter") 803 } 804 // The corpus adapter (Task 6) and the nice and calc adapters (Tasks 7 and 8) append here. 805 // The O*NET adapter is dormant until its key lands: 806 if k := strings.TrimSpace(os.Getenv("ONET_API_KEY")); k != "" { 807 sources = append(sources, adapters.NewONETSource(k, nil)) 808 } else { 809 log.Print("ontology-sync: ONET_API_KEY not set, O*NET adapter is dormant (skipped)") 810 } 811 812 if len(sources) == 0 { 813 log.Fatal("ontology-sync: no sources configured") 814 } 815 816 var grand int 817 for _, src := range sources { 818 res, ierr := ing.Ingest(ctx, src) 819 if ierr != nil { 820 log.Fatalf("ontology-sync: ingest %s: %v", src.Name(), ierr) 821 } 822 log.Printf("ontology-sync: %s fetched %d upserted %d by_kind %v", res.Source, res.Fetched, res.Upserted, res.ByKind) 823 grand += res.Upserted 824 } 825 log.Printf("ontology-sync: done, %d terms upserted into ontology_terms", grand) 826} 827 828func firstNonEmpty(vals ...string) string { 829 for _, v := range vals { 830 if s := strings.TrimSpace(v); s != "" { 831 return s 832 } 833 } 834 return "" 835} 836``` 837 838Note: this references `adapters.NewONETSource`, defined in Task 9. Until Task 9 lands, comment out the O*NET block or implement Task 9 first. The recommended order is Task 6 (corpus) then 7, 8, 9. 839 840- [ ] **Step 2: Add the Makefile build target** 841 842Modify `Makefile`: in the `build` target add the line (after the `opp-sync` line, line 16), and add `ontology-sync` to the `.PHONY` list: 843```makefile 844 CGO_ENABLED=0 GOOS=linux GOARCH=amd64 GOFIPS140=v1.0.0 $(GO) build -trimpath -o ontology-sync ./cmd/ontology-sync 845``` 846 847- [ ] **Step 3: Build it FIPS clean and verify the stamp** 848 849Run: 850```bash 851CGO_ENABLED=0 GOOS=linux GOARCH=amd64 GOFIPS140=v1.0.0 go build -trimpath -o /tmp/ontology-sync ./cmd/ontology-sync 852go version -m /tmp/ontology-sync | grep GOFIPS140 853``` 854Expected: builds clean, prints `GOFIPS140=v1.0.0`. 855 856- [ ] **Step 4: Commit** 857 858```bash 859git add cmd/ontology-sync/main.go Makefile 860git commit -m "feat(ontology): ontology-sync CLI and FIPS build target" 861``` 862 863--- 864 865## Task 6: The corpus adapter (mine our own retained SAM and award text) 866 867**Files:** 868- Create: `internal/modules/ontology/adapters/corpus.go` 869- Test: `internal/modules/ontology/adapters/corpus_test.go` 870 871This adapter needs no external API. It reads the description and raw text we already retain in `sam_opportunities`, `usaspending_awards`, and `opportunity_records`, runs the existing deterministic extractor (`internal/platform/extraction`, `NewDeterministic().ExtractRequirement(ctx, text)`), and yields one `KindSkill` or `KindLaborCategory` VocabRecord per distinct extracted term, deduped, with `SourceName` "Corpus Mining" and `SourceRecordID` the lowercased term. 872 873- [ ] **Step 1: Read `internal/platform/extraction` to confirm the `RawExtraction` field that holds proposed skills (the slice the extractor returns), so the adapter maps it exactly.** 874 875- [ ] **Step 2: Write the failing test** with a stub text reader returning two opportunity blurbs, asserting the adapter yields deduped skill terms. 876 877```go 878package adapters 879 880import ( 881 "context" 882 "testing" 883 884 "github.com/missionworkforce/missionmatch/internal/modules/ontology" 885) 886 887type stubText struct{ blobs []string } 888 889func (s stubText) Texts(_ context.Context, yield func(string) error) error { 890 for _, b := range s.blobs { 891 if err := yield(b); err != nil { 892 return err 893 } 894 } 895 return nil 896} 897 898func TestCorpusYieldsDedupedSkills(t *testing.T) { 899 src := NewCorpusSource(stubText{blobs: []string{ 900 "Seeking a SIGINT analyst with Splunk and AWS experience.", 901 "Cloud engineer with AWS and Kubernetes.", 902 }}) 903 seen := map[string]int{} 904 err := src.Fetch(context.Background(), func(r ontology.VocabRecord) error { 905 seen[r.Term]++ 906 if r.Kind != ontology.KindSkill && r.Kind != ontology.KindLaborCategory { 907 t.Fatalf("unexpected kind %q", r.Kind) 908 } 909 return nil 910 }) 911 if err != nil { 912 t.Fatalf("fetch: %v", err) 913 } 914 if seen["AWS"] != 1 { 915 t.Fatalf("AWS should be yielded once (deduped), got %d", seen["AWS"]) 916 } 917} 918``` 919 920- [ ] **Step 3: Write corpus.go** with a `TextReader` interface (`Texts(ctx, yield func(string) error) error`), a `NewCorpusSource(TextReader)` constructor, a pool backed `TextReader` that selects `description`/`raw` from the three tables, and a `Fetch` that runs the extractor per text and yields each distinct term once. Reuse `extraction.NewDeterministic()`. 921 922- [ ] **Step 4: Run to verify it passes.** Run: `go test ./internal/modules/ontology/adapters/ -run TestCorpus`. Expected: PASS. 923 924- [ ] **Step 5: Wire it into the CLI** (append `adapters.NewCorpusSource(adapters.NewPoolText(pool))` to `sources`), then commit. 925 926```bash 927git add internal/modules/ontology/adapters/corpus.go internal/modules/ontology/adapters/corpus_test.go cmd/ontology-sync/main.go 928git commit -m "feat(ontology): corpus mining adapter over retained federal text" 929``` 930 931--- 932 933## Task 7: The NICE Framework adapter (components JSON) 934 935**Files:** Create `internal/modules/ontology/adapters/nice.go` and `nice_test.go`. 936 937The NICE Framework components are published as a pinned JSON file (`csrc.nist.gov/csrc/media/Projects/cprt/documents/nice/v2-2-0_nf_components.json`), public domain, no auth. The 4 element schema has `elements` (each with an `element_identifier`, `element_type` like Work Role or Competency Area or TKS, and `text`) and `relationships`. Mirror the PSC adapter shape: a `download` plus a pure `parseNICE([]byte) ([]ontology.VocabRecord, error)` that yields KindWorkRole for work roles and KindKSA for TKS statements, keyed on `element_identifier`. Test `parseNICE` on a small literal JSON fixture. Commit. 938 939--- 940 941## Task 8: The GSA CALC adapter (labor category sweep) 942 943**Files:** Create `internal/modules/ontology/adapters/calc.go` and `calc_test.go`. 944 945CALC ceiling rates (`api.gsa.gov/acquisition/calc/v3/api/ceilingrates/`, no auth, optional `DATA_GOV_API_KEY` to lift the gateway rate limit) has no catalog endpoint, so enumerate labor categories by sweeping `?q=<seed>` across a seed list and deduping the `labor_category` field from the JSON results into KindLaborCategory records. Inject the HTTP client and the seed list so the test runs against an `httptest.Server` returning a canned page. Send `DATA_GOV_API_KEY` as the `api_key` query param when set (we already hold it). Commit. 946 947--- 948 949## Task 9: The O*NET adapter (dormant without a key) 950 951**Files:** Create `internal/modules/ontology/adapters/onet.go` and `onet_test.go`. 952 953`NewONETSource(apiKey string, client *http.Client)` sends the key as the `X-API-Key` header (never the query string), lists occupations and pulls skills, knowledge, abilities, and tasks per SOC code, throttling to 5 requests per second with a 200ms backoff on 429, keyed on the content model element id, yielding KindSkill and KindKSA. The CLI only constructs it when `ONET_API_KEY` is set, so it is inert until the key lands. Test the JSON parser on a small literal fixture (no live call). Add the required attribution string ("incorporates information from O*NET Web Services by the U.S. Department of Labor") to the source's records or a run log. Commit. 954 955--- 956 957## Task 10: DCWF crosswalk (open publications), deferred detail 958 959**Files:** Create `internal/modules/ontology/adapters/dcwf.go` and `dcwf_test.go`. 960 961The DCWF KSAT detail is CAC gated and DoD CIO pages reject fetchers, so this adapter ingests the open NICE to DCWF crosswalk (the 72 work roles and their 3 digit codes) into `ontology_crosswalk` rather than scraping the gated tool. Because the open artifact is a PDF, this task carries a real risk: pure Go PDF table extraction is unreliable. The pragmatic implementation is to commit a small curated crosswalk CSV (the 72 role to code mappings, transcribed once from the open factsheet) under `internal/modules/ontology/adapters/data/dcwf_crosswalk.csv`, embedded with `go:embed`, and parsed deterministically. Test the CSV parser on a small fixture. Write to `ontology_crosswalk`. Commit. (This keeps the no Python rule and avoids brittle PDF scraping; it is the deliberate small curation the spec allows for the gated source.) 962 963--- 964 965## Task 11: Validate end to end on M4 staging 966 967- [ ] **Step 1: Apply migration 0021 on a fresh M4 staging database** (not production): 968 969```bash 970ssh joseph@100.94.8.94 'cd ~/missionmatch && \ 971 /opt/homebrew/opt/postgresql@16/bin/psql -U postgres -c "CREATE DATABASE ontology_staging"' 972# provision roles + run the full migration set against ontology_staging via goose as migrator 973``` 974 975- [ ] **Step 2: Run ontology-sync against staging** with `MM_PSC_XLSX_URL` set to the live PSC manual and no O*NET key: 976 977```bash 978MM_MIGRATOR_DATABASE_URL="postgres://missionmatch_migrator:...@100.94.8.94:5432/ontology_staging?sslmode=disable" \ 979MM_ONTOLOGY_RAW_DIR=./data/ontology-raw \ 980MM_PSC_XLSX_URL="https://www.acquisition.gov/sites/default/files/manual/PSC%20April%202025.xlsx" \ 981/tmp/ontology-sync 982``` 983Expected log: `ontology-sync: PSC Manual fetched N upserted N ...`, `ONET_API_KEY not set ... dormant`, `done, N terms upserted`. 984 985- [ ] **Step 3: Verify the store and the audit**: 986 987```bash 988ssh joseph@100.94.8.94 '/opt/homebrew/opt/postgresql@16/bin/psql -U postgres -d ontology_staging -tAc \ 989 "SELECT term_kind, count(*) FROM ontology_terms GROUP BY 1; SELECT count(*) FROM mission_spine; SELECT count(*) FROM audit_logs WHERE action='\''ontology.ingest'\'';"' 990``` 991Expected: psc_code rows present, mission_spine has 10 rows, one ontology.ingest audit per source. 992 993- [ ] **Step 4: Run the full unit suite and the gates**: 994 995```bash 996go test ./internal/modules/ontology/... 997go build ./... 998make verify-fips 999``` 1000Expected: all green. 1001 1002- [ ] **Step 5: Final commit**: 1003 1004```bash 1005git add -A 1006git commit -m "test(ontology): WS1 validated on M4 staging (psc + corpus, onet dormant)" 1007``` 1008 1009--- 1010 1011## Self-Review 1012 1013Run after the plan is written, before execution. 1014 10151. **Spec coverage:** WS1 in `docs/AI_GROUNDING_PROGRAM.md` calls for `cmd/ontology-sync` with six adapters, the staging tables, mission_spine, and M4 validation. Covered: foundation (Tasks 1 to 3), psc (4), CLI and FIPS (5), corpus (6), nice (7), calc (8), onet dormant (9), dcwf crosswalk (10), M4 validation (11). The matching facing promotion into `skills` is correctly deferred to WS2 per the spec. 10162. **Placeholder scan:** Tasks 1 to 5 carry complete code. Tasks 6 to 10 are scoped with concrete interfaces, the exact source mechanics, and a named test, but defer full parser bodies to execution time; Task 6 Step 1 and Task 10 name the precise lookup or artifact rather than hand waving. This is a deliberate decomposition: the machinery (1 to 5) is the novel risk and is fully specified; the remaining adapters repeat the established pattern. 10173. **Type consistency:** `VocabRecord`, `TermKind`, `OntologySource`, `RecordWriter.Upsert(ctx, rec, key, sha)`, `Ingester`, and `NewDBWriter` are used identically across tasks. The CLI references `adapters.NewONETSource` (Task 9) and `adapters.NewCorpusSource`/`NewPoolText` (Task 6), flagged in Task 5 with the recommended build order. 1018 1019Known follow on: split Tasks 7 to 10 into their own short plans if executing subagent driven, so each adapter is one reviewable unit. 1020