diff --git a/docs/architecture.md b/docs/architecture.md index 59558f6..0d99403 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -68,6 +68,7 @@ SIGINT/SIGTERM abort shared loops, stop account runtimes, stop the web server, c - posts, messages, outcomes, activity, outbox, and conversation state repositories; - growth settings, community list, comment history, adaptive comment allowance, and isolated browsing policy/audit with optional submitted-vote state; - Matrix chat cache and sync cursor; +- durable account suspension state; a Matrix `M_FORBIDDEN` or the periodic owned-profile check pauses only the affected runtime; - media index and files; - rate limiter and invitation cooldown; - response checker and conversation automation; diff --git a/docs/data-flow.md b/docs/data-flow.md index 3055be3..22feddd 100644 --- a/docs/data-flow.md +++ b/docs/data-flow.md @@ -46,6 +46,8 @@ An account is eligible for new outreach only when its runtime is running, both i Matrix cooldowns retain the operation that triggered them. A room-creation or invitation limit pauses new outreach but does not suppress replies in existing rooms; a limit returned while sending into an existing room pauses those sends as well. +An account-level Matrix `M_FORBIDDEN` is distinct from a room-level 403: it is treated as a definitive suspension signal. The account is durably marked suspended, its runtime is halted, and its historical content remains readable. Independently, every active account periodically reads its own Reddit profile; an explicit suspended profile result takes the same path. Missing profiles, network errors, and rate limits are not suspension evidence and do not pause an account. + The operator can manually clear the unsent funnel from the authenticated dashboard. The repository atomically moves discovered, screening, ready, and retryable candidates to a terminal rejection with the reason `Cleared manually by operator`, releases their leases, and retains the audit rows so those candidates are not rediscovered. A `sending` lease is never changed because its remote Matrix outcome may already be in flight. Clearing does not change bot mode; running discovery may refill the funnel immediately. The same surface supports individual cancellation and prepared-message editing. Individual cancellation uses the same terminal audit transition and refuses `sending` or completed rows. Editing is limited to `ready` and `retryable` candidates with a preparation snapshot; it atomically replaces only the opener, clears generated strategy/experiment selection metadata, labels the model attribution `operator-edited`, and re-runs the deterministic outbound location/meeting safeguard. Both operations return HTTP 409 when a stale dashboard loses a race to dispatch. diff --git a/docs/database.md b/docs/database.md index 50110e6..a457f13 100644 --- a/docs/database.md +++ b/docs/database.md @@ -10,7 +10,7 @@ The application uses one `better-sqlite3` database, normally `data/reddit-bot.db ### Account registry and account content -- `reddit_accounts`: verified Reddit/Matrix identity, encrypted-at-rest only to the degree provided by the host/volume, cookie JSON, Matrix credentials, active/status timestamps. +- `reddit_accounts`: verified Reddit/Matrix identity, encrypted-at-rest only to the degree provided by the host/volume, cookie JSON, Matrix credentials, active/status timestamps. A confirmed suspension sets `status` to `suspended` and disables the account while retaining credentials and historical content for review. - `reddit_conversations`: account-scoped cached Matrix conversation payloads. - `reddit_account_sync_state`: account-scoped sync cursor, full/delta timestamps, and invitation cooldown. - `reddit_recipient_account_status`: process-global, case-insensitive recipient username status (`active`, `suspended`, or `deleted`) and last confirmed check time. diff --git a/docs/services.md b/docs/services.md index 0d68983..3e7ef69 100644 --- a/docs/services.md +++ b/docs/services.md @@ -13,6 +13,7 @@ This is an ownership map, not an API dump. Read the owning implementation and te | `services/reddit-account-runtime-builder.ts` | Concrete account dependency graph and account loop | Account-specific browser, Matrix, repos, limiter, automation, chat, and media live here. | | `services/account-runtime-event-wiring.ts` | Per-account chat-to-realtime event bridge | Every account runtime publishes reply, media, and sync events into the shared dashboard/push stream; sync IDs remain account-scoped. | | `services/reddit-accounts.ts` | Account import/activation orchestration | Verification must bind Reddit cookies and Matrix identity before activation. | +| `services/account-suspension.ts` | Owned-account suspension monitor | A definitive Matrix `M_FORBIDDEN` or Reddit suspended-profile result durably pauses only that account. | | `services/account-auth-context.ts` | Current verified identity snapshot | Account work pauses unless both sides authenticate. | | `services/session-bound-runtime.ts` | Legacy auth-bound background lifecycle | Prevents background work while logged out. | diff --git a/packages/api-contracts/src/attention.ts b/packages/api-contracts/src/attention.ts index ff72f24..123facb 100644 --- a/packages/api-contracts/src/attention.ts +++ b/packages/api-contracts/src/attention.ts @@ -6,6 +6,7 @@ export const AttentionKindSchema = z.enum([ 'boundary', 'failed-delivery', 'account-login-required', + 'account-suspended', 'runtime-failed', 'invitation-cooldown', 'account-unestablished', diff --git a/packages/api-contracts/src/session.ts b/packages/api-contracts/src/session.ts index 7f28692..dcdb133 100644 --- a/packages/api-contracts/src/session.ts +++ b/packages/api-contracts/src/session.ts @@ -31,7 +31,9 @@ export const RedditAccountSummarySchema = z.strictObject({ redditUsername: z.string().min(1), matrixUserId: z.string().min(1), isRunning: z.boolean(), - status: z.union([z.literal('authenticated'), z.literal('login-required'), z.literal('removed')]), + status: z.union([ + z.literal('authenticated'), z.literal('login-required'), z.literal('suspended'), z.literal('removed'), + ]), createdAt: z.number(), updatedAt: z.number(), lastImportedAt: z.number(), diff --git a/src/database/account-repositories.ts b/src/database/account-repositories.ts index b070d8a..a0c4f01 100644 --- a/src/database/account-repositories.ts +++ b/src/database/account-repositories.ts @@ -2,6 +2,8 @@ import type Database from 'better-sqlite3'; export type RedditAccountStatus = | 'verified' + /** Reddit or Matrix confirmed that this account has been suspended. */ + | 'suspended' | 'login-required' | 'disabled' | 'legacy-unassigned' @@ -161,6 +163,19 @@ export class RedditAccountsRepository { `).run(now, accountId).changes === 1; } + /** + * Disable an account only after Reddit or Matrix has returned a definitive + * suspension signal. Credentials and historical content are retained for + * operator review and a later verified session import. + */ + markSuspended(accountId: string, now: number = Date.now()): boolean { + return this.db.prepare(` + UPDATE reddit_accounts + SET status = 'suspended', is_active = 0, updated_at = ? + WHERE account_id = ? AND status NOT IN ('legacy-unassigned', 'removed') + `).run(now, accountId).changes === 1; + } + private getRequired(accountId: string): RedditAccountPrivateRecord { const account = this.get(accountId); if (!account) throw new Error(`Reddit account ${accountId} was not persisted`); diff --git a/src/database/client.ts b/src/database/client.ts index 1601ece..7e079fe 100644 --- a/src/database/client.ts +++ b/src/database/client.ts @@ -6,7 +6,7 @@ import fs from 'fs'; export const LEGACY_UNASSIGNED_ACCOUNT_ID = '__legacy_unassigned__'; /** Migrations that replace a table other tables reference by foreign key. */ -const FOREIGN_KEY_REBUILD_VERSIONS = new Set([11, 32, 34]); +const FOREIGN_KEY_REBUILD_VERSIONS = new Set([11, 32, 34, 35]); export function createDatabase(config: Pick): Database.Database { const dbPath = config.database.path === ':memory:' @@ -873,6 +873,39 @@ export function runMigrations(db: Database.Database): void { CREATE INDEX idx_learning_trajectories_account ON conversation_learning_trajectories(account_id, conversation_id, updated_at DESC); `), + // A Matrix M_FORBIDDEN or owned-profile suspension must be a durable + // account state, rather than a transient runtime failure. + () => db.exec(` + CREATE TABLE reddit_accounts_with_suspension ( + account_id TEXT PRIMARY KEY, + reddit_username TEXT NOT NULL, + matrix_user_id TEXT UNIQUE, + status TEXT NOT NULL CHECK(status IN ( + 'verified', 'login-required', 'disabled', 'suspended', 'legacy-unassigned', 'removed' + )), + is_active INTEGER NOT NULL DEFAULT 1 CHECK(is_active IN (0, 1)), + cookies_json TEXT NOT NULL, + matrix_credentials_json TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + last_verified_at INTEGER, + last_imported_at INTEGER, + removed_at INTEGER + ); + INSERT INTO reddit_accounts_with_suspension ( + account_id, reddit_username, matrix_user_id, status, is_active, + cookies_json, matrix_credentials_json, created_at, updated_at, + last_verified_at, last_imported_at, removed_at + ) SELECT + account_id, reddit_username, matrix_user_id, status, is_active, + cookies_json, matrix_credentials_json, created_at, updated_at, + last_verified_at, last_imported_at, removed_at + FROM reddit_accounts; + DROP TABLE reddit_accounts; + ALTER TABLE reddit_accounts_with_suspension RENAME TO reddit_accounts; + 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); + `), ]; let currentVersion = db.pragma('user_version', { simple: true }) as number; @@ -884,7 +917,7 @@ export function runMigrations(db: Database.Database): void { const version = index + 1; // SQLite cannot rebuild a referenced table while FK enforcement is on. // Migration 11 replaces the account-produced table graph atomically and - // migration 32 rebuilds `reddit_accounts`, and migration 34 expands the + // migrations 32 and 35 rebuild `reddit_accounts`, and migration 34 expands the // referenced learning-trajectory policy constraint; all validate the graph // before foreign-key enforcement is restored. const rebuildsForeignKeyGraph = FOREIGN_KEY_REBUILD_VERSIONS.has(version); diff --git a/src/services/account-suspension.ts b/src/services/account-suspension.ts new file mode 100644 index 0000000..cb73667 --- /dev/null +++ b/src/services/account-suspension.ts @@ -0,0 +1,73 @@ +import type { Logger } from '../utils/logger.js'; +import { RedditRateLimitError, type RedditAccountProfile } from '../reddit/api.js'; +import { denyAccountAuth, type AccountAuthGate } from './account-auth-context.js'; + +export const ACCOUNT_SUSPENSION_CHECK_INTERVAL_MS = 6 * 60 * 60_000; +const INITIAL_CHECK_DELAY_MS = 30_000; + +export interface AccountSuspensionProfileLookup { + getAccountProfile(username: string): Promise; +} + +/** + * Independently verifies that an owned Reddit account is still usable. This + * is deliberately separate from establishment checks: disabling optional + * establishment policy must never disable suspension protection. + */ +export class AccountSuspensionService { + private intervalHandle: NodeJS.Timeout | null = null; + private initialHandle: NodeJS.Timeout | null = null; + private checking = false; + + constructor( + private readonly redditUsername: string, + private readonly reddit: AccountSuspensionProfileLookup, + private readonly pauseAccount: (reason: string) => void, + private readonly logger: Logger, + private readonly accountAuth: AccountAuthGate = denyAccountAuth, + private readonly intervalMs = ACCOUNT_SUSPENSION_CHECK_INTERVAL_MS, + ) {} + + start(): void { + if (this.intervalHandle) return; + this.initialHandle = setTimeout(() => { + this.initialHandle = null; + void this.checkNow(); + }, Math.min(INITIAL_CHECK_DELAY_MS, this.intervalMs)); + this.intervalHandle = setInterval(() => void this.checkNow(), this.intervalMs); + this.logger.info( + `Account suspension check for u/${this.redditUsername}:` + + ` scheduled every ${Math.round(this.intervalMs / 3_600_000)} hours`, + ); + } + + stop(): void { + if (this.initialHandle) clearTimeout(this.initialHandle); + if (this.intervalHandle) clearInterval(this.intervalHandle); + this.initialHandle = null; + this.intervalHandle = null; + } + + async checkNow(): Promise { + if (this.checking || !this.accountAuth.isFullyAuthenticated()) return false; + this.checking = true; + try { + const profile = await this.reddit.getAccountProfile(this.redditUsername); + if (profile?.isSuspended !== true) return false; + const reason = 'Reddit profile marks this account as suspended'; + this.logger.error(`u/${this.redditUsername} is suspended; pausing the account`); + this.pauseAccount(reason); + return true; + } catch (error) { + if (error instanceof RedditRateLimitError) { + this.logger.warn(`Account suspension check paused for u/${this.redditUsername}: Reddit rate-limited profile lookups`); + return false; + } + const message = error instanceof Error ? error.message : String(error); + this.logger.warn(`Account suspension check failed for u/${this.redditUsername}: ${message}`); + return false; + } finally { + this.checking = false; + } + } +} diff --git a/src/services/chat/index.ts b/src/services/chat/index.ts index a97ed0d..e19450c 100644 --- a/src/services/chat/index.ts +++ b/src/services/chat/index.ts @@ -19,7 +19,7 @@ export type { ScrapedChatMessage, ScrapedRoom, } from './types.js'; -export { MatrixClient, MatrixAuthError } from './matrix/client.js'; +export { MatrixClient, MatrixAuthError, MatrixAccountSuspendedError } from './matrix/client.js'; export type { MatrixCredentials } from './matrix/types.js'; export { loadCachedCredentials, saveCredentials } from './matrix/credential-store.js'; export { diff --git a/src/services/chat/matrix/client.ts b/src/services/chat/matrix/client.ts index f2747c9..6d0a96f 100644 --- a/src/services/chat/matrix/client.ts +++ b/src/services/chat/matrix/client.ts @@ -14,6 +14,14 @@ export class MatrixAuthError extends Error { } } +/** Reddit's Matrix proxy uses this for an account that has been suspended. */ +export class MatrixAccountSuspendedError extends Error { + constructor(message: string) { + super(message); + this.name = 'MatrixAccountSuspendedError'; + } +} + /** * Thrown when Reddit's homeserver explicitly disables a Matrix endpoint. * Callers can remember the missing capability instead of retrying it for @@ -94,6 +102,8 @@ export interface MatrixClientOptions { requestTimeoutMs?: number; /** Test seam: inject a custom fetch (e.g. one that hangs). */ fetchImpl?: typeof fetch; + /** Durable account-level safety hook for a definitive M_FORBIDDEN result. */ + onAccountSuspended?: () => void; } /** @@ -107,6 +117,7 @@ export class MatrixClient { private readonly minRequestIntervalMs: number; private readonly requestTimeoutMs: number; private readonly fetchImpl: typeof fetch; + private readonly onAccountSuspended: (() => void) | undefined; constructor( private creds: MatrixCredentials, @@ -116,6 +127,7 @@ export class MatrixClient { this.minRequestIntervalMs = options.minRequestIntervalMs ?? 30; this.requestTimeoutMs = options.requestTimeoutMs ?? 30_000; this.fetchImpl = options.fetchImpl ?? fetch; + this.onAccountSuspended = options.onAccountSuspended; } /** @@ -208,6 +220,12 @@ export class MatrixClient { if (res.status === 401 || res.status === 403) { const failure = await readMatrixFailure(res); + if (isAccountSuspendedFailure(failure.errcode)) { + this.onAccountSuspended?.(); + throw new MatrixAccountSuspendedError( + `Matrix ${method} ${path} → ${res.status} M_FORBIDDEN (account suspended)`, + ); + } if (isMatrixAuthenticationFailure(res.status, failure.errcode)) { throw new MatrixAuthError( `Matrix ${method} ${path} → ${res.status} ${failure.errcode ?? ''} (token rejected)`, @@ -549,6 +567,10 @@ export class MatrixClient { if (response === null) throw new Error('Matrix media upload timed out'); if (response.status === 401 || response.status === 403) { const failure = await readMatrixFailure(response); + if (isAccountSuspendedFailure(failure.errcode)) { + this.onAccountSuspended?.(); + throw new MatrixAccountSuspendedError('Matrix media upload → M_FORBIDDEN (account suspended)'); + } if (isMatrixAuthenticationFailure(response.status, failure.errcode)) { throw new MatrixAuthError(`Matrix media upload → ${response.status} (token rejected)`); } @@ -667,6 +689,10 @@ export class MatrixClient { } if (res.status === 401 || res.status === 403) { const failure = await readMatrixFailure(res); + if (isAccountSuspendedFailure(failure.errcode)) { + this.onAccountSuspended?.(); + throw new MatrixAccountSuspendedError('Matrix media download → M_FORBIDDEN (account suspended)'); + } if (isMatrixAuthenticationFailure(res.status, failure.errcode)) { throw new MatrixAuthError( `Matrix media download → ${res.status} ${failure.errcode ?? ''} (token rejected)`, @@ -724,6 +750,10 @@ function isMatrixAuthenticationFailure(status: number, errcode?: string): boolea return status === 401 || errcode === 'M_UNKNOWN_TOKEN' || errcode === 'M_MISSING_TOKEN'; } +function isAccountSuspendedFailure(errcode?: string): boolean { + return errcode === 'M_FORBIDDEN'; +} + function isFeatureDisabledFailure(responseText: string): boolean { try { const parsed = JSON.parse(responseText) as { message?: unknown }; diff --git a/src/services/reddit-account-runtime-builder.ts b/src/services/reddit-account-runtime-builder.ts index 92cbb95..fb22db5 100644 --- a/src/services/reddit-account-runtime-builder.ts +++ b/src/services/reddit-account-runtime-builder.ts @@ -70,6 +70,7 @@ 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 { AccountSuspensionService } from './account-suspension.js'; import { allowsConversationSync, allowsSending, type BotMode } from './bot-mode.js'; import { wireAccountChatEvents, @@ -145,8 +146,17 @@ async function buildRuntime( throw new Error(`Stored Reddit cookies no longer verify for u/${session.redditUsername}`); } - const matrixClient = new MatrixClient(session.matrixCredentials, logger); const accountsRepo = new RedditAccountsRepository(db); + let runtime: RedditAccountRuntime | null = null; + const accountAuth = new AccountAuthContext(logger); + const pauseSuspendedAccount = (reason: string): void => { + accountsRepo.markSuspended(session.accountId); + accountAuth?.invalidate(reason); + runtime?.pause(reason); + }; + const matrixClient = new MatrixClient(session.matrixCredentials, logger, { + onAccountSuspended: () => pauseSuspendedAccount('Matrix returned M_FORBIDDEN: account suspended'), + }); const tokenRefresher = new BrowserMatrixTokenRefresher({ context: browserClient.getContext()!, expectedUserId: session.matrixUserId, @@ -172,7 +182,6 @@ async function buildRuntime( throw new Error(`Stored Matrix session no longer verifies for u/${session.redditUsername}`); } - const accountAuth = new AccountAuthContext(logger); accountAuth.markAuthenticated(runtimeAuthenticatedAccount(session, identity)); accountAuth.setAuthenticator(async () => { const current = await browserClient.getAuthenticatedIdentity(); @@ -274,6 +283,14 @@ async function buildRuntime( logger, accountAuth, ); + const suspensionService = new AccountSuspensionService( + session.redditUsername, + redditApi, + pauseSuspendedAccount, + logger, + accountAuth, + accountConfig.accountEstablishment.profileCheckIntervalMs, + ); const responseChecker = new ResponseCheckerService( chatPage, messagesRepo, successfulMessagesRepo, postsRepo, logger, accountAuth, conversationOutcomesRepo, @@ -392,12 +409,13 @@ async function buildRuntime( activityLogRepo, monitorService, processorService, responseChecker, conversationAutomation, chatService, mediaService, mediaSyncService, mediaExchange, rateLimiter, - recipientAccountStatusService, establishmentService, growthService, browsingService, + recipientAccountStatusService, establishmentService, suspensionService, growthService, browsingService, }; const lifecycle = new AccountRuntimeLoop( resources, accountConfig, logger, getBotMode, recipientDeduplicator, ); - return new RedditAccountRuntime(session, resources, lifecycle); + runtime = new RedditAccountRuntime(session, resources, lifecycle); + return runtime; } catch (error) { await browserClient.close(); throw error; @@ -436,6 +454,7 @@ export class AccountRuntimeLoop { this.controller = new AbortController(); this.resources.recipientAccountStatusService.start(); this.resources.establishmentService.start(); + this.resources.suspensionService?.start(); // A comment interrupted mid-post may already be live on Reddit, so those // rows are retired before any new growth work begins. this.resources.growthService?.recoverAfterRestart(); @@ -448,12 +467,22 @@ export class AccountRuntimeLoop { this.resources.mediaSyncService.stop(); this.resources.recipientAccountStatusService.stop(); this.resources.establishmentService.stop(); + this.resources.suspensionService?.stop(); await this.loop?.catch((error) => { if (!(error instanceof DOMException && error.name === 'AbortError')) throw error; }); await this.resources.browserClient.close(); } + pause(_reason: string): void { + this.controller?.abort(); + this.mediaSyncRunning = false; + this.resources.mediaSyncService.stop(); + this.resources.recipientAccountStatusService.stop(); + this.resources.establishmentService.stop(); + this.resources.suspensionService?.stop(); + } + /** * Starts or stops the background media sync only on a mode transition — * `MediaSyncService.start()` re-wraps its own `stop()` on every call, so diff --git a/src/services/reddit-account-runtime.ts b/src/services/reddit-account-runtime.ts index 841ad71..5155009 100644 --- a/src/services/reddit-account-runtime.ts +++ b/src/services/reddit-account-runtime.ts @@ -38,6 +38,7 @@ import type { RecipientAccountStatusService } from './recipient-account-status.j import type { AccountEstablishmentService } from './account-establishment.js'; import type { AccountGrowthService } from './account-growth.js'; import type { AccountBrowsingSimulationService } from './account-browsing-simulation.js'; +import type { AccountSuspensionService } from './account-suspension.js'; import type { ProcessedPost, ProcessResult } from './processor.js'; import type { ConversationEventRepository } from '../database/conversation-event-repository.js'; import type { ConversationLearningRepository } from '../database/conversation-learning-repository.js'; @@ -91,6 +92,8 @@ export interface RedditAccountRuntimeResources { mediaSyncService: MediaSyncService; recipientAccountStatusService: RecipientAccountStatusService; establishmentService: AccountEstablishmentService; + /** Present in concrete runtimes; optional for narrow lifecycle test fixtures. */ + suspensionService?: AccountSuspensionService; /** * Comment-based account growth. Optional because a runtime is fully * functional without it, and it is disabled per account by default. @@ -107,6 +110,8 @@ export interface RedditAccountRuntimeLifecycle { start(): void | Promise; /** Stop and await this account's schedules/loop, then close its browser. */ stop(): void | Promise; + /** Immediately halt account work after a definitive suspension signal. */ + pause?(reason: string): void; /** Process a post assigned by the shared discovery scheduler. */ dispatchPost?(post: RedditPost, signal?: AbortSignal): Promise; preparePost?(post: RedditPost, signal?: AbortSignal): Promise; @@ -123,6 +128,7 @@ export type RedditAccountRuntimeState = | 'starting' | 'running' | 'stopping' + | 'paused' | 'stopped' | 'failed'; @@ -197,6 +203,17 @@ export class RedditAccountRuntime { }); } + /** + * Suspension can be discovered by work that this runtime is currently + * awaiting. Keep this synchronous so it cannot wait on its own loop. + */ + pause(reason: string): void { + if (this.state === 'stopped' || this.state === 'failed') return; + this.failure = reason; + this.state = 'paused'; + this.lifecycle.pause?.(reason); + } + snapshot(): RedditAccountRuntimeSnapshot { return { accountId: this.session.accountId, @@ -294,7 +311,9 @@ export class RedditAccountRuntimeManager { async startEnabled(): Promise { const enabled = this.accounts.listPublic().filter(isEnabledAccount); - await Promise.all(enabled.map((account) => this.start(account.accountId))); + // A broken or suspended session belongs to one account. Do not prevent + // other verified accounts from starting because of it. + await Promise.allSettled(enabled.map((account) => this.start(account.accountId))); return this.snapshot(); } diff --git a/src/services/reddit-accounts.ts b/src/services/reddit-accounts.ts index 4765782..2d642ba 100644 --- a/src/services/reddit-accounts.ts +++ b/src/services/reddit-accounts.ts @@ -51,7 +51,7 @@ export interface RedditAccountSummary { matrixUserId: string; isRunning: boolean; /** `removed` is terminal until a fresh session is imported for the account. */ - status: 'authenticated' | 'login-required' | 'removed'; + status: 'authenticated' | 'login-required' | 'suspended' | 'removed'; createdAt: number; updatedAt: number; lastImportedAt: number; @@ -136,16 +136,20 @@ export class RedditAccountsService { private toSummary(account: RedditAccountPublicRecord): RedditAccountSummary { const runtime = this.runtimes?.get(account.accountId); - const running = account.status === 'removed' ? false : runtime?.snapshot().state === 'running'; + const running = account.status === 'removed' || account.status === 'suspended' + ? false + : runtime?.snapshot().state === 'running'; return { accountId: account.accountId, redditUsername: account.redditUsername, matrixUserId: account.matrixUserId ?? account.accountId, - isRunning: account.status === 'removed' + isRunning: account.status === 'removed' || account.status === 'suspended' ? false : running ?? (account.accountId === this.auth.getSnapshot().accountId), status: account.status === 'removed' ? 'removed' + : account.status === 'suspended' + ? 'suspended' : account.status === 'verified' && (running ?? account.accountId === this.auth.getSnapshot().accountId) ? 'authenticated' : 'login-required', diff --git a/src/web/routes/attention.ts b/src/web/routes/attention.ts index 5385411..7202a77 100644 --- a/src/web/routes/attention.ts +++ b/src/web/routes/attention.ts @@ -3,7 +3,7 @@ import type { WebContext } from '../context.js'; interface AttentionItem { id: string; - kind: 'human-required' | 'boundary' | 'failed-delivery' | 'account-login-required' | 'runtime-failed' | 'invitation-cooldown' | 'account-unestablished'; + kind: 'human-required' | 'boundary' | 'failed-delivery' | 'account-login-required' | 'account-suspended' | 'runtime-failed' | 'invitation-cooldown' | 'account-unestablished'; severity: 'warning' | 'critical'; title: string; detail: string; @@ -38,16 +38,28 @@ export function createAttentionRouter(ctx: WebContext): Router { owner, }); } - if (account.status !== 'login-required') continue; - items.push({ - id: `account:${account.accountId}`, - kind: 'account-login-required', - severity: 'critical', - title: `u/${account.redditUsername} needs a new session`, - detail: 'Import a fresh extension session before this account can resume.', - occurredAt: account.updatedAt, - owner, - }); + if (account.status === 'login-required') { + items.push({ + id: `account:${account.accountId}`, + kind: 'account-login-required', + severity: 'critical', + title: `u/${account.redditUsername} needs a new session`, + detail: 'Import a fresh extension session before this account can resume.', + occurredAt: account.updatedAt, + owner, + }); + } + if (account.status === 'suspended') { + items.push({ + id: `account:${account.accountId}`, + kind: 'account-suspended', + severity: 'critical', + title: `u/${account.redditUsername} is suspended`, + detail: 'Reddit or Matrix confirmed suspension, so this account has been paused.', + occurredAt: account.updatedAt, + owner, + }); + } } const snapshots = ctx.services.accountRuntimes?.list() ?? []; diff --git a/src/web/routes/stats.ts b/src/web/routes/stats.ts index b04043b..1a21e99 100644 --- a/src/web/routes/stats.ts +++ b/src/web/routes/stats.ts @@ -56,6 +56,8 @@ export function createStatsRouter(ctx: WebContext): Router { const runtimeState = snapshot?.state ?? 'stopped'; const sendingState: BotStats['accounts'][number]['sendingState'] = runtimeState === 'failed' ? 'runtime-failed' + : account.status === 'suspended' + ? 'account-suspended' : account.status === 'login-required' || (runtimeState === 'running' && !authenticated) ? 'login-required' : runtimeState !== 'running' diff --git a/src/web/types.ts b/src/web/types.ts index 4d6e7ee..6c6e0d6 100644 --- a/src/web/types.ts +++ b/src/web/types.ts @@ -15,9 +15,9 @@ export interface BotStats { accounts: Array<{ accountId: string; redditUsername: string; - runtimeState: 'created' | 'starting' | 'running' | 'stopping' | 'stopped' | 'failed'; + runtimeState: 'created' | 'starting' | 'running' | 'stopping' | 'paused' | 'stopped' | 'failed'; authenticated: boolean; - sendingState: 'ready' | 'bot-stopped' | 'scan-only' | 'runtime-stopped' | 'runtime-failed' | 'login-required' | 'matrix-cooldown' | 'replies-prioritized' | 'pacing' | 'unestablished'; + sendingState: 'ready' | 'bot-stopped' | 'scan-only' | 'runtime-stopped' | 'runtime-failed' | 'login-required' | 'account-suspended' | 'matrix-cooldown' | 'replies-prioritized' | 'pacing' | 'unestablished'; messagesThisHour: number; maxMessagesPerHour: number; nextMessageAvailableAt: number | null; diff --git a/tests/account-repositories.test.ts b/tests/account-repositories.test.ts index a8a8305..f9532a3 100644 --- a/tests/account-repositories.test.ts +++ b/tests/account-repositories.test.ts @@ -95,6 +95,17 @@ describe('multi-account database ownership', () => { expect(accounts.get('t2_bob')).toMatchObject({ status: 'disabled', isActive: false, updatedAt: 130 }); }); + it('durably pauses a suspended account without destroying its session or history', () => { + addAccount('t2_alice', 'alice-bot', 100); + + expect(accounts.markSuspended('t2_alice', 200)).toBe(true); + expect(accounts.get('t2_alice')).toMatchObject({ + status: 'suspended', isActive: false, updatedAt: 200, + cookiesJson: expect.stringContaining('alice-bot-secret'), + }); + expect(accounts.getActive()).toEqual([]); + }); + it('partitions conversations, sync state and media even when remote IDs match', () => { addAccount('t2_alice', 'alice-bot', 100); addAccount('t2_bob', 'bob-bot', 110); diff --git a/tests/account-suspension.test.ts b/tests/account-suspension.test.ts new file mode 100644 index 0000000..a6ee53f --- /dev/null +++ b/tests/account-suspension.test.ts @@ -0,0 +1,49 @@ +import { describe, expect, it, vi } from 'vitest'; +import { AccountSuspensionService } from '../src/services/account-suspension.js'; +import { RedditRateLimitError } from '../src/reddit/api.js'; +import { createMockLogger } from './mockLogger.js'; + +const activeProfile = { + createdAt: null, linkKarma: null, commentKarma: null, totalKarma: null, + hasVerifiedEmail: null, isSuspended: false, +}; + +describe('AccountSuspensionService', () => { + it('pauses an authenticated account only when its own profile confirms suspension', async () => { + const pause = vi.fn(); + const reddit = { getAccountProfile: vi.fn().mockResolvedValue({ ...activeProfile, isSuspended: true }) }; + const service = new AccountSuspensionService( + 'owned-account', reddit, pause, createMockLogger(), { isFullyAuthenticated: () => true }, + ); + + await expect(service.checkNow()).resolves.toBe(true); + expect(reddit.getAccountProfile).toHaveBeenCalledWith('owned-account'); + expect(pause).toHaveBeenCalledWith('Reddit profile marks this account as suspended'); + }); + + it('does not pause for a missing profile, a transient failure, or a rate limit', async () => { + const pause = vi.fn(); + const reddit = { getAccountProfile: vi.fn() + .mockResolvedValueOnce(null) + .mockRejectedValueOnce(new Error('upstream unavailable')) + .mockRejectedValueOnce(new RedditRateLimitError(1_000)) }; + const service = new AccountSuspensionService( + 'owned-account', reddit, pause, createMockLogger(), { isFullyAuthenticated: () => true }, + ); + + await expect(service.checkNow()).resolves.toBe(false); + await expect(service.checkNow()).resolves.toBe(false); + await expect(service.checkNow()).resolves.toBe(false); + expect(pause).not.toHaveBeenCalled(); + }); + + it('does not probe while the account is not authenticated', async () => { + const reddit = { getAccountProfile: vi.fn() }; + const service = new AccountSuspensionService( + 'owned-account', reddit, vi.fn(), createMockLogger(), { isFullyAuthenticated: () => false }, + ); + + await expect(service.checkNow()).resolves.toBe(false); + expect(reddit.getAccountProfile).not.toHaveBeenCalled(); + }); +}); diff --git a/tests/matrix-client.test.ts b/tests/matrix-client.test.ts index 367ae9b..0a2c1d3 100644 --- a/tests/matrix-client.test.ts +++ b/tests/matrix-client.test.ts @@ -1,6 +1,7 @@ import { describe, it, expect, vi } from 'vitest'; import { MatrixAuthError, + MatrixAccountSuspendedError, MatrixClient, MatrixFeatureDisabledError, MatrixRateLimitError, @@ -186,6 +187,20 @@ describe('MatrixClient direct room registration', () => { }); describe('MatrixClient authentication classification', () => { + it('calls the account pause hook and throws a typed error for M_FORBIDDEN', async () => { + const onAccountSuspended = vi.fn(); + const client = new MatrixClient(creds, createMockLogger(), { + minRequestIntervalMs: 0, + onAccountSuspended, + fetchImpl: vi.fn().mockResolvedValue(new Response(JSON.stringify({ + errcode: 'M_FORBIDDEN', error: 'User is suspended', + }), { status: 403, headers: { 'content-type': 'application/json' } })), + }); + + await expect(client.whoami()).rejects.toBeInstanceOf(MatrixAccountSuspendedError); + expect(onAccountSuspended).toHaveBeenCalledOnce(); + }); + it('classifies an explicitly disabled endpoint as an unavailable feature', async () => { const client = new MatrixClient(creds, createMockLogger(), { minRequestIntervalMs: 0, diff --git a/tests/reddit-account-runtime.test.ts b/tests/reddit-account-runtime.test.ts index 5dd3007..e5f86b4 100644 --- a/tests/reddit-account-runtime.test.ts +++ b/tests/reddit-account-runtime.test.ts @@ -103,6 +103,28 @@ describe('RedditAccountRuntimeManager', () => { expect(manager.list()).toHaveLength(1); }); + it('keeps healthy accounts running when one enabled account cannot start', async () => { + const broken = account('broken'); + const healthy = account('healthy'); + const store = accountStore([broken, healthy]); + const factory = new RedditAccountRuntimeFactory(async (session) => new RedditAccountRuntime( + session, + { ...resources }, + { + start: () => { + if (session.accountId === broken.accountId) throw new Error('Matrix account suspended'); + }, + stop: vi.fn(), + }, + )); + const manager = new RedditAccountRuntimeManager(store.repository, factory); + + await manager.startEnabled(); + + expect(manager.get(broken.accountId)).toBeNull(); + expect(manager.get(healthy.accountId)?.snapshot().state).toBe('running'); + }); + it('replaces only the refreshed account while other accounts keep running', async () => { const first = account('first'); const second = account('second'); @@ -220,6 +242,27 @@ describe('RedditAccountRuntimeManager', () => { }); describe('account runtime isolation', () => { + it('pauses an account immediately and refuses new work', async () => { + const stored = account('suspended'); + const pause = vi.fn(); + const runtime = new RedditAccountRuntime( + { + accountId: stored.accountId, redditUsername: stored.redditUsername, + matrixUserId: stored.matrixUserId!, cookies: [], + matrixCredentials: JSON.parse(stored.matrixCredentialsJson), updatedAt: stored.updatedAt, + }, + { ...resources }, + { start: vi.fn(), stop: vi.fn(), pause }, + ); + await runtime.start(); + + runtime.pause('Matrix returned M_FORBIDDEN: account suspended'); + + expect(runtime.snapshot()).toMatchObject({ state: 'paused', failure: expect.stringContaining('M_FORBIDDEN') }); + expect(pause).toHaveBeenCalledOnce(); + expect(await runtime.canScan()).toBe(false); + }); + it('does not accept new outreach while the account has actionable replies', async () => { const stored = account('reply_first'); const hasActionableReplies = vi.fn().mockResolvedValue(true);