931 lines
39 KiB
TypeScript
931 lines
39 KiB
TypeScript
import type { OpenRouterClient } from '../ai/openrouter.js';
|
|
import type { SubredditConfig } from '../config/index.js';
|
|
import type { ConversationStateRepository, SentMessageRecord } from '../database/repositories.js';
|
|
import type { Logger } from '../utils/logger.js';
|
|
import type { RateLimiter } from '../utils/rate-limiter.js';
|
|
import { findSellerSignal } from '../reddit/api.js';
|
|
import type { ActiveChat, ChatMessage } from './chat/types.js';
|
|
import type { ChatService } from './chat/service.js';
|
|
import type { ConversationAutomationClaimRecord } from '../database/repositories.js';
|
|
import {
|
|
buildConversationState,
|
|
chooseConversationIntent,
|
|
conversationMessageFingerprint,
|
|
createConversationAutomationState,
|
|
hybridMediaAskSafetyFailure,
|
|
LEGACY_AUTOMATED_CONVERSATION_TURNS,
|
|
mergeConversationState,
|
|
r4rReplySafetyFailure,
|
|
remainingAutomationTurns,
|
|
type ConversationIntent,
|
|
type ConversationMessage,
|
|
type ConversationState,
|
|
} from './conversation-state.js';
|
|
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'
|
|
| 'human-required'
|
|
| 'boundary'
|
|
| 'image-received'
|
|
| 'killed';
|
|
|
|
export interface ConversationAutomationResult {
|
|
conversationId: string;
|
|
username: string;
|
|
kind: ConversationAutomationResultKind;
|
|
remainingTurns: number;
|
|
}
|
|
|
|
export interface ConversationOpenerHistory {
|
|
getByChatRoomId(chatRoomId: string): SentMessageRecord | undefined;
|
|
}
|
|
|
|
export interface ConversationAutomationClaims {
|
|
tryClaim(conversationId: string, incomingKey: string, replyText?: string, incomingAt?: number): boolean;
|
|
getClaim?(conversationId: string, incomingKey: string): ConversationAutomationClaimRecord | undefined;
|
|
listPending?(): ConversationAutomationClaimRecord[];
|
|
markAttempt?(conversationId: string, incomingKey: string): boolean;
|
|
releasePending?(conversationId: string, incomingKey: string): boolean;
|
|
markSent(conversationId: string, incomingKey: string, matrixEventId?: string | null): boolean;
|
|
markSuperseded?(conversationId: string, incomingKey: string): boolean;
|
|
}
|
|
|
|
export type ConversationBeforeSendLoader = (
|
|
conversationId: string,
|
|
observed: ActiveChat,
|
|
) => Promise<ActiveChat | undefined>;
|
|
|
|
/** Writes the operator-facing triage status, so killed conversations leave the inbox. */
|
|
export interface ConversationTriageWriter {
|
|
setStatus(conversationId: string, status: 'open' | 'snoozed' | 'done', snoozedUntil?: number | null, now?: number): void;
|
|
}
|
|
|
|
/** Narrow view of the AI client used for second-stage reply screening. */
|
|
export interface ConversationReplyScreener {
|
|
screenConversationReply(
|
|
conversationHistory: { sender: string; text: string; isSelf: boolean }[],
|
|
username: string,
|
|
): Promise<ConversationReplyScreening>;
|
|
}
|
|
|
|
export interface ConversationScreeningOptions {
|
|
enabled: boolean;
|
|
/** At or above this confidence, a seller/off-platform/scam verdict kills the conversation. */
|
|
killConfidence: number;
|
|
/** At or above this confidence, the same verdict hands off to a human instead. */
|
|
flagConfidence: number;
|
|
}
|
|
|
|
export interface ConversationAutomationOptions {
|
|
/** Off-platform kills are a safety control, so they default on regardless of `filterSellers`. */
|
|
offPlatformKill?: boolean;
|
|
/** Whether a configured seller keyword or domain also ends the conversation. */
|
|
killOnSellerKeyword?: boolean;
|
|
triage?: ConversationTriageWriter;
|
|
screener?: ConversationReplyScreener;
|
|
screening?: ConversationScreeningOptions;
|
|
learning?: { provider: ConversationLearningProvider; config: LearningConfig; accountId: string };
|
|
}
|
|
|
|
export interface ReplyPriorityStatus {
|
|
processing: boolean;
|
|
actionableBacklog: number;
|
|
oldestPendingAt: number | null;
|
|
lastScanAt: number | null;
|
|
lastReplySentAt: number | null;
|
|
lastReplyLatencyMs: number | null;
|
|
}
|
|
|
|
const SCREENING_CACHE_LIMIT = 500;
|
|
const SCREENING_REJECT_CLASSIFICATIONS = new Set(['seller', 'off-platform', 'scam', 'bot']);
|
|
|
|
const REPLY_REFRESH_LIMIT = 25;
|
|
const REPLY_SEND_PRIORITY = 100;
|
|
const PENDING_REPLY_RETRY_DELAY_MS = 2 * 60_000;
|
|
|
|
/**
|
|
* Sends at most one reply for each newly received user message. Durable state
|
|
* makes retries and restarts idempotent. The opener subreddit selects a
|
|
* snapshotted legacy, autonomous R4R, hybrid R4R-to-media, or manual policy.
|
|
*/
|
|
export class ConversationAutomationService {
|
|
private processing = false;
|
|
private actionableBacklog = 0;
|
|
private oldestPendingAt: number | null = null;
|
|
private lastScanAt: number | null = null;
|
|
private lastReplySentAt: number | null = null;
|
|
private lastReplyLatencyMs: number | null = null;
|
|
|
|
constructor(
|
|
private readonly chat: ChatService,
|
|
private readonly ai: OpenRouterClient,
|
|
private readonly states: ConversationStateRepository,
|
|
private readonly openers: ConversationOpenerHistory,
|
|
private readonly claims: ConversationAutomationClaims,
|
|
private readonly rateLimiter: RateLimiter,
|
|
private readonly logger: Logger,
|
|
private readonly dryRun = false,
|
|
private readonly sellerKeywords: readonly string[] = [],
|
|
private readonly sellerDomains: readonly string[] = [],
|
|
private readonly configuredSubreddits: () => readonly SubredditConfig[] = () => [],
|
|
private readonly now: () => number = Date.now,
|
|
private readonly pendingReplyRetryDelayMs = PENDING_REPLY_RETRY_DELAY_MS,
|
|
private readonly beforeSendLoader?: ConversationBeforeSendLoader,
|
|
private readonly mediaExchange?: MediaLibraryExchangeService,
|
|
private readonly options: ConversationAutomationOptions = {},
|
|
) {}
|
|
|
|
/** In-process memo so a claim retry does not pay for the same verdict twice. */
|
|
private readonly screeningCache = new Map<string, ConversationReplyScreening>();
|
|
|
|
/**
|
|
* Refreshes chats, recomputes the actionable backlog, and (unless
|
|
* `scanOnly`) generates and sends replies. Scan mode stops after the
|
|
* bookkeeping so the backlog surfaces for a human without spending AI
|
|
* tokens or touching Reddit.
|
|
*/
|
|
async processPendingConversations(
|
|
signal?: AbortSignal,
|
|
{ scanOnly = false }: { scanOnly?: boolean } = {},
|
|
): Promise<ConversationAutomationResult[]> {
|
|
this.processing = true;
|
|
let conversations: ActiveChat[] = [];
|
|
try {
|
|
const previousScanAt = this.lastScanAt;
|
|
await this.chat.refreshTopChats(REPLY_REFRESH_LIMIT);
|
|
conversations = await this.chat.getCachedConversations();
|
|
this.reconcilePendingClaims(conversations);
|
|
const actionable = this.actionableConversations(conversations, previousScanAt);
|
|
this.actionableBacklog = actionable.length;
|
|
this.oldestPendingAt = actionable[0]?.messages?.at(-1)?.timestamp ?? null;
|
|
this.lastScanAt = this.now();
|
|
if (actionable.length > 0) {
|
|
const oldestAgeSeconds = this.oldestPendingAt === null
|
|
? 0
|
|
: Math.max(0, Math.round((this.now() - this.oldestPendingAt) / 1_000));
|
|
this.logger.info(
|
|
`[automation] Reply-first queue has ${actionable.length} actionable conversation(s); oldest ${oldestAgeSeconds}s`,
|
|
);
|
|
}
|
|
|
|
if (scanOnly) {
|
|
if (actionable.length > 0) {
|
|
this.logger.info(
|
|
`[automation] Scan mode: leaving ${actionable.length} actionable conversation(s) for a human`,
|
|
);
|
|
}
|
|
return [];
|
|
}
|
|
|
|
const results: ConversationAutomationResult[] = [];
|
|
for (const conversation of actionable) {
|
|
const result = await this.processConversation(conversation, signal);
|
|
this.actionableBacklog = Math.max(0, this.actionableBacklog - 1);
|
|
this.oldestPendingAt = actionable[actionable.length - this.actionableBacklog]?.messages?.at(-1)?.timestamp ?? null;
|
|
if (result) results.push(result);
|
|
}
|
|
return results;
|
|
} finally {
|
|
this.processing = false;
|
|
const remaining = this.actionableConversations(conversations);
|
|
this.actionableBacklog = remaining.length;
|
|
this.oldestPendingAt = remaining[0]?.messages?.at(-1)?.timestamp ?? null;
|
|
}
|
|
}
|
|
|
|
async hasActionableReplies(): Promise<boolean> {
|
|
if (this.processing || this.actionableBacklog > 0) return true;
|
|
return this.actionableConversations(await this.chat.getCachedConversations()).length > 0;
|
|
}
|
|
|
|
getPriorityStatus(): ReplyPriorityStatus {
|
|
return {
|
|
processing: this.processing,
|
|
actionableBacklog: this.actionableBacklog,
|
|
oldestPendingAt: this.oldestPendingAt,
|
|
lastScanAt: this.lastScanAt,
|
|
lastReplySentAt: this.lastReplySentAt,
|
|
lastReplyLatencyMs: this.lastReplyLatencyMs,
|
|
};
|
|
}
|
|
|
|
async processConversation(
|
|
conversation: ActiveChat,
|
|
signal?: AbortSignal,
|
|
): Promise<ConversationAutomationResult | null> {
|
|
if (!conversation.roomId || !conversation.url) return null;
|
|
const messages = conversation.messages ?? [];
|
|
const latest = latestInboundMessage(messages);
|
|
if (!latest) return null;
|
|
|
|
const conversationId = conversation.roomId;
|
|
const opener = this.openers.getByChatRoomId(conversationId);
|
|
if (!opener || latest.timestamp <= opener.sentAt) {
|
|
this.logger.debug?.(
|
|
`[automation] Ignoring unverified inbound event for u/${conversation.username}: no confirmed post-opener response in this room`,
|
|
);
|
|
return null;
|
|
}
|
|
const mapped = messages.map(({ sender, text, isSelf }) => ({ sender, text, isSelf }));
|
|
const savedState = this.states.get(conversationId);
|
|
const state = mergeConversationState(
|
|
savedState,
|
|
buildConversationState(mapped),
|
|
);
|
|
if (!savedState) {
|
|
const behaviour = this.resolveConversationBehaviour(opener.subreddit);
|
|
const maxTurns = behaviour.mode === 'r4r' || behaviour.mode === 'r4r-media'
|
|
? behaviour.maxAutomatedTurns
|
|
: LEGACY_AUTOMATED_CONVERSATION_TURNS;
|
|
state.automation = createConversationAutomationState(
|
|
behaviour.mode,
|
|
maxTurns,
|
|
behaviour.rapportTurnsBeforeMediaAsk,
|
|
);
|
|
// Snapshot before generation so retries, restarts, dry runs, and later
|
|
// configuration edits cannot change the policy mid-conversation.
|
|
this.states.save(conversationId, state);
|
|
}
|
|
const incomingKey = messageKey(latest);
|
|
const mediaAttribution = this.mediaExchange?.getOutboundAttribution?.(conversationId);
|
|
if (state.automation.status === 'active'
|
|
&& hasUnattributedSelfMessage(
|
|
messages,
|
|
opener,
|
|
state.automation.aiMessageFingerprints,
|
|
mediaAttribution?.eventIds,
|
|
mediaAttribution?.texts,
|
|
)) {
|
|
state.automation.status = 'user-overridden';
|
|
state.automation.handoffReason = 'A manual reply was observed in this conversation.';
|
|
this.states.save(conversationId, state);
|
|
this.claims.markSuperseded?.(conversationId, incomingKey);
|
|
this.logger.info(`[automation] Paused u/${conversation.username}; an operator-authored message was observed`);
|
|
return null;
|
|
}
|
|
const existingClaim = this.claims.getClaim?.(conversationId, incomingKey);
|
|
if (existingClaim?.status === 'pending') {
|
|
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, observedOutbound.eventId ?? null);
|
|
this.logger.info(`[automation] Reconciled interrupted reply for u/${conversation.username} from Matrix history`);
|
|
return null;
|
|
}
|
|
if (!existingClaim.replyText) {
|
|
this.claims.releasePending?.(conversationId, incomingKey);
|
|
return this.finish(
|
|
conversation,
|
|
conversationId,
|
|
state,
|
|
incomingKey,
|
|
'human-required',
|
|
'An interrupted reply has no recoverable message body.',
|
|
);
|
|
}
|
|
if (this.now() - existingClaim.claimedAt < this.pendingReplyRetryDelayMs) return null;
|
|
const requestsMedia = containsMediaRequest(existingClaim.replyText);
|
|
const isHybridMediaAsk = state.automation.policyMode === 'r4r-media' && requestsMedia;
|
|
return this.sendClaimedReply(
|
|
conversation,
|
|
state,
|
|
incomingKey,
|
|
existingClaim.replyText,
|
|
latest.timestamp,
|
|
requestsMedia,
|
|
isHybridMediaAsk,
|
|
signal,
|
|
true,
|
|
);
|
|
}
|
|
if (existingClaim?.status === 'sent' || existingClaim?.status === 'superseded') return null;
|
|
if (state.automation.lastIncomingKey === incomingKey) return null;
|
|
if (state.automation.status !== 'active') return null;
|
|
|
|
if (state.automation.policyMode === 'manual') {
|
|
return this.finish(
|
|
conversation,
|
|
conversationId,
|
|
state,
|
|
incomingKey,
|
|
'human-required',
|
|
'This subreddit is configured for manual conversation handling.',
|
|
);
|
|
}
|
|
|
|
// Off-platform pushes end the conversation outright: there is nothing for an
|
|
// operator to salvage, and leaving them in the manual queue only buries the
|
|
// conversations that do need a human.
|
|
if (this.options.offPlatformKill !== false) {
|
|
const offPlatform = findOffPlatformSignal(latest.text);
|
|
if (offPlatform) {
|
|
return this.kill(conversation, conversationId, state, incomingKey, offPlatformSignalReason(offPlatform));
|
|
}
|
|
}
|
|
const sellerSignal = findSellerSignal(latest.text, [...this.sellerKeywords], [...this.sellerDomains]);
|
|
if (sellerSignal) {
|
|
state.sellerDetected = true;
|
|
const reason = `Potential upsell detected (${sellerSignal.kind}: ${sellerSignal.value}).`;
|
|
return this.options.killOnSellerKeyword === false
|
|
? this.finish(conversation, conversationId, state, incomingKey, 'human-required', reason)
|
|
: this.kill(conversation, conversationId, state, incomingKey, reason);
|
|
}
|
|
// Under-18 and in-person signals are safety handoffs rather than kills, and
|
|
// they apply to every policy mode, not just the R4R ones.
|
|
const safetyReason = safetyHandoffReason(latest.text);
|
|
if (safetyReason) {
|
|
return this.finish(conversation, conversationId, state, incomingKey, 'human-required', safetyReason);
|
|
}
|
|
if (hasReceivedImage(messages)) {
|
|
return this.finish(conversation, conversationId, state, incomingKey, 'image-received', 'The user shared an image.');
|
|
}
|
|
if (state.boundaries.length > 0) {
|
|
return this.finish(conversation, conversationId, state, incomingKey, 'boundary', state.boundaries.join('; '));
|
|
}
|
|
if (this.mediaExchange?.detect(conversationId, latest.text)) {
|
|
const exchangeAttempt = await this.rateLimiter.throttle(
|
|
async () => {
|
|
const refreshed = await this.loadConversationBeforeSend(conversationId, conversation);
|
|
const refreshedLatest = refreshed ? latestInboundMessage(refreshed.messages ?? []) : null;
|
|
if (!refreshed || !refreshedLatest) return { kind: 'superseded' as const };
|
|
const refreshedState = mergeConversationState(
|
|
this.states.get(conversationId),
|
|
buildConversationState((refreshed.messages ?? []).map(({ sender, text, isSelf }) => ({ sender, text, isSelf }))),
|
|
);
|
|
if (refreshedState.boundaries.length > 0 || hasReceivedImage(refreshed.messages ?? [])) {
|
|
return { kind: 'human-required' as const, reason: 'The conversation became ineligible before the image could be sent.' };
|
|
}
|
|
const refreshedOffPlatform = this.options.offPlatformKill !== false
|
|
? findOffPlatformSignal(refreshedLatest.text)
|
|
: null;
|
|
if (refreshedOffPlatform) {
|
|
return { kind: 'killed' as const, reason: offPlatformSignalReason(refreshedOffPlatform) };
|
|
}
|
|
const refreshedSellerSignal = findSellerSignal(
|
|
refreshedLatest.text,
|
|
[...this.sellerKeywords],
|
|
[...this.sellerDomains],
|
|
);
|
|
if (refreshedSellerSignal) {
|
|
return { kind: 'killed' as const, reason: 'Potential upsell appeared before the image could be sent.' };
|
|
}
|
|
if (messageKey(refreshedLatest) !== incomingKey) return { kind: 'superseded' as const };
|
|
return this.mediaExchange!.handle(conversationId, incomingKey, refreshedLatest.text);
|
|
},
|
|
(result) => result.kind === 'sent' || (result.kind === 'human-required' && result.remoteSent === true),
|
|
signal,
|
|
REPLY_SEND_PRIORITY,
|
|
);
|
|
const exchange = exchangeAttempt;
|
|
if (exchange.kind === 'sent') {
|
|
state.askedForMedia = true;
|
|
state.automation.lastIncomingKey = incomingKey;
|
|
this.states.save(conversationId, state);
|
|
return {
|
|
conversationId,
|
|
username: conversation.username,
|
|
kind: 'sent',
|
|
remainingTurns: remainingAutomationTurns(state),
|
|
};
|
|
}
|
|
if (exchange.kind === 'killed') {
|
|
return this.kill(conversation, conversationId, state, incomingKey, exchange.reason);
|
|
}
|
|
if (exchange.kind === 'human-required') {
|
|
return this.finish(conversation, conversationId, state, incomingKey, 'human-required', exchange.reason);
|
|
}
|
|
if (exchange.kind === 'retryable' || exchange.kind === 'dry-run') return null;
|
|
if (exchange.kind === 'superseded') return null;
|
|
}
|
|
if (remainingAutomationTurns(state) === 0) {
|
|
return this.finish(
|
|
conversation,
|
|
conversationId,
|
|
state,
|
|
incomingKey,
|
|
'human-required',
|
|
`${state.automation.maxAutomatedTurns} automated turns have been used.`,
|
|
);
|
|
}
|
|
|
|
const screeningOutcome = await this.screenReply(conversationId, incomingKey, mapped, conversation.username);
|
|
if (screeningOutcome) {
|
|
state.sellerDetected = true;
|
|
return screeningOutcome.kind === 'killed'
|
|
? this.kill(conversation, conversationId, state, incomingKey, screeningOutcome.reason)
|
|
: this.finish(conversation, conversationId, state, incomingKey, 'human-required', screeningOutcome.reason);
|
|
}
|
|
|
|
const { isLegacy, isHybridMediaAsk } = replyPlan(state);
|
|
if (isLegacy && state.askedForMedia) {
|
|
return this.finish(
|
|
conversation,
|
|
conversationId,
|
|
state,
|
|
incomingKey,
|
|
'human-required',
|
|
'A media request has already been made; further autonomous requests are disabled.',
|
|
);
|
|
}
|
|
if (isHybridMediaAsk) {
|
|
state.automation.hybridPhase = 'media-ask';
|
|
this.states.save(conversationId, state);
|
|
}
|
|
const requestsMedia = isLegacy || isHybridMediaAsk;
|
|
const intent = chooseConversationIntent(mapped, state, requestsMedia ? 'media-ask' : undefined);
|
|
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, learningExamples)
|
|
: isHybridMediaAsk
|
|
? 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)
|
|
: r4rReplySafetyFailure(reply);
|
|
if ((isLegacy || isHybridMediaAsk)
|
|
&& safetyFailure === 'media ask does not contain a recognizable image or video request') {
|
|
this.logger.warn?.(
|
|
`[automation] Regenerating media ask for u/${conversation.username}; the first response did not contain a recognizable request`,
|
|
);
|
|
reply = await generatePlannedReply();
|
|
safetyFailure = hybridMediaAskSafetyFailure(reply);
|
|
}
|
|
if (safetyFailure) {
|
|
const outputKind = isLegacy || isHybridMediaAsk ? 'media ask' : 'reply';
|
|
return this.finish(
|
|
conversation,
|
|
conversationId,
|
|
state,
|
|
incomingKey,
|
|
'human-required',
|
|
`The generated ${outputKind} was rejected: ${safetyFailure}.`,
|
|
);
|
|
}
|
|
if (this.dryRun) {
|
|
this.logger.info(`[automation] Dry run prepared turn ${state.automation.turnsUsed + 1} for u/${conversation.username}`);
|
|
return null;
|
|
}
|
|
|
|
if (!this.claims.tryClaim(conversationId, incomingKey, reply, latest.timestamp)) return null;
|
|
|
|
// The durable claim stores the intended body. Send-time validation then
|
|
// snapshots the intended state immediately before the Matrix request.
|
|
return this.sendClaimedReply(
|
|
conversation,
|
|
state,
|
|
incomingKey,
|
|
reply,
|
|
latest.timestamp,
|
|
requestsMedia,
|
|
isHybridMediaAsk,
|
|
signal,
|
|
);
|
|
}
|
|
|
|
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,
|
|
): ActiveChat[] {
|
|
return conversations
|
|
.filter((conversation) => this.isActionableConversation(conversation, observedAfter))
|
|
.sort((left, right) => {
|
|
const leftAt = left.messages?.at(-1)?.timestamp ?? left.lastMessageTime;
|
|
const rightAt = right.messages?.at(-1)?.timestamp ?? right.lastMessageTime;
|
|
return leftAt - rightAt;
|
|
});
|
|
}
|
|
|
|
private reconcilePendingClaims(conversations: ActiveChat[]): void {
|
|
for (const claim of this.claims.listPending?.() ?? []) {
|
|
const conversation = conversations.find((item) => item.roomId === claim.conversationId);
|
|
if (!conversation) continue;
|
|
const messages = conversation.messages ?? [];
|
|
const incomingIndex = messages.findIndex((message) => messageKey(message) === claim.incomingKey);
|
|
if (incomingIndex < 0) continue;
|
|
const laterSelfMessages = messages.slice(incomingIndex + 1).filter((message) => message.isSelf);
|
|
if (laterSelfMessages.length === 0) continue;
|
|
const expectedFingerprint = claim.replyText
|
|
? conversationMessageFingerprint(claim.replyText)
|
|
: null;
|
|
const observedClaimedReply = expectedFingerprint !== null
|
|
? 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`);
|
|
}
|
|
}
|
|
}
|
|
|
|
private isActionableConversation(conversation: ActiveChat, observedAfter: number | null): boolean {
|
|
if (!conversation.roomId || !conversation.url) return false;
|
|
const latest = latestInboundMessage(conversation.messages ?? []);
|
|
if (!latest) return false;
|
|
const opener = this.openers.getByChatRoomId(conversation.roomId);
|
|
if (!opener || latest.timestamp <= opener.sentAt) return false;
|
|
const incomingKey = messageKey(latest);
|
|
const claim = this.claims.getClaim?.(conversation.roomId, incomingKey);
|
|
if (claim?.status === 'pending') return true;
|
|
if (claim?.status === 'sent') return false;
|
|
if (claim?.status === 'superseded') return false;
|
|
// On the first scan after startup, retain Matrix unread as the conservative
|
|
// bootstrap signal so a deployment cannot revive historical read replies.
|
|
// Later scans can recover fresh inbound events even if that flag cleared.
|
|
if (!conversation.unread && (observedAfter === null || latest.timestamp <= observedAfter)) return false;
|
|
const state = this.states.get(conversation.roomId);
|
|
if (state) {
|
|
return state.automation.status === 'active' && state.automation.lastIncomingKey !== incomingKey;
|
|
}
|
|
return true;
|
|
}
|
|
|
|
private advanceStateForReply(
|
|
state: ConversationState,
|
|
conversationId: string,
|
|
incomingKey: string,
|
|
reply: string,
|
|
requestsMedia: boolean,
|
|
isHybridMediaAsk: boolean,
|
|
): boolean {
|
|
if (state.automation.lastIncomingKey !== incomingKey) {
|
|
if (requestsMedia) state.askedForMedia = true;
|
|
if (isHybridMediaAsk) state.automation.hybridPhase = 'post-media-ask';
|
|
state.automation.turnsUsed += 1;
|
|
state.automation.lastIncomingKey = incomingKey;
|
|
state.automation.aiMessageFingerprints.push(conversationMessageFingerprint(reply));
|
|
}
|
|
// Keep the hybrid conversation active after its one media ask even when
|
|
// that send consumes the final turn. The next inbound event still needs
|
|
// to be classified as an image, a decline, or a normal turn-limit handoff.
|
|
const exhausted = state.automation.turnsUsed >= state.automation.maxAutomatedTurns
|
|
&& !isHybridMediaAsk;
|
|
if (exhausted) {
|
|
state.automation.status = 'human-required';
|
|
state.automation.handoffReason = `${state.automation.maxAutomatedTurns} automated turns have been used.`;
|
|
}
|
|
this.states.save(conversationId, state);
|
|
return exhausted;
|
|
}
|
|
|
|
private async sendClaimedReply(
|
|
conversation: ActiveChat,
|
|
state: ConversationState,
|
|
incomingKey: string,
|
|
reply: string,
|
|
incomingAt: number,
|
|
requestsMedia: boolean,
|
|
isHybridMediaAsk: boolean,
|
|
signal?: AbortSignal,
|
|
resumePending = false,
|
|
): Promise<ConversationAutomationResult | null> {
|
|
const conversationId = conversation.roomId!;
|
|
const attempt = await this.rateLimiter.throttle(
|
|
async () => {
|
|
const latestConversation = await this.loadConversationBeforeSend(conversationId, conversation);
|
|
const latest = latestConversation ? latestInboundMessage(latestConversation.messages ?? []) : null;
|
|
const latestKey = latest ? messageKey(latest) : null;
|
|
const freshState = latestConversation
|
|
? mergeConversationState(
|
|
this.states.get(conversationId),
|
|
buildConversationState((latestConversation.messages ?? []).map(({ sender, text, isSelf }) => ({ sender, text, isSelf }))),
|
|
)
|
|
: this.states.get(conversationId) ?? state;
|
|
|
|
const persistedIntentCanResume = resumePending
|
|
&& freshState.automation.lastIncomingKey === incomingKey
|
|
&& freshState.automation.status === 'human-required';
|
|
if (!latestConversation || !latest || latestKey !== incomingKey
|
|
|| (freshState.automation.status !== 'active' && !persistedIntentCanResume)) {
|
|
this.claims.markSuperseded?.(conversationId, incomingKey);
|
|
const result = latestConversation && latest
|
|
? this.applyFreshHandoff(latestConversation, freshState, latest)
|
|
: null;
|
|
if (!result && latestConversation) this.states.save(conversationId, freshState);
|
|
this.logger.info(
|
|
`[automation] Superseded queued reply for u/${conversation.username}; a newer conversation state was observed before send`,
|
|
);
|
|
return { kind: 'superseded' as const, result, state: freshState };
|
|
}
|
|
|
|
if (freshState.boundaries.length > 0 || hasReceivedImage(latestConversation.messages ?? [])) {
|
|
this.claims.markSuperseded?.(conversationId, incomingKey);
|
|
const result = this.applyFreshHandoff(latestConversation, freshState, latest);
|
|
return { kind: 'superseded' as const, result, state: freshState };
|
|
}
|
|
|
|
if (requestsMedia && freshState.askedForMedia
|
|
&& !(resumePending && freshState.automation.lastIncomingKey === incomingKey)) {
|
|
this.claims.markSuperseded?.(conversationId, incomingKey);
|
|
const result = this.finish(
|
|
latestConversation,
|
|
conversationId,
|
|
freshState,
|
|
incomingKey,
|
|
'human-required',
|
|
'A media request has already been made; further autonomous requests are disabled.',
|
|
);
|
|
return { kind: 'superseded' as const, result, state: freshState };
|
|
}
|
|
|
|
const exhausted = this.advanceStateForReply(
|
|
freshState,
|
|
conversationId,
|
|
incomingKey,
|
|
reply,
|
|
requestsMedia,
|
|
isHybridMediaAsk,
|
|
);
|
|
this.claims.markAttempt?.(conversationId, incomingKey);
|
|
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,
|
|
REPLY_SEND_PRIORITY,
|
|
);
|
|
if (attempt.kind === 'superseded') return attempt.result;
|
|
if (attempt.kind === 'failed') return null;
|
|
const sentAt = this.now();
|
|
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(
|
|
`[automation] Sent ${attempt.state.automation.policyMode} turn ${attempt.state.automation.turnsUsed}/${attempt.state.automation.maxAutomatedTurns} `
|
|
+ `to u/${conversation.username}; reply latency ${Math.round(this.lastReplyLatencyMs / 1_000)}s`,
|
|
);
|
|
return {
|
|
conversationId: conversation.roomId!,
|
|
username: conversation.username,
|
|
kind: attempt.exhausted ? 'human-required' : 'sent',
|
|
remainingTurns: remainingAutomationTurns(attempt.state),
|
|
};
|
|
}
|
|
|
|
private async loadConversationBeforeSend(conversationId: string, observed: ActiveChat): Promise<ActiveChat | undefined> {
|
|
if (this.beforeSendLoader) return this.beforeSendLoader(conversationId, observed);
|
|
await this.chat.refreshTopChats(REPLY_REFRESH_LIMIT);
|
|
return (await this.chat.getCachedConversations()).find((item) => item.roomId === conversationId);
|
|
}
|
|
|
|
private applyFreshHandoff(
|
|
conversation: ActiveChat,
|
|
state: ConversationState,
|
|
latest: ChatMessage,
|
|
): ConversationAutomationResult | null {
|
|
if (state.boundaries.length > 0) {
|
|
return this.finish(conversation, conversation.roomId!, state, messageKey(latest), 'boundary', state.boundaries.join('; '));
|
|
}
|
|
if (hasReceivedImage(conversation.messages ?? [])) {
|
|
return this.finish(conversation, conversation.roomId!, state, messageKey(latest), 'image-received', 'The user shared an image.');
|
|
}
|
|
const sellerSignal = findSellerSignal(latest.text, [...this.sellerKeywords], [...this.sellerDomains]);
|
|
if (sellerSignal) {
|
|
state.sellerDetected = true;
|
|
return this.finish(
|
|
conversation,
|
|
conversation.roomId!,
|
|
state,
|
|
messageKey(latest),
|
|
'human-required',
|
|
`Potential upsell detected (${sellerSignal.kind}: ${sellerSignal.value}).`,
|
|
);
|
|
}
|
|
return null;
|
|
}
|
|
|
|
private resolveConversationBehaviour(subreddit: string | undefined): SubredditConfig['conversation'] {
|
|
const configured = subreddit
|
|
? this.configuredSubreddits().find((entry) => entry.name.toLowerCase() === subreddit.toLowerCase())
|
|
: undefined;
|
|
return configured?.conversation ?? {
|
|
mode: 'legacy',
|
|
maxAutomatedTurns: LEGACY_AUTOMATED_CONVERSATION_TURNS,
|
|
rapportTurnsBeforeMediaAsk: LEGACY_AUTOMATED_CONVERSATION_TURNS,
|
|
};
|
|
}
|
|
|
|
/**
|
|
* Runs the optional second-stage classifier. Returns null when the reply
|
|
* should proceed — including on any classifier failure, because an
|
|
* OpenRouter outage must not stall the reply path.
|
|
*/
|
|
private async screenReply(
|
|
conversationId: string,
|
|
incomingKey: string,
|
|
messages: ConversationMessage[],
|
|
username: string,
|
|
): Promise<{ kind: 'killed' | 'human-required'; reason: string } | null> {
|
|
const screening = this.options.screening;
|
|
const screener = this.options.screener;
|
|
if (!screening?.enabled || !screener) return null;
|
|
|
|
const cacheKey = `${conversationId}:${incomingKey}`;
|
|
let verdict = this.screeningCache.get(cacheKey);
|
|
if (!verdict) {
|
|
try {
|
|
verdict = await screener.screenConversationReply(messages, username);
|
|
} catch (error) {
|
|
this.logger.warn(
|
|
`[automation] Reply screening unavailable for u/${username}; proceeding: ${error instanceof Error ? error.message : String(error)}`,
|
|
);
|
|
return null;
|
|
}
|
|
if (this.screeningCache.size >= SCREENING_CACHE_LIMIT) {
|
|
const oldest = this.screeningCache.keys().next().value;
|
|
if (oldest !== undefined) this.screeningCache.delete(oldest);
|
|
}
|
|
this.screeningCache.set(cacheKey, verdict);
|
|
}
|
|
|
|
if (!SCREENING_REJECT_CLASSIFICATIONS.has(verdict.classification)) return null;
|
|
const reason = `AI screening classified this conversation as ${verdict.classification}`
|
|
+ ` (confidence ${verdict.confidence.toFixed(2)})`
|
|
+ (verdict.evidence ? `: ${verdict.evidence}` : '.');
|
|
if (verdict.confidence >= screening.killConfidence) return { kind: 'killed', reason };
|
|
if (verdict.confidence >= screening.flagConfidence) return { kind: 'human-required', reason };
|
|
return null;
|
|
}
|
|
|
|
/**
|
|
* Ends a conversation permanently. Unlike a handoff this also resolves the
|
|
* operator's triage state, so a killed conversation drops out of every
|
|
* actionable queue while staying reachable through the seller and done views.
|
|
*/
|
|
private kill(
|
|
conversation: ActiveChat,
|
|
conversationId: string,
|
|
state: ConversationState,
|
|
incomingKey: string,
|
|
reason: string,
|
|
): ConversationAutomationResult {
|
|
state.sellerDetected = true;
|
|
state.automation.status = 'killed';
|
|
state.automation.lastIncomingKey = incomingKey;
|
|
state.automation.handoffReason = reason;
|
|
this.states.save(conversationId, state);
|
|
this.claims.markSuperseded?.(conversationId, incomingKey);
|
|
try {
|
|
this.options.triage?.setStatus(conversationId, 'done', null, this.now());
|
|
} catch (error) {
|
|
this.logger.warn(
|
|
`[automation] Killed u/${conversation.username} but could not resolve triage state: ${error instanceof Error ? error.message : String(error)}`,
|
|
);
|
|
}
|
|
this.logger.info(`[automation] Killed conversation with u/${conversation.username}: ${reason}`);
|
|
return {
|
|
conversationId,
|
|
username: conversation.username,
|
|
kind: 'killed',
|
|
remainingTurns: 0,
|
|
};
|
|
}
|
|
|
|
private finish(
|
|
conversation: ActiveChat,
|
|
conversationId: string,
|
|
state: ConversationState,
|
|
incomingKey: string,
|
|
status: 'image-received' | 'human-required' | 'boundary',
|
|
reason: string,
|
|
): ConversationAutomationResult {
|
|
state.automation.status = status;
|
|
state.automation.lastIncomingKey = incomingKey;
|
|
state.automation.handoffReason = reason;
|
|
this.states.save(conversationId, state);
|
|
return {
|
|
conversationId,
|
|
username: conversation.username,
|
|
kind: status,
|
|
remainingTurns: remainingAutomationTurns(state),
|
|
};
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Matrix unread counts are presentation state and can return to zero before
|
|
* the automation worker observes an inbound event. The durable message list,
|
|
* opener timestamp, scan watermark, conversation state, and event claim decide
|
|
* whether work remains; only a latest inbound event can enter that flow.
|
|
*/
|
|
function latestInboundMessage(messages: ChatMessage[]): ChatMessage | null {
|
|
const latest = messages.at(-1);
|
|
return latest && !latest.isSelf ? latest : null;
|
|
}
|
|
|
|
function hasReceivedImage(messages: ChatMessage[]): boolean {
|
|
return messages.some((message) => !message.isSelf && (message.media?.length ?? 0) > 0);
|
|
}
|
|
|
|
function messageKey(message: ChatMessage): string {
|
|
return message.eventId ?? `${message.timestamp}:${message.sender}:${message.text}`;
|
|
}
|
|
|
|
function hasUnattributedSelfMessage(
|
|
messages: readonly ChatMessage[],
|
|
opener: SentMessageRecord,
|
|
automatedFingerprints: readonly string[],
|
|
automatedEventIds: readonly string[] = [],
|
|
automatedTexts: readonly string[] = [],
|
|
): boolean {
|
|
const knownEventIds = new Set(automatedEventIds);
|
|
const knownFingerprints = new Map<string, number>();
|
|
const addKnown = (fingerprint: string) => {
|
|
knownFingerprints.set(fingerprint, (knownFingerprints.get(fingerprint) ?? 0) + 1);
|
|
};
|
|
addKnown(conversationMessageFingerprint(opener.message));
|
|
for (const fingerprint of automatedFingerprints) addKnown(fingerprint);
|
|
for (const text of automatedTexts) addKnown(conversationMessageFingerprint(text));
|
|
|
|
for (const message of messages) {
|
|
if (!message.isSelf || message.timestamp < opener.sentAt) continue;
|
|
if (message.eventId && knownEventIds.has(message.eventId)) continue;
|
|
const fingerprint = conversationMessageFingerprint(message.text);
|
|
const remaining = knownFingerprints.get(fingerprint) ?? 0;
|
|
if (remaining === 0) return true;
|
|
knownFingerprints.set(fingerprint, remaining - 1);
|
|
}
|
|
return false;
|
|
}
|
|
|
|
function replyPlan(state: ConversationState): { isLegacy: boolean; isHybridMediaAsk: boolean } {
|
|
const isLegacy = state.automation.policyMode === 'legacy';
|
|
const isHybridMediaAsk = state.automation.policyMode === 'r4r-media'
|
|
&& !state.askedForMedia
|
|
&& state.automation.turnsUsed >= state.automation.rapportTurnsBeforeMediaAsk;
|
|
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
|
|
* earlier by the off-platform kill, which runs before this check.
|
|
*/
|
|
function safetyHandoffReason(message: string): string | null {
|
|
if (/\b(?:i(?:'m| am)|im|age(?:d)?|turning)\s*(?:1[0-7]|[0-9])\b/i.test(message)
|
|
|| /\b(?:1[0-7]|[0-9])\s*[fm]\b/i.test(message)
|
|
|| /\b(?:minor|underage)\b/i.test(message)) {
|
|
return 'The participant may be under 18.';
|
|
}
|
|
if (/\b(?:meet|meetup|meet up|come over|address|phone number)\b/i.test(message)) {
|
|
return 'The conversation reached contact details or an in-person meeting.';
|
|
}
|
|
return null;
|
|
}
|