feat: add agent API and operational analytics

This commit is contained in:
jjbrunton committed 2026-08-08 22:08:14 +01:00
1 parent 720ea0652a
commit e8b730b592
65 files changed
+2650 -1214

No files matched your search

+30 -16
View File
@@ -1,10 +1,12 @@
import type { Logger } from '../../utils/logger.js';
import type {
ConversationOutcomesRepository,
MessagesRepository,
RedditUsersRepository,
SentMessageRecord,
SuccessfulMessagesRepository,
} from '../../database/repositories.js';
import { recordConversationReply } from '../conversation-outcome-recorder.js';
import type { MediaService } from '../media.js';
import { ChatCache } from './cache.js';
import {
@@ -151,6 +153,7 @@ export class ChatService {
private messagesRepo?: MessagesRepository,
private redditUsersRepo?: RedditUsersRepository,
private accountAuth?: AccountAuthGate,
private outcomesRepo?: ConversationOutcomesRepository,
) {
this.cache = cache ?? new ChatCache(logger);
if (this.redditUsersRepo) {
@@ -1007,6 +1010,7 @@ export class ChatService {
(message) => !message.isSelf && (message.media?.length ?? 0) > 0,
);
let performanceMessage: SentMessageRecord | undefined;
let firstAttributedReply = false;
if (firstSelf && chat.roomId && this.messagesRepo) {
try {
@@ -1017,7 +1021,7 @@ export class ChatService {
if (existing && hasPerformanceContext) {
performanceMessage = existing;
if (hasReply && !existing.gotResponse) {
this.messagesRepo.markGotResponse(existing.id);
firstAttributedReply = true;
}
if (hasReceivedImage && !existing.gotImage) {
this.messagesRepo.markGotImage(existing.id);
@@ -1028,19 +1032,26 @@ export class ChatService {
}
}
if (firstSelf && hasReply && performanceMessage && this.successfulMessagesRepo) {
if (!this.successfulMessagesRepo.existsByMessage(firstSelf.text)) {
const firstReply = chat.messages.find((m) => !m.isSelf);
try {
this.successfulMessagesRepo.insert({
sentMessageId: performanceMessage.id,
originalMessage: firstSelf.text,
imageAnalysis: performanceMessage.postImageAnalysis ?? null,
postTitle: performanceMessage.postTitle ?? null,
responseSnippet: firstReply?.text.slice(0, 200) || null,
subreddit: performanceMessage.subreddit ?? null,
});
this.logger.info(`[chat] stored successful chat example from ${chat.username}`);
if (firstSelf && hasReply && performanceMessage && this.successfulMessagesRepo && this.outcomesRepo && this.messagesRepo) {
const firstReply = chat.messages.find((m) => !m.isSelf);
const selfCount = chat.messages.filter((m) => m.isSelf).length;
const otherCount = chat.messages.length - selfCount;
try {
recordConversationReply(
performanceMessage,
firstReply?.text.slice(0, 200) ?? '',
{
messageCount: chat.messages.length,
reciprocal: selfCount >= 2 && otherCount >= 2,
},
{
messages: this.messagesRepo,
successful: this.successfulMessagesRepo,
outcomes: this.outcomesRepo,
logger: this.logger,
},
);
if (firstAttributedReply) {
try {
this.newReplyListener?.({
conversationId: chat.roomId,
@@ -1051,10 +1062,13 @@ export class ChatService {
} catch (err) {
this.logger.warn('[chat] newReplyListener threw:', err);
}
} catch (err) {
this.logger.warn('[chat] failed to store successful chat example:', err);
}
} catch (err) {
this.logger.warn('[chat] failed to record conversation outcome:', err);
}
} else if (hasReply && performanceMessage && this.messagesRepo && !performanceMessage.gotResponse) {
this.messagesRepo.markGotResponse(performanceMessage.id);
this.logger.warn('[chat] response recorded without canonical outcome repository; excluded from learning');
}
}
@@ -0,0 +1,46 @@
import type {
ConversationOutcomesRepository,
MessagesRepository,
SentMessageRecord,
SuccessfulMessagesRepository,
} from '../database/repositories.js';
import type { Logger } from '../utils/logger.js';
import { classifyConversationOutcome, type OutcomeContext, type ConversationOutcome } from './conversation-outcome.js';
export interface ConversationOutcomeDependencies {
messages: MessagesRepository;
successful: SuccessfulMessagesRepository;
outcomes: ConversationOutcomesRepository;
logger: Logger;
}
/** Canonical reply/outcome path shared by inbox and Matrix ingestion. */
export function recordConversationReply(
message: SentMessageRecord,
responseSnippet: string,
context: OutcomeContext,
deps: ConversationOutcomeDependencies,
): ConversationOutcome {
const outcome = classifyConversationOutcome(responseSnippet, context);
deps.messages.markGotResponse(message.id);
deps.outcomes.record(message.id, outcome, responseSnippet || null);
if (outcome === 'boundary') {
deps.outcomes.suppressRecipient(message.recipient, 'boundary');
}
if (outcome === 'engaged' && !deps.successful.existsByMessage(message.message)) {
deps.successful.insert({
sentMessageId: message.id,
originalMessage: message.message,
imageAnalysis: message.postImageAnalysis ?? null,
postTitle: message.postTitle ?? null,
responseSnippet: responseSnippet || null,
subreddit: message.subreddit ?? null,
});
deps.logger.info(`Engaged response from u/${message.recipient}; stored for learning.`);
} else if (outcome !== 'engaged') {
deps.logger.info(`Response from u/${message.recipient} classified as ${outcome}; excluded from learning examples.`);
}
return outcome;
}
+12 -2
View File
@@ -1,15 +1,25 @@
export const CONVERSATION_OUTCOMES = ['engaged', 'declined', 'boundary', 'upsell', 'unknown'] as const;
export type ConversationOutcome = (typeof CONVERSATION_OUTCOMES)[number];
export interface OutcomeContext {
/** Total human messages in the thread from both participants. */
messageCount?: number;
/** True when both participants have sent at least two messages. */
reciprocal?: boolean;
}
/**
* Deterministic first-pass outcome classifier. It is deliberately conservative:
* only replies with no decline, boundary, or sales signal become learning examples.
*/
export function classifyConversationOutcome(reply: string): ConversationOutcome {
export function classifyConversationOutcome(
reply: string,
context: OutcomeContext = {},
): ConversationOutcome {
const text = reply.toLowerCase();
if (/\b(onlyfans|fansly|manyvids|link in bio|cash\.?app|venmo)\b/.test(text)) return 'upsell';
if (/\b(stop|leave me alone|do not message|don't message)\b/.test(text)) return 'boundary';
if (/\b(no thanks|not interested|nope|don't send|do not send)\b/.test(text)) return 'declined';
if (text.trim().length > 0) return 'engaged';
if (context.reciprocal && (context.messageCount ?? 0) >= 4 && text.trim().length > 0) return 'engaged';
return 'unknown';
}
+3 -1
View File
@@ -42,7 +42,9 @@ export class MonitorService {
// to two profile requests for every otherwise-valid post in the listing.
const newestFirst = [...posts].sort((a, b) => b.created_utc - a.created_utc);
for (const post of newestFirst) {
if (this.postsRepo.isProcessed(post.id)) {
// Older adapters used in integrations may only expose account-local
// history; production repositories provide the shared lookup.
if (this.postsRepo.isProcessedAnywhere?.(post.id) ?? this.postsRepo.isProcessed(post.id)) {
continue;
}
+15
View File
@@ -13,6 +13,11 @@ export interface ProcessedPost {
detectedGender: DetectedGender;
generatedMessage: string;
strategyId: string | null;
experimentId: string | null;
promptVersion: string | null;
selectionReason: 'explore' | 'exploit' | null;
modelId: string;
sourceType: 'image' | 'chat';
toneId: ToneKey;
scripted: boolean;
}
@@ -213,6 +218,11 @@ export class ProcessorService {
detectedGender: analysis.detectedGender,
generatedMessage: message,
strategyId: selectedStrategy?.id || null,
experimentId: selectedStrategy?.experimentId ?? null,
promptVersion: selectedStrategy?.promptVersion ?? null,
selectionReason: selectedStrategy?.reason ?? null,
modelId: this.config.openrouter.imageAnalysisModel,
sourceType: 'image',
toneId: tone.profile.id,
scripted: tone.scripted !== null,
};
@@ -290,6 +300,11 @@ export class ProcessorService {
detectedGender: detectedGender ?? 'unknown',
generatedMessage: message,
strategyId: selectedStrategy?.id || null,
experimentId: selectedStrategy?.experimentId ?? null,
promptVersion: selectedStrategy?.promptVersion ?? null,
selectionReason: selectedStrategy?.reason ?? null,
modelId: this.config.openrouter.chatModel,
sourceType: 'chat',
toneId: tone.profile.id,
scripted: tone.scripted !== null,
};
+35 -29
View File
@@ -12,7 +12,8 @@ import {
RecipientHistoryRepository,
} from '../database/repositories.js';
import { AccountContentRepository } from '../database/account-repositories.js';
import { RedditApiClient } from '../reddit/api.js';
import { RedditApiClient, RedditRateLimitError, type RedditPost } from '../reddit/api.js';
import { RedditRequestCoordinator } from '../reddit/request-coordinator.js';
import { OpenRouterClient } from '../ai/openrouter.js';
import { MonitorService } from './monitor.js';
import { ProcessorService, isSkip } from './processor.js';
@@ -52,8 +53,9 @@ export function createConcreteRedditAccountRuntimeFactory(
const recipientDeduplicator = new RecipientDeduplicator(
new RecipientHistoryRepository(options.db),
);
const redditRequestCoordinator = new RedditRequestCoordinator();
return new RedditAccountRuntimeFactory(
(session) => buildRuntime(options, session, recipientDeduplicator),
(session) => buildRuntime(options, session, recipientDeduplicator, redditRequestCoordinator),
);
}
@@ -61,6 +63,7 @@ async function buildRuntime(
{ config, logger, db, isBotRunning = () => true }: ConcreteRuntimeFactoryOptions,
session: VerifiedAccountSession,
recipientDeduplicator: RecipientDeduplicator,
redditRequestCoordinator: RedditRequestCoordinator,
): Promise<RedditAccountRuntime> {
const browserProfileDir = accountBrowserProfileDir(
path.resolve(process.cwd(), config.browser.sessionDir),
@@ -107,7 +110,7 @@ async function buildRuntime(
const accountContentRepo = new AccountContentRepository(db);
const activityLogRepo = new ActivityLogRepository(db, session.accountId);
const redditApi = new RedditApiClient(accountConfig, logger);
const redditApi = new RedditApiClient(accountConfig, logger, redditRequestCoordinator);
redditApi.setBrowserContext(browserClient.getContext()!);
const chatPage = await browserClient.getContext()!.newPage();
const aiClient = new OpenRouterClient(accountConfig, logger);
@@ -135,7 +138,7 @@ async function buildRuntime(
const chatService = new ChatService(
matrixClient, logger, successfulMessagesRepo, mediaService,
new MatrixTokenRefresher(), usernameResolver, chatCache,
messagesRepo, redditUsersRepo, accountAuth,
messagesRepo, redditUsersRepo, accountAuth, conversationOutcomesRepo,
);
await chatService.loadCache();
const mediaSyncService = new MediaSyncService(chatService, logger, accountConfig.media.syncIntervalMs, accountAuth);
@@ -227,23 +230,23 @@ class AccountRuntimeLoop {
nextResponseCheckAt = Date.now() + responseInterval;
await this.resources.responseChecker.checkForResponses();
}
if (!this.resources.chatService.getInvitationCooldown()) {
await this.processPosts(signal);
}
await delay(this.config.reddit.pollIntervalMs, signal);
} catch (error) {
if (signal.aborted) return;
if (error instanceof RedditRateLimitError) {
await delay(error.retryAfterMs + 1_000, signal);
continue;
}
this.logger.error(`Account runtime loop failed for ${this.resources.accountAuth.getSnapshot().redditUsername}:`, error);
await delay(60_000, signal);
}
}
}
private async processPosts(signal: AbortSignal): Promise<void> {
const { monitorService, postsRepo, messagesRepo, redditApi, processorService, rateLimiter, chatService, responseChecker } = this.resources;
const posts = await monitorService.getUnprocessedPosts();
for (const post of posts) {
if (signal.aborted) return;
async dispatchPost(post: RedditPost, signal?: AbortSignal): Promise<void> {
const { postsRepo, messagesRepo, redditApi, processorService, rateLimiter, chatService, responseChecker } = this.resources;
if (signal?.aborted) return;
try {
const subreddit = this.config.reddit.subreddits.find((entry) => entry.name.toLowerCase() === post.subreddit.toLowerCase());
const isChat = subreddit?.type === 'chat';
const imageUrl = isChat ? null : redditApi.getImageUrl(post);
@@ -251,25 +254,26 @@ class AccountRuntimeLoop {
id: post.id, subreddit: post.subreddit, author: post.author, title: post.title,
postUrl: redditApi.getPostUrl(post), imageUrl, processedAt: Date.now(),
});
try {
if (
this.config.reddit.filterSellers
&& (this.config.reddit.sellerKeywords.length > 0 || this.config.reddit.sellerDomains.length > 0)
&& await redditApi.isSellerProfile(post.author, this.config.reddit.sellerKeywords, this.config.reddit.sellerDomains)
) {
postsRepo.setError(post.id, 'Seller profile detected');
this.resources.activityLogRepo.insert({ timestamp: Date.now(), type: 'skipped', message: `Skipped post by u/${post.author} (seller profile)`, recipient: post.author });
continue;
}
// Seller screening is performed once by the shared MonitorService before
// assignment. Do not repeat its profile request for every recipient.
processorService.setSuccessfulExamples(responseChecker.getSuccessfulExamples(3, post.subreddit));
const processed = await processorService.processPost(post);
if (isSkip(processed)) {
postsRepo.setError(post.id, processed.skipReason);
this.resources.activityLogRepo.insert({ timestamp: Date.now(), type: 'skipped', message: `Skipped post by u/${post.author} (${processed.skipReason})`, recipient: post.author });
continue;
return;
}
postsRepo.updateAnalysis(post.id, processed.imageAnalysis, processed.generatedMessage);
await rateLimiter.throttle(async () => {
if (this.config.browser.dryRun) {
const reason = 'Dry run: message prepared but not sent';
postsRepo.setError(post.id, reason);
this.resources.activityLogRepo.insert({
timestamp: Date.now(), type: 'skipped',
message: `${reason} to u/${post.author}`, recipient: post.author,
});
return false;
}
if (!this.recipientDeduplicator.tryReserve(post.author)) {
postsRepo.setError(post.id, 'User already contacted across accounts');
this.resources.activityLogRepo.insert({
@@ -295,6 +299,9 @@ class AccountRuntimeLoop {
postId: post.id, recipient: post.author, subject: post.author,
message: processed.generatedMessage, sentAt: Date.now(),
strategyId: processed.strategyId, chatRoomId: chat.roomId,
experimentId: processed.experimentId, promptVersion: processed.promptVersion,
modelId: processed.modelId, selectionReason: processed.selectionReason,
sourceType: processed.sourceType,
});
this.resources.activityLogRepo.insert({
timestamp: Date.now(), type: 'sent', message: `Sent message to u/${post.author} (r/${post.subreddit})`,
@@ -303,12 +310,11 @@ class AccountRuntimeLoop {
void chatService.refreshTopChats(5);
return true;
}, (success) => success, signal);
} catch (error) {
if (signal.aborted) return;
postsRepo.setError(post.id, error instanceof Error ? error.message : String(error));
this.resources.activityLogRepo.insert({ timestamp: Date.now(), type: 'error', message: `Failed processing post by u/${post.author}`, recipient: post.author });
this.logger.error(`Account runtime failed processing post ${post.id}:`, error);
}
} catch (error) {
if (signal?.aborted) return;
postsRepo.setError(post.id, error instanceof Error ? error.message : String(error));
this.resources.activityLogRepo.insert({ timestamp: Date.now(), type: 'error', message: `Failed processing post by u/${post.author}`, recipient: post.author });
this.logger.error(`Account runtime failed processing post ${post.id}:`, error);
}
}
}
+21
View File
@@ -29,6 +29,7 @@ import type { MediaSyncService } from './media-sync.js';
import type { MonitorService } from './monitor.js';
import type { ProcessorService } from './processor.js';
import type { ResponseCheckerService } from './response-checker.js';
import type { RedditPost } from '../reddit/api.js';
/** Credentials passed to a factory. They must never be exposed by list/snapshot APIs. */
export interface VerifiedAccountSession {
@@ -80,6 +81,8 @@ export interface RedditAccountRuntimeLifecycle {
start(): void | Promise<void>;
/** Stop and await this account's schedules/loop, then close its browser. */
stop(): void | Promise<void>;
/** Process a post assigned by the shared discovery scheduler. */
dispatchPost?(post: RedditPost, signal?: AbortSignal): Promise<void>;
}
export type RedditAccountRuntimeState =
@@ -172,6 +175,24 @@ export class RedditAccountRuntime {
};
}
async canAcceptPost(): Promise<boolean> {
if (this.state !== 'running') return false;
if (!this.resources.accountAuth.isFullyAuthenticated()
&& !(await this.resources.accountAuth.ensureFullyAuthenticated())) return false;
if (this.resources.chatService.getInvitationCooldown()) return false;
return this.resources.rateLimiter.getTimers().nextMessageAvailableAt === null;
}
dispatchPost(post: RedditPost, signal?: AbortSignal): Promise<void> {
if (this.state !== 'running') {
throw new Error(`Cannot assign a post to ${this.session.redditUsername}: runtime is not running`);
}
if (!this.lifecycle.dispatchPost) {
throw new Error(`Runtime ${this.session.redditUsername} does not support shared post dispatch`);
}
return this.lifecycle.dispatchPost(post, signal);
}
private enqueue(operation: () => Promise<void>): Promise<void> {
const result = this.transition.then(operation, operation);
this.transition = result.catch(() => undefined);
+21 -18
View File
@@ -2,7 +2,7 @@ import type { Page } from 'playwright';
import type { Logger } from '../utils/logger.js';
import type { MessagesRepository, SuccessfulMessagesRepository, SentMessageRecord, ConversationOutcomesRepository } from '../database/repositories.js';
import type { PostsRepository } from '../database/repositories.js';
import { classifyConversationOutcome } from './conversation-outcome.js';
import { recordConversationReply } from './conversation-outcome-recorder.js';
import { denyAccountAuth, type AccountAuthGate } from './account-auth-context.js';
export interface ResponseCheckResult {
@@ -68,25 +68,28 @@ export class ResponseCheckerService {
continue;
}
this.messagesRepo.markGotResponse(message.id);
// Get the original post info
const post = message.postId ? this.postsRepo.getById(message.postId) : undefined;
const outcome = classifyConversationOutcome(result.responseSnippet ?? '');
this.outcomesRepo?.record(message.id, outcome, result.responseSnippet ?? null);
if (outcome === 'engaged') {
this.successfulMessagesRepo.insert({
sentMessageId: message.id,
originalMessage: message.message,
imageAnalysis: post?.imageAnalysis || null,
postTitle: post?.title || null,
responseSnippet: result.responseSnippet || null,
subreddit: post?.subreddit || null,
});
this.logger.info(`Engaged response from u/${message.recipient}; stored for learning.`);
const attributedMessage = {
...message,
subreddit: message.subreddit ?? post?.subreddit,
postTitle: message.postTitle ?? post?.title,
postImageAnalysis: message.postImageAnalysis ?? post?.imageAnalysis ?? undefined,
};
if (this.outcomesRepo) {
recordConversationReply(
attributedMessage,
result.responseSnippet ?? '',
{ messageCount: 2, reciprocal: false },
{
messages: this.messagesRepo,
successful: this.successfulMessagesRepo,
outcomes: this.outcomesRepo,
logger: this.logger,
},
);
} else {
this.logger.info(`Response from u/${message.recipient} classified as ${outcome}; excluded from learning examples.`);
this.messagesRepo.markGotResponse(message.id);
this.logger.warn('Outcome repository unavailable; response recorded but excluded from learning.');
}
} else {
// Just mark as checked
+39 -7
View File
@@ -6,12 +6,16 @@ export interface SelectedStrategy {
id: string;
promptOverride: string;
reason: 'exploit' | 'explore';
experimentId?: string;
promptVersion?: string;
}
export class StrategySelector {
private variants: Config['strategies']['variants'];
private explorationRate: number;
private minSamples: number;
private responseWindowMs: number;
private section: 'strategies' | 'chatStrategies';
constructor(
private config: Config,
@@ -19,10 +23,12 @@ export class StrategySelector {
private logger: Logger,
section: 'strategies' | 'chatStrategies' = 'strategies',
) {
this.section = section;
const cfg = config[section];
this.variants = cfg.variants;
this.explorationRate = cfg.explorationRate;
this.minSamples = cfg.minSamples;
this.responseWindowMs = cfg.responseWindowHours * 60 * 60 * 1000;
}
selectStrategy(): SelectedStrategy | null {
@@ -32,13 +38,13 @@ export class StrategySelector {
// Exploration: random strategy
if (Math.random() < this.explorationRate) {
const variant = this.variants[Math.floor(Math.random() * this.variants.length)];
const variant = this.weightedChoice(this.variants);
this.logger.debug(`Strategy selected (explore): ${variant.id}`);
return { id: variant.id, promptOverride: variant.promptOverride, reason: 'explore' };
return this.selected(variant, 'explore');
}
// Exploitation: pick the best-performing strategy
const stats = this.messagesRepo.getStrategyStats();
const stats = this.messagesRepo.getStrategyStats(Date.now() - this.responseWindowMs);
const statsMap = new Map(stats.map(s => [s.strategyId, s]));
// Find strategies with insufficient samples - prioritize exploring them
@@ -48,9 +54,9 @@ export class StrategySelector {
});
if (unexplored.length > 0) {
const variant = unexplored[Math.floor(Math.random() * unexplored.length)];
const variant = this.weightedChoice(unexplored);
this.logger.debug(`Strategy selected (under-sampled): ${variant.id}`);
return { id: variant.id, promptOverride: variant.promptOverride, reason: 'explore' };
return this.selected(variant, 'explore');
}
// All strategies have enough samples - pick the best response rate
@@ -66,11 +72,11 @@ export class StrategySelector {
}
this.logger.debug(`Strategy selected (exploit, ${bestRate}% response rate): ${bestVariant.id}`);
return { id: bestVariant.id, promptOverride: bestVariant.promptOverride, reason: 'exploit' };
return this.selected(bestVariant, 'exploit');
}
getStats(): (StrategyStats & { configured: boolean })[] {
const dbStats = this.messagesRepo.getStrategyStats();
const dbStats = this.messagesRepo.getStrategyStats(Date.now() - this.responseWindowMs);
const statsMap = new Map(dbStats.map(s => [s.strategyId, s]));
// Include all configured variants, even if they have no data yet
@@ -102,4 +108,30 @@ export class StrategySelector {
hasStrategies(): boolean {
return this.variants.length > 0;
}
private selected(
variant: Config['strategies']['variants'][number],
reason: 'explore' | 'exploit',
): SelectedStrategy {
const promptVersion = createHash('sha256').update(variant.promptOverride).digest('hex').slice(0, 12);
return {
id: variant.id,
promptOverride: variant.promptOverride,
reason,
promptVersion,
experimentId: `${this.section}:${variant.id}:${promptVersion}`,
};
}
private weightedChoice<T extends { weight?: number }>(variants: readonly T[]): T {
const total = variants.reduce((sum, variant) => sum + Math.max(0, variant.weight ?? 1), 0);
if (total <= 0) return variants[Math.floor(Math.random() * variants.length)];
let cursor = Math.random() * total;
for (const variant of variants) {
cursor -= Math.max(0, variant.weight ?? 1);
if (cursor < 0) return variant;
}
return variants[variants.length - 1];
}
}
import { createHash } from 'node:crypto';