Janady13/missionmatch / internal/modules/ai/curation_apply.go

1package ai 2 3// curation_apply.go is the WS8 Phase 4 TWO STEP human in the loop apply path (LIVE). It is the only place a 4// curation proposal turns into a curated ontology edge, and it does so in two distinct, separately authorized 5// human actions, never one click and never automatically: 6// 7// 1. APPROVE (the first human step). A compliance or operator user approves a pending proposal. The proposal 8// row is moved to approved with the approver and approved_at recorded, and for a substitutes_for_edge a 9// skill_relationships row is INSERTED with verified_at NULL. That edge is STAGED and INERT: the resolver 10// loads only verified_at IS NOT NULL edges (aliasresolver.go), so a staged edge changes NO score. The 11// proposal approved event is audited through System. 12// 13// 2. VERIFY (a SECOND, distinct human step). A compliance or operator user verifies the staged edge, setting 14// verified_at and the verifier. Only this single write makes the edge live: from now on the resolver 15// reads it and it changes FUTURE scores (never an already served and audited score, and never the engine 16// math, which is byte identical, only the resolver reference set grew). The edge added event is audited 17// through System carrying the from, the to, the relation, the strength, and the verifier. 18// 19// Every write here goes through the MIGRATOR pool, never the app role, because curation_proposals and 20// skill_relationships are global migrator owned tables the app role can only SELECT. The actor on each step is 21// the validated session principal, never a client supplied id, and the two events are global (NULL org) 22// because a proposal and an ontology edge are global objects, not per tenant ones. 23 24import ( 25 "context" 26 "encoding/json" 27 "errors" 28 29 "github.com/google/uuid" 30 "github.com/jackc/pgx/v5" 31 "github.com/jackc/pgx/v5/pgxpool" 32 33 "github.com/missionworkforce/missionmatch/internal/platform/audit" 34 "github.com/missionworkforce/missionmatch/internal/platform/database/dbgen" 35) 36 37// Curation apply errors. They map to typed HTTP statuses at the handler so a double submit or a missing 38// proposal is a clear client error, not a 500. 39var ( 40 // ErrProposalNotFound is returned when no proposal exists for the id. 41 ErrProposalNotFound = errors.New("curation: proposal not found") 42 // ErrProposalNotPending is returned when an approve targets a proposal that is not pending (already 43 // approved, already rejected, or concurrently moved), so an approval never re fires. 44 ErrProposalNotPending = errors.New("curation: proposal is not pending") 45 // ErrProposalNotApproved is returned when a verify targets a proposal that is not approved yet, enforcing 46 // that verify is genuinely the second step after approve. 47 ErrProposalNotApproved = errors.New("curation: proposal is not approved") 48 // ErrNotASubstitutionEdge is returned when an approve or verify targets a proposal whose kind is not a 49 // substitutes_for_edge, because only that kind stages and verifies an ontology edge. The evidence only 50 // kinds (reciprocity_review, recalibration_evidence) are not applied to the resolver. 51 ErrNotASubstitutionEdge = errors.New("curation: proposal is not a substitutes_for_edge") 52 // ErrNoStagedEdge is returned when a verify finds no staged edge to activate (the approve never ran, or 53 // the edge is already verified). 54 ErrNoStagedEdge = errors.New("curation: no staged edge to verify") 55 // ErrNoActor is returned when the request has no validated session principal to record as the actor. 56 ErrNoActor = errors.New("curation: no validated actor") 57 // ErrNoMigratorPool is returned when the apply service has no migrator pool configured, so the governance 58 // write surface reports unavailable rather than silently using the app role. 59 ErrNoMigratorPool = errors.New("curation: migrator pool not configured") 60 // ErrVerifierMustDifferFromApprover is returned when separation of duties is enforced and the same user 61 // who approved a proposal tries to verify its staged edge. It makes the two person control real rather 62 // than procedural. A single operator pilot may turn enforcement off in config to self verify. 63 ErrVerifierMustDifferFromApprover = errors.New("curation: the verifier must differ from the approver") 64) 65 66// CurationApplyService runs the two step apply on the migrator pool. It holds the migrator pool directly (the 67// global write seam) and the audit writer (System events). It is constructed only when a migrator DSN is 68// present; a nil pool makes every action return ErrNoMigratorPool so the surface fails closed. 69type CurationApplyService struct { 70 migrator *pgxpool.Pool 71 audit *audit.Writer 72 enforceSeparation bool // when true, the verifier must differ from the approver (the two person control) 73} 74 75// NewCurationApplyService wires the apply service. migrator is the BYPASSRLS migrator pool (the only pool that 76// may write the global curation tables); auditW records the global proposal approved and edge added events. 77// enforceSeparation requires the verifier to differ from the approver (config MM_CURATION_SEPARATION_OF_DUTIES, 78// default true); a single operator pilot may pass false to self verify, an explicit audited choice. 79func NewCurationApplyService(migrator *pgxpool.Pool, auditW *audit.Writer, enforceSeparation bool) *CurationApplyService { 80 return &CurationApplyService{migrator: migrator, audit: auditW, enforceSeparation: enforceSeparation} 81} 82 83// ProposalView is the read shape the /compliance surface lists: a pending proposal with its kind, the skill 84// pair (ids only, never PII), the proposed strength, and the evidence counts. The frontend renders the 85// approve and verify actions over this. 86type ProposalView struct { 87 ID string `json:"id"` 88 Kind string `json:"kind"` 89 FromSkillID string `json:"from_skill_id"` 90 ToSkillID string `json:"to_skill_id"` 91 Relation string `json:"relation"` 92 ProposedStrength string `json:"proposed_strength"` 93 EvidenceObservations int `json:"evidence_observations"` 94 EvidenceSuccesses int `json:"evidence_successes"` 95 ModelVersion string `json:"model_version"` 96 Status string `json:"status"` 97 CreatedAt string `json:"created_at"` 98} 99 100// ListPending returns the proposals awaiting the first human step, read through the migrator pool. 101func (s *CurationApplyService) ListPending(ctx context.Context) ([]ProposalView, error) { 102 if s == nil || s.migrator == nil { 103 return nil, ErrNoMigratorPool 104 } 105 rows, err := dbgen.New(s.migrator).ListPendingProposals(ctx) 106 if err != nil { 107 return nil, err 108 } 109 out := make([]ProposalView, 0, len(rows)) 110 for _, r := range rows { 111 out = append(out, toProposalView(r)) 112 } 113 return out, nil 114} 115 116// Approve is the FIRST human step. It moves a pending substitutes_for_edge proposal to approved (recording the 117// approver and approved_at) and stages the INERT skill_relationships edge (verified_at NULL) in one migrator 118// transaction, so the proposal transition and the staged edge commit together. The staged edge is read by NO 119// score (the resolver ignores unverified edges), so this step changes nothing live yet. It audits 120// teaching.proposal.approved through System. actor is the validated session principal. 121func (s *CurationApplyService) Approve(ctx context.Context, proposalID, actor uuid.UUID) (ProposalView, error) { 122 if s == nil || s.migrator == nil { 123 return ProposalView{}, ErrNoMigratorPool 124 } 125 var view ProposalView 126 err := pgx.BeginFunc(ctx, s.migrator, func(tx pgx.Tx) error { 127 q := dbgen.New(tx) 128 129 prop, gerr := q.GetProposal(ctx, proposalID) 130 if errors.Is(gerr, pgx.ErrNoRows) { 131 return ErrProposalNotFound 132 } 133 if gerr != nil { 134 return gerr 135 } 136 if prop.Kind != kindSubstitutesForEdge { 137 return ErrNotASubstitutionEdge 138 } 139 if prop.Status != "pending" { 140 return ErrProposalNotPending 141 } 142 if prop.FromSkillID == nil || prop.ToSkillID == nil || prop.ProposedStrength == nil { 143 return ErrNotASubstitutionEdge 144 } 145 146 approved, aerr := q.ApproveProposalTx(ctx, dbgen.ApproveProposalTxParams{ID: proposalID, Approver: &actor}) 147 if errors.Is(aerr, pgx.ErrNoRows) { 148 return ErrProposalNotPending // a concurrent approval won the race 149 } 150 if aerr != nil { 151 return aerr 152 } 153 154 // Stage the INERT edge. verified_at stays NULL, so the resolver does not read it; a SECOND human action 155 // verifies it later to make it live. ON CONFLICT keeps approve idempotent against an existing edge. 156 if _, serr := q.StageSubstitutesForEdge(ctx, dbgen.StageSubstitutesForEdgeParams{ 157 FromSkillID: *approved.FromSkillID, 158 ToSkillID: *approved.ToSkillID, 159 Strength: *approved.ProposedStrength, 160 }); serr != nil { 161 return serr 162 } 163 164 payload, merr := json.Marshal(map[string]any{ 165 "from_skill_id": approved.FromSkillID.String(), 166 "to_skill_id": approved.ToSkillID.String(), 167 "relation": substitutionRelation, 168 "staged": true, // staged and inert, not yet verified 169 }) 170 if merr != nil { 171 return merr 172 } 173 if auderr := s.audit.System(ctx, audit.Event{ 174 Action: "teaching.proposal.approved", 175 ActorUserID: &actor, 176 EntityType: "curation_proposal", 177 EntityID: proposalID.String(), 178 NewValues: payload, 179 }); auderr != nil { 180 return auderr 181 } 182 view = toProposalView(approved) 183 return nil 184 }) 185 if err != nil { 186 return ProposalView{}, err 187 } 188 return view, nil 189} 190 191// Verify is the SECOND, distinct human step. It activates the staged edge that an approval created by setting 192// verified_at and the verifier, ONLY when the proposal is approved and the edge is still unverified. Only this 193// write makes the edge live: the resolver now reads it and it changes FUTURE scores. The engine ModelVersion 194// stays v1.4.0, because the math is unchanged and only the resolver reference set grew (recorded in the model 195// card). It audits ontology.edge.added through System carrying the from, the to, the relation, the strength, 196// and the verifier. actor is the validated session principal, distinct from the approver in practice. 197func (s *CurationApplyService) Verify(ctx context.Context, proposalID, actor uuid.UUID) (ProposalView, error) { 198 if s == nil || s.migrator == nil { 199 return ProposalView{}, ErrNoMigratorPool 200 } 201 var view ProposalView 202 err := pgx.BeginFunc(ctx, s.migrator, func(tx pgx.Tx) error { 203 q := dbgen.New(tx) 204 205 prop, gerr := q.GetProposal(ctx, proposalID) 206 if errors.Is(gerr, pgx.ErrNoRows) { 207 return ErrProposalNotFound 208 } 209 if gerr != nil { 210 return gerr 211 } 212 if prop.Kind != kindSubstitutesForEdge { 213 return ErrNotASubstitutionEdge 214 } 215 if prop.Status != "approved" { 216 return ErrProposalNotApproved 217 } 218 // Separation of duties: the human who verifies an edge live must not be the human who approved it, so 219 // the two person control is enforced, not merely procedural. A single operator pilot may turn this off 220 // in config. The approver is recorded on the proposal, so no extra read is needed. 221 if s.enforceSeparation && prop.Approver != nil && *prop.Approver == actor { 222 return ErrVerifierMustDifferFromApprover 223 } 224 if prop.FromSkillID == nil || prop.ToSkillID == nil { 225 return ErrNotASubstitutionEdge 226 } 227 228 // Find the staged edge for this pair and activate it. A double submit finds the edge already verified 229 // (VerifyStagedEdge only matches verified_at IS NULL), so it returns no row and we report ErrNoStagedEdge, 230 // keeping verify a single transition. 231 edgeID, found, ferr := stagedEdgeID(ctx, tx, *prop.FromSkillID, *prop.ToSkillID) 232 if ferr != nil { 233 return ferr 234 } 235 if !found { 236 return ErrNoStagedEdge 237 } 238 edge, verr := q.VerifyStagedEdge(ctx, dbgen.VerifyStagedEdgeParams{ID: edgeID, Verifier: &actor}) 239 if errors.Is(verr, pgx.ErrNoRows) { 240 return ErrNoStagedEdge // already verified by a concurrent action 241 } 242 if verr != nil { 243 return verr 244 } 245 246 payload, merr := json.Marshal(map[string]any{ 247 "from_skill_id": edge.FromSkillID.String(), 248 "to_skill_id": edge.ToSkillID.String(), 249 "relation": edge.Relation, 250 "strength": edge.Strength.String(), 251 "verifier": actor.String(), 252 }) 253 if merr != nil { 254 return merr 255 } 256 if auderr := s.audit.System(ctx, audit.Event{ 257 Action: "ontology.edge.added", 258 ActorUserID: &actor, 259 EntityType: "skill_relationship", 260 EntityID: edge.ID.String(), 261 NewValues: payload, 262 }); auderr != nil { 263 return auderr 264 } 265 view = toProposalView(prop) 266 view.Status = "verified" // surface the live state to the caller; the proposal row itself stays approved 267 return nil 268 }) 269 if err != nil { 270 return ProposalView{}, err 271 } 272 return view, nil 273} 274 275// stagedEdgeID finds the unverified staged skill_relationships edge for a substitutes_for pair, so verify can 276// activate exactly that row. It reads through the same migrator tx. An edge that is already verified is not 277// returned (found false), so verify is a single transition. 278func stagedEdgeID(ctx context.Context, tx pgx.Tx, fromSkillID, toSkillID uuid.UUID) (uuid.UUID, bool, error) { 279 var id uuid.UUID 280 err := tx.QueryRow(ctx, 281 `SELECT id FROM skill_relationships 282 WHERE from_skill_id = $1 AND to_skill_id = $2 AND relation = 'substitutes_for' 283 AND verified_at IS NULL`, 284 fromSkillID, toSkillID).Scan(&id) 285 if errors.Is(err, pgx.ErrNoRows) { 286 return uuid.Nil, false, nil 287 } 288 if err != nil { 289 return uuid.Nil, false, err 290 } 291 return id, true, nil 292} 293 294// toProposalView maps a stored proposal to the read shape, with empty strings for the nullable id and strength 295// fields rather than a nil pointer, so the frontend renders a stable row. 296func toProposalView(p dbgen.CurationProposal) ProposalView { 297 v := ProposalView{ 298 ID: p.ID.String(), 299 Kind: p.Kind, 300 EvidenceObservations: int(p.EvidenceObservations), 301 EvidenceSuccesses: int(p.EvidenceSuccesses), 302 ModelVersion: p.ModelVersion, 303 Status: p.Status, 304 CreatedAt: p.CreatedAt.Format("2006-01-02"), 305 } 306 if p.FromSkillID != nil { 307 v.FromSkillID = p.FromSkillID.String() 308 } 309 if p.ToSkillID != nil { 310 v.ToSkillID = p.ToSkillID.String() 311 } 312 if p.Relation != nil { 313 v.Relation = *p.Relation 314 } 315 if p.ProposedStrength != nil { 316 v.ProposedStrength = p.ProposedStrength.String() 317 } 318 return v 319} 320