diff --git a/THIRD_PARTY_NOTICES.md b/THIRD_PARTY_NOTICES.md index b4740d9e2..f03fb8866 100644 --- a/THIRD_PARTY_NOTICES.md +++ b/THIRD_PARTY_NOTICES.md @@ -295,3 +295,12 @@ Exact Voice Brain source, model and build-tool pins are recorded in `config/upst - License copy: `third_party/licenses/agent-lightning-MIT.txt` - Upstream dependency lock: exact `uv.lock` Git blob `5a98a2ac121b050b0a82f6ac8dc207577ce3af4e` - Yance integration: source-module CORE + APO only, downstream of Learning and Model Brain authority, returning `CANDIDATE_ONLY`. +## Vowpal Wabbit + +- Project: Vowpal Wabbit +- Upstream: `VowpalWabbit/vowpal_wabbit` +- Version: `9.11.2` +- Frozen commit: `122bae254a5b8bc2b774d13b33d53e6dbc2cfba7` +- License: `BSD-3-Clause` +- License copy: `third_party/licenses/vowpal-wabbit-BSD-3-Clause.txt` +- Yance integration: sealed Learning runtime contextual-bandit ADF policy head only. P1 is deterministic (`actionProbability=1.0`, `exploration=false`); Model Brain/LiteLLM remains the final reply-generation authority. diff --git a/backend/services/contextAwareReplyBrain.js b/backend/services/contextAwareReplyBrain.js index ee443d2f7..d244142fc 100644 --- a/backend/services/contextAwareReplyBrain.js +++ b/backend/services/contextAwareReplyBrain.js @@ -22,6 +22,8 @@ const aiDirectorStrategyAuthority = require('./aiDirectorStrategyAuthority').sin const aiWorkbenchDirectorRuleAuthority = require('./aiWorkbenchDirectorRuleAuthority'); const goalDrivenMemoryRecall = require('./goalDrivenMemoryRecallService'); const { singleton: platformCoreRepository } = require('../repositories/platformCoreRepository'); +const { createLearningPolicyRuntimeAdapter } = require('./learningPolicyRuntimeAdapter'); +const { createLearningPolicyDecisionContract } = require('./learningPolicyDecisionContract'); function clean(value) { return String(value == null ? '' : value).trim(); @@ -98,6 +100,9 @@ function memoryCandidatesForRecall(packet = {}) { function branchNameForVariant(value = '') { const variant = clean(value).toLowerCase(); + const exactAction = ['natural_hook', 'playful_attraction', 'direct_advance', 'screen_and_advance', 'leave_aftertaste'] + .find(action => action === variant); + if (exactAction) return exactAction; if (/情趣|暧昧|俏皮|妩媚|女人味|feminine|flirt/u.test(variant)) return 'playful_attraction'; if (/边界|筛选|screen/u.test(variant)) return 'screen_and_advance'; if (/直接|强势|direct/u.test(variant)) return 'direct_advance'; @@ -106,6 +111,17 @@ function branchNameForVariant(value = '') { return 'natural_hook'; } +function learnedPolicyInteractionBand(context = {}) { + const interaction = context.interaction && typeof context.interaction === 'object' ? context.interaction : {}; + const explicit = clean(interaction.engagementBand || interaction.responseBand || interaction.band).toLowerCase(); + if (['low', 'balanced', 'high'].includes(explicit)) return explicit; + const score = Number(interaction.engagementScore ?? interaction.responseRate ?? interaction.reciprocity); + if (!Number.isFinite(score)) return 'balanced'; + if (score < 0.35) return 'low'; + if (score > 0.7) return 'high'; + return 'balanced'; +} + function candidateBranchPlanForCount(value = 3) { const count = Math.max(1, Math.min(5, Number(value || 3))); return ['natural_hook', 'playful_attraction', 'direct_advance', 'screen_and_advance', 'leave_aftertaste'].slice(0, count); @@ -674,10 +690,19 @@ async function contextStillCurrent(previous, current) { return clean(previous.contactId) === clean(current.contactId); } -function createContextAwareReplyBrain({ storeManager, aiGateway, personaBrain, resolveContactId }) { +function createContextAwareReplyBrain({ + storeManager, + aiGateway, + personaBrain, + resolveContactId, + learningPolicyRuntimeAdapter, + learningPolicyDecisionContract +}) { if (!storeManager?.select || !storeManager?.dispatch) throw new TypeError('storeManager is required'); if (!aiGateway?.execute) throw new TypeError('aiGateway is required'); const persona = personaBrain || personaBrainModule.createPersonaBrain(); + const learnedPolicyRuntime = learningPolicyRuntimeAdapter || createLearningPolicyRuntimeAdapter(); + const learnedPolicyDecisions = learningPolicyDecisionContract || createLearningPolicyDecisionContract(); function currentConversationRevision(conversationId) { const selected = storeManager.select(state => Number(state.conversations?.byId?.[conversationId]?.version || 0)); @@ -1081,7 +1106,75 @@ function createContextAwareReplyBrain({ storeManager, aiGateway, personaBrain, r targetLanguage: languageAuthority.code, learningWeights: {}, }); - const branchApplication = applyCandidateBranch(effectiveDirector, candidatePlan.plan, variant); + const policyFeatureBundle = Object.freeze({ + interactionBand: learnedPolicyInteractionBand(socialContext), + performanceMode, + questionPolicy: Number(automaticDirectorPlan.maxQuestions) === 0 ? 'none' : 'optional', + relationshipStage: clean(socialContext.relationshipPotential?.relationshipStage || 'unknown') || 'unknown', + targetLanguage: clean(languageAuthority.code || 'unknown') || 'unknown' + }); + const allowedPolicyActions = (candidatePlan.plan.branches || []) + .map(row => clean(row.strategy)) + .filter(Boolean); + const requestedBaselineAction = branchNameForVariant(variant); + const baselinePolicyAction = allowedPolicyActions.includes(requestedBaselineAction) + ? requestedBaselineAction + : allowedPolicyActions[0]; + const learnedPolicySelection = await learnedPolicyRuntime.selectLearnedPolicyAction({ + featureBundle: policyFeatureBundle, + allowedActions: allowedPolicyActions, + baselineAction: baselinePolicyAction + }); + const branchApplication = applyCandidateBranch( + effectiveDirector, + candidatePlan.plan, + learnedPolicySelection.candidateStrategyBranch + ); + let decisionRecord = null; + let learningPolicyEvidenceEligible = true; + let learningPolicyDegradation = learnedPolicySelection.degradation || null; + try { + decisionRecord = learnedPolicyDecisions.createDecisionRecord({ + contactId, + conversationId, + personaProfileId: personaCtx.profileId || 'owner', + featureBundle: policyFeatureBundle, + allowedActionSet: allowedPolicyActions, + candidateStrategyBranch: learnedPolicySelection.candidateStrategyBranch, + policyVersion: learnedPolicySelection.policyVersion, + behaviorPolicyVersion: learnedPolicySelection.policyVersion, + policyArtifactId: learnedPolicySelection.policyArtifactId, + generation: { + candidatePlanId: candidatePlan.plan.planId, + directorStrategyId: directorStrategy.strategy.strategyId, + contextVersion: socialContext.contextVersion, + conversationRevision + }, + contributingPolicyVersions: { + relationship: `relationship-v${Number(socialContext.entityVersions?.relationship || 0)}`, + memory: `memory-v${Number(socialContext.entityVersions?.memory || 0)}`, + strategy: `director-strategy-v${Number(directorStrategy.strategy.strategyVersion || 1)}`, + candidateRanker: 'candidate-strategy-branch-v1', + routing: 'model-brain-routing-current-v1', + promptProgram: 'context-aware-reply-current-v1' + } + }); + } catch (error) { + learningPolicyEvidenceEligible = false; + learningPolicyDegradation = Object.freeze({ + reasonCode: clean(error.reasonCode || error.code) || 'LEARNING_POLICY_DECISION_EVIDENCE_BLOCKED' + }); + } + const learningPolicyReceipt = Object.freeze({ + authority: 'LearningPolicyRuntimeAdapter', + policyVersion: clean(learnedPolicySelection.policyVersion), + policyArtifactId: clean(learnedPolicySelection.policyArtifactId), + candidateStrategyBranch: clean(learnedPolicySelection.candidateStrategyBranch), + actionProbability: 1, + exploration: false, + evidenceEligible: learningPolicyEvidenceEligible, + degradation: learningPolicyDegradation + }); effectiveDirector = { ...branchApplication.director, strategyId: directorStrategy.strategy.strategyId, @@ -1265,6 +1358,8 @@ function createContextAwareReplyBrain({ storeManager, aiGateway, personaBrain, r emergencyMode, learningEligible, highCapabilityPath, + decisionRecord, + learningPolicy: learningPolicyReceipt, directorRuleStackReceipt: directorRuleStack.receipt, director: { plan: automaticDirectorPlan, diff --git a/backend/services/learningDeepTrainingContract.js b/backend/services/learningDeepTrainingContract.js index 9df1f8f73..518c3d8cb 100644 --- a/backend/services/learningDeepTrainingContract.js +++ b/backend/services/learningDeepTrainingContract.js @@ -27,7 +27,7 @@ function createLearningDeepTrainingContract(options = {}) { } function isLearningEligible(signal = {}) { - return signal.learning_eligible === true || signal.learning_eligible === 1; + return signal.learning_eligible === true || signal.learning_eligible === 1 || signal.learningEligible === true; } function isDoNotLearn(signal = {}) { @@ -35,7 +35,7 @@ function createLearningDeepTrainingContract(options = {}) { } function hasRawPrivatePersistence(signal = {}) { - return signal.signal?.metadata?.rawPrivateChatPersisted === true; + return signal.signal?.metadata?.rawPrivateChatPersisted === true || signal.signal?.rawPrivateChatPersisted === true; } function hasValidScoreSubject(score = {}) { @@ -195,6 +195,110 @@ function createLearningDeepTrainingContract(options = {}) { }); } + function exactOutcomeIds(rows = []) { + return [...new Set(rows.flatMap(row => Array.isArray(row.signal?.outcomes) ? row.signal.outcomes : []) + .map(outcome => clean(outcome?.outcomeId)).filter(Boolean))].sort(); + } + + function hasReplayableDecision(decision = {}) { + const probability = Number(decision.actionProbability); + return Boolean( + clean(decision.decisionId) && clean(decision.candidateStrategyBranch) && + decision.featureBundle && typeof decision.featureBundle === 'object' && !Array.isArray(decision.featureBundle) && + clean(decision.actionId) && clean(decision.actionSetRef) && clean(decision.actionEncodingVersion) && + clean(decision.behaviorPolicyVersion || decision.policyVersion) && + Number.isFinite(probability) && probability > 0 && probability <= 1 && decision.exploration !== true + ); + } + + async function listPolicyOutcomes(decisionIds, scopeType, scopeId) { + if (typeof repository.listPolicyOutcomeSignals === 'function') { + const listed = await repository.listPolicyOutcomeSignals({ decisionIds }); + return Array.isArray(listed) ? listed : []; + } + const listed = await repository.listLearningSignals({ scopeType, scopeId, learningLevel: 'L1', learningEligible: false }); + return (Array.isArray(listed) ? listed : []).filter(row => + clean(row.signal_type || row.signalType) === 'policy_outcome_observed' && decisionIds.includes(clean(row.signal?.decisionId)) + ); + } + + async function projectPolicy(input = {}) { + requireProjectionDependencies(); + const scopeType = clean(input.scopeType); + const scopeId = clean(input.scopeId); + const listed = await repository.listLearningSignals({ scopeType, scopeId, learningLevel: 'L1', learningEligible: true }); + const sources = (Array.isArray(listed) ? listed : []).filter(signal => + isLearningEligible(signal) && !isDoNotLearn(signal) && !hasRawPrivatePersistence(signal) + && clean(signal.signal_type || signal.signalType) === 'candidate_sent' + && clean(signal.signal?.decisionRecord?.decisionId) + ); + assertCanonicalScope(sources, scopeType, scopeId, 'LEARNING_POLICY_SOURCE_SCOPE_MISMATCH'); + const decisionIds = sources.map(row => clean(row.signal.decisionRecord.decisionId)); + const rawOutcomes = await listPolicyOutcomes(decisionIds, scopeType, scopeId); + const trajectory = []; + + for (const source of sources) { + const sourceSignalId = clean(source.signal_id || source.signalId); + const decision = source.signal.decisionRecord; + const decisionId = clean(decision.decisionId); + const joined = rawOutcomes.filter(row => { + const raw = row.signal || {}; + const rawFalseEligible = row.learning_eligible === false || row.learning_eligible === 0 || row.learningEligible === false; + if (!rawFalseEligible || clean(row.signal_type || row.signalType) !== 'policy_outcome_observed') return false; + if (clean(raw.decisionId) !== decisionId) return false; + if (clean(raw.sourceSignalId) && clean(raw.sourceSignalId) !== sourceSignalId) return false; + if (clean(raw.personId) && clean(decision.personId) && clean(raw.personId) !== clean(decision.personId)) return false; + if (clean(raw.conversationId) && clean(decision.conversationId) && clean(raw.conversationId) !== clean(decision.conversationId)) return false; + return true; + }); + const outcomes = Object.freeze(joined.flatMap(row => Array.isArray(row.signal?.outcomes) ? row.signal.outcomes.map(value => Object.freeze({ ...value })) : [])); + const outcomeIds = exactOutcomeIds(joined); + const score = approvedScoreFor(sourceSignalId, input.approvedScoresBySignalId); + const scoreSourceId = clean(score.sourceSignalId || score.eligibleSourceSignalId); + const scoreOutcomeIds = [...new Set((Array.isArray(score.outcomeIds) ? score.outcomeIds : []).map(clean).filter(Boolean))].sort(); + const exactOutcomeSet = outcomeIds.length === scoreOutcomeIds.length && outcomeIds.every((id, index) => id === scoreOutcomeIds[index]); + if ( + scoreSourceId !== sourceSignalId || clean(score.decisionId) !== decisionId || !exactOutcomeSet || + !clean(score.outcomeEvidenceSetRef) || !clean(score.rewardPolicyVersion) + ) { + throw contractError('LEARNING_POLICY_SCORE_EVIDENCE_BINDING_REQUIRED', `Score for ${sourceSignalId} must bind exact source/decision/outcome evidence.`); + } + const minimized = await dataPolicy.minimize({ + text: clean(input.contentBySignalId?.[sourceSignalId]), + signalId: sourceSignalId, + scopeType, + scopeId, + learningEligible: true, + featureBundle: decision.featureBundle + }); + if (!minimized || minimized.allowed !== true) continue; + trajectory.push(Object.freeze({ + sourceSignalId, + signalId: sourceSignalId, + decisionId, + decision: Object.freeze({ ...decision }), + featureBundle: Object.freeze({ ...(decision.featureBundle || {}) }), + outcomes, + approvedScore: score, + score, + minimizedContent: String(minimized.text ?? minimized.minimizedText ?? ''), + vwTrainingEligible: hasReplayableDecision(decision) + })); + } + + const projection = Object.freeze({ + authority: 'Learning', + readOnly: true, + scopeType, + scopeId, + learningLevel: 'L1', + policyProjection: true, + trajectory: Object.freeze(trajectory) + }); + issuedProjections.add(projection); + return projection; + } + async function bindExperimentEvidence(input = {}) { if (!evidenceAdapter || typeof evidenceAdapter.bindTrainingEvidence !== 'function') { throw contractError('LEARNING_DEEP_TRAINING_LANGFUSE_EVIDENCE_REQUIRED', 'Langfuse Dataset/Score evidence adapter is required.'); @@ -231,10 +335,11 @@ function createLearningDeepTrainingContract(options = {}) { return Object.freeze({ projectRelationship, projectGlobal, + projectPolicy, bindExperimentEvidence, rollbackPromotion, authority: 'Learning read-only Deep Training projection' }); } -module.exports = { createLearningDeepTrainingContract }; +module.exports = { createLearningDeepTrainingContract }; \ No newline at end of file diff --git a/backend/services/learningOutcomeAttributionService.js b/backend/services/learningOutcomeAttributionService.js new file mode 100644 index 000000000..8fb9099b7 --- /dev/null +++ b/backend/services/learningOutcomeAttributionService.js @@ -0,0 +1,266 @@ +'use strict'; + +const { canonicalHash } = require('./canonicalSerialization'); +const eventBus = require('./eventBus'); +const personContextAuthority = require('./personContextAuthority').singleton; +const { singleton: platformCoreRepository } = require('../repositories/platformCoreRepository'); + +const AUTHORITY = 'LearningOutcomeAttribution'; +const RAW_SIGNAL_TYPE = 'policy_outcome_observed'; + +function clean(value) { return String(value == null ? '' : value).trim(); } +function attributionError(reasonCode, message, details = {}) { + return Object.assign(new Error(message || reasonCode), { reasonCode, code: reasonCode, ...details }); +} +function deepFreeze(value) { + if (!value || typeof value !== 'object' || Object.isFrozen(value)) return value; + Object.freeze(value); + for (const child of Object.values(value)) deepFreeze(child); + return value; +} +function timestamp(value) { + const ms = Date.parse(clean(value)); + return Number.isFinite(ms) ? new Date(ms).toISOString() : ''; +} +function unique(values = []) { return [...new Set((Array.isArray(values) ? values : []).map(clean).filter(Boolean))]; } + +function normalizeOutcomes(values = []) { + if (!Array.isArray(values)) throw attributionError('LEARNING_POLICY_OUTCOMES_INVALID', 'outcomes must be an array.'); + const seen = new Set(); + const rows = values.map((row, index) => { + if (!row || typeof row !== 'object' || Array.isArray(row)) { + throw attributionError('LEARNING_POLICY_OUTCOME_INVALID', `Outcome ${index} must be an object.`); + } + const outcomeId = clean(row.outcomeId); + const type = clean(row.type); + const evidenceRef = clean(row.evidenceRef); + if (!outcomeId || !type || !evidenceRef || seen.has(outcomeId)) { + throw attributionError('LEARNING_POLICY_OUTCOME_PROVENANCE_REQUIRED', 'Each outcome requires a unique outcomeId, type and evidenceRef.'); + } + seen.add(outcomeId); + const value = row.value; + if (typeof value !== 'boolean' && !(typeof value === 'number' && Number.isFinite(value)) && value !== null) { + throw attributionError('LEARNING_POLICY_OUTCOME_VALUE_INVALID', 'Outcome values must be finite numbers, booleans or null.'); + } + return deepFreeze({ outcomeId, type, value, evidenceRef }); + }); + return deepFreeze(rows); +} + +function createLearningOutcomeAttributionService(options = {}) { + const repository = options.repository || platformCoreRepository; + const identityAuthority = options.personContextAuthority || personContextAuthority; + const bus = options.eventBus || eventBus; + let started = false; + let listener = null; + + function createOutcomeVector(input = {}) { + const decisionId = clean(input.decisionId); + if (!decisionId) throw attributionError('LEARNING_POLICY_DECISION_ID_REQUIRED', 'OutcomeVector requires decisionId.'); + const observedAt = timestamp(input.observedAt) || clean(input.observedAt); + if (!observedAt) throw attributionError('LEARNING_POLICY_OUTCOME_OBSERVED_AT_REQUIRED', 'OutcomeVector requires observedAt.'); + const outcomes = normalizeOutcomes(input.outcomes || []); + const sourceSignalId = clean(input.sourceSignalId); + const contactId = clean(input.contactId); + const conversationId = clean(input.conversationId); + const personId = clean(input.personId); + const observationWindow = input.observationWindow && typeof input.observationWindow === 'object' + ? deepFreeze({ start: timestamp(input.observationWindow.start) || clean(input.observationWindow.start), end: timestamp(input.observationWindow.end) || clean(input.observationWindow.end) }) + : undefined; + const signals = input.signals && typeof input.signals === 'object' && !Array.isArray(input.signals) + ? deepFreeze(JSON.parse(JSON.stringify(input.signals))) + : undefined; + const outcomeId = clean(input.outcomeId) || `outcome:${canonicalHash({ decisionId, sourceSignalId, observedAt, outcomes, signals: signals || null })}`; + const vector = { + schemaVersion: 1, + authority: AUTHORITY, + signalType: RAW_SIGNAL_TYPE, + outcomeId, + decisionId, + sourceSignalId, + scopeType: conversationId ? 'conversation' : clean(input.scopeType) || 'conversation', + scopeId: conversationId || clean(input.scopeId), + contactId, + personId, + conversationId, + observedAt, + outcomes, + learningEligible: false, + rawPrivateChatPersisted: false + }; + if (observationWindow) vector.observationWindow = observationWindow; + if (signals) vector.signals = signals; + return deepFreeze(vector); + } + + function bindTrainableOutcome(input = {}) { + const source = input.eligibleSourceSignal || {}; + const vector = input.outcomeVector || {}; + const score = input.score || {}; + const sourceSignalId = clean(source.signal_id || source.signalId); + const sourceDecisionId = clean(source.signal?.decisionRecord?.decisionId); + if ( + !sourceSignalId || clean(source.signal_type || source.signalType) !== 'candidate_sent' || + !(source.learning_eligible === true || source.learning_eligible === 1 || source.learningEligible === true) || + !sourceDecisionId + ) { + throw attributionError('LEARNING_POLICY_ELIGIBLE_SOURCE_SIGNAL_REQUIRED', 'A learning-eligible immutable candidate_sent source signal is required.'); + } + if (vector.signalType !== RAW_SIGNAL_TYPE || vector.learningEligible !== false || clean(vector.decisionId) !== sourceDecisionId) { + throw attributionError('LEARNING_POLICY_OUTCOME_DECISION_BINDING_MISMATCH', 'Raw outcome must remain false-eligible and bind the source DecisionRecord.'); + } + const scoreSourceId = clean(score.eligibleSourceSignalId || score.sourceSignalId); + const outcomeIds = unique((vector.outcomes || []).map(row => row.outcomeId)); + const scoreOutcomeIds = unique(score.outcomeIds); + const exactOutcomeSet = outcomeIds.length === scoreOutcomeIds.length && outcomeIds.every(id => scoreOutcomeIds.includes(id)); + if ( + score.authority !== 'Langfuse' || score.approvedByLearning !== true || !clean(score.scoreId) || + scoreSourceId !== sourceSignalId || clean(score.decisionId) !== sourceDecisionId || + !exactOutcomeSet || !clean(score.outcomeEvidenceSetRef) || !clean(score.rewardPolicyVersion) || + !Number.isFinite(Number(score.value)) + ) { + throw attributionError('LEARNING_POLICY_APPROVED_SCORE_BINDING_REQUIRED', 'Learning-approved Langfuse Score must bind the exact source/decision/outcome evidence set.'); + } + return deepFreeze({ + eligibleSourceSignalId: sourceSignalId, + sourceSignalId, + decisionId: sourceDecisionId, + outcomeIds, + outcomeEvidenceSetRef: clean(score.outcomeEvidenceSetRef), + rewardPolicyVersion: clean(score.rewardPolicyVersion), + outcomeVector: vector, + reward: deepFreeze({ + authority: 'Langfuse', + scoreId: clean(score.scoreId), + value: Number(score.value), + approvedByLearning: true + }) + }); + } + + function eligibleSentSources(conversationId) { + if (!repository || typeof repository.listLearningSignals !== 'function') return []; + const rows = repository.listLearningSignals({ + scopeType: 'conversation', scopeId: conversationId, learningLevel: 'L1', learningEligible: true + }); + return (Array.isArray(rows) ? rows : []).filter(row => + clean(row.signal_type || row.signalType) === 'candidate_sent' && row.signal?.decisionRecord?.decisionId + ); + } + + function resolveInboundAttribution(message = {}) { + const conversationId = clean(message.conversationId || message.sessionKey); + const contactId = clean(message.contactId); + const inboundId = clean(message.externalMessageId || message.messageId || message.id); + const observedAt = timestamp(message.sentAt || message.timestamp || message.createdAt); + if (!conversationId || !contactId || !inboundId || !observedAt) return null; + const direction = clean(message.direction).toLowerCase(); + if (message.fromMe === true || ['outbound', 'outgoing'].includes(direction)) return null; + + const identity = identityAuthority.resolve({ contactId, conversationId }); + if (identity?.authority !== 'PersonContextAuthority' || identity?.found !== true || !clean(identity.personId)) { + throw attributionError('LEARNING_POLICY_OUTCOME_IDENTITY_REQUIRED', 'Inbound attribution requires canonical PersonContextAuthority identity.'); + } + const personId = clean(identity.personId); + const candidates = eligibleSentSources(conversationId).filter(row => { + const decision = row.signal?.decisionRecord || {}; + const sentAt = timestamp(row.created_at || row.createdAt); + return sentAt && sentAt < observedAt + && clean(decision.contactId) === contactId + && clean(decision.conversationId) === conversationId + && clean(decision.personId) === personId; + }).sort((a, b) => timestamp(b.created_at || b.createdAt).localeCompare(timestamp(a.created_at || a.createdAt))); + if (!candidates.length) return null; + const latestAt = timestamp(candidates[0].created_at || candidates[0].createdAt); + const latest = candidates.filter(row => timestamp(row.created_at || row.createdAt) === latestAt); + if (latest.length !== 1) { + throw attributionError('LEARNING_POLICY_OUTCOME_ATTRIBUTION_AMBIGUOUS', 'Inbound outcome matches more than one latest eligible decision.', { conversationId, contactId, inboundId }); + } + return { source: latest[0], identity, inboundId, observedAt }; + } + + function persistInboundOutcome(message = {}) { + const match = resolveInboundAttribution(message); + if (!match) return { skipped: true, reasonCode: 'NO_ELIGIBLE_SENT_DECISION' }; + const source = match.source; + const decision = source.signal.decisionRecord; + const sourceSignalId = clean(source.signal_id || source.signalId); + const sentAt = timestamp(source.created_at || source.createdAt); + const latencyMs = Math.max(0, Date.parse(match.observedAt) - Date.parse(sentAt)); + const messagePairRef = canonicalHash({ sourceSignalId, inboundMessageId: match.inboundId, sentAt, observedAt: match.observedAt }); + const inboundEvidenceRef = canonicalHash({ inboundMessageId: match.inboundId, conversationId: decision.conversationId, contactId: decision.contactId, observedAt: match.observedAt }); + const nextDayWindowEnd = new Date(Date.parse(match.observedAt) + 24 * 60 * 60 * 1000).toISOString(); + const outcomeId = `outcome:${canonicalHash({ decisionId: decision.decisionId, sourceSignalId, inboundMessageId: match.inboundId })}`; + const vector = createOutcomeVector({ + outcomeId, + decisionId: decision.decisionId, + sourceSignalId, + contactId: decision.contactId, + personId: decision.personId, + conversationId: decision.conversationId, + observedAt: match.observedAt, + observationWindow: { start: sentAt, end: match.observedAt }, + outcomes: [ + { outcomeId: `${outcomeId}:reply`, type: 'reply_received', value: 1, evidenceRef: inboundEvidenceRef }, + { outcomeId: `${outcomeId}:continued`, type: 'conversation_continued', value: 1, evidenceRef: messagePairRef } + ], + signals: { + replyLatencyMs: { value: latencyMs, status: 'observed', observedAt: match.observedAt, windowStart: sentAt, windowEnd: match.observedAt, sourceType: 'message-pair', sourceId: `${sourceSignalId}:${match.inboundId}`, provenanceRef: messagePairRef }, + conversationContinued: { value: true, status: 'observed', observedAt: match.observedAt, windowStart: sentAt, windowEnd: match.observedAt, sourceType: 'inbound-message', sourceId: match.inboundId, provenanceRef: inboundEvidenceRef }, + nextDayReinitiation: { value: null, status: 'pending', observedAt: '', windowStart: match.observedAt, windowEnd: nextDayWindowEnd, sourceType: 'conversation-window', sourceId: decision.conversationId, provenanceRef: '' } + } + }); + if (!repository || typeof repository.insertLearningSignal !== 'function') { + return { skipped: true, reasonCode: 'LEARNING_SIGNAL_LEDGER_UNAVAILABLE', outcomeVector: vector }; + } + const idempotencyKey = `policy-outcome:${decision.decisionId}:${sourceSignalId}:${match.inboundId}`; + const persisted = repository.insertLearningSignal({ + signalId: `learning-signal-${canonicalHash({ idempotencyKey }).slice(0, 24)}`, + idempotencyKey, + learningLevel: 'L1', + scopeType: 'conversation', + scopeId: decision.conversationId, + contactId: decision.contactId, + conversationId: decision.conversationId, + candidateId: clean(source.candidate_id || source.candidateId), + outboxId: clean(source.outbox_id || source.outboxId), + signalType: RAW_SIGNAL_TYPE, + signal: vector, + qualityTier: clean(source.quality_tier || source.qualityTier), + emergencyMode: false, + learningEligible: false, + createdAt: match.observedAt + }); + return { skipped: false, persisted, outcomeVector: vector }; + } + + function start() { + if (started) return status(); + listener = event => { + Promise.resolve().then(() => persistInboundOutcome(event?.payload?.message || {})).catch(error => { + bus.publish('learning-policy:outcome-attribution-failed', { + reasonCode: clean(error.reasonCode || error.code) || 'LEARNING_POLICY_OUTCOME_ATTRIBUTION_FAILED', + message: clean(error.message), + conversationId: clean(event?.payload?.message?.conversationId || event?.payload?.message?.sessionKey), + contactId: clean(event?.payload?.message?.contactId) + }); + }); + }; + bus.on('message:inserted', listener); + started = true; + return status(); + } + + function stop() { + if (listener) bus.off('message:inserted', listener); + listener = null; + started = false; + return { stopped: true }; + } + function status() { return Object.freeze({ started, authority: AUTHORITY, signalType: RAW_SIGNAL_TYPE, rawOutcomeLearningEligible: false }); } + + return Object.freeze({ createOutcomeVector, bindTrainableOutcome, resolveInboundAttribution, persistInboundOutcome, start, stop, status }); +} + +const singleton = createLearningOutcomeAttributionService(); +module.exports = { AUTHORITY, RAW_SIGNAL_TYPE, createLearningOutcomeAttributionService, singleton }; diff --git a/backend/services/learningPolicyDecisionContract.js b/backend/services/learningPolicyDecisionContract.js new file mode 100644 index 000000000..13152dee6 --- /dev/null +++ b/backend/services/learningPolicyDecisionContract.js @@ -0,0 +1,201 @@ +'use strict'; + +const { canonicalHash } = require('./canonicalSerialization'); +const personContextAuthority = require('./personContextAuthority').singleton; + +const AUTHORITY = 'LearningPolicyDecisionContract'; +const ACTION_ENCODING_VERSION = 'candidate-strategy-branch-v1'; +const ALLOWED_ACTIONS = Object.freeze([ + 'natural_hook', + 'playful_attraction', + 'direct_advance', + 'screen_and_advance', + 'leave_aftertaste' +]); +const FEATURE_SCHEMA = Object.freeze([ + 'interactionBand', + 'performanceMode', + 'questionPolicy', + 'relationshipStage', + 'targetLanguage' +]); +const PRIVATE_FEATURE_PATTERN = /(raw|chat|message|body|text|memory|name|email|phone|address|credential|api[_-]?key|secret|token|prompt)/iu; +const SAFE_ENUM_TOKEN = /^[\p{L}\p{N}_-]{1,64}$/u; + +function clean(value) { return String(value == null ? '' : value).trim(); } +function policyError(reasonCode, message, details = {}) { + return Object.assign(new Error(message || reasonCode), { reasonCode, code: reasonCode, ...details }); +} +function unique(values = []) { return [...new Set((Array.isArray(values) ? values : []).map(clean).filter(Boolean))]; } +function deepFreeze(value) { + if (!value || typeof value !== 'object' || Object.isFrozen(value)) return value; + Object.freeze(value); + for (const child of Object.values(value)) deepFreeze(child); + return value; +} + +function normalizeFeatureBundle(input = {}) { + if (!input || typeof input !== 'object' || Array.isArray(input)) { + throw policyError('LEARNING_POLICY_FEATURE_BUNDLE_INVALID', 'Learned Policy featureBundle must be a plain object.'); + } + const keys = Object.keys(input); + for (const key of keys) { + if (PRIVATE_FEATURE_PATTERN.test(key)) { + throw policyError( + 'LEARNING_POLICY_FEATURE_BUNDLE_PRIVATE_BODY_FORBIDDEN', + `Private/free-form feature field ${key} is forbidden.` + ); + } + if (!FEATURE_SCHEMA.includes(key)) { + throw policyError('LEARNING_POLICY_FEATURE_BUNDLE_FIELD_FORBIDDEN', `Feature field ${key} is outside the fixed P1 schema.`); + } + } + const output = {}; + for (const key of FEATURE_SCHEMA) { + if (!Object.prototype.hasOwnProperty.call(input, key)) continue; + const value = input[key]; + if (typeof value === 'boolean') output[key] = value; + else if (typeof value === 'number' && Number.isFinite(value)) output[key] = value; + else { + const token = clean(value); + if (!SAFE_ENUM_TOKEN.test(token)) { + throw policyError('LEARNING_POLICY_FEATURE_BUNDLE_VALUE_INVALID', `Feature ${key} must be a bounded enum/numeric/boolean value.`); + } + output[key] = token; + } + } + return deepFreeze(output); +} + +function normalizeGeneration(input = {}) { + const source = input && typeof input === 'object' && !Array.isArray(input) ? input : {}; + const output = {}; + for (const key of ['modelBrainExecutionId', 'candidatePlanId', 'directorStrategyId', 'contextVersion', 'conversationRevision']) { + if (source[key] == null || source[key] === '') continue; + if (['contextVersion', 'conversationRevision'].includes(key)) { + const numeric = Number(source[key]); + if (Number.isFinite(numeric)) output[key] = numeric; + } else { + const token = clean(source[key]); + if (token) output[key] = token.slice(0, 256); + } + } + return deepFreeze(output); +} + +function createLearningPolicyDecisionContract(options = {}) { + const identityAuthority = options.personContextAuthority || personContextAuthority; + if (!identityAuthority || typeof identityAuthority.resolve !== 'function') { + throw new TypeError('personContextAuthority.resolve is required'); + } + + function createDecisionRecord(input = {}) { + const contactId = clean(input.contactId); + const conversationId = clean(input.conversationId); + const personaProfileId = clean(input.personaProfileId); + if (!contactId || !conversationId || !personaProfileId) { + throw policyError('LEARNING_POLICY_DECISION_SCOPE_REQUIRED', 'contactId, conversationId and personaProfileId are required.'); + } + + const resolved = identityAuthority.resolve({ contactId, conversationId }); + const personId = clean(resolved?.personId); + const contactIds = unique(resolved?.contactIds); + const conversationIds = unique(resolved?.conversationIds); + const expectedPersonId = clean(input.expectedPersonId); + if ( + resolved?.authority !== 'PersonContextAuthority' || resolved?.found !== true || !personId || + !contactIds.includes(contactId) || !conversationIds.includes(conversationId) || + (expectedPersonId && expectedPersonId !== personId) + ) { + throw policyError( + 'LEARNING_POLICY_IDENTITY_BINDING_MISMATCH', + 'Learned Policy decision identity must resolve to one canonical Person/Contact/Conversation binding.', + { contactId, conversationId, resolvedPersonId: personId, expectedPersonId } + ); + } + + const featureBundle = normalizeFeatureBundle(input.featureBundle || {}); + const candidateStrategyBranch = clean(input.candidateStrategyBranch); + const allowedActionSet = unique(input.allowedActionSet?.length ? input.allowedActionSet : ALLOWED_ACTIONS); + if ( + !allowedActionSet.length || + allowedActionSet.some(action => !ALLOWED_ACTIONS.includes(action)) || + !allowedActionSet.includes(candidateStrategyBranch) + ) { + throw policyError('LEARNING_POLICY_ACTION_NOT_ALLOWED', 'candidateStrategyBranch/action-set is outside the exact P1 action authority.', { candidateStrategyBranch, allowedActionSet }); + } + const behaviorPolicyVersion = clean(input.behaviorPolicyVersion || input.policyVersion) || 'vw-p1-baseline-v1'; + const policyVersion = clean(input.policyVersion) || behaviorPolicyVersion; + const policyArtifactId = clean(input.policyArtifactId) || 'baseline'; + const generation = normalizeGeneration(input.generation); + const contributingPolicyVersions = deepFreeze({ + relationship: clean(input.contributingPolicyVersions?.relationship) || 'relationship-current-v1', + memory: clean(input.contributingPolicyVersions?.memory) || 'memory-current-v1', + strategy: clean(input.contributingPolicyVersions?.strategy) || 'director-strategy-current-v1', + candidateRanker: clean(input.contributingPolicyVersions?.candidateRanker) || ACTION_ENCODING_VERSION, + routing: clean(input.contributingPolicyVersions?.routing) || 'model-brain-routing-current-v1', + promptProgram: clean(input.contributingPolicyVersions?.promptProgram) || 'context-aware-reply-current-v1', + behaviorPolicy: behaviorPolicyVersion + }); + const stateSnapshotRef = canonicalHash({ + personId, contactId, conversationId, personaProfileId, featureBundle, + generation, contributingPolicyVersions + }); + const featureSchemaRef = canonicalHash({ schemaVersion: 1, fields: FEATURE_SCHEMA }); + const contextCandidateSetRef = canonicalHash({ + candidatePlanId: generation.candidatePlanId || '', + directorStrategyId: generation.directorStrategyId || '', + actions: allowedActionSet + }); + const actionSetRef = canonicalHash({ encoding: ACTION_ENCODING_VERSION, actions: allowedActionSet }); + const actionId = `candidateStrategyBranch:${candidateStrategyBranch}`; + const recordCore = { + schemaVersion: 1, + authority: AUTHORITY, + scopeType: 'conversation', + scopeId: conversationId, + contactId, + conversationId, + personId, + contactIds, + conversationIds, + personaProfileId, + featureBundle, + stateSnapshotRef, + featureSchemaRef, + contextCandidateSetRef, + actionId, + actionEncodingVersion: ACTION_ENCODING_VERSION, + allowedActionSet: deepFreeze([...allowedActionSet]), + actionSetRef, + chosenAction: { kind: 'candidateStrategyBranch', value: candidateStrategyBranch }, + candidateStrategyBranch, + behaviorPolicyVersion, + policyVersion, + policyArtifactId, + actionProbability: 1, + exploration: false, + generation, + contributingPolicyVersions, + rawPrivateChatPersisted: false + }; + const decisionId = `decision:${canonicalHash(recordCore)}`; + return deepFreeze({ ...recordCore, decisionId }); + } + + return Object.freeze({ + authority: AUTHORITY, + allowedActions: ALLOWED_ACTIONS, + featureSchema: FEATURE_SCHEMA, + createDecisionRecord + }); +} + +module.exports = { + AUTHORITY, + ACTION_ENCODING_VERSION, + ALLOWED_ACTIONS, + FEATURE_SCHEMA, + createLearningPolicyDecisionContract, + normalizeFeatureBundle +}; diff --git a/backend/services/learningPolicyRuntimeAdapter.js b/backend/services/learningPolicyRuntimeAdapter.js new file mode 100644 index 000000000..cbd0913da --- /dev/null +++ b/backend/services/learningPolicyRuntimeAdapter.js @@ -0,0 +1,99 @@ +'use strict'; + +const { ALLOWED_ACTIONS, normalizeFeatureBundle } = require('./learningPolicyDecisionContract'); + +const AUTHORITY = 'LearningPolicyRuntimeAdapter'; +const BASELINE_POLICY_VERSION = 'vw-p1-baseline-v1'; + +function clean(value) { return String(value == null ? '' : value).trim(); } +function runtimeError(reasonCode, message, details = {}) { + return Object.assign(new Error(message || reasonCode), { reasonCode, code: reasonCode, ...details }); +} +function deepFreeze(value) { + if (!value || typeof value !== 'object' || Object.isFrozen(value)) return value; + Object.freeze(value); + for (const child of Object.values(value)) deepFreeze(child); + return value; +} +function exactAllowedActions(values) { + const actions = Array.isArray(values) ? values.map(clean).filter(Boolean) : []; + if (!actions.length || actions.some(action => !ALLOWED_ACTIONS.includes(action)) || new Set(actions).size !== actions.length) { + throw runtimeError('LEARNING_POLICY_ACTION_SET_INVALID', 'Learned Policy requires a non-empty exact subset of the frozen P1 action set.'); + } + return actions; +} + +function createLearningPolicyRuntimeAdapter(options = {}) { + const invokeVowpalWabbit = typeof options.invokeVowpalWabbit === 'function' ? options.invokeVowpalWabbit : null; + const resolveActivePolicy = typeof options.resolveActivePolicy === 'function' ? options.resolveActivePolicy : null; + const onDegradation = typeof options.onDegradation === 'function' ? options.onDegradation : null; + + function baseline(input, reasonCode = 'NO_PROMOTED_POLICY') { + const actions = exactAllowedActions(input.allowedActions); + const requested = clean(input.baselineAction); + const candidateStrategyBranch = actions.includes(requested) ? requested : actions[0]; + return deepFreeze({ + authority: AUTHORITY, + candidateStrategyBranch, + policyVersion: BASELINE_POLICY_VERSION, + policyArtifactId: 'baseline', + actionProbability: 1, + exploration: false, + degradation: reasonCode === 'NO_PROMOTED_POLICY' ? null : { reasonCode }, + executedPolicy: 'baseline' + }); + } + + async function selectLearnedPolicyAction(input = {}) { + const featureBundle = normalizeFeatureBundle(input.featureBundle || {}); + const allowedActions = exactAllowedActions(input.allowedActions || ALLOWED_ACTIONS); + let activePolicy = input.activePolicy && typeof input.activePolicy === 'object' ? input.activePolicy : null; + if (!activePolicy && resolveActivePolicy) activePolicy = await resolveActivePolicy({ featureBundle, allowedActions }); + + // Production constructs this adapter without an arbitrary request-supplied + // runtime. Tests/UAT may inject the already-sealed VW operation directly. + // No runtime injection means an explicit deterministic baseline. + if (!invokeVowpalWabbit && !activePolicy) return baseline({ ...input, allowedActions }, 'NO_PROMOTED_POLICY'); + if (!invokeVowpalWabbit) { + onDegradation?.({ reasonCode: 'SEALED_VW_RUNTIME_UNAVAILABLE', policyArtifactId: clean(activePolicy?.policyArtifactId) }); + return baseline({ ...input, allowedActions }, 'SEALED_VW_RUNTIME_UNAVAILABLE'); + } + + try { + const result = await invokeVowpalWabbit({ + operation: 'policy_predict', + featureBundle, + allowedActions, + policyArtifactId: clean(activePolicy?.policyArtifactId || input.policyArtifactId), + policyVersion: clean(activePolicy?.policyVersion || input.policyVersion) || 'vw-p1-v1' + }); + const action = clean(result?.action || result?.candidateStrategyBranch); + if (!allowedActions.includes(action)) { + throw runtimeError('LEARNING_POLICY_RUNTIME_ACTION_INVALID', 'Sealed VW runtime returned an action outside the supplied exact action set.', { action }); + } + const probability = Number(result?.probability ?? result?.actionProbability ?? 1); + if (probability !== 1 || result?.exploration === true) { + throw runtimeError('LEARNING_POLICY_RUNTIME_P1_NONDETERMINISTIC', 'P1 Learned Policy must remain deterministic with probability 1 and exploration disabled.'); + } + return deepFreeze({ + authority: AUTHORITY, + candidateStrategyBranch: action, + policyVersion: clean(result?.policyVersion || activePolicy?.policyVersion || input.policyVersion) || 'vw-p1-v1', + policyArtifactId: clean(result?.policyArtifactId || result?.policyArtifactVersion || activePolicy?.policyArtifactId || input.policyArtifactId) || 'baseline', + actionProbability: 1, + exploration: false, + degradation: null, + executedPolicy: 'vowpalwabbit' + }); + } catch (error) { + const reasonCode = clean(error?.reasonCode || error?.code) || 'SEALED_VW_POLICY_PREDICTION_FAILED'; + onDegradation?.({ reasonCode, policyArtifactId: clean(activePolicy?.policyArtifactId), message: clean(error?.message) }); + if (input.failClosed === true) throw error; + return baseline({ ...input, allowedActions }, reasonCode); + } + } + + return Object.freeze({ authority: AUTHORITY, selectLearnedPolicyAction, baseline }); +} + +module.exports = { AUTHORITY, BASELINE_POLICY_VERSION, createLearningPolicyRuntimeAdapter }; diff --git a/backend/services/replyFeedbackLearningService.js b/backend/services/replyFeedbackLearningService.js index f2438939b..536ede6c0 100644 --- a/backend/services/replyFeedbackLearningService.js +++ b/backend/services/replyFeedbackLearningService.js @@ -37,6 +37,10 @@ function buildImmutableFeedbackSignal(input = {}) { const evidenceKey = clean(input.evidenceId) || (eventType === 'sent' ? outboxId : candidateId); const idempotencyKey = 'reply-feedback:' + eventType + ':' + evidenceKey; const branch = input.candidateStrategyBranch || metadata.candidateStrategyBranch || {}; + const decisionRecord = input.decisionRecord || metadata.decisionRecord || null; + const immutableDecisionRecord = decisionRecord && typeof decisionRecord === 'object' && !Array.isArray(decisionRecord) + ? Object.freeze({ ...decisionRecord }) + : null; return Object.freeze({ skipped: false, signalId: signalId(idempotencyKey), @@ -61,6 +65,7 @@ function buildImmutableFeedbackSignal(input = {}) { hasExplicitRejectionReason: input.hasExplicitRejectionReason === true, adjustments: normalizeAdjustments(metadata, input.styleVariant), strategyBranch: clean(metadata.candidateStrategyBranchId || branch.strategy), + ...(immutableDecisionRecord ? { decisionRecord: immutableDecisionRecord } : {}), metadata: Object.freeze({ source: clean(input.source) || 'reply-outcome-transaction', replyTask: clean(input.replyTask || metadata.replyTask), @@ -118,4 +123,4 @@ function status() { }); } -module.exports = { AUTHORITY, start, stop, status, waitForIdle, isLearningDisabled, normalizeAdjustments, buildImmutableFeedbackSignal, persistImmutableLearningSignal }; +module.exports = { AUTHORITY, start, stop, status, waitForIdle, isLearningDisabled, normalizeAdjustments, buildImmutableFeedbackSignal, persistImmutableLearningSignal }; \ No newline at end of file diff --git a/backend/services/storeManagerService.js b/backend/services/storeManagerService.js index c860aeaae..51e2bf592 100644 --- a/backend/services/storeManagerService.js +++ b/backend/services/storeManagerService.js @@ -11,6 +11,7 @@ const { registerRuntimeStateCommands } = require('../store/commands/registerRunt const { StoreIntegrityMonitor } = require('../store/StoreIntegrityMonitor'); const typingStateService = require('./typingStateService'); const replyFeedbackLearningService = require('./replyFeedbackLearningService'); +const learningOutcomeAttributionService = require('./learningOutcomeAttributionService').singleton; const conversationTurnCoordinator = require('./conversationTurnCoordinator'); const aiTaskRuntimeRegistry = require('./aiTaskRuntimeRegistry'); const personaBrainModule = require('../personaBrain'); @@ -101,6 +102,10 @@ async function initialize(options = {}) { }); conversationTurnCoordinator.start(); replyFeedbackLearningService.start({ storeManager, personaBrain }); + // Outcome attribution deliberately starts only after authoritative StoreManager + // hydration. It reuses the existing message:inserted event and immutable + // learning_signal_ledger; it owns no second inbound pipeline or scheduler. + learningOutcomeAttributionService.start(); integrityMonitor = new StoreIntegrityMonitor({ storeManager, logger, @@ -135,6 +140,7 @@ function status() { } function stop() { + learningOutcomeAttributionService.stop(); replyFeedbackLearningService.stop(); conversationTurnCoordinator.stop(); typingStateService.stop(); @@ -143,4 +149,4 @@ function stop() { started = false; } -module.exports = { initialize, status, stop, ensureCustomerContext }; +module.exports = { initialize, status, stop, ensureCustomerContext }; \ No newline at end of file diff --git a/config/upstreams/v21-learning-growth-brain-p0.json b/config/upstreams/v21-learning-growth-brain-p0.json index 548a6e240..53fe96b6f 100644 --- a/config/upstreams/v21-learning-growth-brain-p0.json +++ b/config/upstreams/v21-learning-growth-brain-p0.json @@ -8,6 +8,7 @@ "opentelemetryPython": { "repository": "open-telemetry/opentelemetry-python", "version": "1.44.0", "commit": "53a5a40c9604583c501bcf13970a635f00e62df4", "license": "Apache-2.0" }, "dspy": { "repository": "stanfordnlp/dspy", "version": "3.3.0", "commit": "e4e97aae29b8ad8aa2fb7e99ffae6fd52970fad8", "license": "MIT" }, "gepa": { "repository": "gepa-ai/gepa", "version": "0.1.1", "commit": "b4dbb55b7601dac448cdb836d5a401ca7d9eb920", "license": "MIT" }, + "vowpalWabbit": { "repository": "VowpalWabbit/vowpal_wabbit", "version": "9.11.2", "commit": "122bae254a5b8bc2b774d13b33d53e6dbc2cfba7", "license": "BSD-3-Clause", "mode": "contextual-bandit-adf-offline-candidate-policy" }, "promptfoo": { "repository": "promptfoo/promptfoo", "version": "0.122.0", "commit": "7b898cbdb16205cb7f0e2994baa807d131eb2326", "license": "MIT", "mode": "precomputed-provider-only" }, "apscheduler": { "repository": "agronholm/apscheduler", "version": "3.11.3", "commit": "4308ec95b94069f5dbdddb6c60fb792dfc8c40a4", "license": "MIT" }, "presidio": { "repository": "microsoft/presidio", "version": "2.2.364", "commit": "779dbd286d5ef4d1fbe2514275fb1bce358f2417", "license": "MIT", "p0Nlp": "NoOpNlpEngine" }, diff --git a/runtime/learning-growth/python/learning_entrypoint.py b/runtime/learning-growth/python/learning_entrypoint.py index b606e8992..a479d3aad 100644 --- a/runtime/learning-growth/python/learning_entrypoint.py +++ b/runtime/learning-growth/python/learning_entrypoint.py @@ -2,15 +2,23 @@ """Thin Learning Growth Brain OSS composition entrypoint. DSPy + GEPA own optimization, APScheduler owns scheduling, Presidio owns -source-side PII minimization, and all live model execution remains delegated -to Yance Model Brain V4. This process never owns model provider credentials. +source-side PII minimization, Vowpal Wabbit owns the bounded learned-policy +action head, and all live model execution remains delegated to Yance Model +Brain V4. This process never owns model provider credentials or reply text. """ from __future__ import annotations +import hashlib import json +import math import sys from dataclasses import dataclass -from typing import Any, Mapping +from importlib.metadata import version as package_version +from pathlib import Path +from typing import Any, Mapping, Sequence + +LEARNED_POLICY_ACTION_ENCODING = "candidate-strategy-branch-v1" +LEARNED_POLICY_FEATURES = ("interactionBand", "performanceMode", "questionPolicy", "relationshipStage", "targetLanguage") @dataclass(frozen=True) @@ -58,18 +66,238 @@ def optimizer_contract() -> dict[str, str]: } +def _assert_allowed_request_fields(payload: Mapping[str, Any], allowed: set[str]) -> None: + unexpected = set(payload.keys()) - allowed + if unexpected: + # Never echo unknown request keys: the sealed action head accepts only + # its exact bounded schema and does not inspect arbitrary extensions. + raise ValueError(f"LEARNED_POLICY_REQUEST_SCHEMA_MISMATCH:{len(unexpected)}") + + +def _safe_token(value: Any) -> str: + text = str(value if value is not None else "").strip() + if not text or len(text) > 64 or any(ch.isspace() for ch in text): + raise ValueError("LEARNED_POLICY_FEATURE_VALUE_INVALID") + return "".join(ch if ch.isalnum() or ch in "_-" else "_" for ch in text) + + +def _feature_bundle(value: Any) -> dict[str, Any]: + if not isinstance(value, Mapping): + raise ValueError("LEARNED_POLICY_FEATURE_BUNDLE_INVALID") + unknown = sorted(set(value.keys()) - set(LEARNED_POLICY_FEATURES)) + if unknown: + raise ValueError("LEARNED_POLICY_FEATURE_FIELD_FORBIDDEN:" + ",".join(map(str, unknown))) + result: dict[str, Any] = {} + for key in LEARNED_POLICY_FEATURES: + if key not in value: + continue + item = value[key] + if isinstance(item, bool): + result[key] = item + elif isinstance(item, (int, float)) and not isinstance(item, bool): + numeric = float(item) + if not math.isfinite(numeric): + raise ValueError("LEARNED_POLICY_FEATURE_VALUE_INVALID") + result[key] = numeric + else: + result[key] = _safe_token(item) + return result + + +def _actions(value: Any) -> list[str]: + if not isinstance(value, Sequence) or isinstance(value, (str, bytes)): + raise ValueError("LEARNED_POLICY_ACTION_SET_INVALID") + actions = [_safe_token(item) for item in value] + if not actions or len(actions) != len(set(actions)): + raise ValueError("LEARNED_POLICY_ACTION_SET_INVALID") + return actions + + +def _shared_features(features: Mapping[str, Any]) -> str: + tokens: list[str] = [] + for key in LEARNED_POLICY_FEATURES: + if key not in features: + continue + value = features[key] + if isinstance(value, bool): + tokens.append(f"{key}_{str(value).lower()}") + elif isinstance(value, (int, float)) and not isinstance(value, bool): + tokens.append(f"{key}:{float(value)}") + else: + tokens.append(f"{key}_{_safe_token(value)}") + return " ".join(tokens) or "bias" + + +def _adf_lines( + features: Mapping[str, Any], + actions: Sequence[str], + chosen_action: str = "", + cost: float | None = None, + probability: float | None = None, +) -> list[str]: + lines = [f"shared |c {_shared_features(features)}"] + for action in actions: + prefix = "" + if action == chosen_action: + if cost is None or probability is None: + raise ValueError("LEARNED_POLICY_LOGGED_FEEDBACK_REQUIRED") + prefix = f"0:{cost}:{probability} " + lines.append(f"{prefix}|a action_{_safe_token(action)}") + return lines + + +def _artifact_sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +def _policy_rows(payload: Mapping[str, Any]) -> list[Mapping[str, Any]]: + rows = payload.get("rows") + if not isinstance(rows, list) or not rows: + raise ValueError("LEARNED_POLICY_TRAINING_ROWS_REQUIRED") + return [row for row in rows if isinstance(row, Mapping)] + + +def policy_runtime_contract() -> dict[str, Any]: + import vowpalwabbit + + return { + "status": "READY", + "authority": "Vowpal Wabbit", + "vowpalwabbit": package_version("vowpalwabbit"), + "mode": "contextual-bandit-adf-offline-candidate-policy", + "actionEncodingVersion": LEARNED_POLICY_ACTION_ENCODING, + "operations": ["policy_runtime_contract", "policy_train", "policy_predict"], + "exploration": False, + "textGeneration": False, + "workspace": getattr(vowpalwabbit, "Workspace").__name__, + } + + +def policy_train(payload: Mapping[str, Any]) -> dict[str, Any]: + _assert_allowed_request_fields(payload, {"operation", "rows", "artifactPath"}) + from vowpalwabbit import Workspace + + artifact_path = Path(str(payload.get("artifactPath") or "").strip()) + if not str(artifact_path) or str(artifact_path) == ".": + raise ValueError("LEARNED_POLICY_ARTIFACT_PATH_REQUIRED") + artifact_path.parent.mkdir(parents=True, exist_ok=True) + workspace = Workspace(arg_list=["--cb_adf", "--rank_all", "--quiet"]) + learned = 0 + try: + for row in _policy_rows(payload): + decision = row.get("decision") if isinstance(row.get("decision"), Mapping) else {} + features = _feature_bundle(row.get("featureBundle") or decision.get("featureBundle") or {}) + actions = _actions(decision.get("allowedActionSet") or row.get("allowedActions") or []) + chosen = str(decision.get("candidateStrategyBranch") or decision.get("chosenAction", {}).get("value") or "").strip() + if chosen not in actions: + raise ValueError("LEARNED_POLICY_CHOSEN_ACTION_INVALID") + probability = float(decision.get("actionProbability")) + if not math.isfinite(probability) or probability <= 0 or probability > 1: + raise ValueError("LEARNED_POLICY_LOGGED_PROPENSITY_INVALID") + score = row.get("approvedScore") if isinstance(row.get("approvedScore"), Mapping) else row.get("score") + if not isinstance(score, Mapping) or score.get("approvedByLearning") is not True: + raise ValueError("LEARNED_POLICY_APPROVED_SCORE_REQUIRED") + approved_score = float(score.get("value")) + if not math.isfinite(approved_score): + raise ValueError("LEARNED_POLICY_SCORE_NONFINITE") + cost = -float(approved_score) + workspace.learn(_adf_lines(features, actions, chosen, cost, probability)) + learned += 1 + workspace.save(str(artifact_path)) + finally: + workspace.finish() + digest = _artifact_sha256(artifact_path) + return { + "status": "READY", + "rowCount": learned, + "policyArtifactVersion": digest, + "policyArtifactId": digest, + "actionEncodingVersion": LEARNED_POLICY_ACTION_ENCODING, + "rewardToCost": "cost=-float(approved_score)", + "probability": 1.0, + "exploration": False, + } + + +def _prediction_action_index(prediction: Any, action_count: int) -> int: + if isinstance(prediction, list) and prediction: + first = prediction[0] + if hasattr(first, "action"): + index = int(first.action) + return index if 0 <= index < action_count else index - 1 + if isinstance(first, (tuple, list)) and first: + index = int(first[0]) + return index if 0 <= index < action_count else index - 1 + if all(isinstance(value, (int, float)) for value in prediction): + return min(range(len(prediction)), key=lambda idx: float(prediction[idx])) + if isinstance(prediction, int): + return prediction if 0 <= prediction < action_count else prediction - 1 + raise ValueError("LEARNED_POLICY_PREDICTION_INVALID") + + +def policy_predict(payload: Mapping[str, Any]) -> dict[str, Any]: + _assert_allowed_request_fields(payload, { + "operation", "featureBundle", "allowedActions", "artifactPath", + "policyArtifactId", "policyArtifactVersion", "policyVersion" + }) + from vowpalwabbit import Workspace + + artifact_path = Path(str(payload.get("artifactPath") or "").strip()) + if not artifact_path.is_file(): + raise ValueError("LEARNED_POLICY_ARTIFACT_MISSING") + digest = _artifact_sha256(artifact_path) + expected = str(payload.get("policyArtifactId") or payload.get("policyArtifactVersion") or "").strip() + if expected and expected != digest: + raise ValueError("LEARNED_POLICY_ARTIFACT_IDENTITY_MISMATCH") + features = _feature_bundle(payload.get("featureBundle") or {}) + actions = _actions(payload.get("allowedActions") or []) + workspace = Workspace(arg_list=["--cb_adf", "--rank_all", "--quiet", "-i", str(artifact_path)]) + try: + prediction = workspace.predict(_adf_lines(features, actions)) + finally: + workspace.finish() + index = _prediction_action_index(prediction, len(actions)) + if index < 0 or index >= len(actions): + raise ValueError("LEARNED_POLICY_PREDICTION_ACTION_OUT_OF_RANGE") + return { + "status": "READY", + "action": actions[index], + "candidateStrategyBranch": actions[index], + "policyArtifactId": digest, + "policyArtifactVersion": digest, + "policyVersion": str(payload.get("policyVersion") or "vw-p1-v1"), + "probability": 1.0, + "exploration": False, + "textGeneration": False, + } + + def main() -> int: request = json.load(sys.stdin) operation = str(request.get("operation") or "evaluate") - if operation == "runtime_contract": - json.dump(optimizer_contract(), sys.stdout, sort_keys=True) - return 0 - if operation == "evaluate": - json.dump(evaluate_precomputed(request), sys.stdout, sort_keys=True) - return 0 - json.dump({"status": "UNSUPPORTED_OPERATION", "operation": operation}, sys.stdout, sort_keys=True) - return 2 + try: + if operation == "runtime_contract": + json.dump(optimizer_contract(), sys.stdout, sort_keys=True) + return 0 + if operation == "evaluate": + json.dump(evaluate_precomputed(request), sys.stdout, sort_keys=True) + return 0 + if operation == "policy_runtime_contract": + _assert_allowed_request_fields(request, {"operation"}) + json.dump(policy_runtime_contract(), sys.stdout, sort_keys=True) + return 0 + if operation == "policy_train": + json.dump(policy_train(request), sys.stdout, sort_keys=True) + return 0 + if operation == "policy_predict": + json.dump(policy_predict(request), sys.stdout, sort_keys=True) + return 0 + json.dump({"status": "UNSUPPORTED_OPERATION", "operation": operation}, sys.stdout, sort_keys=True) + return 2 + except Exception as error: # fail closed; caller receives a structured sealed-runtime RED + json.dump({"status": "ERROR", "operation": operation, "error": str(error)}, sys.stdout, sort_keys=True) + return 1 if __name__ == "__main__": - raise SystemExit(main()) + raise SystemExit(main()) \ No newline at end of file diff --git a/runtime/learning-growth/python/pyproject.toml b/runtime/learning-growth/python/pyproject.toml index e192f84a2..e0c683fe1 100644 --- a/runtime/learning-growth/python/pyproject.toml +++ b/runtime/learning-growth/python/pyproject.toml @@ -10,8 +10,9 @@ dependencies = [ "opentelemetry-sdk==1.44.0", "opentelemetry-exporter-otlp-proto-http==1.44.0", "presidio-analyzer==2.2.364", - "presidio-anonymizer==2.2.364" + "presidio-anonymizer==2.2.364", + "vowpalwabbit==9.11.2" ] [tool.uv] -package = false +package = false \ No newline at end of file diff --git a/runtime/learning-growth/python/uv.lock b/runtime/learning-growth/python/uv.lock index 845df8d5a..cf2aa6e32 100644 --- a/runtime/learning-growth/python/uv.lock +++ b/runtime/learning-growth/python/uv.lock @@ -1460,6 +1460,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/7f/3e/5db95bcf282c52709639744ca2a8b149baccf648e39c8cc87553df9eae0c/urllib3-2.7.0-py3-none-any.whl", hash = "sha256:9fb4c81ebbb1ce9531cce37674bbc6f1360472bc18ca9a553ede278ef7276897", size = 131087, upload-time = "2026-05-07T16:13:17.151Z" }, ] +[[package]] +name = "vowpalwabbit" +version = "9.11.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/4d/ff/bd0ffb16f5792a00151136741cfad876777777db7d14aae9a9b126d03c6f/vowpalwabbit-9.11.2.tar.gz", hash = "sha256:c0ecf8d773c0174a9ab7b0f9207a2608c894bc2fbec50380ca276e89553fd379", size = 35442893, upload-time = "2026-03-07T16:25:32.093Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/33/5b/f88c516cd411adb4636c9fa8df828c2e5338de3dc36accc68ea0143d4469/vowpalwabbit-9.11.2-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:6ee450a5dd40a60a9a12abfe16972410e8e5d06a7ee92d8c6ec15be4a7ec0ecb", size = 2283281, upload-time = "2026-03-07T16:25:07.933Z" }, + { url = "https://files.pythonhosted.org/packages/9a/a1/c8d4ab89a2a3346645683ccc1b4e3c1fbbf68dc70d9901f896cfa03ac3f8/vowpalwabbit-9.11.2-cp312-cp312-macosx_11_0_x86_64.whl", hash = "sha256:fb962bce215757e4cd49d8d90341617416d5b360abdf476daa109b3eb4312825", size = 2479835, upload-time = "2026-03-07T16:25:09.984Z" }, + { url = "https://files.pythonhosted.org/packages/ec/2b/bbc242bdd69d65daf130b65ae08046fa5ddac4effe85deb8dcd7de3cad7c/vowpalwabbit-9.11.2-cp312-cp312-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:690353cc881bbd26ac81e758ed4b7fe0fecae7edb534a1a86c26246e38c5aecd", size = 3010926, upload-time = "2026-03-07T16:25:11.727Z" }, + { url = "https://files.pythonhosted.org/packages/8e/cf/62ae2a81385ca007a2b1db52a9a1d9a2a27afd2c183b14a8e522661295ed/vowpalwabbit-9.11.2-cp312-cp312-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:62d117df1f5d2994130ee5b9f145d710c8198d065b061f0480ba0cab78aa3141", size = 3478813, upload-time = "2026-03-07T16:25:13.461Z" }, + { url = "https://files.pythonhosted.org/packages/06/3f/9231a026e3344118605a752933df5e935602efaa0281e50fe3d60361f3f3/vowpalwabbit-9.11.2-cp312-cp312-win_amd64.whl", hash = "sha256:bdf3267faec119ce69dd147b11da234507cc1acd38a6ad61521188918e06d382", size = 1961008, upload-time = "2026-03-07T16:25:14.747Z" }, +] + [[package]] name = "wasabi" version = "1.1.3" @@ -1525,6 +1538,7 @@ dependencies = [ { name = "opentelemetry-sdk" }, { name = "presidio-analyzer" }, { name = "presidio-anonymizer" }, + { name = "vowpalwabbit" }, ] [package.metadata] @@ -1537,6 +1551,7 @@ requires-dist = [ { name = "opentelemetry-sdk", specifier = "==1.44.0" }, { name = "presidio-analyzer", specifier = "==2.2.364" }, { name = "presidio-anonymizer", specifier = "==2.2.364" }, + { name = "vowpalwabbit", specifier = "==9.11.2" }, ] [[package]] diff --git a/tests/wp0/v21-learning-policy-p1-decision-record.test.js b/tests/wp0/v21-learning-policy-p1-decision-record.test.js new file mode 100644 index 000000000..c22e79db6 --- /dev/null +++ b/tests/wp0/v21-learning-policy-p1-decision-record.test.js @@ -0,0 +1,94 @@ +'use strict'; + +const test = require('node:test'); +const assert = require('node:assert/strict'); + +const { createLearningPolicyDecisionContract } = require('../../backend/services/learningPolicyDecisionContract'); + +test('Learned Policy DecisionRecord binds canonical identity, persona, features, action and generation provenance', () => { + const authorityCalls = []; + const contract = createLearningPolicyDecisionContract({ + personContextAuthority: { + resolve(input) { + authorityCalls.push(input); + return { + authority: 'PersonContextAuthority', + found: true, + personId: 'person-1', + contactIds: ['contact-1'], + conversationIds: ['conversation-1'] + }; + } + } + }); + + const record = contract.createDecisionRecord({ + contactId: 'contact-1', + conversationId: 'conversation-1', + personaProfileId: 'persona-owner-v7', + featureBundle: { + relationshipStage: 'warming', + interactionBand: 'balanced', + targetLanguage: 'en' + }, + candidateStrategyBranch: 'screen_and_advance', + policyVersion: 'vw-p1-v1', + policyArtifactId: 'artifact-sha256:abc123', + generation: { + modelBrainExecutionId: 'exec-1', + candidatePlanId: 'plan-1' + } + }); + + assert.equal(authorityCalls.length, 1); + assert.equal(record.personId, 'person-1'); + assert.deepEqual(record.contactIds, ['contact-1']); + assert.deepEqual(record.conversationIds, ['conversation-1']); + assert.equal(record.personaProfileId, 'persona-owner-v7'); + assert.equal(record.candidateStrategyBranch, 'screen_and_advance'); + assert.equal(record.actionProbability, 1); + assert.equal(record.exploration, false); + assert.equal(record.policyVersion, 'vw-p1-v1'); + assert.equal(record.policyArtifactId, 'artifact-sha256:abc123'); + assert.match(record.decisionId, /^decision:/); + assert.equal(Object.isFrozen(record), true); +}); + +test('DecisionRecord fails closed on identity/persona ambiguity and rejects raw chat bodies in features', () => { + const ambiguous = createLearningPolicyDecisionContract({ + personContextAuthority: { + resolve() { + return { authority: 'PersonContextAuthority', found: true, personId: 'person-1', contactIds: ['contact-1'], conversationIds: ['conversation-2'] }; + } + } + }); + + assert.throws(() => ambiguous.createDecisionRecord({ + contactId: 'contact-1', + conversationId: 'conversation-1', + personaProfileId: 'persona-owner-v7', + featureBundle: { relationshipStage: 'warming' }, + candidateStrategyBranch: 'natural_hook', + policyVersion: 'vw-p1-v1', + policyArtifactId: 'artifact-sha256:abc123', + generation: { modelBrainExecutionId: 'exec-1' } + }), error => error?.reasonCode === 'LEARNING_POLICY_IDENTITY_BINDING_MISMATCH'); + + const valid = createLearningPolicyDecisionContract({ + personContextAuthority: { + resolve() { + return { authority: 'PersonContextAuthority', found: true, personId: 'person-1', contactIds: ['contact-1'], conversationIds: ['conversation-1'] }; + } + } + }); + assert.throws(() => valid.createDecisionRecord({ + contactId: 'contact-1', + conversationId: 'conversation-1', + personaProfileId: 'persona-owner-v7', + featureBundle: { rawChatBody: 'private text' }, + candidateStrategyBranch: 'natural_hook', + policyVersion: 'vw-p1-v1', + policyArtifactId: 'artifact-sha256:abc123', + generation: { modelBrainExecutionId: 'exec-1' } + }), error => error?.reasonCode === 'LEARNING_POLICY_FEATURE_BUNDLE_PRIVATE_BODY_FORBIDDEN'); +}); diff --git a/tests/wp0/v21-learning-policy-p1-outcome-binding.test.js b/tests/wp0/v21-learning-policy-p1-outcome-binding.test.js new file mode 100644 index 000000000..3db7b6960 --- /dev/null +++ b/tests/wp0/v21-learning-policy-p1-outcome-binding.test.js @@ -0,0 +1,62 @@ +'use strict'; + +const test = require('node:test'); +const assert = require('node:assert/strict'); + +const { createLearningOutcomeAttributionService } = require('../../backend/services/learningOutcomeAttributionService'); + +test('raw OutcomeVector is immutable, decision-bound and permanently non-trainable', () => { + const service = createLearningOutcomeAttributionService(); + const vector = service.createOutcomeVector({ + decisionId: 'decision:abc', + outcomes: [ + { outcomeId: 'outcome-1', type: 'reply_received', value: 1, evidenceRef: 'message-99' }, + { outcomeId: 'outcome-2', type: 'conversation_continued', value: 1, evidenceRef: 'conversation-1' } + ], + observedAt: '2026-08-14T11:00:00.000Z' + }); + + assert.equal(vector.signalType, 'policy_outcome_observed'); + assert.equal(vector.learningEligible, false); + assert.equal(vector.decisionId, 'decision:abc'); + assert.equal(Object.isFrozen(vector), true); + assert.equal(Object.isFrozen(vector.outcomes), true); +}); + +test('trainable binding requires the immutable eligible candidate_sent anchor and Learning-approved Langfuse Score', () => { + const service = createLearningOutcomeAttributionService(); + const vector = service.createOutcomeVector({ + decisionId: 'decision:abc', + outcomes: [{ outcomeId: 'outcome-1', type: 'reply_received', value: 1, evidenceRef: 'message-99' }], + observedAt: '2026-08-14T11:00:00.000Z' + }); + const source = { + signal_id: 'signal-1', + signal_type: 'candidate_sent', + learning_eligible: true, + signal: { decisionRecord: { decisionId: 'decision:abc' } } + }; + const score = { + authority: 'Langfuse', + approvedByLearning: true, + scoreId: 'score-1', + eligibleSourceSignalId: 'signal-1', + decisionId: 'decision:abc', + outcomeIds: ['outcome-1'], + outcomeEvidenceSetRef: 'evidence-set-1', + rewardPolicyVersion: 'reward-v1', + value: 1 + }; + + const bound = service.bindTrainableOutcome({ eligibleSourceSignal: source, outcomeVector: vector, score }); + assert.equal(bound.eligibleSourceSignalId, 'signal-1'); + assert.equal(bound.decisionId, 'decision:abc'); + assert.equal(bound.reward.authority, 'Langfuse'); + assert.equal(vector.learningEligible, false, 'binding may never upgrade the raw outcome row'); + + assert.throws(() => service.bindTrainableOutcome({ + eligibleSourceSignal: { ...source, learning_eligible: false }, + outcomeVector: vector, + score + }), error => error?.reasonCode === 'LEARNING_POLICY_ELIGIBLE_SOURCE_SIGNAL_REQUIRED'); +}); diff --git a/tests/wp0/v21-learning-policy-p1-production-consumption.test.js b/tests/wp0/v21-learning-policy-p1-production-consumption.test.js new file mode 100644 index 000000000..a629ababf --- /dev/null +++ b/tests/wp0/v21-learning-policy-p1-production-consumption.test.js @@ -0,0 +1,24 @@ +'use strict'; + +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const path = require('node:path'); + +const ROOT = path.resolve(__dirname, '..', '..'); +const source = () => fs.readFileSync(path.join(ROOT, 'backend/services/contextAwareReplyBrain.js'), 'utf8'); + +test('production reply brain consumes Learned Policy before existing frontier generation', () => { + const text = source(); + assert.match(text, /require\(['"]\.\/learningPolicyRuntimeAdapter['"]\)/); + assert.match(text, /selectLearnedPolicyAction\s*\(/); + assert.match(text, /candidateStrategyBranch/); + assert.match(text, /aiGateway\.execute\s*\(/, 'existing Model Brain gateway must remain the final generator'); +}); + +test('Learned Policy production consumption cannot take provider/model credential or final-text authority', () => { + const text = source(); + assert.doesNotMatch(text, /learnedPolicy[^\n]{0,120}(apiKey|credential|providerCredential)/i); + assert.doesNotMatch(text, /learnedPolicy[^\n]{0,120}modelId\s*:/i); + assert.doesNotMatch(text, /learnedPolicy[^\n]{0,120}(finalReply|finalText)\s*:/i); +}); diff --git a/tests/wp0/v21-learning-policy-p1-projection.test.js b/tests/wp0/v21-learning-policy-p1-projection.test.js new file mode 100644 index 000000000..97050511f --- /dev/null +++ b/tests/wp0/v21-learning-policy-p1-projection.test.js @@ -0,0 +1,72 @@ +'use strict'; + +const test = require('node:test'); +const assert = require('node:assert/strict'); + +const { createLearningDeepTrainingContract } = require('../../backend/services/learningDeepTrainingContract'); + +test('projectPolicy reads only immutable eligible source signals and joins raw outcomes by decisionId', async () => { + const repository = { + async listLearningSignals(query) { + assert.equal(query.learningEligible, true); + return [{ + signal_id: 'signal-1', + signal_type: 'candidate_sent', + learning_eligible: true, + scope_type: 'relationship', + scope_id: 'person-1', + signal: { + decisionRecord: { + decisionId: 'decision:abc', + candidateStrategyBranch: 'natural_hook', + featureBundle: { relationshipStage: 'warming' }, + policyVersion: 'vw-p1-v1', + policyArtifactId: 'artifact-sha256:abc123' + } + } + }]; + }, + async listPolicyOutcomeSignals({ decisionIds }) { + assert.deepEqual(decisionIds, ['decision:abc']); + return [{ + signal_id: 'raw-outcome-1', + signal_type: 'policy_outcome_observed', + learning_eligible: false, + signal: { + decisionId: 'decision:abc', + outcomes: [{ outcomeId: 'outcome-1', type: 'reply_received', value: 1, evidenceRef: 'message-99' }] + } + }]; + } + }; + const contract = createLearningDeepTrainingContract({ repository, dataPolicy: { minimize: async value => ({ allowed: true, text: value.text || '' }) } }); + const projection = await contract.projectPolicy({ + scopeType: 'relationship', + scopeId: 'person-1', + approvedScoresBySignalId: { + 'signal-1': { + authority: 'Langfuse', approvedByLearning: true, scoreId: 'score-1', name: 'policy_reward', value: 1, traceId: 'trace-1', + eligibleSourceSignalId: 'signal-1', decisionId: 'decision:abc', outcomeIds: ['outcome-1'], outcomeEvidenceSetRef: 'evidence-set-1', rewardPolicyVersion: 'reward-v1' + } + } + }); + + assert.equal(projection.authority, 'Learning'); + assert.equal(projection.readOnly, true); + assert.equal(projection.trajectory.length, 1); + assert.equal(projection.trajectory[0].decisionId, 'decision:abc'); + assert.equal(projection.trajectory[0].outcomes[0].outcomeId, 'outcome-1'); +}); + +test('projectPolicy never upgrades raw outcome eligibility', async () => { + const raw = { signal_id: 'raw-outcome-1', signal_type: 'policy_outcome_observed', learning_eligible: false, signal: { decisionId: 'decision:abc', outcomes: [] } }; + const repository = { + async listLearningSignals() { + return [{ signal_id: 'signal-1', signal_type: 'candidate_sent', learning_eligible: true, scope_type: 'relationship', scope_id: 'person-1', signal: { decisionRecord: { decisionId: 'decision:abc', candidateStrategyBranch: 'natural_hook', featureBundle: {}, policyVersion: 'vw-p1-v1', policyArtifactId: 'artifact-sha256:abc123' } } }]; + }, + async listPolicyOutcomeSignals() { return [raw]; } + }; + const contract = createLearningDeepTrainingContract({ repository, dataPolicy: { minimize: async () => ({ allowed: true, text: '' }) } }); + await contract.projectPolicy({ scopeType: 'relationship', scopeId: 'person-1', approvedScoresBySignalId: { 'signal-1': { authority: 'Langfuse', approvedByLearning: true, scoreId: 'score-1', name: 'policy_reward', value: 1, traceId: 'trace-1', eligibleSourceSignalId: 'signal-1', decisionId: 'decision:abc', outcomeIds: [], outcomeEvidenceSetRef: 'evidence-set-1', rewardPolicyVersion: 'reward-v1' } } }); + assert.equal(raw.learning_eligible, false); +}); diff --git a/tests/wp0/v21-learning-policy-p1-supply-chain.test.js b/tests/wp0/v21-learning-policy-p1-supply-chain.test.js new file mode 100644 index 000000000..3360b8dad --- /dev/null +++ b/tests/wp0/v21-learning-policy-p1-supply-chain.test.js @@ -0,0 +1,25 @@ +'use strict'; + +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const path = require('node:path'); + +const ROOT = path.resolve(__dirname, '..', '..'); +const read = p => fs.readFileSync(path.join(ROOT, p), 'utf8'); + +test('Vowpal Wabbit supply chain is pinned to 9.11.2 with exact upstream provenance and BSD-3-Clause notice', () => { + const pyproject = read('runtime/learning-growth/python/pyproject.toml'); + const lock = read('runtime/learning-growth/python/uv.lock'); + const upstream = JSON.parse(read('config/upstreams/v21-learning-growth-brain-p0.json')); + const notices = read('THIRD_PARTY_NOTICES.md'); + const license = read('third_party/licenses/vowpal-wabbit-BSD-3-Clause.txt'); + + assert.match(pyproject, /vowpalwabbit\s*==\s*9\.11\.2/i); + assert.match(lock, /name\s*=\s*["']vowpalwabbit["']/i); + assert.match(lock, /version\s*=\s*["']9\.11\.2["']/i); + assert.match(JSON.stringify(upstream), /122bae254a5b8bc2b774d13b33d53e6dbc2cfba7/); + assert.match(notices, /Vowpal Wabbit/i); + assert.match(notices, /BSD-3-Clause/i); + assert.match(license, /Redistribution and use in source and binary forms/i); +}); diff --git a/tests/wp0/v21-learning-policy-p1-vw-runtime.test.js b/tests/wp0/v21-learning-policy-p1-vw-runtime.test.js new file mode 100644 index 000000000..d0cfb13fd --- /dev/null +++ b/tests/wp0/v21-learning-policy-p1-vw-runtime.test.js @@ -0,0 +1,40 @@ +'use strict'; + +const test = require('node:test'); +const assert = require('node:assert/strict'); +const fs = require('node:fs'); +const path = require('node:path'); + +const ROOT = path.resolve(__dirname, '..', '..'); +const read = p => fs.readFileSync(path.join(ROOT, p), 'utf8'); + +const { createLearningPolicyRuntimeAdapter } = require('../../backend/services/learningPolicyRuntimeAdapter'); + +test('Learning runtime adapter delegates the action head to sealed Vowpal Wabbit and keeps P1 deterministic', async () => { + const calls = []; + const adapter = createLearningPolicyRuntimeAdapter({ + invokeVowpalWabbit: async input => { + calls.push(input); + return { action: 'playful_attraction', policyVersion: 'vw-p1-v1', policyArtifactId: 'artifact-sha256:abc123' }; + } + }); + const decision = await adapter.selectLearnedPolicyAction({ + featureBundle: { relationshipStage: 'warming', interactionBand: 'balanced' }, + allowedActions: ['natural_hook', 'playful_attraction', 'direct_advance', 'screen_and_advance', 'leave_aftertaste'] + }); + + assert.equal(calls.length, 1); + assert.equal(decision.candidateStrategyBranch, 'playful_attraction'); + assert.equal(decision.actionProbability, 1); + assert.equal(decision.exploration, false); + assert.equal(decision.providerRoutingAuthority, undefined); + assert.equal(decision.finalReply, undefined); +}); + +test('sealed Learning Python entrypoint exposes the VW policy action mode without provider credentials or local text generation', () => { + const source = read('runtime/learning-growth/python/learning_entrypoint.py'); + assert.match(source, /vowpalwabbit/i); + assert.match(source, /learned[_-]policy/i); + assert.doesNotMatch(source, /(OPENAI_API_KEY|ANTHROPIC_API_KEY|provider[_-]?credential)/i); + assert.doesNotMatch(source, /(generate[_-]?reply|final[_-]?reply)/i); +}); diff --git a/third_party/licenses/vowpal-wabbit-BSD-3-Clause.txt b/third_party/licenses/vowpal-wabbit-BSD-3-Clause.txt new file mode 100644 index 000000000..5b46f6b80 --- /dev/null +++ b/third_party/licenses/vowpal-wabbit-BSD-3-Clause.txt @@ -0,0 +1,46 @@ +Copyright © Microsoft Corp 2012-2014, Yahoo! Inc. 2007-2012, and many +individual contributors. + +All rights reserved. + +Redistribution and use in source and binary forms, with or without + +modification, are permitted provided that the following conditions are met: + + * Redistributions of source code must retain the above copyright + + notice, this list of conditions and the following disclaimer. + + * Redistributions in binary form must reproduce the above copyright + + notice, this list of conditions and the following disclaimer in the + + documentation and/or other materials provided with the distribution. + + * Neither the name of the Microsoft Corp nor the + + names of its contributors may be used to endorse or promote products + + derived from this software without specific prior written permission. + + + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND + +ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED + +WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE + +DISCLAIMED. IN NO EVENT SHALL BE LIABLE FOR ANY + +DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES + +(INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; + +LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND + +ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + +(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS + +SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. diff --git a/tools/uat/v21LearningPolicyClosedLoopEvidence.js b/tools/uat/v21LearningPolicyClosedLoopEvidence.js new file mode 100644 index 000000000..dddeb23ee --- /dev/null +++ b/tools/uat/v21LearningPolicyClosedLoopEvidence.js @@ -0,0 +1,296 @@ +'use strict'; + +const assert = require('node:assert/strict'); +const cp = require('node:child_process'); +const fs = require('node:fs'); +const os = require('node:os'); +const path = require('node:path'); + +const root = fs.mkdtempSync(path.join(os.tmpdir(), 'yance-v21-policy-uat-')); +process.env.YANCE_DATA_DIR = root; +process.env.NODE_ENV = 'test'; + +const { R32SqliteStore } = require('../../backend/lib/r32SqliteStore'); +const { createPlatformCoreRepository } = require('../../backend/repositories/platformCoreRepository'); +const feedback = require('../../backend/services/replyFeedbackLearningService'); +const { createLearningPolicyDecisionContract } = require('../../backend/services/learningPolicyDecisionContract'); +const { createLearningOutcomeAttributionService } = require('../../backend/services/learningOutcomeAttributionService'); +const { createLearningDeepTrainingContract } = require('../../backend/services/learningDeepTrainingContract'); +const { createLearningPolicyRuntimeAdapter } = require('../../backend/services/learningPolicyRuntimeAdapter'); +const { createLearningPromotionAdapter } = require('../../backend/services/learningPromotionAdapter'); + +function runPolicyPython(payload) { + const pythonDir = path.resolve(__dirname, '../../runtime/learning-growth/python'); + const entrypoint = path.join(pythonDir, 'learning_entrypoint.py'); + const result = cp.spawnSync( + 'uv', + ['run', '--frozen', '--offline', 'python', entrypoint], + { cwd: pythonDir, input: JSON.stringify(payload), encoding: 'utf8' } + ); + if (result.status !== 0) { + throw new Error(`sealed policy runtime failed (${result.status}): ${result.stderr}\n${result.stdout}`); + } + return JSON.parse(result.stdout); +} + +async function main() { + const store = new R32SqliteStore({ dbPath: path.join(root, 'policy-uat.db') }); + const repository = createPlatformCoreRepository({ storeProvider: () => store }); + const identityAuthority = { + resolve({ contactId, conversationId }) { + return { + authority: 'PersonContextAuthority', + found: true, + personId: 'person-uat', + contactIds: [contactId], + conversationIds: [conversationId] + }; + } + }; + + const decisionContract = createLearningPolicyDecisionContract({ personContextAuthority: identityAuthority }); + const decision = decisionContract.createDecisionRecord({ + contactId: 'contact-uat', + conversationId: 'conversation-uat', + personaProfileId: 'owner', + featureBundle: { + interactionBand: 'balanced', + performanceMode: 'balanced', + questionPolicy: 'optional', + relationshipStage: 'trust_building', + targetLanguage: 'en' + }, + allowedActionSet: ['natural_hook', 'playful_attraction', 'direct_advance'], + candidateStrategyBranch: 'playful_attraction', + policyVersion: 'vw-p1-baseline-v1', + policyArtifactId: 'baseline', + generation: { + candidatePlanId: 'plan-uat', + directorStrategyId: 'strategy-uat', + contextVersion: 7, + conversationRevision: 3 + } + }); + assert.equal(decision.actionProbability, 1); + assert.equal(decision.exploration, false); + assert.equal(decision.rawPrivateChatPersisted, false); + + const sent = feedback.buildImmutableFeedbackSignal({ + eventType: 'sent', + evidenceId: 'outbox-uat', + outboxId: 'outbox-uat', + candidateId: 'candidate-uat', + contactId: 'contact-uat', + conversationId: 'conversation-uat', + observedAt: '2026-08-14T01:00:00.000Z', + learningMode: 'send_and_learn', + learningEligible: true, + personaTruthReceipt: { pass: true, receiptSha256: 'truth-uat' }, + generationMetadata: { + learningEligible: true, + personaTruthReceipt: { pass: true, receiptSha256: 'truth-uat' }, + decisionRecord: decision, + candidateStrategyBranchId: decision.candidateStrategyBranch + } + }); + assert.equal(sent.learningEligible, true); + assert.equal(sent.signal.decisionRecord.decisionId, decision.decisionId); + repository.insertLearningSignal(sent); + + const outcomeService = createLearningOutcomeAttributionService({ + repository, + personContextAuthority: identityAuthority + }); + const outcomeResult = outcomeService.persistInboundOutcome({ + id: 'inbound-uat-1', + externalMessageId: 'inbound-uat-1', + contactId: 'contact-uat', + conversationId: 'conversation-uat', + direction: 'inbound', + sentAt: '2026-08-14T01:02:00.000Z' + }); + assert.equal(outcomeResult.skipped, false); + assert.equal(outcomeResult.outcomeVector.learningEligible, false); + assert.equal(outcomeResult.outcomeVector.signals.replyLatencyMs.status, 'observed'); + assert.equal(outcomeResult.outcomeVector.signals.nextDayReinitiation.status, 'pending'); + + const rawRows = repository.listLearningSignals({ + scopeType: 'conversation', + scopeId: 'conversation-uat', + learningLevel: 'L1', + learningEligible: false + }); + const rawOutcome = rawRows.find(row => row.signal_type === 'policy_outcome_observed'); + assert.ok(rawOutcome); + assert.equal(Number(rawOutcome.learning_eligible), 0); + + const sourceRows = repository.listLearningSignals({ + scopeType: 'conversation', + scopeId: 'conversation-uat', + learningLevel: 'L1', + learningEligible: true + }); + const source = sourceRows.find(row => row.signal_type === 'candidate_sent'); + assert.ok(source); + const outcomeIds = rawOutcome.signal.outcomes.map(row => row.outcomeId).sort(); + const approvedScore = { + authority: 'Langfuse', + approvedByLearning: true, + scoreId: 'score-uat', + name: 'policy_reward', + value: 0.75, + traceId: 'trace-uat', + sourceSignalId: source.signal_id, + eligibleSourceSignalId: source.signal_id, + decisionId: decision.decisionId, + outcomeIds, + outcomeEvidenceSetRef: 'uat-evidence-set-v1', + rewardPolicyVersion: 'reward-policy-v1' + }; + + const deep = createLearningDeepTrainingContract({ + repository, + dataPolicy: { + async minimize() { return { allowed: true, text: '' }; } + } + }); + const projection = await deep.projectPolicy({ + scopeType: 'conversation', + scopeId: 'conversation-uat', + approvedScoresBySignalId: { [source.signal_id]: approvedScore }, + contentBySignalId: { [source.signal_id]: '' } + }); + assert.equal(projection.readOnly, true); + assert.equal(projection.trajectory.length, 1); + assert.equal(projection.trajectory[0].decision.decisionId, decision.decisionId); + assert.equal(projection.trajectory[0].vwTrainingEligible, true); + + const artifactPath = path.join(root, 'policy.vw'); + const trained = runPolicyPython({ + operation: 'policy_train', + rows: projection.trajectory, + artifactPath + }); + assert.match(trained.policyArtifactVersion, /^[a-f0-9]{64}$/u); + assert.equal(trained.exploration, false); + + const predicted = runPolicyPython({ + operation: 'policy_predict', + featureBundle: decision.featureBundle, + allowedActions: decision.allowedActionSet, + artifactPath, + policyArtifactId: trained.policyArtifactVersion, + policyVersion: 'vw-p1-uat-v1' + }); + assert.ok(decision.allowedActionSet.includes(predicted.action)); + assert.equal(predicted.probability, 1); + assert.equal(predicted.exploration, false); + + const runtimeAdapter = createLearningPolicyRuntimeAdapter({ + async invokeVowpalWabbit(input) { + return runPolicyPython({ + ...input, + artifactPath, + policyArtifactId: trained.policyArtifactVersion + }); + } + }); + const consumed = await runtimeAdapter.selectLearnedPolicyAction({ + featureBundle: decision.featureBundle, + allowedActions: decision.allowedActionSet, + baselineAction: 'natural_hook' + }); + assert.ok(decision.allowedActionSet.includes(consumed.candidateStrategyBranch)); + assert.equal(consumed.actionProbability, 1); + assert.equal(consumed.exploration, false); + + const degraded = await createLearningPolicyRuntimeAdapter({ + async invokeVowpalWabbit() { + const error = new Error('corrupt artifact'); + error.code = 'LEARNING_POLICY_ARTIFACT_IDENTITY_MISMATCH'; + throw error; + } + }).selectLearnedPolicyAction({ + featureBundle: decision.featureBundle, + allowedActions: decision.allowedActionSet, + baselineAction: 'natural_hook' + }); + assert.equal(degraded.executedPolicy, 'baseline'); + assert.equal(degraded.degradation.reasonCode, 'LEARNING_POLICY_ARTIFACT_IDENTITY_MISMATCH'); + + const promotion = createLearningPromotionAdapter({ + openFeature: { + setEvaluationContext() {} + }, + flagd: { + mode: 'in-process-offline' + } + }); + + const proposal = { + status: 'READY_FOR_REVIEW', + Candidate: { + id: `policy:${trained.policyArtifactVersion}`, + version: trained.policyArtifactVersion, + exposure: 0 + }, + Regression: { passed: true }, + Shadow: { passed: true } + }; + + const active = await promotion.promote(proposal, { + approved: true, + evidence: { id: 'uat-promotion-evidence' } + }); + + assert.equal(active.kind, 'LEARNING_ROLLOUT'); + assert.equal(active.automaticPromotion, false); + assert.equal(active.flagd, 'in-process-offline'); + + const rolledBack = await promotion.rollback(active, { + approved: true, + evidence: { id: 'uat-rollback-evidence' } + }); + + assert.equal(rolledBack.kind, 'LEARNING_ROLLBACK'); + assert.equal(rolledBack.automaticPromotion, false); + assert.equal(rolledBack.evidenceId, 'uat-rollback-evidence'); + + const brainSource = fs.readFileSync( + path.resolve(__dirname, '../../backend/services/contextAwareReplyBrain.js'), + 'utf8' + ); + const selectionIndex = brainSource.indexOf('selectLearnedPolicyAction('); + const finalFrontierIndex = brainSource.indexOf('let modelResult = await aiGateway.execute', selectionIndex); + assert.ok(selectionIndex >= 0 && finalFrontierIndex > selectionIndex); + + const receipt = { + workPackage: 'V21-LEARNING-POLICY-P1-DECISION-OUTCOME-CLOSED-LOOP-V2-SUCCESSOR', + canonicalIdentityBound: true, + storedHistoricalFeatureBundle: true, + behaviorPropensityLoggedAtDecision: true, + eligibleSourceSignalAnchored: true, + rawOutcomeEligibilityImmutableFalse: true, + outcomeWindowsAndMissingnessExplicit: true, + scoreEvidenceBindingVerified: true, + decisionOutcomeBound: true, + rawOutcomeIsNotReward: true, + providerPrivacyBoundaryProved: true, + learner: 'VowpalWabbit 9.11.2', + frontierGenerationAuthority: 'Model Brain / LiteLLM', + policyConsumedBeforeGeneration: true, + promotionAuthority: 'Learning', + availabilityFallbackProved: true, + rollbackProved: true, + localReplyModelUsed: false, + policyArtifactVersion: trained.policyArtifactVersion, + selectedAction: consumed.candidateStrategyBranch + }; + process.stdout.write(`${JSON.stringify(receipt, null, 2)}\n`); + store.close?.(); +} + +main().catch(error => { + console.error(error); + process.exitCode = 1; +}); \ No newline at end of file