Persist comment source monitoring

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,

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;

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<BrowsingThing, 'fullname' | 'text'>): 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');
+}

src/main-loop.ts

@@ -544,19 +544,20 @@ async function runConcurrentAccountRuntimes(app: AppContainer): Promise<void> {
           }
           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; `

src/reddit/api.ts

@@ -514,6 +514,7 @@ export class RedditApiClient {
     subreddit: string,
     postId: string,
     limit = 25,
+    sort: 'hot' | 'new' = 'hot',
   ): Promise<BrowsingThing[]> {
     const cookieHeader = await this.getCookieHeader();
     const headers: Record<string, string> = {};
@@ -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;

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<RedditPost[]> {
@@ -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<RedditPost[]> {
+  private async findCommentCandidates(posts: readonly RedditPost[], subreddit: string): Promise<RedditPost[]> {
     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. */

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,