diff --git a/src/database/client.ts b/src/database/client.ts index 8d518b5..d851110 100644 --- a/src/database/client.ts +++ b/src/database/client.ts @@ -312,6 +312,30 @@ export function runMigrations(db: Database.Database): void { updated_at INTEGER NOT NULL ); `), + () => { + db.exec(` + CREATE TABLE IF NOT EXISTS conversation_automation_claims ( + account_id TEXT NOT NULL, + conversation_id TEXT NOT NULL, + incoming_key TEXT NOT NULL, + status TEXT NOT NULL CHECK(status IN ('pending', 'sent')), + claimed_at INTEGER NOT NULL, + sent_at INTEGER, + reply_text TEXT, + incoming_at INTEGER, + last_attempt_at INTEGER, + attempt_count INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (account_id, conversation_id, incoming_key), + FOREIGN KEY (account_id) REFERENCES reddit_accounts(account_id) ON DELETE CASCADE + ); + CREATE INDEX IF NOT EXISTS idx_conversation_automation_claims_status + ON conversation_automation_claims(account_id, status, claimed_at); + `); + addColumnIfNotExists(db, 'conversation_automation_claims', 'reply_text', 'TEXT'); + addColumnIfNotExists(db, 'conversation_automation_claims', 'incoming_at', 'INTEGER'); + addColumnIfNotExists(db, 'conversation_automation_claims', 'last_attempt_at', 'INTEGER'); + addColumnIfNotExists(db, 'conversation_automation_claims', 'attempt_count', 'INTEGER NOT NULL DEFAULT 0'); + }, ]; let currentVersion = db.pragma('user_version', { simple: true }) as number; diff --git a/src/database/repositories.ts b/src/database/repositories.ts index 368cdbb..3a20b96 100644 --- a/src/database/repositories.ts +++ b/src/database/repositories.ts @@ -1173,19 +1173,96 @@ export class ConversationStateRepository { } /** Durable idempotency claims for AI replies to individual inbound events. */ +export interface ConversationAutomationClaimRecord { + conversationId: string; + incomingKey: string; + status: 'pending' | 'sent'; + replyText: string | null; + incomingAt: number | null; + claimedAt: number; + lastAttemptAt: number | null; + attemptCount: number; + sentAt: number | null; +} + export class ConversationAutomationClaimRepository { constructor( private readonly db: Database.Database, private readonly accountId: string = LEGACY_UNASSIGNED_ACCOUNT_ID, ) {} - tryClaim(conversationId: string, incomingKey: string, now: number = Date.now()): boolean { + tryClaim( + conversationId: string, + incomingKey: string, + replyTextOrNow: string | number = '', + incomingAt: number | null = null, + now: number = Date.now(), + ): boolean { + const replyText = typeof replyTextOrNow === 'string' ? replyTextOrNow : ''; + const claimedAt = typeof replyTextOrNow === 'number' ? replyTextOrNow : now; return this.db.prepare(` INSERT INTO conversation_automation_claims ( - account_id, conversation_id, incoming_key, status, claimed_at - ) VALUES (?, ?, ?, 'pending', ?) + account_id, conversation_id, incoming_key, status, claimed_at, reply_text, incoming_at + ) VALUES (?, ?, ?, 'pending', ?, ?, ?) ON CONFLICT(account_id, conversation_id, incoming_key) DO NOTHING - `).run(this.accountId, conversationId, incomingKey, now).changes === 1; + `).run(this.accountId, conversationId, incomingKey, claimedAt, replyText || null, incomingAt).changes === 1; + } + + getClaim(conversationId: string, incomingKey: string): ConversationAutomationClaimRecord | undefined { + const row = this.db.prepare(` + SELECT conversation_id, incoming_key, status, reply_text, incoming_at, + claimed_at, last_attempt_at, attempt_count, sent_at + FROM conversation_automation_claims + WHERE account_id = ? AND conversation_id = ? AND incoming_key = ? + `).get(this.accountId, conversationId, incomingKey) as Record | undefined; + if (!row) return undefined; + return { + conversationId: row.conversation_id as string, + incomingKey: row.incoming_key as string, + status: row.status as 'pending' | 'sent', + replyText: (row.reply_text as string | null) ?? null, + incomingAt: (row.incoming_at as number | null) ?? null, + claimedAt: row.claimed_at as number, + lastAttemptAt: (row.last_attempt_at as number | null) ?? null, + attemptCount: row.attempt_count as number, + sentAt: (row.sent_at as number | null) ?? null, + }; + } + + listPending(): ConversationAutomationClaimRecord[] { + const rows = this.db.prepare(` + SELECT conversation_id, incoming_key, status, reply_text, incoming_at, + claimed_at, last_attempt_at, attempt_count, sent_at + FROM conversation_automation_claims + WHERE account_id = ? AND status = 'pending' + ORDER BY claimed_at ASC + `).all(this.accountId) as Array>; + return rows.map((row) => ({ + conversationId: row.conversation_id as string, + incomingKey: row.incoming_key as string, + status: 'pending', + replyText: (row.reply_text as string | null) ?? null, + incomingAt: (row.incoming_at as number | null) ?? null, + claimedAt: row.claimed_at as number, + lastAttemptAt: (row.last_attempt_at as number | null) ?? null, + attemptCount: row.attempt_count as number, + sentAt: (row.sent_at as number | null) ?? null, + })); + } + + markAttempt(conversationId: string, incomingKey: string, now: number = Date.now()): boolean { + return this.db.prepare(` + UPDATE conversation_automation_claims + SET last_attempt_at = ?, attempt_count = attempt_count + 1 + WHERE account_id = ? AND conversation_id = ? AND incoming_key = ? AND status = 'pending' + `).run(now, this.accountId, conversationId, incomingKey).changes === 1; + } + + releasePending(conversationId: string, incomingKey: string): boolean { + return this.db.prepare(` + DELETE FROM conversation_automation_claims + WHERE account_id = ? AND conversation_id = ? AND incoming_key = ? AND status = 'pending' + `).run(this.accountId, conversationId, incomingKey).changes === 1; } markSent(conversationId: string, incomingKey: string, now: number = Date.now()): boolean { diff --git a/src/services/conversation-automation.ts b/src/services/conversation-automation.ts index f0ff6db..b1f6489 100644 --- a/src/services/conversation-automation.ts +++ b/src/services/conversation-automation.ts @@ -6,6 +6,7 @@ import type { RateLimiter } from '../utils/rate-limiter.js'; import { findSellerSignal } from '../reddit/api.js'; import type { ActiveChat, ChatMessage } from './chat/types.js'; import type { ChatService } from './chat/service.js'; +import type { ConversationAutomationClaimRecord } from '../database/repositories.js'; import { buildConversationState, chooseConversationIntent, @@ -37,16 +38,40 @@ export interface ConversationOpenerHistory { } export interface ConversationAutomationClaims { - tryClaim(conversationId: string, incomingKey: string): boolean; + tryClaim(conversationId: string, incomingKey: string, replyText?: string, incomingAt?: number): boolean; + getClaim?(conversationId: string, incomingKey: string): ConversationAutomationClaimRecord | undefined; + listPending?(): ConversationAutomationClaimRecord[]; + markAttempt?(conversationId: string, incomingKey: string): boolean; + releasePending?(conversationId: string, incomingKey: string): boolean; markSent(conversationId: string, incomingKey: string): boolean; } +export interface ReplyPriorityStatus { + processing: boolean; + actionableBacklog: number; + oldestPendingAt: number | null; + lastScanAt: number | null; + lastReplySentAt: number | null; + lastReplyLatencyMs: number | null; +} + +const REPLY_REFRESH_LIMIT = 25; +const REPLY_SEND_PRIORITY = 100; +const PENDING_REPLY_RETRY_DELAY_MS = 2 * 60_000; + /** * Sends at most one reply for each newly received user message. Durable state * makes retries and restarts idempotent. The opener subreddit selects a * snapshotted legacy, autonomous R4R, hybrid R4R-to-media, or manual policy. */ export class ConversationAutomationService { + private processing = false; + private actionableBacklog = 0; + private oldestPendingAt: number | null = null; + private lastScanAt: number | null = null; + private lastReplySentAt: number | null = null; + private lastReplyLatencyMs: number | null = null; + constructor( private readonly chat: ChatService, private readonly ai: OpenRouterClient, @@ -59,17 +84,60 @@ export class ConversationAutomationService { private readonly sellerKeywords: readonly string[] = [], private readonly sellerDomains: readonly string[] = [], private readonly configuredSubreddits: () => readonly SubredditConfig[] = () => [], + private readonly now: () => number = Date.now, + private readonly pendingReplyRetryDelayMs = PENDING_REPLY_RETRY_DELAY_MS, ) {} async processUnreadConversations(signal?: AbortSignal): Promise { - await this.chat.refreshTopChats(5); - const conversations = await this.chat.getCachedConversations(); - const results: ConversationAutomationResult[] = []; - for (const conversation of conversations) { - const result = await this.processConversation(conversation, signal); - if (result) results.push(result); + this.processing = true; + let conversations: ActiveChat[] = []; + try { + await this.chat.refreshTopChats(REPLY_REFRESH_LIMIT); + conversations = await this.chat.getCachedConversations(); + this.reconcilePendingClaims(conversations); + const actionable = this.actionableConversations(conversations); + this.actionableBacklog = actionable.length; + this.oldestPendingAt = actionable[0]?.messages?.at(-1)?.timestamp ?? null; + this.lastScanAt = this.now(); + if (actionable.length > 0) { + const oldestAgeSeconds = this.oldestPendingAt === null + ? 0 + : Math.max(0, Math.round((this.now() - this.oldestPendingAt) / 1_000)); + this.logger.info( + `[automation] Reply-first queue has ${actionable.length} actionable conversation(s); oldest ${oldestAgeSeconds}s`, + ); + } + + const results: ConversationAutomationResult[] = []; + for (const conversation of actionable) { + const result = await this.processConversation(conversation, signal); + this.actionableBacklog = Math.max(0, this.actionableBacklog - 1); + this.oldestPendingAt = actionable[actionable.length - this.actionableBacklog]?.messages?.at(-1)?.timestamp ?? null; + if (result) results.push(result); + } + return results; + } finally { + this.processing = false; + const remaining = this.actionableConversations(conversations); + this.actionableBacklog = remaining.length; + this.oldestPendingAt = remaining[0]?.messages?.at(-1)?.timestamp ?? null; } - return results; + } + + async hasActionableReplies(): Promise { + if (this.processing || this.actionableBacklog > 0) return true; + return this.actionableConversations(await this.chat.getCachedConversations()).length > 0; + } + + getPriorityStatus(): ReplyPriorityStatus { + return { + processing: this.processing, + actionableBacklog: this.actionableBacklog, + oldestPendingAt: this.oldestPendingAt, + lastScanAt: this.lastScanAt, + lastReplySentAt: this.lastReplySentAt, + lastReplyLatencyMs: this.lastReplyLatencyMs, + }; } async processConversation( @@ -110,6 +178,46 @@ export class ConversationAutomationService { this.states.save(conversationId, state); } const incomingKey = messageKey(latest); + const existingClaim = this.claims.getClaim?.(conversationId, incomingKey); + if (existingClaim?.status === 'pending') { + const observedOutbound = messages.some((message) => message.isSelf && message.timestamp > latest.timestamp); + if (observedOutbound) { + this.claims.markSent(conversationId, incomingKey); + this.logger.info(`[automation] Reconciled interrupted reply for u/${conversation.username} from Matrix history`); + return null; + } + if (!existingClaim.replyText) { + this.claims.releasePending?.(conversationId, incomingKey); + return this.finish( + conversation, + conversationId, + state, + incomingKey, + 'human-required', + 'An interrupted reply has no recoverable message body.', + ); + } + if (this.now() - existingClaim.claimedAt < this.pendingReplyRetryDelayMs) return null; + const retryPlan = replyPlan(state); + const exhausted = this.advanceStateForReply( + state, + conversationId, + incomingKey, + existingClaim.replyText, + retryPlan.isLegacy, + retryPlan.isHybridMediaAsk, + ); + return this.sendClaimedReply( + conversation, + state, + incomingKey, + existingClaim.replyText, + latest.timestamp, + exhausted, + signal, + ); + } + if (existingClaim?.status === 'sent') return null; if (state.automation.lastIncomingKey === incomingKey) return null; if (state.automation.status !== 'active') return null; @@ -159,11 +267,7 @@ export class ConversationAutomationService { ); } - const isLegacy = state.automation.policyMode === 'legacy'; - const isHybrid = state.automation.policyMode === 'r4r-media'; - const isHybridMediaAsk = isHybrid - && !state.askedForMedia - && state.automation.turnsUsed >= state.automation.rapportTurnsBeforeMediaAsk; + const { isLegacy, isHybridMediaAsk } = replyPlan(state); if (isHybridMediaAsk) { state.automation.hybridPhase = 'media-ask'; this.states.save(conversationId, state); @@ -194,16 +298,84 @@ export class ConversationAutomationService { return null; } - if (!this.claims.tryClaim(conversationId, incomingKey)) return null; + if (!this.claims.tryClaim(conversationId, incomingKey, reply, latest.timestamp)) return null; // Persist the full intended state before the remote send. If the process // dies after Matrix accepts the message, both this state and the durable // event claim make the retry fail closed instead of sending twice. - if (isLegacy || isHybridMediaAsk) state.askedForMedia = true; - if (isHybridMediaAsk) state.automation.hybridPhase = 'post-media-ask'; - state.automation.turnsUsed += 1; - state.automation.lastIncomingKey = incomingKey; - state.automation.aiMessageFingerprints.push(conversationMessageFingerprint(reply)); + const exhausted = this.advanceStateForReply( + state, + conversationId, + incomingKey, + reply, + isLegacy, + isHybridMediaAsk, + ); + return this.sendClaimedReply( + conversation, + state, + incomingKey, + reply, + latest.timestamp, + exhausted, + signal, + ); + } + + private actionableConversations(conversations: ActiveChat[]): ActiveChat[] { + return conversations + .filter((conversation) => this.isActionableConversation(conversation)) + .sort((left, right) => { + const leftAt = left.messages?.at(-1)?.timestamp ?? left.lastMessageTime; + const rightAt = right.messages?.at(-1)?.timestamp ?? right.lastMessageTime; + return leftAt - rightAt; + }); + } + + private reconcilePendingClaims(conversations: ActiveChat[]): void { + for (const claim of this.claims.listPending?.() ?? []) { + const conversation = conversations.find((item) => item.roomId === claim.conversationId); + if (!conversation) continue; + const messages = conversation.messages ?? []; + const incomingIndex = messages.findIndex((message) => messageKey(message) === claim.incomingKey); + if (incomingIndex < 0) continue; + if (!messages.slice(incomingIndex + 1).some((message) => message.isSelf)) continue; + if (this.claims.markSent(claim.conversationId, claim.incomingKey)) { + this.logger.info(`[automation] Reconciled interrupted reply for u/${conversation.username} from Matrix history`); + } + } + } + + private isActionableConversation(conversation: ActiveChat): boolean { + if (!conversation.unread || conversation.isFromUs || !conversation.roomId || !conversation.url) return false; + const latest = conversation.messages?.at(-1); + if (!latest || latest.isSelf) return false; + const opener = this.openers.getByChatRoomId(conversation.roomId); + if (!opener || latest.timestamp <= opener.sentAt) return false; + const incomingKey = messageKey(latest); + const claim = this.claims.getClaim?.(conversation.roomId, incomingKey); + if (claim?.status === 'pending') return true; + if (claim?.status === 'sent') return false; + const state = this.states.get(conversation.roomId); + if (!state) return true; + return state.automation.status === 'active' && state.automation.lastIncomingKey !== incomingKey; + } + + private advanceStateForReply( + state: ConversationState, + conversationId: string, + incomingKey: string, + reply: string, + isLegacy: boolean, + isHybridMediaAsk: boolean, + ): boolean { + if (state.automation.lastIncomingKey !== incomingKey) { + if (isLegacy || isHybridMediaAsk) state.askedForMedia = true; + if (isHybridMediaAsk) state.automation.hybridPhase = 'post-media-ask'; + state.automation.turnsUsed += 1; + state.automation.lastIncomingKey = incomingKey; + state.automation.aiMessageFingerprints.push(conversationMessageFingerprint(reply)); + } // Keep the hybrid conversation active after its one media ask even when // that send consumes the final turn. The next inbound event still needs // to be classified as an image, a decline, or a normal turn-limit handoff. @@ -214,19 +386,36 @@ export class ConversationAutomationService { state.automation.handoffReason = `${state.automation.maxAutomatedTurns} automated turns have been used.`; } this.states.save(conversationId, state); + return exhausted; + } + private async sendClaimedReply( + conversation: ActiveChat, + state: ConversationState, + incomingKey: string, + reply: string, + incomingAt: number, + exhausted: boolean, + signal?: AbortSignal, + ): Promise { + this.claims.markAttempt?.(conversation.roomId!, incomingKey); const sent = await this.rateLimiter.throttle( () => this.chat.sendChatMessage(conversation.url!, reply), (success) => success, signal, + REPLY_SEND_PRIORITY, ); if (!sent) return null; - this.claims.markSent(conversationId, incomingKey); + const sentAt = this.now(); + this.claims.markSent(conversation.roomId!, incomingKey); + this.lastReplySentAt = sentAt; + this.lastReplyLatencyMs = Math.max(0, sentAt - incomingAt); this.logger.info( - `[automation] Sent ${state.automation.policyMode} turn ${state.automation.turnsUsed}/${state.automation.maxAutomatedTurns} to u/${conversation.username}`, + `[automation] Sent ${state.automation.policyMode} turn ${state.automation.turnsUsed}/${state.automation.maxAutomatedTurns} ` + + `to u/${conversation.username}; reply latency ${Math.round(this.lastReplyLatencyMs / 1_000)}s`, ); return { - conversationId, + conversationId: conversation.roomId!, username: conversation.username, kind: exhausted ? 'human-required' : 'sent', remainingTurns: remainingAutomationTurns(state), @@ -273,6 +462,14 @@ function messageKey(message: ChatMessage): string { return message.eventId ?? `${message.timestamp}:${message.sender}:${message.text}`; } +function replyPlan(state: ConversationState): { isLegacy: boolean; isHybridMediaAsk: boolean } { + const isLegacy = state.automation.policyMode === 'legacy'; + const isHybridMediaAsk = state.automation.policyMode === 'r4r-media' + && !state.askedForMedia + && state.automation.turnsUsed >= state.automation.rapportTurnsBeforeMediaAsk; + return { isLegacy, isHybridMediaAsk }; +} + function r4rSafetyHandoffReason(message: string): string | null { if (/\b(?:i(?:'m| am)|im|age(?:d)?|turning)\s*(?:1[0-7]|[0-9])\b/i.test(message) || /\b(?:1[0-7]|[0-9])\s*[fm]\b/i.test(message) diff --git a/src/services/reddit-account-runtime-builder.ts b/src/services/reddit-account-runtime-builder.ts index 260c1bd..20112ec 100644 --- a/src/services/reddit-account-runtime-builder.ts +++ b/src/services/reddit-account-runtime-builder.ts @@ -265,8 +265,10 @@ class AccountRuntimeLoop { } private async run(signal: AbortSignal): Promise { - let nextResponseCheckAt = 0; - const responseInterval = Math.max(this.config.reddit.pollIntervalMs * 5, 60_000); + let nextReplyCheckAt = 0; + let nextPerformanceCheckAt = 0; + const replyInterval = 30_000; + const performanceInterval = Math.max(this.config.reddit.pollIntervalMs * 5, 60_000); while (!signal.aborted) { try { if (!this.isBotRunning()) { @@ -278,9 +280,8 @@ class AccountRuntimeLoop { await delay(1_000, signal); continue; } - if (Date.now() >= nextResponseCheckAt) { - nextResponseCheckAt = Date.now() + responseInterval; - await this.resources.responseChecker.checkForResponses(); + if (Date.now() >= nextReplyCheckAt) { + nextReplyCheckAt = Date.now() + replyInterval; const automationResults = await this.resources.conversationAutomation.processUnreadConversations(signal); for (const result of automationResults) { if (result.kind === 'sent') continue; @@ -296,7 +297,11 @@ class AccountRuntimeLoop { }); } } - await delay(this.config.reddit.pollIntervalMs, signal); + if (Date.now() >= nextPerformanceCheckAt) { + nextPerformanceCheckAt = Date.now() + performanceInterval; + await this.resources.responseChecker.checkForResponses(); + } + await delay(5_000, signal); } catch (error) { if (signal.aborted) return; if (error instanceof RedditRateLimitError) { diff --git a/src/services/reddit-account-runtime.ts b/src/services/reddit-account-runtime.ts index dbf6678..572174f 100644 --- a/src/services/reddit-account-runtime.ts +++ b/src/services/reddit-account-runtime.ts @@ -189,6 +189,7 @@ export class RedditAccountRuntime { if (this.state !== 'running') return false; if (!this.resources.accountAuth.isFullyAuthenticated() && !(await this.resources.accountAuth.ensureFullyAuthenticated())) return false; + if (await this.resources.conversationAutomation.hasActionableReplies()) return false; if (this.resources.chatService.getInvitationCooldown()) return false; return this.resources.rateLimiter.getTimers().nextMessageAvailableAt === null; } diff --git a/src/utils/rate-limiter.ts b/src/utils/rate-limiter.ts index 4b9cd0b..bf05d27 100644 --- a/src/utils/rate-limiter.ts +++ b/src/utils/rate-limiter.ts @@ -54,6 +54,7 @@ export class RateLimiter { fn: () => Promise, shouldCount: (result: T) => boolean = () => true, signal?: AbortSignal, + priority = 0, ): Promise { return this.queue.add(async () => { await this.initialize(); @@ -92,7 +93,7 @@ export class RateLimiter { } return result; - }) as Promise; + }, { priority }) as Promise; } private async loadHistory(provider: SentMessageHistoryProvider): Promise { diff --git a/src/web/agent-api-contracts.ts b/src/web/agent-api-contracts.ts index e7eb278..f395265 100644 --- a/src/web/agent-api-contracts.ts +++ b/src/web/agent-api-contracts.ts @@ -26,6 +26,14 @@ export const AgentAccountSchema = z.strictObject({ redditUsername: z.string(), runtimeState: z.string(), authenticated: z.boolean(), + replyPriority: z.strictObject({ + processing: z.boolean(), + actionableBacklog: z.number().int().nonnegative(), + oldestPendingAt: z.number().nullable(), + lastScanAt: z.number().nullable(), + lastReplySentAt: z.number().nullable(), + lastReplyLatencyMs: z.number().nonnegative().nullable(), + }).nullable(), failure: z.string().nullable(), updatedAt: z.number(), }); diff --git a/src/web/routes/agent.ts b/src/web/routes/agent.ts index 42b2922..627fbb4 100644 --- a/src/web/routes/agent.ts +++ b/src/web/routes/agent.ts @@ -121,6 +121,7 @@ function buildOverview(ctx: WebContext) { redditUsername: account.redditUsername, runtimeState: runtime?.state ?? 'stopped', authenticated: live?.resources.accountAuth.isFullyAuthenticated() ?? false, + replyPriority: live?.resources.conversationAutomation.getPriorityStatus() ?? null, failure: runtime?.failure ?? null, updatedAt: account.updatedAt, }; diff --git a/tests/conversation-automation.test.ts b/tests/conversation-automation.test.ts index fd82c0e..faaf3a4 100644 --- a/tests/conversation-automation.test.ts +++ b/tests/conversation-automation.test.ts @@ -12,7 +12,8 @@ function harness( rapportTurnsBeforeMediaAsk?: number; }, ) { - let saved: ConversationState | undefined; + const saved = new Map(); + let lastSaved: ConversationState | undefined; const chat = { refreshTopChats: vi.fn().mockResolvedValue(undefined), getCachedConversations: vi.fn().mockResolvedValue([]), @@ -24,8 +25,11 @@ function harness( generateReply: vi.fn().mockResolvedValue('That sounds fun. What got you into it?'), }; const states = { - get: vi.fn(() => saved), - save: vi.fn((_id: string, state: ConversationState) => { saved = structuredClone(state); }), + get: vi.fn((id: string) => saved.get(id)), + save: vi.fn((id: string, state: ConversationState) => { + lastSaved = structuredClone(state); + saved.set(id, lastSaved); + }), }; const openers = { getByChatRoomId: vi.fn(() => openerSentAt === null @@ -34,6 +38,10 @@ function harness( }; const claims = { tryClaim: vi.fn(() => true), + getClaim: vi.fn(() => undefined), + listPending: vi.fn(() => []), + markAttempt: vi.fn(() => true), + releasePending: vi.fn(() => true), markSent: vi.fn(() => true), }; const rateLimiter = { @@ -49,8 +57,10 @@ function harness( rapportTurnsBeforeMediaAsk: subredditConversation.rapportTurnsBeforeMediaAsk ?? 3, }, }] : [], + () => 1_000, + 0, ); - return { service, chat, ai, states, openers, claims, getSaved: () => saved }; + return { service, chat, ai, states, openers, claims, rateLimiter, getSaved: () => lastSaved }; } function incoming(turn: number, text = `reply ${turn}`, media?: Array<{ originalUrl: string; localPath: string; fileName: string }>) { @@ -151,10 +161,118 @@ describe('conversation automation', () => { await service.processConversation(incoming(1)); - expect(claims.tryClaim).toHaveBeenCalledWith('room-alice', 'in-1'); + expect(claims.tryClaim).toHaveBeenCalledWith( + 'room-alice', + 'in-1', + 'Would you be comfortable sharing a selfie?', + 3, + ); expect(claims.tryClaim.mock.invocationCallOrder[0]) .toBeLessThan(chat.sendChatMessage.mock.invocationCallOrder[0]); expect(claims.markSent).toHaveBeenCalledWith('room-alice', 'in-1'); + expect(claims.markAttempt).toHaveBeenCalledWith('room-alice', 'in-1'); + }); + + it('refreshes a wider reply window and drains actionable conversations oldest first', async () => { + const { service, chat } = harness(); + const newer = { ...incoming(2), username: 'newer', roomId: 'room-newer', url: '/room/newer' }; + const older = { ...incoming(1), username: 'older', roomId: 'room-older', url: '/room/older' }; + chat.getCachedConversations.mockResolvedValue([newer, older]); + + await service.processUnreadConversations(); + + expect(chat.refreshTopChats).toHaveBeenCalledWith(25); + expect(chat.sendChatMessage.mock.calls.map(([url]) => url)).toEqual(['/room/older', '/room/newer']); + }); + + it('blocks outreach while an actionable reply exists and clears after processing', async () => { + const { service, chat } = harness(); + chat.getCachedConversations.mockResolvedValue([incoming(1)]); + + expect(await service.hasActionableReplies()).toBe(true); + await service.processUnreadConversations(); + + expect(await service.hasActionableReplies()).toBe(false); + expect(service.getPriorityStatus()).toMatchObject({ + processing: false, + actionableBacklog: 0, + lastScanAt: 1_000, + lastReplySentAt: 1_000, + lastReplyLatencyMs: 997, + }); + }); + + it('submits replies to the rate limiter at high priority', async () => { + const { service, rateLimiter } = harness(); + + await service.processConversation(incoming(1)); + + expect(rateLimiter.throttle).toHaveBeenCalledWith( + expect.any(Function), + expect.any(Function), + undefined, + 100, + ); + }); + + it('retries the same durable reply after an interrupted send', async () => { + const { service, chat, ai, claims } = harness(); + chat.sendChatMessage.mockRejectedValueOnce(new Error('interrupted')); + await expect(service.processConversation(incoming(1))).rejects.toThrow('interrupted'); + claims.getClaim.mockReturnValue({ + conversationId: 'room-alice', incomingKey: 'in-1', status: 'pending', + replyText: 'Would you be comfortable sharing a selfie?', incomingAt: 3, + claimedAt: 0, lastAttemptAt: 10, attemptCount: 1, sentAt: null, + } as never); + chat.sendChatMessage.mockResolvedValue(true); + + const result = await service.processConversation(incoming(1)); + + expect(result?.kind).toBe('sent'); + expect(ai.generateMediaAsk).toHaveBeenCalledTimes(1); + expect(chat.sendChatMessage).toHaveBeenCalledTimes(2); + expect(claims.markSent).toHaveBeenCalledWith('room-alice', 'in-1'); + }); + + it('hands legacy pending claims without a stored reply to a human without blocking outreach forever', async () => { + const { service, chat, claims, getSaved } = harness(); + claims.getClaim.mockReturnValue({ + conversationId: 'room-alice', incomingKey: 'in-1', status: 'pending', + replyText: null, incomingAt: null, claimedAt: 0, + lastAttemptAt: null, attemptCount: 0, sentAt: null, + } as never); + chat.getCachedConversations.mockResolvedValue([incoming(1)]); + + const result = await service.processConversation(incoming(1)); + + expect(result?.kind).toBe('human-required'); + expect(claims.releasePending).toHaveBeenCalledWith('room-alice', 'in-1'); + expect(getSaved()?.automation.handoffReason).toContain('no recoverable message body'); + claims.getClaim.mockReturnValue(undefined); + expect(await service.hasActionableReplies()).toBe(false); + }); + + it('reconciles an interrupted claim when Matrix already contains the reply', async () => { + const { service, chat, claims } = harness(); + const conversation = { + ...incoming(1), unread: false, isFromUs: true, lastMessageTime: 4, + messages: [ + ...incoming(1).messages, + { eventId: 'out-reply', sender: 'me', text: 'reply', timestamp: 4, isSelf: true }, + ], + }; + const pending = { + conversationId: 'room-alice', incomingKey: 'in-1', status: 'pending' as const, + replyText: 'Would you be comfortable sharing a selfie?', incomingAt: 3, + claimedAt: 0, lastAttemptAt: 10, attemptCount: 1, sentAt: null, + }; + claims.listPending.mockReturnValue([pending] as never); + chat.getCachedConversations.mockResolvedValue([conversation]); + + await service.processUnreadConversations(); + + expect(claims.markSent).toHaveBeenCalledWith('room-alice', 'in-1'); + expect(chat.sendChatMessage).not.toHaveBeenCalled(); }); it('does not send when another worker already claimed the inbound event', async () => { diff --git a/tests/database.test.ts b/tests/database.test.ts index 4ae9141..a4fe749 100644 --- a/tests/database.test.ts +++ b/tests/database.test.ts @@ -160,7 +160,7 @@ describe('Database Client', () => { DROP TABLE conversation_automation_claims; DROP TABLE ai_usage; `); - db.pragma(`user_version = ${latestVersion - 4}`); + db.pragma(`user_version = ${latestVersion - 5}`); runMigrations(db); expect(db.prepare('SELECT recipient, reason FROM recipient_suppressions ORDER BY recipient COLLATE NOCASE').all()) diff --git a/tests/rate-limiter.test.ts b/tests/rate-limiter.test.ts index 3216e4d..446e35e 100644 --- a/tests/rate-limiter.test.ts +++ b/tests/rate-limiter.test.ts @@ -105,6 +105,25 @@ describe('RateLimiter', () => { expect(order).toEqual([1, 2, 3]); }); + it('runs a queued high-priority reply before queued outreach', async () => { + const order: string[] = []; + let release!: () => void; + const blocker = new Promise((resolve) => { release = resolve; }); + const active = limiter.throttle(async () => { + order.push('active'); + await blocker; + return 'active'; + }); + await vi.waitFor(() => expect(order).toEqual(['active'])); + + const outreach = limiter.throttle(async () => { order.push('outreach'); return 'outreach'; }); + const reply = limiter.throttle(async () => { order.push('reply'); return 'reply'; }, () => true, undefined, 100); + release(); + await Promise.all([active, outreach, reply]); + + expect(order).toEqual(['active', 'reply', 'outreach']); + }); + describe('getStats', () => { it('returns initial stats', () => { const stats = limiter.getStats(); diff --git a/tests/reddit-account-runtime.test.ts b/tests/reddit-account-runtime.test.ts index 5e28e04..bb82095 100644 --- a/tests/reddit-account-runtime.test.ts +++ b/tests/reddit-account-runtime.test.ts @@ -220,6 +220,36 @@ describe('RedditAccountRuntimeManager', () => { }); describe('account runtime isolation', () => { + it('does not accept new outreach while the account has actionable replies', async () => { + const stored = account('reply_first'); + const hasActionableReplies = vi.fn().mockResolvedValue(true); + const runtime = new RedditAccountRuntime( + { + accountId: stored.accountId, + redditUsername: stored.redditUsername, + matrixUserId: stored.matrixUserId!, + cookies: [], + matrixCredentials: JSON.parse(stored.matrixCredentialsJson), + updatedAt: stored.updatedAt, + }, + { + accountAuth: { + isFullyAuthenticated: () => true, + ensureFullyAuthenticated: async () => true, + }, + conversationAutomation: { hasActionableReplies }, + chatService: { getInvitationCooldown: () => null }, + rateLimiter: { getTimers: () => ({ nextMessageAvailableAt: null }) }, + } as unknown as RedditAccountRuntimeResources, + { start: vi.fn(), stop: vi.fn() }, + ); + await runtime.start(); + + expect(await runtime.canAcceptPost()).toBe(false); + hasActionableReplies.mockResolvedValue(false); + expect(await runtime.canAcceptPost()).toBe(true); + }); + it('uses the Reddit fullname for auth while retaining the Matrix runtime owner', () => { expect(runtimeAuthenticatedAccount( { matrixUserId: '@t2_1u57iqxxbv:reddit.com' }, diff --git a/tests/repository-account-isolation.test.ts b/tests/repository-account-isolation.test.ts index 2c17dfb..28f40fa 100644 --- a/tests/repository-account-isolation.test.ts +++ b/tests/repository-account-isolation.test.ts @@ -113,6 +113,21 @@ describe('account-bound domain repositories', () => { expect(claimsA.markSent('same-room', 'same-event', 14)).toBe(false); }); + it('persists reply bodies and attempts so interrupted automation can resume', () => { + const claims = new ConversationAutomationClaimRepository(db, ACCOUNT_A); + + expect(claims.tryClaim('room', 'event', 'durable reply', 50, 60)).toBe(true); + expect(claims.markAttempt('room', 'event', 70)).toBe(true); + expect(claims.getClaim('room', 'event')).toEqual({ + conversationId: 'room', incomingKey: 'event', status: 'pending', + replyText: 'durable reply', incomingAt: 50, claimedAt: 60, + lastAttemptAt: 70, attemptCount: 1, sentAt: null, + }); + expect(claims.listPending()).toEqual([expect.objectContaining({ incomingKey: 'event' })]); + expect(claims.releasePending('room', 'event')).toBe(true); + expect(claims.listPending()).toEqual([]); + }); + it('claims recipients globally and blocks usernames with existing chat history', () => { db.prepare(` INSERT INTO reddit_conversations (