jjbrunton/RedditBot
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Make conversation queues mutually exclusive

Browse Source
This commit is contained in:
Jamie Brunton committed 2026-08-14 09:25:45 +01:00
1 parent 3bb1dda6bb
commit 2447563c60
20 files changed
+386 -220
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/services/conversation-automation.ts
+99 -39
View File
@@ -3,7 +3,6 @@ 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';
@@ -22,7 +21,7 @@ import {
type ConversationState,
} from './conversation-state.js';
import { containsMediaRequest } from './conversation-boundaries.js';
import { findOffPlatformSignal, offPlatformSignalReason } from './off-platform-signals.js';
import { findDeterministicConversationRisk } from './conversation-risk.js';
import type { MediaLibraryExchangeService } from './media-library-exchange.js';
import type { ConversationReplyScreening } from '../ai/openrouter.js';
import type { LearningConfig } from '../config/index.js';
@@ -121,6 +120,8 @@ export class ConversationAutomationService {
private lastScanAt: number | null = null;
private lastReplySentAt: number | null = null;
private lastReplyLatencyMs: number | null = null;
/** Avoid rescanning unchanged cached transcripts every 30 seconds. */
private readonly deterministicScreeningKeys = new Map<string, string>();
constructor(
private readonly chat: ChatService,
@@ -161,6 +162,7 @@ export class ConversationAutomationService {
await this.chat.refreshTopChats(REPLY_REFRESH_LIMIT);
conversations = await this.chat.getCachedConversations();
this.reconcilePendingClaims(conversations);
const deterministicResults = scanOnly ? [] : this.screenCachedConversationRisks(conversations);
const actionable = this.actionableConversations(conversations, previousScanAt);
this.actionableBacklog = actionable.length;
this.oldestPendingAt = actionable[0]?.messages?.at(-1)?.timestamp ?? null;
@@ -183,7 +185,7 @@ export class ConversationAutomationService {
return [];
}
const results: ConversationAutomationResult[] = [];
const results: ConversationAutomationResult[] = [...deterministicResults];
for (const conversation of actionable) {
const result = await this.processConversation(conversation, signal);
this.actionableBacklog = Math.max(0, this.actionableBacklog - 1);
@@ -325,22 +327,17 @@ export class ConversationAutomationService {
);
}
// 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) {
const deterministicRisk = findDeterministicConversationRisk(
latest.text,
this.sellerKeywords,
this.sellerDomains,
{ includeOffPlatform: this.options.offPlatformKill !== false },
);
if (deterministicRisk) {
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);
return deterministicRisk.terminal || this.options.killOnSellerKeyword !== false
? this.kill(conversation, conversationId, state, incomingKey, deterministicRisk.reason)
: this.finish(conversation, conversationId, state, incomingKey, 'human-required', deterministicRisk.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.
@@ -368,19 +365,14 @@ export class ConversationAutomationService {
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(
const refreshedRisk = findDeterministicConversationRisk(
refreshedLatest.text,
[...this.sellerKeywords],
[...this.sellerDomains],
this.sellerKeywords,
this.sellerDomains,
{ includeOffPlatform: this.options.offPlatformKill !== false },
);
if (refreshedSellerSignal) {
return { kind: 'killed' as const, reason: 'Potential upsell appeared before the image could be sent.' };
if (refreshedRisk) {
return { kind: 'killed' as const, reason: refreshedRisk.reason };
}
if (messageKey(refreshedLatest) !== incomingKey) return { kind: 'superseded' as const };
const refreshedPreviousSelfText = previousSelfMessageBefore(
@@ -570,6 +562,67 @@ export class ConversationAutomationService {
}
}
/**
* Retrospectively applies the same deterministic rules as the live reply
* path to cached inbound history. This catches conversations first observed
* before a restart, after an operator reply, or outside the automation
* actionability window without spending classifier tokens.
*/
private screenCachedConversationRisks(
conversations: ActiveChat[],
): ConversationAutomationResult[] {
const results: ConversationAutomationResult[] = [];
for (const conversation of conversations) {
if (!conversation.roomId) continue;
const inbound = (conversation.messages ?? []).filter((message) => !message.isSelf);
const latest = inbound.at(-1);
if (!latest) continue;
const inboundText = inbound.map((message) => message.text).join('\n');
const screeningKey = `${messageKey(latest)}:${conversationMessageFingerprint(inboundText)}`;
if (this.deterministicScreeningKeys.get(conversation.roomId) === screeningKey) continue;
this.deterministicScreeningKeys.set(conversation.roomId, screeningKey);
const savedState = this.states.get(conversation.roomId);
if (savedState?.sellerDetected || savedState?.automation.status === 'killed') continue;
const risk = findDeterministicConversationRisk(
inboundText,
this.sellerKeywords,
this.sellerDomains,
{ includeOffPlatform: this.options.offPlatformKill !== false },
);
if (!risk) continue;
const mapped = (conversation.messages ?? []).map(({ sender, text, isSelf }) => ({ sender, text, isSelf }));
const state = mergeConversationState(savedState, buildConversationState(mapped));
if (!savedState) {
const opener = this.openers.getByChatRoomId(conversation.roomId);
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,
);
}
state.sellerDetected = true;
results.push(
risk.terminal || this.options.killOnSellerKeyword !== false
? this.kill(conversation, conversation.roomId, state, messageKey(latest), risk.reason)
: this.finish(
conversation,
conversation.roomId,
state,
messageKey(latest),
'human-required',
risk.reason,
),
);
}
return results;
}
private isActionableConversation(conversation: ActiveChat, observedAfter: number | null): boolean {
if (!conversation.roomId || !conversation.url) return false;
const latest = latestInboundMessage(conversation.messages ?? []);
@@ -738,17 +791,24 @@ export class ConversationAutomationService {
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) {
const risk = findDeterministicConversationRisk(
latest.text,
this.sellerKeywords,
this.sellerDomains,
{ includeOffPlatform: this.options.offPlatformKill !== false },
);
if (risk) {
state.sellerDetected = true;
return this.finish(
conversation,
conversation.roomId!,
state,
messageKey(latest),
'human-required',
`Potential upsell detected (${sellerSignal.kind}: ${sellerSignal.value}).`,
);
return risk.terminal || this.options.killOnSellerKeyword !== false
? this.kill(conversation, conversation.roomId!, state, messageKey(latest), risk.reason)
: this.finish(
conversation,
conversation.roomId!,
state,
messageKey(latest),
'human-required',
risk.reason,
);
}
return null;
}
src/services/conversation-risk.ts
+43
View File
@@ -0,0 +1,43 @@
import { findSellerSignal } from '../reddit/api.js';
import { findOffPlatformSignal, offPlatformSignalReason } from './off-platform-signals.js';
export type DeterministicConversationRiskSource = 'keyword' | 'domain' | 'funnel' | 'off-platform';
export interface DeterministicConversationRisk {
source: DeterministicConversationRiskSource;
reason: string;
/** Off-platform and seller-funnel signals are terminal safety outcomes. */
terminal: boolean;
}
/**
* Shared deterministic seller/off-platform decision used by live replies,
* send-time revalidation, automatic cached-history screening, and the manual
* inbox scan. Keeping one decision here prevents those paths from disagreeing.
*/
export function findDeterministicConversationRisk(
text: string,
sellerKeywords: readonly string[],
sellerDomains: readonly string[],
{ includeOffPlatform = true }: { includeOffPlatform?: boolean } = {},
): DeterministicConversationRisk | null {
if (includeOffPlatform) {
const offPlatform = findOffPlatformSignal(text);
if (offPlatform) {
return {
source: offPlatform.kind === 'seller-link' ? 'funnel' : 'off-platform',
reason: offPlatformSignalReason(offPlatform),
terminal: true,
};
}
}
const configured = findSellerSignal(text, [...sellerKeywords], [...sellerDomains]);
return configured
? {
source: configured.kind,
reason: `Potential upsell detected (${configured.kind}: ${configured.value}).`,
terminal: false,
}
: null;
}
src/services/reddit-account-runtime-builder.ts
+2
View File
@@ -514,6 +514,8 @@ export class AccountRuntimeLoop {
? `Human follow-up required for u/${result.username}`
: result.kind === 'image-received'
? `Image received from u/${result.username}`
: result.kind === 'killed'
? `Seller or off-platform conversation ended for u/${result.username}`
: `Automation stopped for u/${result.username} after a boundary`,
recipient: result.username,
});
src/services/seller-sweep.ts
+22 -19
View File
@@ -1,5 +1,7 @@
import { findSellerSignal } from '../reddit/api.js';
import { findOffPlatformSignal, offPlatformSignalReason } from './off-platform-signals.js';
import {
findDeterministicConversationRisk,
type DeterministicConversationRiskSource,
} from './conversation-risk.js';
import {
buildConversationState,
mergeConversationState,
@@ -38,7 +40,7 @@ export interface SellerSweepConversation {
watermark: string;
}
export type SellerSweepSource = 'keyword' | 'domain' | 'funnel' | 'ai';
export type SellerSweepSource = DeterministicConversationRiskSource | 'ai';
export interface SellerSweepFlag {
conversationId: string;
@@ -97,6 +99,8 @@ export interface SellerSweepOptions {
flagConfidence: number;
/** Mirrors `conversations.killOnSellerKeyword` for deterministic hits. */
killOnSellerKeyword: boolean;
/** Mirrors `conversations.offPlatformKill` for app/contact/payment pushes. */
offPlatformKill?: boolean;
/** Conversations per classifier request. */
batchSize?: number;
/** Newest messages per conversation sent to the classifier. */
@@ -144,21 +148,20 @@ function inboundText(messages: ConversationMessage[]): string {
function deterministicVerdict(
text: string,
options: SellerSweepOptions,
): { source: SellerSweepSource; reason: string } | null {
const configured = findSellerSignal(text, [...options.sellerKeywords], [...options.sellerDomains]);
if (configured) {
return {
source: configured.kind,
reason: `Potential upsell detected (${configured.kind}: ${configured.value}).`,
};
}
const signal = findOffPlatformSignal(text);
// Only the seller funnel patterns belong to a seller sweep. The other
// off-platform kinds are the live reply loop's business, not this one's.
if (signal?.kind === 'seller-link') {
return { source: 'funnel', reason: offPlatformSignalReason(signal) };
}
return null;
): { source: SellerSweepSource; reason: string; kill: boolean } | null {
const risk = findDeterministicConversationRisk(
text,
options.sellerKeywords,
options.sellerDomains,
{ includeOffPlatform: options.offPlatformKill !== false },
);
return risk
? {
source: risk.source,
reason: risk.reason,
kill: risk.terminal || options.killOnSellerKeyword,
}
: null;
}
export async function sweepConversationsForSellers(
@@ -238,7 +241,7 @@ export async function sweepConversationsForSellers(
const deterministic = deterministicVerdict(text, options);
if (deterministic) {
summary.deterministic += 1;
flag(conversation, state, deterministic.source, deterministic.reason, options.killOnSellerKeyword);
flag(conversation, state, deterministic.source, deterministic.reason, deterministic.kill);
continue;
}
// Deterministically clean and unchanged since the last sweep: the
src/web/routes/conversations.ts
+1
View File
@@ -834,6 +834,7 @@ export function createConversationsRouter(ctx: WebContext): Router {
killConfidence: ctx.config.conversations.aiReplyScreening.killConfidence,
flagConfidence: ctx.config.conversations.aiReplyScreening.flagConfidence,
killOnSellerKeyword: ctx.config.conversations.killOnSellerKeyword,
offPlatformKill: ctx.config.conversations.offPlatformKill,
};
}