Add human-led conversation learning loop

src/services/chat/service.ts

@@ -23,6 +23,7 @@ export type { InvitationCooldown } from './types.js';
 import { fetchWithTimeout } from '../../utils/fetch-with-timeout.js';
 import type { AccountAuthGate } from '../account-auth-context.js';
 import { containsMediaRequest } from '../conversation-boundaries.js';
+import type { ConversationLearningObserver } from '../conversation-learning-observer.js';
 
 const ACTIVE_WINDOW_MS = 96 * 60 * 60 * 1000;
 const ROOM_URL_PREFIX = 'https://chat.reddit.com/room/';
@@ -60,6 +61,11 @@ export type ConversationStartResult =
   | { status: 'terminal'; reason: string }
   | { status: 'retryable'; reason: string; retryAt: number };
 
+export interface ChatSendReceipt {
+  eventId: string;
+  sentAt: number;
+}
+
 interface BuildChatOpts {
   knownUsernames: Map<string, string>;
   includeTimelineMessages?: boolean;
@@ -170,6 +176,7 @@ export class ChatService {
     private accountAuth?: AccountAuthGate,
     private outcomesRepo?: ConversationOutcomesRepository,
     private conversationStateRepo?: ConversationStateRepository,
+    private conversationLearningObserver?: ConversationLearningObserver,
   ) {
     this.cache = cache ?? new ChatCache(logger);
     if (this.redditUsersRepo) {
@@ -608,38 +615,48 @@ export class ChatService {
    * a raw Matrix room ID like `!abc:reddit.com`.
    */
   async sendChatMessage(chatUrlOrRoomId: string, message: string, txnId?: string): Promise<boolean> {
-    if (!this.accountAuth?.isFullyAuthenticated()) return false;
+    return (await this.sendChatMessageDetailed(chatUrlOrRoomId, message, txnId)) !== null;
+  }
+
+  /** Send a reply and retain the authoritative Matrix event identity. */
+  async sendChatMessageDetailed(
+    chatUrlOrRoomId: string,
+    message: string,
+    txnId?: string,
+  ): Promise<ChatSendReceipt | null> {
+    if (!this.accountAuth?.isFullyAuthenticated()) return null;
     const active = this.getInvitationCooldown();
     if (active && this.blocksExistingRoomSends(active)) {
       const secondsLeft = Math.ceil((active.until - Date.now()) / 1000);
       this.logger.warn(
         `[chat] skipping sendChatMessage — send cooldown active for ${secondsLeft}s (${active.redditCode ?? 'rate-limited'})`,
       );
-      return false;
+      return null;
     }
     return this.enqueue(async () => {
       try {
         const roomId = ChatService.extractRoomId(chatUrlOrRoomId);
         if (!roomId) {
           this.logger.error(`[chat] cannot extract room ID from "${chatUrlOrRoomId}"`);
-          return false;
+          return null;
         }
         const eventId = await this.withAuth(() => this.matrix.sendTextMessage(roomId, message, txnId));
         this.logger.info(`[chat] sent message to ${roomId} (event_id=${eventId})`);
         const chat = this.cache.get().find((c) => c.roomId === roomId);
+        const sentAt = Date.now();
         if (chat) {
           chat.lastMessageSnippet = message.slice(0, 200);
-          chat.lastMessageTime = Date.now();
+          chat.lastMessageTime = sentAt;
           chat.isFromUs = true;
         }
-        return true;
+        return { eventId, sentAt };
       } catch (error) {
         if (error instanceof MatrixRateLimitError) {
           this.setInvitationCooldown(error, 'all-sends');
-          return false;
+          return null;
         }
         this.logger.error('[chat] sendChatMessage failed:', error);
-        return false;
+        return null;
       }
     });
   }
@@ -1252,11 +1269,17 @@ export class ChatService {
         this.logger.warn('[chat] failed to record conversation outcome:', err);
         return false;
       }
+      this.conversationLearningObserver?.observe(chat, performanceMessage, conversationState);
       return wasMissingOutcome;
     } else if (hasReply && !performanceMessage.gotResponse) {
       this.messagesRepo.markGotResponse(performanceMessage.id);
       this.logger.warn('[chat] response recorded without canonical outcome repository; excluded from learning');
     }
+    this.conversationLearningObserver?.observe(
+      chat,
+      performanceMessage,
+      this.conversationStateRepo?.get(chat.roomId),
+    );
     return false;
   }
 

src/services/conversation-attribution.ts

@@ -0,0 +1,138 @@
+import { createHash } from 'node:crypto';
+import type {
+  ConversationEventRepository,
+  AttributionConfidence,
+  ConversationEventKind,
+} from '../database/conversation-event-repository.js';
+import type { ConversationAutomationClaimRecord, SentMessageRecord } from '../database/repositories.js';
+import type { MediaLibrarySend } from './media-library-types.js';
+import type { ChatMessage } from './chat/types.js';
+import { conversationMessageFingerprint } from './conversation-state.js';
+import type { AttributedConversationMessage } from './conversation-learning-types.js';
+
+export interface ConversationAttributionInput {
+  conversationId: string;
+  messages: readonly ChatMessage[];
+  opener: SentMessageRecord | undefined;
+  claims: readonly ConversationAutomationClaimRecord[];
+  mediaSends: readonly MediaLibrarySend[];
+}
+
+/**
+ * Deterministically attributes a Matrix timeline. Event IDs win over all
+ * fingerprint fallbacks; remaining self-authored events are human-owned with
+ * inferred confidence, matching the automation safety boundary.
+ */
+export function attributeConversationMessages(
+  input: ConversationAttributionInput,
+  repository?: ConversationEventRepository,
+  observedAt = Date.now(),
+): AttributedConversationMessage[] {
+  const claimsByEventId = new Map(input.claims.flatMap((claim) =>
+    claim.matrixEventId ? [[claim.matrixEventId, claim] as const] : []));
+  const claimsByFingerprint = new Map(input.claims.flatMap((claim) =>
+    claim.replyText ? [[conversationMessageFingerprint(claim.replyText), claim] as const] : []));
+  const mediaByEventId = new Map<string, { send: MediaLibrarySend; kind: ConversationEventKind }>();
+  for (const send of input.mediaSends) {
+    if (send.matrixMediaEventId) {
+      mediaByEventId.set(send.matrixMediaEventId, {
+        send,
+        kind: send.initiatedBy === 'human' ? 'manual-library-media' : 'automated-library-media',
+      });
+    }
+    if (send.matrixTextEventId && !send.matrixTextEventId.startsWith('manual-') && send.matrixTextEventId !== 'unconfirmed') {
+      mediaByEventId.set(send.matrixTextEventId, { send, kind: 'library-reciprocal' });
+    }
+  }
+
+  return input.messages.map((message, index) => {
+    const eventId = message.eventId ?? syntheticEventId(message, index);
+    const existing = repository?.get(input.conversationId, eventId);
+    let actor: AttributedConversationMessage['actor'];
+    let kind: ConversationEventKind;
+    let confidence: AttributionConfidence;
+    let sourceReference: string | null = null;
+
+    if (!message.isSelf) {
+      actor = 'recipient';
+      kind = (message.media?.length ?? 0) > 0 ? 'inbound-media' : 'inbound-text';
+      confidence = message.eventId ? 'exact' : 'inferred';
+    } else if (existing?.actor === 'human' && existing.attributionConfidence === 'exact') {
+      actor = 'human';
+      kind = existing.eventKind;
+      confidence = 'exact';
+      sourceReference = existing.sourceReference;
+    } else if (message.eventId && input.opener?.matrixEventId === message.eventId) {
+      actor = 'ai';
+      kind = 'opener';
+      confidence = 'exact';
+      sourceReference = input.opener ? `sent-message:${input.opener.id}` : null;
+    } else if (message.eventId && claimsByEventId.has(message.eventId)) {
+      const claim = claimsByEventId.get(message.eventId)!;
+      actor = 'ai';
+      kind = 'automation-reply';
+      confidence = 'exact';
+      sourceReference = `claim:${claim.incomingKey}`;
+    } else if (message.eventId && mediaByEventId.has(message.eventId)) {
+      const matched = mediaByEventId.get(message.eventId)!;
+      actor = matched.send.initiatedBy === 'human' ? 'human' : 'ai';
+      kind = matched.kind;
+      confidence = 'exact';
+      sourceReference = `media-library-send:${matched.send.id}`;
+    } else if (isOpenerFingerprint(message, input.opener)) {
+      actor = 'ai';
+      kind = 'opener';
+      confidence = 'fingerprint';
+      sourceReference = input.opener ? `sent-message:${input.opener.id}` : null;
+    } else {
+      const claim = claimsByFingerprint.get(conversationMessageFingerprint(message.text));
+      if (claim) {
+        actor = 'ai';
+        kind = 'automation-reply';
+        confidence = 'fingerprint';
+        sourceReference = `claim:${claim.incomingKey}`;
+      } else {
+        actor = 'human';
+        kind = 'manual-text';
+        confidence = message.eventId ? 'inferred' : 'inferred';
+      }
+    }
+
+    repository?.upsert({
+      conversationId: input.conversationId,
+      eventId,
+      eventTimestamp: message.timestamp,
+      direction: message.isSelf ? 'outbound' : 'inbound',
+      actor,
+      eventKind: kind,
+      text: message.text,
+      sourceReference,
+      attributionConfidence: confidence,
+      observedAt,
+    });
+    return {
+      eventId,
+      timestamp: message.timestamp,
+      isSelf: message.isSelf,
+      text: message.text,
+      mediaCount: message.media?.length ?? 0,
+      actor,
+      kind,
+      confidence,
+    };
+  });
+}
+
+function isOpenerFingerprint(message: ChatMessage, opener: SentMessageRecord | undefined): boolean {
+  if (!opener || !message.isSelf) return false;
+  if (Math.abs(message.timestamp - opener.sentAt) > 60_000) return false;
+  return conversationMessageFingerprint(message.text) === conversationMessageFingerprint(opener.message);
+}
+
+function syntheticEventId(message: ChatMessage, index: number): string {
+  const digest = createHash('sha256')
+    .update(`${message.timestamp}\u0000${message.isSelf ? 'self' : 'recipient'}\u0000${message.text}\u0000${index}`)
+    .digest('hex')
+    .slice(0, 20);
+  return `legacy:${digest}`;
+}

src/services/conversation-automation.ts

@@ -17,6 +17,7 @@ import {
   mergeConversationState,
   r4rReplySafetyFailure,
   remainingAutomationTurns,
+  type ConversationIntent,
   type ConversationMessage,
   type ConversationState,
 } from './conversation-state.js';
@@ -24,6 +25,8 @@ import { containsMediaRequest } from './conversation-boundaries.js';
 import { findOffPlatformSignal, offPlatformSignalReason } from './off-platform-signals.js';
 import type { MediaLibraryExchangeService } from './media-library-exchange.js';
 import type { ConversationReplyScreening } from '../ai/openrouter.js';
+import type { LearningConfig } from '../config/index.js';
+import type { ConversationLearningProvider, LearningStage, RetrievedLearningExample } from './conversation-learning-types.js';
 
 export type ConversationAutomationResultKind =
   | 'sent'
@@ -49,7 +52,7 @@ export interface ConversationAutomationClaims {
   listPending?(): ConversationAutomationClaimRecord[];
   markAttempt?(conversationId: string, incomingKey: string): boolean;
   releasePending?(conversationId: string, incomingKey: string): boolean;
-  markSent(conversationId: string, incomingKey: string): boolean;
+  markSent(conversationId: string, incomingKey: string, matrixEventId?: string | null): boolean;
   markSuperseded?(conversationId: string, incomingKey: string): boolean;
 }
 
@@ -87,6 +90,7 @@ export interface ConversationAutomationOptions {
   triage?: ConversationTriageWriter;
   screener?: ConversationReplyScreener;
   screening?: ConversationScreeningOptions;
+  learning?: { provider: ConversationLearningProvider; config: LearningConfig; accountId: string };
 }
 
 export interface ReplyPriorityStatus {
@@ -267,9 +271,16 @@ export class ConversationAutomationService {
     }
     const existingClaim = this.claims.getClaim?.(conversationId, incomingKey);
     if (existingClaim?.status === 'pending') {
-      const observedOutbound = messages.some((message) => message.isSelf && message.timestamp > latest.timestamp);
+      const expectedReplyFingerprint = existingClaim.replyText
+        ? conversationMessageFingerprint(existingClaim.replyText)
+        : null;
+      const observedOutbound = messages.find((message) =>
+        message.isSelf
+        && message.timestamp > latest.timestamp
+        && expectedReplyFingerprint !== null
+        && conversationMessageFingerprint(message.text) === expectedReplyFingerprint);
       if (observedOutbound) {
-        this.claims.markSent(conversationId, incomingKey);
+        this.claims.markSent(conversationId, incomingKey, observedOutbound.eventId ?? null);
         this.logger.info(`[automation] Reconciled interrupted reply for u/${conversation.username} from Matrix history`);
         return null;
       }
@@ -437,11 +448,20 @@ export class ConversationAutomationService {
     if (isLegacy && intent.kind !== 'media-ask') {
       return this.finish(conversation, conversationId, state, incomingKey, 'human-required', intent.reason);
     }
+    const learningExamples = this.learningExamples({
+      accountId: this.options.learning?.accountId ?? '',
+      conversationId,
+      subreddit: opener.subreddit ?? null,
+      policyMode: state.automation.policyMode,
+      stage: stageForIntent(intent.kind),
+      purpose: requestsMedia ? 'media-ask' : 'reply',
+      incomingEventId: incomingKey,
+    });
     const generatePlannedReply = () => isLegacy
-      ? this.ai.generateMediaAsk(mapped, conversation.username, state, intent)
+      ? this.ai.generateMediaAsk(mapped, conversation.username, state, intent, learningExamples)
       : isHybridMediaAsk
-        ? this.ai.generateR4RMediaAsk(mapped, conversation.username, state)
-        : this.ai.generateReply(mapped, conversation.username, state, intent);
+        ? this.ai.generateR4RMediaAsk(mapped, conversation.username, state, learningExamples)
+        : this.ai.generateReply(mapped, conversation.username, state, intent, learningExamples);
     let reply = await generatePlannedReply();
     let safetyFailure = isLegacy || isHybridMediaAsk
       ? hybridMediaAskSafetyFailure(reply)
@@ -486,6 +506,20 @@ export class ConversationAutomationService {
     );
   }
 
+  private learningExamples(query: Parameters<ConversationLearningProvider['getExamples']>[0]): readonly RetrievedLearningExample[] {
+    const learning = this.options.learning;
+    if (!learning) return [];
+    const examples = learning.provider.getExamples(query);
+    const injected = learning.config.generationEnabled && !learning.config.shadowMode;
+    learning.provider.recordGeneration({
+      query,
+      mode: injected ? 'injected' : 'shadow',
+      policyVersion: learning.config.policyVersion,
+      exampleIds: examples.map((example) => example.decisionId),
+    });
+    return injected ? examples : [];
+  }
+
   private actionableConversations(
     conversations: ActiveChat[],
     observedAfter: number | null = this.lastScanAt,
@@ -512,8 +546,13 @@ export class ConversationAutomationService {
         ? conversationMessageFingerprint(claim.replyText)
         : null;
       const observedClaimedReply = expectedFingerprint !== null
-        && laterSelfMessages.some((message) => conversationMessageFingerprint(message.text) === expectedFingerprint);
-      if (observedClaimedReply && this.claims.markSent(claim.conversationId, claim.incomingKey)) {
+        ? laterSelfMessages.find((message) => conversationMessageFingerprint(message.text) === expectedFingerprint)
+        : undefined;
+      if (observedClaimedReply && this.claims.markSent(
+        claim.conversationId,
+        claim.incomingKey,
+        observedClaimedReply.eventId ?? null,
+      )) {
         this.logger.info(`[automation] Reconciled interrupted reply for u/${conversation.username} from Matrix history`);
       } else if (!observedClaimedReply && this.claims.markSuperseded?.(claim.conversationId, claim.incomingKey)) {
         this.logger.info(`[automation] Superseded queued reply for u/${conversation.username}; an operator-authored message was observed`);
@@ -640,8 +679,14 @@ export class ConversationAutomationService {
           isHybridMediaAsk,
         );
         this.claims.markAttempt?.(conversationId, incomingKey);
-        const sent = await this.chat.sendChatMessage(latestConversation.url!, reply);
-        return { kind: sent ? 'sent' as const : 'failed' as const, exhausted, state: freshState };
+        const receipt = typeof this.chat.sendChatMessageDetailed === 'function'
+          ? await this.chat.sendChatMessageDetailed(latestConversation.url!, reply)
+          : await this.chat.sendChatMessage(latestConversation.url!, reply)
+            ? { eventId: null, sentAt: this.now() }
+            : null;
+        return receipt
+          ? { kind: 'sent' as const, exhausted, state: freshState, eventId: receipt.eventId }
+          : { kind: 'failed' as const, exhausted, state: freshState };
       },
       (result) => result.kind === 'sent',
       signal,
@@ -650,7 +695,8 @@ export class ConversationAutomationService {
     if (attempt.kind === 'superseded') return attempt.result;
     if (attempt.kind === 'failed') return null;
     const sentAt = this.now();
-    this.claims.markSent(conversation.roomId!, incomingKey);
+    if (attempt.eventId) this.claims.markSent(conversation.roomId!, incomingKey, attempt.eventId);
+    else this.claims.markSent(conversation.roomId!, incomingKey);
     this.lastReplySentAt = sentAt;
     this.lastReplyLatencyMs = Math.max(0, sentAt - incomingAt);
     this.logger.info(
@@ -859,6 +905,13 @@ function replyPlan(state: ConversationState): { isLegacy: boolean; isHybridMedia
   return { isLegacy, isHybridMediaAsk };
 }
 
+function stageForIntent(intent: ConversationIntent['kind']): LearningStage {
+  if (intent === 'media-ask') return 'media-invitation';
+  if (intent === 'answer-question') return 'answer';
+  if (intent === 'respect-boundary') return 'handoff';
+  return 'rapport';
+}
+
 /**
  * Handoffs that need a person rather than a kill, and that apply to every
  * policy mode. Messaging-app names are deliberately absent: those are handled

src/services/conversation-learning-evaluator.ts

@@ -0,0 +1,50 @@
+import { createHash } from 'node:crypto';
+import type { ConversationLearningRepository, LearningEvaluationRun } from '../database/conversation-learning-repository.js';
+
+const DIRECT_IDENTIFIER = /https?:\/\/|\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b|(?:^|\s)(?:u\/|@)[A-Za-z0-9_-]{2,}/i;
+
+/** Offline hard gates for a versioned snapshot of eligible learning examples. */
+export function runConversationLearningEvaluation(
+  repository: ConversationLearningRepository,
+  policyVersion: string,
+  baselinePolicyVersion: string | null = null,
+  now = Date.now(),
+): LearningEvaluationRun {
+  const trajectories = repository.listTrajectories({ status: 'eligible', limit: 500 });
+  const details = trajectories.flatMap((trajectory) => {
+    const detail = repository.getTrajectory(trajectory.id);
+    return detail ? [detail] : [];
+  });
+  const decisions = details.flatMap((detail) => detail.decisions);
+  const identifierLeaks = decisions.filter((decision) =>
+    DIRECT_IDENTIFIER.test(decision.chosenResponse)
+      || decision.context.some((turn) => DIRECT_IDENTIFIER.test(turn.text)));
+  const invalidStatuses = decisions.filter((decision) => decision.qualityStatus !== 'eligible');
+  const emptyDecisions = decisions.filter((decision) => decision.chosenResponse.trim().length === 0);
+  const hardGateFailures = identifierLeaks.length + invalidStatuses.length + emptyDecisions.length;
+  const datasetVersion = createHash('sha256')
+    .update(details.map((detail) => `${detail.id}:${detail.updatedAt}`).sort().join('\n'))
+    .digest('hex').slice(0, 16);
+  const run: LearningEvaluationRun = {
+    id: createHash('sha256').update(`${policyVersion}\0${datasetVersion}\0${now}`).digest('hex').slice(0, 24),
+    policyVersion,
+    datasetVersion,
+    baselinePolicyVersion,
+    status: hardGateFailures === 0 ? 'passed' : 'failed',
+    totalCases: decisions.length,
+    hardGateFailures,
+    summary: {
+      trajectories: details.length,
+      decisions: decisions.length,
+      identifierLeaks: identifierLeaks.length,
+      invalidStatuses: invalidStatuses.length,
+      emptyDecisions: emptyDecisions.length,
+      humanLed: details.filter((item) => item.creditClass.startsWith('human-')).length,
+      autonomous: details.filter((item) => item.creditClass === 'autonomous-success').length,
+    },
+    createdAt: now,
+    completedAt: now,
+  };
+  repository.saveEvaluationRun(run);
+  return run;
+}

src/services/conversation-learning-extractor.ts

@@ -0,0 +1,134 @@
+import type { LearningConfig } from '../config/index.js';
+import type { AttributionConfidence } from '../database/conversation-event-repository.js';
+import {
+  stableLearningId,
+  type ConversationLearningRepository,
+  type UpsertLearningDecision,
+} from '../database/conversation-learning-repository.js';
+import type { SentMessageRecord } from '../database/repositories.js';
+import { containsMediaRequest } from './conversation-boundaries.js';
+import type { ConversationState } from './conversation-state.js';
+import type {
+  AttributedConversationMessage,
+  LearningCreditClass,
+  LearningStage,
+  LearningTrajectoryStatus,
+  LearningTurn,
+} from './conversation-learning-types.js';
+import { redactLearningText } from './conversation-learning-redactor.js';
+
+const CONFIDENCE_RANK: Record<AttributionConfidence, number> = { inferred: 1, fingerprint: 2, exact: 3 };
+
+export interface LearningExtractionInput {
+  conversationId: string;
+  messages: readonly AttributedConversationMessage[];
+  opener: SentMessageRecord;
+  state: ConversationState | undefined;
+  canonicalOutcome: 'engaged' | 'declined' | 'boundary' | 'upsell' | 'unknown' | undefined;
+  now?: number;
+}
+
+/** Derives revocable decision examples from an authoritative attributed timeline. */
+export function extractConversationLearning(
+  input: LearningExtractionInput,
+  repository: ConversationLearningRepository,
+  config: LearningConfig,
+): number {
+  if (!config.captureEnabled) return 0;
+  const now = input.now ?? Date.now();
+  const negative = input.canonicalOutcome && ['declined', 'boundary', 'upsell'].includes(input.canonicalOutcome);
+  const inboundImages = input.messages.filter((message) =>
+    message.actor === 'recipient' && message.kind === 'inbound-media' && message.mediaCount > 0);
+  let extracted = 0;
+
+  for (const image of inboundImages) {
+    const timeline = input.messages.filter((message) => message.timestamp <= image.timestamp);
+    const humans = timeline.filter((message) => message.actor === 'human');
+    const asks = timeline.filter((message) => message.isSelf && containsMediaRequest(message.text));
+    const directHumanAsk = [...humans].reverse().find((message) => containsMediaRequest(message.text));
+    const priorAiAsk = asks.find((message) => message.actor === 'ai');
+    const confidence = minimumConfidence(timeline);
+    const creditClass = classifyCredit(humans.length > 0, directHumanAsk !== undefined, priorAiAsk !== undefined, asks.length > 0);
+    const maturityAt = image.timestamp + config.maturityHours * 60 * 60 * 1_000;
+    const status: LearningTrajectoryStatus = negative
+      ? 'invalidated'
+      : CONFIDENCE_RANK[confidence] < CONFIDENCE_RANK[config.minimumAttributionConfidence]
+        ? 'needs-review'
+        : now < maturityAt ? 'provisional' : 'eligible';
+    const trajectoryId = stableLearningId(input.conversationId, image.eventId);
+    repository.upsertTrajectory({
+      id: trajectoryId,
+      conversationId: input.conversationId,
+      sentMessageId: input.opener.id,
+      subreddit: input.opener.subreddit ?? null,
+      sourceType: input.opener.sourceType ?? null,
+      policyMode: input.state?.automation.policyMode ?? null,
+      firstHumanEventId: humans[0]?.eventId ?? null,
+      outcomeEventId: image.eventId,
+      outcomeType: 'image-received',
+      creditClass,
+      attributionConfidence: confidence,
+      status,
+      maturityAt,
+      invalidatedAt: negative ? now : null,
+      invalidationReason: negative ? `Canonical outcome changed to ${input.canonicalOutcome}` : null,
+      now,
+    });
+
+    const decisionMessages = humans.filter((message) => message.kind === 'manual-text');
+    const decisions = decisionMessages.map((decision): UpsertLearningDecision => ({
+      id: stableLearningId(trajectoryId, decision.eventId),
+      trajectoryId,
+      decisionEventId: decision.eventId,
+      stage: inferStage(decision.text),
+      context: buildContext(timeline, decision, config.maxTurnsPerExample),
+      chosenResponse: redactLearningText(decision.text),
+      turnsUntilOutcome: timeline.filter((message) => message.timestamp > decision.timestamp).length,
+      isDirectMediaInvitation: containsMediaRequest(decision.text),
+      qualityStatus: status === 'provisional' ? 'eligible' : status,
+      now,
+    }));
+    repository.replaceDecisions(trajectoryId, decisions);
+    extracted += 1;
+  }
+  return extracted;
+}
+
+function classifyCredit(hasHuman: boolean, directHumanAsk: boolean, priorAiAsk: boolean, anyAsk: boolean): LearningCreditClass {
+  if (!hasHuman) return anyAsk ? 'autonomous-success' : 'recipient-initiated-media';
+  if (directHumanAsk) return priorAiAsk ? 'human-improved-success' : 'human-created-success';
+  return 'human-completed-ai-trajectory';
+}
+
+function minimumConfidence(messages: readonly AttributedConversationMessage[]): AttributionConfidence {
+  return messages.reduce<AttributionConfidence>((lowest, message) =>
+    CONFIDENCE_RANK[message.confidence] < CONFIDENCE_RANK[lowest] ? message.confidence : lowest, 'exact');
+}
+
+function inferStage(text: string): LearningStage {
+  if (containsMediaRequest(text)) return 'media-invitation';
+  if (/\b(sorry|understand|no pressure|comfortable|boundary)\b/i.test(text)) return 'handoff';
+  if (/\?/.test(text)) return 'answer';
+  return 'rapport';
+}
+
+function buildContext(
+  timeline: readonly AttributedConversationMessage[],
+  decision: AttributedConversationMessage,
+  maxTurns: number,
+): LearningTurn[] {
+  return timeline
+    .filter((message) => message.timestamp < decision.timestamp && message.kind !== 'inbound-media')
+    .slice(-maxTurns)
+    .flatMap((message): LearningTurn[] => {
+      if (!['inbound-text', 'opener', 'automation-reply', 'manual-text', 'library-reciprocal'].includes(message.kind)) return [];
+      return [{
+        eventId: message.eventId,
+        timestamp: message.timestamp,
+        role: message.actor === 'recipient' ? 'recipient' : 'operator',
+        actor: message.actor === 'unknown' ? 'ai' : message.actor,
+        kind: message.kind as LearningTurn['kind'],
+        text: redactLearningText(message.text),
+      }];
+    });
+}

src/services/conversation-learning-observer.ts

@@ -0,0 +1,53 @@
+import type { LearningConfig } from '../config/index.js';
+import type { ConversationEventRepository } from '../database/conversation-event-repository.js';
+import type { ConversationLearningRepository } from '../database/conversation-learning-repository.js';
+import type {
+  ConversationAutomationClaimRepository,
+  ConversationOutcomesRepository,
+  SentMessageRecord,
+} from '../database/repositories.js';
+import type { MediaLibraryRepository } from '../database/media-library-repository.js';
+import type { Logger } from '../utils/logger.js';
+import type { ActiveChat } from './chat/types.js';
+import { attributeConversationMessages } from './conversation-attribution.js';
+import type { ConversationState } from './conversation-state.js';
+import { extractConversationLearning } from './conversation-learning-extractor.js';
+
+export interface ConversationLearningObserver {
+  observe(chat: ActiveChat, opener: SentMessageRecord, state: ConversationState | undefined): void;
+}
+
+export class DurableConversationLearningObserver implements ConversationLearningObserver {
+  constructor(
+    private readonly accountId: string,
+    private readonly events: ConversationEventRepository,
+    private readonly learning: ConversationLearningRepository,
+    private readonly claims: ConversationAutomationClaimRepository,
+    private readonly media: MediaLibraryRepository,
+    private readonly outcomes: ConversationOutcomesRepository,
+    private readonly config: LearningConfig,
+    private readonly logger: Logger,
+  ) {}
+
+  observe(chat: ActiveChat, opener: SentMessageRecord, state: ConversationState | undefined): void {
+    if (!chat.roomId || !chat.messages) return;
+    try {
+      const attributed = attributeConversationMessages({
+        conversationId: chat.roomId,
+        messages: chat.messages,
+        opener,
+        claims: this.claims.listConversation(chat.roomId),
+        mediaSends: this.media.listConversationSends(this.accountId, chat.roomId),
+      }, this.events);
+      extractConversationLearning({
+        conversationId: chat.roomId,
+        messages: attributed,
+        opener,
+        state,
+        canonicalOutcome: this.outcomes.get(opener.id)?.outcome,
+      }, this.learning, this.config);
+    } catch (error) {
+      this.logger.warn('[learning] failed to attribute or extract conversation:', error);
+    }
+  }
+}

src/services/conversation-learning-provider.ts

@@ -0,0 +1,38 @@
+import type { LearningConfig } from '../config/index.js';
+import type { ConversationLearningRepository } from '../database/conversation-learning-repository.js';
+import type { ConversationLearningProvider, ConversationLearningQuery } from './conversation-learning-types.js';
+
+/** Account-scoped retrieval with explicit human/autonomous quotas and an audit ledger. */
+export class RepositoryConversationLearningProvider implements ConversationLearningProvider {
+  constructor(
+    private readonly repository: ConversationLearningRepository,
+    private readonly config: LearningConfig,
+  ) {}
+
+  getExamples(query: ConversationLearningQuery) {
+    this.repository.promoteMature();
+    return this.repository.getEligibleExamples({
+      conversationId: query.conversationId,
+      policyMode: query.policyMode,
+      stage: query.stage,
+      subreddit: query.subreddit,
+      minimumConfidence: this.config.minimumAttributionConfidence,
+      humanLimit: this.config.humanExampleSlots,
+      autonomousLimit: this.config.autonomousExampleSlots,
+      totalLimit: this.config.maxContextExamples,
+    });
+  }
+
+  recordGeneration(input: Parameters<ConversationLearningProvider['recordGeneration']>[0]): void {
+    this.repository.recordGeneration({
+      conversationId: input.query.conversationId,
+      incomingEventId: input.query.incomingEventId,
+      purpose: input.query.purpose,
+      mode: input.mode,
+      policyVersion: input.policyVersion,
+      exampleIds: input.exampleIds,
+      promptSha256: input.promptSha256,
+      generatedAt: input.generatedAt,
+    });
+  }
+}

src/services/conversation-learning-redactor.ts

@@ -0,0 +1,16 @@
+const URL = /https?:\/\/\S+/gi;
+const EMAIL = /\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b/gi;
+const PHONE = /(?<!\w)(?:\+?\d[\d ()-]{7,}\d)(?!\w)/g;
+const USERNAME = /(?:^|\s)(?:u\/|@)[A-Za-z0-9_-]{2,}/g;
+
+/** Removes direct identifiers before durable example storage or prompt use. */
+export function redactLearningText(value: string, maxLength = 500): string {
+  return value
+    .replace(URL, '[url]')
+    .replace(EMAIL, '[email]')
+    .replace(PHONE, '[phone]')
+    .replace(USERNAME, (match) => `${match[0] === ' ' ? ' ' : ''}[username]`)
+    .replace(/\s+/g, ' ')
+    .trim()
+    .slice(0, maxLength);
+}

src/services/conversation-learning-types.ts

@@ -0,0 +1,129 @@
+import { z } from 'zod';
+import type { AttributionConfidence, ConversationEventKind } from '../database/conversation-event-repository.js';
+import type { ConversationPolicyMode } from './conversation-state.js';
+
+export const LearningStageSchema = z.enum([
+  'rapport',
+  'answer',
+  'flirt',
+  'disclosure',
+  'media-invitation',
+  'handoff',
+  'unknown',
+]);
+export type LearningStage = z.infer<typeof LearningStageSchema>;
+
+export const LearningTurnSchema = z.strictObject({
+  eventId: z.string(),
+  timestamp: z.number(),
+  role: z.enum(['recipient', 'operator']),
+  actor: z.enum(['recipient', 'ai', 'human']),
+  kind: z.enum([
+    'inbound-text',
+    'opener',
+    'automation-reply',
+    'manual-text',
+    'library-reciprocal',
+  ]),
+  text: z.string(),
+});
+export type LearningTurn = z.infer<typeof LearningTurnSchema>;
+
+export const LearningCreditClasses = [
+  'human-created-success',
+  'human-improved-success',
+  'human-completed-ai-trajectory',
+  'autonomous-success',
+  'recipient-initiated-media',
+  'uncertain',
+] as const;
+export type LearningCreditClass = (typeof LearningCreditClasses)[number];
+
+export type LearningTrajectoryStatus = 'provisional' | 'eligible' | 'excluded' | 'invalidated' | 'needs-review';
+export type LearningDecisionStatus = 'eligible' | 'excluded' | 'invalidated' | 'needs-review';
+
+export interface LearningTrajectoryRecord {
+  id: string;
+  accountId: string;
+  conversationId: string;
+  sentMessageId: number | null;
+  subreddit: string | null;
+  sourceType: 'image' | 'chat' | null;
+  policyMode: ConversationPolicyMode | null;
+  firstHumanEventId: string | null;
+  outcomeEventId: string;
+  outcomeType: 'image-received' | 'reciprocal-engagement';
+  creditClass: LearningCreditClass;
+  attributionConfidence: AttributionConfidence;
+  status: LearningTrajectoryStatus;
+  maturityAt: number;
+  invalidatedAt: number | null;
+  invalidationReason: string | null;
+  reviewedAt: number | null;
+  createdAt: number;
+  updatedAt: number;
+}
+
+export interface LearningDecisionRecord {
+  id: string;
+  trajectoryId: string;
+  accountId: string;
+  decisionEventId: string;
+  stage: LearningStage;
+  context: LearningTurn[];
+  chosenResponse: string;
+  turnsUntilOutcome: number;
+  isDirectMediaInvitation: boolean;
+  qualityStatus: LearningDecisionStatus;
+  createdAt: number;
+  updatedAt: number;
+}
+
+export interface LearningTrajectoryDetail extends LearningTrajectoryRecord {
+  decisions: LearningDecisionRecord[];
+}
+
+export interface RetrievedLearningExample {
+  decisionId: string;
+  trajectoryId: string;
+  creditClass: LearningCreditClass;
+  stage: LearningStage;
+  context: LearningTurn[];
+  chosenResponse: string;
+  turnsUntilOutcome: number;
+  subreddit: string | null;
+  policyMode: ConversationPolicyMode | null;
+}
+
+export interface ConversationLearningQuery {
+  accountId: string;
+  conversationId: string;
+  subreddit: string | null;
+  policyMode: ConversationPolicyMode;
+  stage: LearningStage;
+  purpose: 'reply' | 'media-ask';
+  incomingEventId: string;
+}
+
+export interface ConversationLearningProvider {
+  getExamples(query: ConversationLearningQuery): readonly RetrievedLearningExample[];
+  recordGeneration(input: {
+    query: ConversationLearningQuery;
+    mode: 'shadow' | 'injected';
+    policyVersion: string;
+    exampleIds: readonly string[];
+    promptSha256?: string | null;
+    generatedAt?: number;
+  }): void;
+}
+
+export interface AttributedConversationMessage {
+  eventId: string;
+  timestamp: number;
+  isSelf: boolean;
+  text: string;
+  mediaCount: number;
+  actor: 'recipient' | 'ai' | 'human' | 'unknown';
+  kind: ConversationEventKind;
+  confidence: AttributionConfidence;
+}

src/services/media-library-exchange.ts

@@ -12,6 +12,7 @@ export interface MediaExchangeChat {
     existingMxcUri?: string | null,
   ): Promise<{ eventId: string; mxcUri: string } | null>;
   sendChatMessage(roomId: string, message: string, txnId?: string): Promise<boolean>;
+  sendChatMessageDetailed?(roomId: string, message: string, txnId?: string): Promise<{ eventId: string } | null>;
 }
 
 export interface MediaExchangeOutboundAttribution {
@@ -125,7 +126,7 @@ export class MediaLibraryExchangeService {
       if (!chosen || !chosen.enabled) {
         return { kind: 'human-required', reason: 'That library image is disabled or unavailable.' };
       }
-      send = this.repository.tryReserve(assetId, this.accountId, conversationId, incomingKey, 'generic') ?? undefined;
+      send = this.repository.tryReserve(assetId, this.accountId, conversationId, incomingKey, 'generic', 'human') ?? undefined;
       if (!send) {
         return { kind: 'human-required', reason: 'That library image has already been shared in this conversation.' };
       }
@@ -205,20 +206,27 @@ export class MediaLibraryExchangeService {
       if ('failure' in delivered) return delivered.failure;
       send = delivered.send;
 
-      const textSent = send.matrixTextEventId
-        ? true
-        : await this.chat.sendChatMessage(
-            conversationId,
-            RECIPROCAL_MESSAGE,
-            `rbot-library-reciprocal-${send.id}`,
-          );
+      const textReceipt = send.matrixTextEventId
+        ? { eventId: send.matrixTextEventId }
+        : this.chat.sendChatMessageDetailed
+          ? await this.chat.sendChatMessageDetailed(
+              conversationId,
+              RECIPROCAL_MESSAGE,
+              `rbot-library-reciprocal-${send.id}`,
+            )
+          : await this.chat.sendChatMessage(
+              conversationId,
+              RECIPROCAL_MESSAGE,
+              `rbot-library-reciprocal-${send.id}`,
+            ) ? { eventId: `rbot-library-reciprocal-${send.id}` } : null;
+      const textSent = textReceipt !== null;
       if (!textSent) {
         // The image event is already confirmed. Consume the asset and surface
         // a handoff rather than risking a duplicate image on a later retry.
         this.repository.markSent(send.id, 'unconfirmed');
         return { kind: 'human-required', reason: 'The image was sent, but the reciprocal text was not confirmed.', remoteSent: true };
       }
-      if (!send.matrixTextEventId) this.repository.markSent(send.id, `rbot-library-reciprocal-${send.id}`);
+      if (!send.matrixTextEventId) this.repository.markSent(send.id, textReceipt!.eventId);
       this.logger.info(`[media-library] Sent ${requestKind} asset ${asset.id} in ${conversationId}`);
       return { kind: 'sent', assetId: asset.id, requestKind };
     } catch (error) {

src/services/media-library-types.ts

@@ -48,4 +48,5 @@ export interface MediaLibrarySend {
   matrixTextEventId: string | null;
   mxcUri: string | null;
   error: string | null;
+  initiatedBy: 'automation' | 'human';
 }

src/services/reddit-account-runtime-builder.ts

@@ -58,7 +58,11 @@ import { createConfigBackedSubredditPollingControl } from './subreddit-polling-c
 import { ConversationAutomationService } from './conversation-automation.js';
 import { AiUsageRepository } from '../database/ai-usage-repository.js';
 import { MediaLibraryRepository } from '../database/media-library-repository.js';
+import { ConversationEventRepository } from '../database/conversation-event-repository.js';
+import { ConversationLearningRepository } from '../database/conversation-learning-repository.js';
 import { MediaLibraryExchangeService } from './media-library-exchange.js';
+import { DurableConversationLearningObserver } from './conversation-learning-observer.js';
+import { RepositoryConversationLearningProvider } from './conversation-learning-provider.js';
 import { RecipientAccountStatusService } from './recipient-account-status.js';
 import { AccountEstablishmentService } from './account-establishment.js';
 import { AccountGrowthService, type GrowthActivityEvent } from './account-growth.js';
@@ -185,6 +189,9 @@ async function buildRuntime(
     const conversationReadStateRepo = new ConversationReadStateRepository(db, session.accountId);
     const conversationAutomationClaimRepo = new ConversationAutomationClaimRepository(db, session.accountId);
     const conversationOutcomesRepo = new ConversationOutcomesRepository(db, session.accountId);
+    const conversationEventRepo = new ConversationEventRepository(db, session.accountId);
+    const conversationLearningRepo = new ConversationLearningRepository(db, session.accountId);
+    const mediaLibraryRepo = new MediaLibraryRepository(db);
     const redditUsersRepo = new RedditUsersRepository(db);
     const accountContentRepo = new AccountContentRepository(db);
     const activityLogRepo = new ActivityLogRepository(db, session.accountId);
@@ -228,10 +235,21 @@ async function buildRuntime(
       resolveUsernameToT2: (username) => redditApi.resolveUsernameToT2(username),
     };
     const chatCache = new ChatCache(logger, undefined, accountContentRepo, session.accountId);
+    const learningObserver = new DurableConversationLearningObserver(
+      session.accountId,
+      conversationEventRepo,
+      conversationLearningRepo,
+      conversationAutomationClaimRepo,
+      mediaLibraryRepo,
+      conversationOutcomesRepo,
+      accountConfig.learning,
+      logger,
+    );
     const chatService = new ChatService(
       matrixClient, logger, successfulMessagesRepo, mediaService,
       tokenRefresher, usernameResolver, chatCache,
       messagesRepo, redditUsersRepo, accountAuth, conversationOutcomesRepo, conversationStateRepo,
+      learningObserver,
     );
     if (events) wireAccountChatEvents(chatService, events, session.accountId);
     await chatService.loadCache();
@@ -322,7 +340,7 @@ async function buildRuntime(
     });
     await rateLimiter.initialize();
     const mediaExchange = new MediaLibraryExchangeService(
-      new MediaLibraryRepository(db),
+      mediaLibraryRepo,
       chatService,
       logger,
       session.accountId,
@@ -354,6 +372,11 @@ async function buildRuntime(
         triage: conversationReadStateRepo,
         screener: aiClient,
         get screening() { return accountConfig.conversations.aiReplyScreening; },
+        learning: {
+          provider: new RepositoryConversationLearningProvider(conversationLearningRepo, accountConfig.learning),
+          config: accountConfig.learning,
+          accountId: session.accountId,
+        },
       },
     );
 
@@ -361,6 +384,7 @@ async function buildRuntime(
       browserProfileDir, browserClient, accountAuth, matrixClient, chatCache, redditApi,
       postsRepo, messagesRepo, successfulMessagesRepo, deliveryOutboxRepo,
       conversationStateRepo, conversationReadStateRepo, conversationOutcomesRepo,
+      conversationEventRepo, conversationLearningRepo,
       redditUsersRepo, accountContentRepo, recipientAccountStatusRepo,
       activityLogRepo,
       monitorService, processorService, responseChecker, conversationAutomation, chatService, mediaService,
@@ -626,6 +650,7 @@ export class AccountRuntimeLoop {
         experimentId: processed.experimentId, promptVersion: processed.promptVersion,
         modelId: processed.modelId, selectionReason: processed.selectionReason,
         sourceType: processed.sourceType,
+        matrixEventId: start.chat.messages?.[0]?.eventId ?? null,
       });
       this.recipientDeduplicator.commit(post.author);
       this.resources.activityLogRepo.insert({

src/services/reddit-account-runtime.ts

@@ -39,6 +39,8 @@ import type { AccountEstablishmentService } from './account-establishment.js';
 import type { AccountGrowthService } from './account-growth.js';
 import type { AccountBrowsingSimulationService } from './account-browsing-simulation.js';
 import type { ProcessedPost, ProcessResult } from './processor.js';
+import type { ConversationEventRepository } from '../database/conversation-event-repository.js';
+import type { ConversationLearningRepository } from '../database/conversation-learning-repository.js';
 
 /** Credentials passed to a factory. They must never be exposed by list/snapshot APIs. */
 export interface VerifiedAccountSession {
@@ -73,6 +75,8 @@ export interface RedditAccountRuntimeResources {
   conversationStateRepo: ConversationStateRepository;
   conversationReadStateRepo: ConversationReadStateRepository;
   conversationOutcomesRepo: ConversationOutcomesRepository;
+  conversationEventRepo: ConversationEventRepository;
+  conversationLearningRepo: ConversationLearningRepository;
   redditUsersRepo: RedditUsersRepository;
   accountContentRepo: AccountContentRepository;
   recipientAccountStatusRepo: RedditRecipientAccountStatusRepository;