diff --git a/docs/configuration.md b/docs/configuration.md index 8ed89e5..696f077 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -35,7 +35,7 @@ Never document real values. `.env` is ignored and may contain production credent - `userAgent`: Reddit JSON request identifier. - `targetGenders`: allowed parsed values. - `filterSellers`, `sellerKeywords`, `sellerDomains`: seller policy inputs; see also the `conversations` section, whose off-platform kill is independent of this flag. The defaults treat Telegram references, including `t.me` and `telegram.me` links, as seller signals alongside paid-content platforms and payment prompts. -- `prioritizeRecentPosts`, `recentWindowMinutes`: recency/fallback behavior. +- `prioritizeRecentPosts`, `recentWindowMinutes`: candidate recency/fallback behavior. `recentWindowMinutes` also controls how long comment-author source posts remain in the durable rotating watchlist. - `maxReadyCandidates`: hard ceiling for the active candidate target; valid range 1–500. Runtime lowers the actual target to what authenticated running accounts can send within `recentWindowMinutes` under current rolling-hour and message-spacing limits. - `physiqueTargeting`: optional image-candidate filter with `enabled`, inclusive `minScore`, and inclusive `maxScore` values from 1 to 10. It is disabled by default. When enabled, a model score outside the range or an image without enough visible body evidence to score is rejected before opener generation. Saving the setting also reconciles prepared image candidates in the ready and delivery-retry stages: candidates that no longer qualify are rejected, while retained candidates receive the newly applied range snapshot shown in the funnel. A candidate already leased for sending is not changed mid-send. The setting is hot-read by existing account runtimes; chat sources are unaffected. diff --git a/docs/data-flow.md b/docs/data-flow.md index 4e8f424..02f2400 100644 --- a/docs/data-flow.md +++ b/docs/data-flow.md @@ -23,7 +23,7 @@ stateDiagram-v2 ### 1. Discovery -The discovery loop rotates through enabled subreddits and authenticated runtimes. A runtime fetches Reddit `/new` data using the process-shared `RedditRequestCoordinator`. In the default `post-author` targeting mode those posts enter intake directly. An image subreddit configured for `comment-author` instead reads top-level comments from at most five newly listed posts and batches previously unseen eligible comment text through the reaction classifier. Candidates require a verdict at or above 0.8 confidence that the commenter wants to receive or experience the depicted person's role; comments describing what their author would do to the pictured person do not qualify. When `reddit.targetGenders` is non-empty, the expressed gender perspective must also be included. An `unknown` perspective fails closed unless explicitly targeted. A bounded in-process cache prevents unchanged comments from repeatedly consuming model calls. The candidate ID, author, timestamp, and permalink come from the comment while the original post image and title remain attached as preparation context; the usual image screen later verifies the depicted gender against the same target list. Intake then deduplicates candidate IDs and authors, excludes processed/contacted history, stores the full snapshot, source type, weight, and timestamps, and logs exact intake counts. Queued comments carry the classifier-policy version; active candidates admitted by the broader historical role policy are rejected on upgrade rather than allowed to send. +The discovery loop rotates through enabled subreddits and authenticated runtimes. A runtime fetches Reddit `/new` data using the process-shared `RedditRequestCoordinator`. In the default `post-author` targeting mode those posts enter intake directly. An image subreddit configured for `comment-author` admits each newly listed source post to a durable SQLite watchlist until the post leaves `reddit.recentWindowMinutes`. Each poll rotates to the least-recently checked source post and requests a bounded batch of up to 100 comments sorted newest-first, so a source remains eligible to produce later commenter candidates after its first scan and across process restarts. Comment IDs and body hashes retain prompt-free classifier verdicts; unchanged comments do not consume another model call, while an edited body is reclassified. Candidates require a verdict at or above 0.8 confidence that the commenter wants to receive or experience the depicted person's role; comments describing what their author would do to the pictured person do not qualify. When `reddit.targetGenders` is non-empty, the expressed gender perspective must also be included. An `unknown` perspective fails closed unless explicitly targeted. The candidate ID, author, timestamp, and permalink come from the comment while the original post image and title remain attached as preparation context; the usual image screen later verifies the depicted gender against the same target list. Intake then deduplicates candidate IDs and authors, excludes processed/contacted history, stores the full snapshot, source type, weight, and timestamps, and logs exact intake and comment-gate counts. Queued comments carry the classifier-policy version; active candidates admitted by the broader historical role policy are rejected on upgrade rather than allowed to send. The queue grows to a demand-based target rather than a fixed backlog. The target is the smaller of configured `reddit.maxReadyCandidates` and the work that authenticated running accounts can still send within the recent-post window, accounting for rolling hourly capacity and minimum message spacing. Intake admits only the current deficit, newest Reddit posts first, pauses at zero outbound capacity, and reports fetched posts left unevaluated because the funnel is full. Reddit request pacing, not a tight application poll, remains the primary request limiter. diff --git a/docs/database.md b/docs/database.md index fb4a3c8..50110e6 100644 --- a/docs/database.md +++ b/docs/database.md @@ -38,6 +38,8 @@ The special legacy-unassigned account exists only to preserve rows created befor ### Durable work and idempotency - `candidate_queue`: global candidate snapshot, source/priority, state, prepared payload, worker lease, retry timing, selected account, and rejection/error details. Image preparation snapshots retain the generated opener, grounded analysis, both AI scores, the physique range used, source-image URL, and experiment/model/tone metadata for delivery and operator review. +- `comment_source_watches`: durable rotating source-post snapshots with last-check and expiry timestamps for comment-author discovery. +- `comment_source_comment_observations`: prompt-free comment ID/body hash and classifier verdict cache, retained across restarts so unchanged comments avoid another model call and edited bodies can be reconsidered. - `delivery_outbox`: account-scoped prepared delivery and idempotency status retained for legacy/recovery paths. - `conversation_states`: account/conversation policy and derived state JSON. - `conversation_automation_claims`: account/conversation/inbound-event claim with intended reply and `pending`, `sent`, or terminal `superseded` state plus audit timestamps. diff --git a/docs/product-and-safety.md b/docs/product-and-safety.md index a78113e..f1f51ad 100644 --- a/docs/product-and-safety.md +++ b/docs/product-and-safety.md @@ -31,7 +31,7 @@ Configured subreddits have a source type: - `image`: image-led posts are downloaded and analyzed before an opener is generated. - `chat`: text-led posts require a parseable self-declared gender and adult age, then use the chat-opener prompt. -Image subreddits may separately target either the post author (the default) or a qualifying commenter. Comment-author targeting scans a bounded set of top-level comments on newly listed image posts and uses a fail-closed AI classification. A comment qualifies only when it personally expresses a desire to receive or experience the depicted person's role; describing what the commenter would do to the pictured person does not qualify. Confidence must be at least 0.8, and the gender perspective expressed by the comment must be included in `reddit.targetGenders` when that list is non-empty. `unknown` therefore fails closed unless it is explicitly targeted. The normal image screen also verifies that the depicted role matches `reddit.targetGenders`. The comment verdict is an interpretation of public text, not a claim about the author's actual gender or identity. Generic praise, ambiguous reactions, deleted/moderator comments, and the post author's own comments do not qualify. The original image remains scene context, but commenter openers use a separate prompt contract and must ground the message in what the commenter expressed. Drafts fail closed when they attribute the pictured body, pose, post, or image to the commenter, request another image, or omit a clear connection to the comment. +Image subreddits may separately target either the post author (the default) or a qualifying commenter. Comment-author targeting durably watches newly listed image posts throughout the configured recent window and checks their newest comments in rotation, rather than treating a source post as finished after one pass. A comment qualifies only when it personally expresses a desire to receive or experience the depicted person's role; describing what the commenter would do to the pictured person does not qualify. Confidence must be at least 0.8, and the gender perspective expressed by the comment must be included in `reddit.targetGenders` when that list is non-empty. `unknown` therefore fails closed unless it is explicitly targeted. The normal image screen also verifies that the depicted role matches `reddit.targetGenders`. The comment verdict is an interpretation of public text, not a claim about the author's actual gender or identity. Generic praise, ambiguous reactions, deleted/moderator comments, and the post author's own comments do not qualify. The original image remains scene context, but commenter openers use a separate prompt contract and must ground the message in what the commenter expressed. Drafts fail closed when they attribute the pictured body, pose, post, or image to the commenter, request another image, or omit a clear connection to the comment. Image analysis may produce a subjective 1–10 visible-physique score. It is not a medical assessment and remains unknown when the image does not show enough of the body for a grounded judgment. The scale is deliberately critical: 5 is broadly average, 6 requires visible above-average leanness or definition, and 8–10 are reserved for rare athletic or model-like physiques. Nudity and photographic presentation do not increase the score. Physique targeting is disabled by default. When an operator enables an inclusive minimum/maximum range, image candidates outside that range—and unscorable image candidates—are rejected before opener generation; chat candidates are unaffected. diff --git a/docs/services.md b/docs/services.md index c73280c..0d68983 100644 --- a/docs/services.md +++ b/docs/services.md @@ -20,7 +20,7 @@ This is an ownership map, not an API dump. Read the owning implementation and te | Module | Owns | Important constraints | | --- | --- | --- | -| `services/monitor.ts` | Subreddit ordering, post/comment candidate fetching, reaction/age/gender/seller screening, rejection summaries | Comment reactions fail closed at the confidence gate and, when configured, the expressed-gender gate; cheap checks run first; preserve exact rejection/deferred counts; listing 404 disables the subreddit. | +| `services/monitor.ts` | Subreddit ordering, post/comment candidate fetching, reaction/age/gender/seller screening, rejection summaries | Comment sources rotate durably and fetch newest comments; reactions fail closed at recipient-role, confidence, and configured expressed-gender gates; cheap checks run first; preserve exact rejection/deferred counts; listing 404 disables the subreddit. | | `reddit/api.ts` | Reddit JSON calls, post/comment parsing, seller/profile evidence, username resolution, error types | All reads must use the coordinator; redact error metadata. | | `reddit/request-coordinator.ts` | Process-wide Reddit request spacing and cooldown | Space failed requests too; 429 cooldown is global. | | `services/processor.ts` | Source-specific AI preparation and outbound safety | Returns prepared or explicit skip; performs no remote chat send. | @@ -30,6 +30,7 @@ This is an ownership map, not an API dump. Read the owning implementation and te | `services/recipient-deduplicator.ts` | Global recipient history plus live reservation | Case-insensitive; release failures, commit successes, recover stale claims. | | `utils/rate-limiter.ts` | Per-account send spacing/hour limits/jitter | Increment only for confirmed sends; reply priority may use the same limiter. | | `database/candidate-queue-repository.ts` | Durable candidate state machine, leases, retries, prepared-message replacement, counts, and manual clearing | Mutations require an eligible state or expected worker lease; recover expired/interrupted work; reject active comments admitted under a superseded eligibility policy; operator edits clear AI attribution; manual removals preserve sending leases and terminal audit rows. | +| `database/comment-source-watch-repository.ts` | Durable comment-source rotation plus comment body-hash/verdict observations | Watch lifetime follows the configured recent window; persist no comment body; reclassify when the body hash changes. | ## Account growth diff --git a/src/bootstrap.ts b/src/bootstrap.ts index 700b65f..ea7da45 100644 --- a/src/bootstrap.ts +++ b/src/bootstrap.ts @@ -58,6 +58,7 @@ import { importLegacyAccountContent } from './database/legacy-content-import.js' import { RecipientDeduplicator } from './services/recipient-deduplicator.js'; import { createConfigBackedSubredditPollingControl } from './services/subreddit-polling-control.js'; import { CandidateQueueRepository, RuntimeStateRepository } from './database/candidate-queue-repository.js'; +import { CommentSourceWatchRepository } from './database/comment-source-watch-repository.js'; import { AccountGrowthRepository } from './database/growth-repository.js'; import { AccountBrowsingSimulationRepository } from './database/browsing-simulation-repository.js'; import { MediaLibraryRepository } from './database/media-library-repository.js'; @@ -82,6 +83,7 @@ export interface AppContainer { conversationOutcomesRepo: ConversationOutcomesRepository; recipientDeduplicator: RecipientDeduplicator; candidateQueueRepo: CandidateQueueRepository; + commentSourceWatchRepo: CommentSourceWatchRepository; runtimeStateRepo: RuntimeStateRepository; growthRepo: AccountGrowthRepository; browsingRepo: AccountBrowsingSimulationRepository; @@ -151,6 +153,7 @@ export async function bootstrap( const conversationOutcomesRepo = new ConversationOutcomesRepository(db, bootstrapAccountId); const recipientDeduplicator = new RecipientDeduplicator(new RecipientHistoryRepository(db)); const candidateQueueRepo = new CandidateQueueRepository(db); + const commentSourceWatchRepo = new CommentSourceWatchRepository(db); const runtimeStateRepo = new RuntimeStateRepository(db); const growthRepo = new AccountGrowthRepository(db); const browsingRepo = new AccountBrowsingSimulationRepository(db); @@ -210,6 +213,7 @@ export async function bootstrap( redditApi, postsRepo, recipientDeduplicator, config, logger, 1000, createConfigBackedSubredditPollingControl(config, new ConfigStore(db), logger), aiClient, + commentSourceWatchRepo, ); const processorService = new ProcessorService( redditApi, @@ -516,6 +520,7 @@ export async function bootstrap( conversationOutcomesRepo, recipientDeduplicator, candidateQueueRepo, + commentSourceWatchRepo, runtimeStateRepo, growthRepo, browsingRepo, diff --git a/src/database/client.ts b/src/database/client.ts index 36f77ee..aef59d4 100644 --- a/src/database/client.ts +++ b/src/database/client.ts @@ -809,6 +809,31 @@ export function runMigrations(db: Database.Database): void { CREATE INDEX idx_reddit_accounts_username ON reddit_accounts(reddit_username COLLATE NOCASE); CREATE INDEX idx_reddit_accounts_active ON reddit_accounts(is_active, status); `), + () => db.exec(` + CREATE TABLE IF NOT EXISTS comment_source_watches ( + source_post_id TEXT PRIMARY KEY, + subreddit TEXT NOT NULL COLLATE NOCASE, + payload_json TEXT NOT NULL, + post_created_at INTEGER NOT NULL, + discovered_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + last_checked_at INTEGER, + expires_at INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_comment_source_watches_due + ON comment_source_watches(subreddit COLLATE NOCASE, expires_at, last_checked_at, post_created_at DESC); + + CREATE TABLE IF NOT EXISTS comment_source_comment_observations ( + comment_id TEXT PRIMARY KEY, + source_post_id TEXT NOT NULL, + body_sha256 TEXT NOT NULL, + verdict_json TEXT NOT NULL, + first_seen_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_comment_source_observations_source + ON comment_source_comment_observations(source_post_id, updated_at DESC); + `), ]; let currentVersion = db.pragma('user_version', { simple: true }) as number; diff --git a/src/database/comment-source-watch-repository.ts b/src/database/comment-source-watch-repository.ts new file mode 100644 index 0000000..9eaef8a --- /dev/null +++ b/src/database/comment-source-watch-repository.ts @@ -0,0 +1,110 @@ +import { createHash } from 'node:crypto'; +import type Database from 'better-sqlite3'; +import { z } from 'zod'; +import type { CommentReactionVerdict } from '../ai/openrouter.js'; +import type { BrowsingThing, RedditPost } from '../reddit/api.js'; + +const StoredVerdictSchema = z.object({ + id: z.string(), + qualifies: z.boolean(), + expressedGender: z.enum(['male', 'female', 'other', 'unknown']), + confidence: z.number().min(0).max(1), + evidence: z.string(), +}); + +export interface CommentWatchRefreshResult { + added: number; + active: number; + expired: number; +} + +/** Durable source-post rotation and prompt-free comment classification cache. */ +export class CommentSourceWatchRepository { + constructor(private readonly db: Database.Database) {} + + refreshSources(posts: readonly RedditPost[], watchWindowMs: number, now = Date.now()): CommentWatchRefreshResult { + const expired = this.db.prepare('DELETE FROM comment_source_watches WHERE expires_at <= ?').run(now).changes; + const existing = this.db.prepare('SELECT 1 FROM comment_source_watches WHERE source_post_id = ?'); + const upsert = this.db.prepare(` + INSERT INTO comment_source_watches ( + source_post_id, subreddit, payload_json, post_created_at, + discovered_at, updated_at, expires_at + ) VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(source_post_id) DO UPDATE SET + subreddit = excluded.subreddit, + payload_json = excluded.payload_json, + updated_at = excluded.updated_at, + expires_at = MAX(comment_source_watches.expires_at, excluded.expires_at) + `); + let added = 0; + this.db.transaction(() => { + for (const post of posts) { + const createdAt = Math.round(post.created_utc * 1_000); + const expiresAt = createdAt + watchWindowMs; + if (expiresAt <= now) continue; + if (!existing.get(post.id)) added += 1; + upsert.run(post.id, post.subreddit, JSON.stringify(post), createdAt, now, now, expiresAt); + } + })(); + const active = Number((this.db.prepare(` + SELECT COUNT(*) AS count FROM comment_source_watches WHERE expires_at > ? + `).get(now) as { count: number }).count); + return { added, active, expired }; + } + + nextSource(subreddit: string, now = Date.now()): RedditPost | null { + const row = this.db.prepare(` + SELECT payload_json FROM comment_source_watches + WHERE subreddit = ? COLLATE NOCASE AND expires_at > ? + ORDER BY COALESCE(last_checked_at, 0) ASC, post_created_at DESC + LIMIT 1 + `).get(subreddit, now) as { payload_json: string } | undefined; + if (!row) return null; + return JSON.parse(row.payload_json) as RedditPost; + } + + markChecked(sourcePostId: string, now = Date.now()): void { + this.db.prepare(` + UPDATE comment_source_watches SET last_checked_at = ?, updated_at = ? + WHERE source_post_id = ? + `).run(now, now, sourcePostId); + } + + getVerdict(comment: Pick): CommentReactionVerdict | null { + const row = this.db.prepare(` + SELECT body_sha256, verdict_json FROM comment_source_comment_observations + WHERE comment_id = ? + `).get(comment.fullname) as { body_sha256: string; verdict_json: string } | undefined; + if (!row || row.body_sha256 !== hashBody(comment.text)) return null; + try { + const parsed = StoredVerdictSchema.safeParse(JSON.parse(row.verdict_json)); + return parsed.success ? parsed.data : null; + } catch { + return null; + } + } + + recordVerdict(comment: BrowsingThing, verdict: CommentReactionVerdict, now = Date.now()): void { + this.db.prepare(` + INSERT INTO comment_source_comment_observations ( + comment_id, source_post_id, body_sha256, verdict_json, first_seen_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(comment_id) DO UPDATE SET + source_post_id = excluded.source_post_id, + body_sha256 = excluded.body_sha256, + verdict_json = excluded.verdict_json, + updated_at = excluded.updated_at + `).run( + comment.fullname, + comment.parentPostId, + hashBody(comment.text), + JSON.stringify({ ...verdict, evidence: '' }), + now, + now, + ); + } +} + +function hashBody(body: string): string { + return createHash('sha256').update(body).digest('hex'); +} diff --git a/src/main-loop.ts b/src/main-loop.ts index 1ac9346..480006d3 100644 --- a/src/main-loop.ts +++ b/src/main-loop.ts @@ -544,19 +544,20 @@ async function runConcurrentAccountRuntimes(app: AppContainer): Promise { } const target = candidateFunnelTarget(); const active = candidateQueueRepo.countActive(); - if (target === 0 || active >= target) { + const subreddit = enabled[nextSubredditIndex % enabled.length]; + nextSubredditIndex += 1; + const hasCandidateCapacity = target > 0 && active < target; + if (!hasCandidateCapacity && subreddit.targetingMode !== 'comment-author') { await sleep(5_000, controller.signal).catch(() => undefined); continue; } - const subreddit = enabled[nextSubredditIndex % enabled.length]; - nextSubredditIndex += 1; const posts = await scanner.resources.monitorService.fetchCandidates(subreddit); const intake = candidateQueueRepo.enqueue( posts, subreddit.type, subreddit.weight ?? 1, Date.now(), - Math.max(0, target - active), + hasCandidateCapacity ? target - active : 0, ); logger.info( `Candidate intake r/${subreddit.name}: ${posts.length} fetched; ${intake.discovered} queued; ` diff --git a/src/reddit/api.ts b/src/reddit/api.ts index 06e29d6..131e441 100644 --- a/src/reddit/api.ts +++ b/src/reddit/api.ts @@ -514,6 +514,7 @@ export class RedditApiClient { subreddit: string, postId: string, limit = 25, + sort: 'hot' | 'new' = 'hot', ): Promise { const cookieHeader = await this.getCookieHeader(); const headers: Record = {}; @@ -522,7 +523,7 @@ export class RedditApiClient { try { const response = await this.scheduleRequest(() => this.client.get( `/r/${subreddit}/comments/${postId}.json`, - { params: { limit, depth: 1, sort: 'hot', raw_json: 1 }, headers }, + { params: { limit, depth: 1, sort, raw_json: 1 }, headers }, )); const listings = Array.isArray(response.data) ? response.data : []; const children = (listings[1] as { data?: { children?: unknown } } | undefined)?.data?.children; diff --git a/src/services/monitor.ts b/src/services/monitor.ts index 04fedd1..cb08df0 100644 --- a/src/services/monitor.ts +++ b/src/services/monitor.ts @@ -7,6 +7,7 @@ import { } from '../reddit/api.js'; import type { CommentReactionCandidate, CommentReactionVerdict } from '../ai/openrouter.js'; import type { PostsRepository } from '../database/repositories.js'; +import type { CommentSourceWatchRepository } from '../database/comment-source-watch-repository.js'; import type { Config } from '../config/index.js'; import type { Logger } from '../utils/logger.js'; import { parseGenderTag } from '../utils/gender-tag.js'; @@ -42,6 +43,7 @@ export interface CommentReactionScreener { const COMMENT_REACTION_CONFIDENCE = 0.8; const COMMENT_POST_SCAN_LIMIT = 5; const COMMENTS_PER_POST_LIMIT = 20; +const WATCHED_COMMENTS_PER_POST_LIMIT = 100; export class MonitorService { // Flips on each call so that when both pools have candidates the bot @@ -59,6 +61,7 @@ export class MonitorService { private delayBetweenFetchesMs: number = 1000, private readonly pollingControl?: SubredditPollingControl, private readonly commentReactionScreener?: CommentReactionScreener, + private readonly commentSourceWatches?: CommentSourceWatchRepository, ) { } async getUnprocessedPosts(): Promise { @@ -79,7 +82,7 @@ export class MonitorService { this.logger.debug(`Checking r/${subreddit.name} (${subreddit.type}) for new posts...`); const listedPosts = await this.redditApi.getNewPosts(subreddit.name, 25, subreddit.type); const posts = subreddit.targetingMode === 'comment-author' - ? await this.findCommentCandidates(listedPosts) + ? await this.findCommentCandidates(listedPosts, subreddit.name) : listedPosts; const summary: ListingScreeningSummary = { scanned: 0, @@ -226,9 +229,13 @@ export class MonitorService { if (!this.isSubredditEnabled(subreddit.name)) return []; try { this.logger.debug(`Collecting candidate posts from r/${subreddit.name} (${subreddit.type})...`); - const posts = await this.redditApi.getNewPosts(subreddit.name, 25, subreddit.type); + const posts = await this.redditApi.getNewPosts( + subreddit.name, + subreddit.targetingMode === 'comment-author' ? 100 : 25, + subreddit.type, + ); if (subreddit.targetingMode !== 'comment-author') return posts; - return await this.findCommentCandidates(posts); + return await this.findCommentCandidates(posts, subreddit.name); } catch (error) { if (error instanceof RedditRateLimitError) throw error; if (error instanceof RedditSubredditUnavailableError) { @@ -242,18 +249,34 @@ export class MonitorService { } } - private async findCommentCandidates(posts: readonly RedditPost[]): Promise { + private async findCommentCandidates(posts: readonly RedditPost[], subreddit: string): Promise { if (!this.commentReactionScreener) { this.logger.warn('Comment-author targeting is configured but no reaction screener is available'); return []; } + let sourcePosts = posts.slice(0, COMMENT_POST_SCAN_LIMIT); + if (this.commentSourceWatches) { + const watch = this.commentSourceWatches.refreshSources( + posts, + this.config.reddit.recentWindowMinutes * 60_000, + ); + const due = this.commentSourceWatches.nextSource(subreddit); + sourcePosts = due ? [due] : []; + this.logger.debug( + `Comment watch r/${subreddit}: ${watch.added} source(s) added; ${watch.expired} expired; ${watch.active} active`, + ); + } const pairs: Array<{ post: RedditPost; comment: BrowsingThing }> = []; - for (const post of posts.slice(0, COMMENT_POST_SCAN_LIMIT)) { + let commentsRead = 0; + for (const post of sourcePosts) { const comments = await this.redditApi.getBrowsingComments( post.subreddit, post.id.replace(/^t3_/, ''), - COMMENTS_PER_POST_LIMIT, + this.commentSourceWatches ? WATCHED_COMMENTS_PER_POST_LIMIT : COMMENTS_PER_POST_LIMIT, + this.commentSourceWatches ? 'new' : 'hot', ); + commentsRead += comments.length; + this.commentSourceWatches?.markChecked(post.id); for (const comment of comments) { if (comment.removed || comment.author === '[deleted]' || comment.author === 'AutoModerator') continue; if (comment.author.toLowerCase() === post.author.toLowerCase()) continue; @@ -261,16 +284,23 @@ export class MonitorService { pairs.push({ post, comment }); } } - if (pairs.length === 0) return []; + if (pairs.length === 0) { + if (this.commentSourceWatches && sourcePosts.length > 0) { + this.logger.info( + `Comment watch r/${subreddit}: ${commentsRead} newest comment(s); 0 eligible for classification; 0 accepted`, + ); + } + return []; + } - const unseen = pairs.filter(({ comment }) => !this.commentReactionCache.has(comment.fullname)); + const unseen = pairs.filter(({ comment }) => this.getCommentReactionVerdict(comment) === null); if (unseen.length > 0) { const verdicts = await this.commentReactionScreener.screenCommentReactions( unseen.map(({ comment }) => ({ id: comment.fullname, text: comment.text })), ); const returned = new Map(verdicts.map((verdict) => [verdict.id, verdict])); for (const { comment } of unseen) { - this.commentReactionCache.set(comment.fullname, returned.get(comment.fullname) ?? { + this.storeCommentReactionVerdict(comment, returned.get(comment.fullname) ?? { id: comment.fullname, qualifies: false, expressedGender: 'unknown', @@ -278,17 +308,32 @@ export class MonitorService { evidence: 'Classifier returned no verdict', }); } - while (this.commentReactionCache.size > 2_000) { + while (!this.commentSourceWatches && this.commentReactionCache.size > 2_000) { const oldest = this.commentReactionCache.keys().next().value; if (typeof oldest !== 'string') break; this.commentReactionCache.delete(oldest); } } - return pairs.flatMap(({ post, comment }): RedditPost[] => { - const verdict = this.commentReactionCache.get(comment.fullname); - if (!verdict?.qualifies || verdict.confidence < COMMENT_REACTION_CONFIDENCE) return []; + let roleRejected = 0; + let lowConfidence = 0; + let genderRejected = 0; + const genderRejectedBy = { male: 0, female: 0, other: 0, unknown: 0 }; + const candidates = pairs.flatMap(({ post, comment }): RedditPost[] => { + const verdict = this.getCommentReactionVerdict(comment); + if (!verdict?.qualifies) { + roleRejected += 1; + return []; + } + if (verdict.confidence < COMMENT_REACTION_CONFIDENCE) { + lowConfidence += 1; + return []; + } const targetGenders = this.config.reddit.targetGenders; - if (targetGenders.length > 0 && !targetGenders.includes(verdict.expressedGender)) return []; + if (targetGenders.length > 0 && !targetGenders.includes(verdict.expressedGender)) { + genderRejected += 1; + genderRejectedBy[verdict.expressedGender] += 1; + return []; + } return [{ ...post, id: comment.fullname, @@ -308,6 +353,29 @@ export class MonitorService { }, }]; }); + if (this.commentSourceWatches && sourcePosts.length > 0) { + this.logger.info( + `Comment watch r/${subreddit}: ${commentsRead} newest comment(s); ${commentsRead - pairs.length} basic rejected; ` + + `${pairs.length} eligible for classification; ${unseen.length} newly classified; ` + + `${roleRejected} role rejected; ${lowConfidence} low confidence; ${genderRejected} gender rejected; ` + + `gender detail male=${genderRejectedBy.male}, female=${genderRejectedBy.female}, ` + + `other=${genderRejectedBy.other}, unknown=${genderRejectedBy.unknown}; ${candidates.length} accepted`, + ); + } + return candidates; + } + + private getCommentReactionVerdict(comment: BrowsingThing): CommentReactionVerdict | null { + if (this.commentSourceWatches) return this.commentSourceWatches.getVerdict(comment); + return this.commentReactionCache.get(comment.fullname) ?? null; + } + + private storeCommentReactionVerdict(comment: BrowsingThing, verdict: CommentReactionVerdict): void { + if (this.commentSourceWatches) { + this.commentSourceWatches.recordVerdict(comment, verdict); + return; + } + this.commentReactionCache.set(comment.fullname, verdict); } /** Apply cheap eligibility checks before any profile or AI work. */ diff --git a/src/services/reddit-account-runtime-builder.ts b/src/services/reddit-account-runtime-builder.ts index 7c2ff4c..23a4282 100644 --- a/src/services/reddit-account-runtime-builder.ts +++ b/src/services/reddit-account-runtime-builder.ts @@ -68,6 +68,7 @@ import { AccountEstablishmentService } from './account-establishment.js'; import { AccountGrowthService, type GrowthActivityEvent } from './account-growth.js'; import { AccountGrowthRepository } from '../database/growth-repository.js'; import { AccountBrowsingSimulationRepository } from '../database/browsing-simulation-repository.js'; +import { CommentSourceWatchRepository } from '../database/comment-source-watch-repository.js'; import { AccountBrowsingSimulationService } from './account-browsing-simulation.js'; import { allowsConversationSync, allowsSending, type BotMode } from './bot-mode.js'; import { @@ -207,6 +208,7 @@ async function buildRuntime( redditApi, postsRepo, recipientDeduplicator, accountConfig, logger, 1000, createConfigBackedSubredditPollingControl(config, new ConfigStore(db), logger), aiClient, + new CommentSourceWatchRepository(db), ); const processorService = new ProcessorService( redditApi, diff --git a/tests/comment-source-watch.test.ts b/tests/comment-source-watch.test.ts new file mode 100644 index 0000000..6ee6c76 --- /dev/null +++ b/tests/comment-source-watch.test.ts @@ -0,0 +1,64 @@ +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import Database from 'better-sqlite3'; +import { runMigrations } from '../src/database/client.js'; +import { CommentSourceWatchRepository } from '../src/database/comment-source-watch-repository.js'; +import type { BrowsingThing, RedditPost } from '../src/reddit/api.js'; + +function post(id: string, createdAtMs: number): RedditPost { + return { + id, name: id, subreddit: 'scene', author: `author-${id}`, title: id, + url: `https://example.test/${id}.jpg`, permalink: `/r/scene/comments/${id}/`, thumbnail: '', + is_self: false, created_utc: createdAtMs / 1_000, over_18: true, spoiler: false, + }; +} + +function comment(text: string): BrowsingThing { + return { + fullname: 't1_comment', kind: 'comment', subreddit: 'scene', author: 'reader', + permalink: '/r/scene/comments/source/post/comment/', createdAt: 2_000, score: 1, + text, nsfw: true, parentPostId: 'source', distinguished: null, removed: false, + }; +} + +describe('CommentSourceWatchRepository', () => { + let db: Database.Database; + let repo: CommentSourceWatchRepository; + + beforeEach(() => { + db = new Database(':memory:'); + runMigrations(db); + repo = new CommentSourceWatchRepository(db); + }); + + afterEach(() => db.close()); + + it('durably rotates source posts until their configured watch window expires', () => { + expect(repo.refreshSources([post('older', 1_000), post('newer', 2_000)], 10_000, 3_000)) + .toEqual({ added: 2, active: 2, expired: 0 }); + expect(repo.nextSource('SCENE', 3_000)?.id).toBe('newer'); + + repo.markChecked('newer', 3_100); + expect(repo.nextSource('scene', 3_200)?.id).toBe('older'); + + const restored = new CommentSourceWatchRepository(db); + expect(restored.nextSource('scene', 3_200)?.id).toBe('older'); + expect(restored.refreshSources([], 10_000, 12_001)).toEqual({ added: 0, active: 0, expired: 2 }); + }); + + it('reuses unchanged verdicts across restarts and reclassifies edited bodies', () => { + const original = comment('I would love that done to me'); + const verdict = { + id: original.fullname, qualifies: true, expressedGender: 'female' as const, + confidence: 0.94, evidence: 'recipient role', + }; + repo.recordVerdict(original, verdict, 4_000); + + expect(new CommentSourceWatchRepository(db).getVerdict(original)).toEqual({ ...verdict, evidence: '' }); + const stored = db.prepare(` + SELECT body_sha256, verdict_json FROM comment_source_comment_observations WHERE comment_id = ? + `).get(original.fullname); + expect(JSON.stringify(stored)).not.toContain(original.text); + expect(JSON.stringify(stored)).not.toContain(verdict.evidence); + expect(repo.getVerdict(comment('Edited: I would really love that done to me'))).toBeNull(); + }); +}); diff --git a/tests/database.test.ts b/tests/database.test.ts index 4f841a0..85115f6 100644 --- a/tests/database.test.ts +++ b/tests/database.test.ts @@ -108,9 +108,11 @@ describe('Database Client', () => { .all() as { name: string }[]; // Existing domain/config tables plus account, conversation, sync, media // ownership and account-growth stores. - expect(tables.length).toBe(33); + expect(tables.length).toBe(35); expect(tables.map((table) => table.name)).toEqual(expect.arrayContaining([ 'candidate_queue', + 'comment_source_watches', + 'comment_source_comment_observations', 'conversation_events', 'conversation_learning_trajectories', 'conversation_learning_decisions', diff --git a/tests/monitor.test.ts b/tests/monitor.test.ts index a9e16b9..d1bbbac 100644 --- a/tests/monitor.test.ts +++ b/tests/monitor.test.ts @@ -9,6 +9,9 @@ import { import type { PostsRepository, MessagesRepository } from '../src/database/repositories.js'; import { createMockLogger } from '../tests/mockLogger.js'; import { getTestConfig } from '../tests/testUtils.js'; +import Database from 'better-sqlite3'; +import { runMigrations } from '../src/database/client.js'; +import { CommentSourceWatchRepository } from '../src/database/comment-source-watch-repository.js'; function createMockRedditApi(posts: RedditPost[] = []): RedditApiClient { return { @@ -677,6 +680,60 @@ describe('MonitorService', () => { }); describe('comment-author targeting', () => { + it('keeps watching a durable source and classifies only new comments newest-first', async () => { + const db = new Database(':memory:'); + runMigrations(db); + try { + const source: RedditPost = { + id: 'post1', name: 't3_post1', subreddit: 'scene', author: 'poster', title: 'A scene', + url: 'https://example.com/scene.jpg', permalink: '/r/scene/comments/post1/', thumbnail: '', + is_self: false, created_utc: Date.now() / 1_000, over_18: true, spoiler: false, + }; + const first = { + fullname: 't1_first', kind: 'comment' as const, subreddit: 'scene', author: 'first', + permalink: '/first', createdAt: Date.now(), score: 1, text: 'As a woman I would love that done to me', + nsfw: true, parentPostId: 'post1', distinguished: null, removed: false, + }; + const second = { ...first, fullname: 't1_second', author: 'second', permalink: '/second' }; + const redditApi = createMockRedditApi([source]); + redditApi.getNewPosts = vi.fn() + .mockResolvedValueOnce([source]) + .mockResolvedValueOnce([]); + redditApi.getBrowsingComments = vi.fn() + .mockResolvedValueOnce([first]) + .mockResolvedValueOnce([second, first]); + const screener = { screenCommentReactions: vi.fn(async (comments) => comments.map((item) => ({ + id: item.id, qualifies: true, expressedGender: 'female' as const, confidence: 0.95, evidence: 'recipient role', + }))) }; + const config = getTestConfig({ reddit: { + subreddits: [{ name: 'scene', type: 'image', targetingMode: 'comment-author' }], + targetGenders: ['female'], recentWindowMinutes: 600, + } }); + monitor = new MonitorService( + redditApi, createMockPostsRepo(), createMockMessagesRepo(), config, logger, 0, undefined, screener, + new CommentSourceWatchRepository(db), + ); + + await expect(monitor.fetchCandidates(config.reddit.subreddits[0])) + .resolves.toEqual([expect.objectContaining({ id: 't1_first' })]); + await expect(monitor.fetchCandidates(config.reddit.subreddits[0])) + .resolves.toEqual(expect.arrayContaining([ + expect.objectContaining({ id: 't1_first' }), + expect.objectContaining({ id: 't1_second' }), + ])); + expect(screener.screenCommentReactions).toHaveBeenNthCalledWith(1, [ + { id: 't1_first', text: first.text }, + ]); + expect(screener.screenCommentReactions).toHaveBeenNthCalledWith(2, [ + { id: 't1_second', text: second.text }, + ]); + expect(redditApi.getNewPosts).toHaveBeenCalledWith('scene', 100, 'image'); + expect(redditApi.getBrowsingComments).toHaveBeenCalledWith('scene', 'post1', 100, 'new'); + } finally { + db.close(); + } + }); + it('turns a high-confidence targeted reaction into a commenter candidate with source media intact', async () => { const source: RedditPost = { id: 'post1', name: 't3_post1', subreddit: 'scene', author: 'poster', title: 'A scene', diff --git a/tests/reddit-api.test.ts b/tests/reddit-api.test.ts index 13257c8..276f9a7 100644 --- a/tests/reddit-api.test.ts +++ b/tests/reddit-api.test.ts @@ -54,6 +54,20 @@ describe('RedditApiClient', () => { params: { limit: 10, depth: 1, sort: 'hot', raw_json: 1 }, headers: {}, }); }); + + it('can request newest comments for a durable source-post watch', async () => { + const get = vi.fn().mockResolvedValue({ + data: [{ data: { children: [] } }, { data: { children: [] } }], + }); + vi.mocked(axios.create).mockReturnValue({ get } as unknown as AxiosInstance); + const client2 = new RedditApiClient(getTestConfig(), logger); + + await client2.getBrowsingComments('scene', 'source', 100, 'new'); + + expect(get).toHaveBeenCalledWith('/r/scene/comments/source.json', { + params: { limit: 100, depth: 1, sort: 'new', raw_json: 1 }, headers: {}, + }); + }); }); describe('hasImage', () => {