feat: application-level read state and inbox triage for conversations

Conversations were hard to manage because "unread" came straight from
Matrix's notification_count, and nothing in the app ever sends a read
receipt — so opening a conversation in the dashboard could never clear it
and the unread set only ever grew. The list also collapsed three
orthogonal facts into one word, labelling "they spoke last and we have
not replied" as "Up to date".

Read state is now ours:

- conversation_read_state table (migration 19) plus
  ConversationReadStateRepository, kept separate from conversation_states
  so a read write can never be lost to mergeConversationState. The
  watermark is monotonic, so a stale poll cannot resurrect an unread
  badge.
- hasUnreadInbound and needsReply are derived server-side and returned on
  every conversation. Matrix's `unread` stays on the contract but the
  dashboard no longer reads it.
- POST /conversations/{id}/read|unread|status and
  POST /conversations/bulk-read, with optimistic cache updates.

Inbox rework:

- List rows are a real grid (checkbox, open button, chip strip) instead of
  a full-bleed absolute overlay button, which previously forced a
  pointer-events escape hatch for anything interactive in a row. Rows use
  aria-current rather than aria-pressed. Not a listbox: options must not
  contain interactive descendants.
- Counted filter pills replace the select, independent state chips replace
  the collapsed status word, plus a sort control (including oldest
  unanswered), a per-account filter, age-coloured timestamps, and an
  explicit count of rows hidden by the seller filter.
- Keyboard: j/k/arrows, Enter, u, e, / and Escape. Shortcuts yield to any
  open dialog so Escape dismisses the modal, not the conversation behind
  it. Multi-select with bulk mark-read and mark-done.
- Detail panel: auto-mark-read on open, a "New messages" divider pinned at
  mount (deriving it from the live watermark made it vanish a second after
  opening), triage actions in the header, the automation banner moved out
  of the composer, day dividers with time-only stamps, and a flex column
  so history scrolls in its own region instead of painting over the
  composer.
- Row order freezes while a conversation is open, with a "N new — refresh"
  pill, so a background poll cannot reorder rows mid-triage.

Two database tests rewound user_version by an offset from the latest
version, which silently re-targets a different migration whenever one is
appended. They now pin absolute versions.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

src/bootstrap.ts

@@ -10,6 +10,7 @@ import {
   ActivityLogRepository,
   RedditUsersRepository,
   ConversationStateRepository,
+  ConversationReadStateRepository,
   ConversationOutcomesRepository,
   DeliveryOutboxRepository,
   RecipientHistoryRepository,
@@ -70,6 +71,7 @@ export interface AppContainer {
   successfulMessagesRepo: SuccessfulMessagesRepository;
   deliveryOutboxRepo: DeliveryOutboxRepository;
   conversationStateRepo: ConversationStateRepository;
+  conversationReadStateRepo: ConversationReadStateRepository;
   conversationOutcomesRepo: ConversationOutcomesRepository;
   recipientDeduplicator: RecipientDeduplicator;
   candidateQueueRepo: CandidateQueueRepository;
@@ -130,6 +132,7 @@ export async function bootstrap(
   const redditUsersRepo = new RedditUsersRepository(db);
   const deliveryOutboxRepo = new DeliveryOutboxRepository(db, bootstrapAccountId);
   const conversationStateRepo = new ConversationStateRepository(db, bootstrapAccountId);
+  const conversationReadStateRepo = new ConversationReadStateRepository(db, bootstrapAccountId);
   const conversationOutcomesRepo = new ConversationOutcomesRepository(db, bootstrapAccountId);
   const recipientDeduplicator = new RecipientDeduplicator(new RecipientHistoryRepository(db));
   const candidateQueueRepo = new CandidateQueueRepository(db);
@@ -447,6 +450,7 @@ export async function bootstrap(
     webServer.setRedditAccountsService(redditAccounts);
     webServer.setAccountRuntimeManager(accountRuntimeManager);
     webServer.setConversationStateRepository(conversationStateRepo);
+    webServer.setConversationReadStateRepository(conversationReadStateRepo);
     webServer.setCandidateQueueRepository(candidateQueueRepo);
     webServer.setConfigStore(new ConfigStore(db));
     webServer.setOnConfigChange(() => logger.info('Runtime config updated via dashboard'));
@@ -480,6 +484,7 @@ export async function bootstrap(
     successfulMessagesRepo,
     deliveryOutboxRepo,
     conversationStateRepo,
+    conversationReadStateRepo,
     conversationOutcomesRepo,
     recipientDeduplicator,
     candidateQueueRepo,

src/database/client.ts

@@ -412,6 +412,19 @@ export function runMigrations(db: Database.Database): void {
       CREATE INDEX IF NOT EXISTS idx_media_library_sends_conversation
         ON media_library_sends(account_id, conversation_id, requested_at);
     `),
+    () => db.exec(`
+      CREATE TABLE IF NOT EXISTS conversation_read_state (
+        account_id TEXT NOT NULL,
+        conversation_id TEXT NOT NULL,
+        last_read_message_ts INTEGER NOT NULL DEFAULT 0,
+        last_read_at INTEGER NOT NULL,
+        status TEXT NOT NULL DEFAULT 'open' CHECK(status IN ('open', 'snoozed', 'done')),
+        snoozed_until INTEGER,
+        PRIMARY KEY (account_id, conversation_id)
+      );
+      CREATE INDEX IF NOT EXISTS idx_conversation_read_state_status
+        ON conversation_read_state(account_id, status, snoozed_until);
+    `),
   ];
 
   let currentVersion = db.pragma('user_version', { simple: true }) as number;

src/database/repositories.ts

@@ -1180,6 +1180,127 @@ export class ConversationStateRepository {
   }
 }
 
+/**
+ * Operator-facing triage state for a conversation.
+ *
+ * This is deliberately separate from `conversation_states`: that blob is owned
+ * by the automation pipeline and goes through `mergeConversationState`, whereas
+ * read state is written by the dashboard on every open and must never lose a
+ * write to a merge. Matrix's own `unread_notifications` count is not usable for
+ * this — nothing in the app sends read receipts, so it never clears.
+ */
+export type ConversationTriageStatus = 'open' | 'snoozed' | 'done';
+
+export interface ConversationReadStateRecord {
+  conversationId: string;
+  lastReadMessageTs: number;
+  lastReadAt: number;
+  status: ConversationTriageStatus;
+  snoozedUntil: number | null;
+}
+
+interface ConversationReadStateRow {
+  conversation_id: string;
+  last_read_message_ts: number;
+  last_read_at: number;
+  status: ConversationTriageStatus;
+  snoozed_until: number | null;
+}
+
+function toReadStateRecord(row: ConversationReadStateRow): ConversationReadStateRecord {
+  return {
+    conversationId: row.conversation_id,
+    lastReadMessageTs: row.last_read_message_ts,
+    lastReadAt: row.last_read_at,
+    status: row.status,
+    snoozedUntil: row.snoozed_until,
+  };
+}
+
+export class ConversationReadStateRepository {
+  constructor(
+    private readonly db: Database.Database,
+    private readonly accountId: string = LEGACY_UNASSIGNED_ACCOUNT_ID,
+  ) {}
+
+  get(conversationId: string): ConversationReadStateRecord | undefined {
+    const row = this.db.prepare(`
+      SELECT conversation_id, last_read_message_ts, last_read_at, status, snoozed_until
+      FROM conversation_read_state WHERE account_id = ? AND conversation_id = ?
+    `).get(this.accountId, conversationId) as ConversationReadStateRow | undefined;
+    return row ? toReadStateRecord(row) : undefined;
+  }
+
+  list(): ConversationReadStateRecord[] {
+    const rows = this.db.prepare(`
+      SELECT conversation_id, last_read_message_ts, last_read_at, status, snoozed_until
+      FROM conversation_read_state WHERE account_id = ?
+    `).all(this.accountId) as ConversationReadStateRow[];
+    return rows.map(toReadStateRecord);
+  }
+
+  /**
+   * Advances the read watermark. Never moves backwards, so a stale poll landing
+   * after the operator has read further cannot resurrect an unread badge.
+   */
+  markRead(conversationId: string, throughTimestamp: number, now = Date.now()): void {
+    this.db.prepare(`
+      INSERT INTO conversation_read_state (account_id, conversation_id, last_read_message_ts, last_read_at)
+      VALUES (?, ?, ?, ?)
+      ON CONFLICT(account_id, conversation_id) DO UPDATE SET
+        last_read_message_ts = MAX(last_read_message_ts, excluded.last_read_message_ts),
+        last_read_at = excluded.last_read_at
+    `).run(this.accountId, conversationId, Math.max(0, throughTimestamp), now);
+  }
+
+  /**
+   * Pushes a conversation back onto the queue by rewinding the watermark to just
+   * before the message the operator wants to see as new again.
+   */
+  markUnread(conversationId: string, beforeTimestamp: number, now = Date.now()): void {
+    const watermark = Math.max(0, beforeTimestamp - 1);
+    this.db.prepare(`
+      INSERT INTO conversation_read_state (account_id, conversation_id, last_read_message_ts, last_read_at, status)
+      VALUES (?, ?, ?, ?, 'open')
+      ON CONFLICT(account_id, conversation_id) DO UPDATE SET
+        last_read_message_ts = excluded.last_read_message_ts,
+        last_read_at = excluded.last_read_at,
+        status = 'open',
+        snoozed_until = NULL
+    `).run(this.accountId, conversationId, watermark, now);
+  }
+
+  setStatus(
+    conversationId: string,
+    status: ConversationTriageStatus,
+    snoozedUntil: number | null = null,
+    now = Date.now(),
+  ): void {
+    this.db.prepare(`
+      INSERT INTO conversation_read_state (account_id, conversation_id, last_read_message_ts, last_read_at, status, snoozed_until)
+      VALUES (?, ?, 0, ?, ?, ?)
+      ON CONFLICT(account_id, conversation_id) DO UPDATE SET
+        status = excluded.status,
+        snoozed_until = excluded.snoozed_until,
+        last_read_at = excluded.last_read_at
+    `).run(this.accountId, conversationId, now, status, status === 'snoozed' ? snoozedUntil : null);
+  }
+
+  /** Returns snoozed conversations whose snooze has expired, so they can reopen. */
+  expireSnoozes(now = Date.now()): string[] {
+    const rows = this.db.prepare(`
+      SELECT conversation_id FROM conversation_read_state
+      WHERE account_id = ? AND status = 'snoozed' AND snoozed_until IS NOT NULL AND snoozed_until <= ?
+    `).all(this.accountId, now) as Array<{ conversation_id: string }>;
+    if (rows.length === 0) return [];
+    this.db.prepare(`
+      UPDATE conversation_read_state SET status = 'open', snoozed_until = NULL, last_read_at = ?
+      WHERE account_id = ? AND status = 'snoozed' AND snoozed_until IS NOT NULL AND snoozed_until <= ?
+    `).run(now, this.accountId, now);
+    return rows.map((row) => row.conversation_id);
+  }
+}
+
 /** Durable idempotency claims for AI replies to individual inbound events. */
 export interface ConversationAutomationClaimRecord {
   conversationId: string;

src/services/reddit-account-runtime-builder.ts

@@ -7,6 +7,7 @@ import { BrowserClient } from '../browser/client.js';
 import {
   PostsRepository, MessagesRepository, SuccessfulMessagesRepository,
   DeliveryOutboxRepository, ConversationStateRepository,
+  ConversationReadStateRepository,
   ConversationAutomationClaimRepository,
   ConversationOutcomesRepository, RedditUsersRepository,
   ActivityLogRepository,
@@ -158,6 +159,7 @@ async function buildRuntime(
     const successfulMessagesRepo = new SuccessfulMessagesRepository(db, session.accountId);
     const deliveryOutboxRepo = new DeliveryOutboxRepository(db, session.accountId);
     const conversationStateRepo = new ConversationStateRepository(db, session.accountId);
+    const conversationReadStateRepo = new ConversationReadStateRepository(db, session.accountId);
     const conversationAutomationClaimRepo = new ConversationAutomationClaimRepository(db, session.accountId);
     const conversationOutcomesRepo = new ConversationOutcomesRepository(db, session.accountId);
     const redditUsersRepo = new RedditUsersRepository(db);
@@ -247,7 +249,8 @@ async function buildRuntime(
     const resources: RedditAccountRuntimeResources = {
       browserProfileDir, browserClient, accountAuth, matrixClient, chatCache, redditApi,
       postsRepo, messagesRepo, successfulMessagesRepo, deliveryOutboxRepo,
-      conversationStateRepo, conversationOutcomesRepo, redditUsersRepo, accountContentRepo,
+      conversationStateRepo, conversationReadStateRepo, conversationOutcomesRepo,
+      redditUsersRepo, accountContentRepo,
       activityLogRepo,
       monitorService, processorService, responseChecker, conversationAutomation, chatService, mediaService,
       mediaSyncService, mediaExchange, rateLimiter,

src/services/reddit-account-runtime.ts

@@ -11,6 +11,7 @@ import type {
   ConversationOutcomesRepository,
   ActivityLogRepository,
   ConversationStateRepository,
+  ConversationReadStateRepository,
   DeliveryOutboxRepository,
   MessagesRepository,
   PostsRepository,
@@ -65,6 +66,7 @@ export interface RedditAccountRuntimeResources {
   successfulMessagesRepo: SuccessfulMessagesRepository;
   deliveryOutboxRepo: DeliveryOutboxRepository;
   conversationStateRepo: ConversationStateRepository;
+  conversationReadStateRepo: ConversationReadStateRepository;
   conversationOutcomesRepo: ConversationOutcomesRepository;
   redditUsersRepo: RedditUsersRepository;
   accountContentRepo: AccountContentRepository;

src/web/api-catalog.ts

@@ -86,6 +86,29 @@ export const API_OPERATIONS: readonly ApiOperation[] = [
   operation('getConversationMediaOptions', 'GET', '/api/conversations/{id}/shareable-media', 'List every enabled exchange-library image not yet shared in this conversation', ['Conversations']),
   operation('shareConversationMedia', 'POST', '/api/conversations/{id}/share-media', 'Send a selected exchange-library image', ['Conversations']),
   operation('startConversation', 'POST', '/api/conversations/start', 'Start a new conversation', ['Conversations']),
+  operation('markConversationRead', 'POST', '/api/conversations/{id}/read', 'Advance the operator read watermark', ['Conversations'], {
+    requestSchema: z.strictObject({ throughTimestamp: z.number().int().nonnegative().optional() }),
+    responseSchema: z.strictObject({ success: z.literal(true) }),
+  }),
+  operation('markConversationUnread', 'POST', '/api/conversations/{id}/unread', 'Return a conversation to the unread queue', ['Conversations'], {
+    requestSchema: z.strictObject({ beforeTimestamp: z.number().int().nonnegative().optional() }),
+    responseSchema: z.strictObject({ success: z.literal(true) }),
+  }),
+  operation('setConversationStatus', 'POST', '/api/conversations/{id}/status', 'Snooze, resolve or reopen a conversation', ['Conversations'], {
+    requestSchema: z.strictObject({
+      status: z.enum(['open', 'snoozed', 'done']),
+      snoozeUntil: z.number().int().positive().nullable().optional(),
+    }),
+    responseSchema: z.strictObject({ success: z.literal(true) }),
+  }),
+  operation('bulkMarkConversationsRead', 'POST', '/api/conversations/bulk-read', 'Mark many conversations read in one request', ['Conversations'], {
+    requestSchema: z.strictObject({ conversationIds: z.array(z.string().min(1)).min(1).max(500) }),
+    responseSchema: z.strictObject({
+      success: z.literal(true),
+      updated: z.number().int().nonnegative(),
+      failed: z.array(z.string()),
+    }),
+  }),
   operation('getMedia', 'GET', '/api/media', 'List received private media', ['Media']),
   operation('getMediaSync', 'GET', '/api/media/sync', 'Get media sync state', ['Media']),
   operation('triggerMediaSync', 'POST', '/api/media/sync', 'Trigger media sync', ['Media']),

src/web/context.ts

@@ -5,6 +5,7 @@ import type {
   MessagesRepository,
   SuccessfulMessagesRepository,
   ConversationStateRepository,
+  ConversationReadStateRepository,
 } from '../database/repositories.js';
 import type { ChatService } from '../services/chat.js';
 import type { MediaService } from '../services/media.js';
@@ -55,6 +56,7 @@ export interface WebContext {
     sessionImporter?: RedditSessionImporter;
     accountAuth?: AccountAuthContext;
     conversationState?: ConversationStateRepository;
+    conversationReadState?: ConversationReadStateRepository;
     configStore?: ConfigStore;
     redditAccounts?: RedditAccountsService;
     accountRuntimes?: RedditAccountRuntimeManager;

src/web/routes/conversations.ts

@@ -10,11 +10,73 @@ import {
   remainingAutomationTurns,
 } from '../../services/conversation-state.js';
 import type { MediaLibraryAsset } from '../../services/media-library-types.js';
+import type {
+  ConversationReadStateRecord,
+  ConversationReadStateRepository,
+} from '../../database/repositories.js';
 
 const ShareConversationMediaRequestSchema = z.strictObject({ assetId: z.string().uuid() });
+const MarkReadRequestSchema = z.strictObject({
+  throughTimestamp: z.number().int().nonnegative().optional(),
+});
+const MarkUnreadRequestSchema = z.strictObject({
+  beforeTimestamp: z.number().int().nonnegative().optional(),
+});
+const SetStatusRequestSchema = z.strictObject({
+  status: z.enum(['open', 'snoozed', 'done']),
+  snoozeUntil: z.number().int().positive().nullable().optional(),
+});
+const BulkMarkReadRequestSchema = z.strictObject({
+  conversationIds: z.array(z.string().min(1)).min(1).max(500),
+});
 
 type OwnedActiveChat = ActiveChat & { owner?: { accountId: string; redditUsername: string } };
 
+/** An owned conversation with operator triage state projected onto it. */
+type TriagedActiveChat = OwnedActiveChat & {
+  hasUnreadInbound: boolean;
+  needsReply: boolean;
+  triageStatus: 'open' | 'snoozed' | 'done';
+  snoozedUntil: number | null;
+  lastReadMessageTs: number;
+};
+
+/**
+ * Projects operator read state onto a conversation summary.
+ *
+ * `hasUnreadInbound` is deliberately derived from the summary's last-message
+ * fields rather than the full message list — the list endpoint does not always
+ * have message history loaded, and a conversation whose newest message is ours
+ * is by definition one we have seen.
+ */
+function annotateTriage(
+  conversation: OwnedActiveChat,
+  readState: ConversationReadStateRecord | undefined,
+): TriagedActiveChat {
+  const lastReadMessageTs = readState?.lastReadMessageTs ?? 0;
+  const status = readState?.status ?? 'open';
+  return {
+    ...conversation,
+    hasUnreadInbound: !conversation.isFromUs && conversation.lastMessageTime > lastReadMessageTs,
+    needsReply: !conversation.isFromUs && status === 'open',
+    triageStatus: status,
+    snoozedUntil: readState?.snoozedUntil ?? null,
+    lastReadMessageTs,
+  };
+}
+
+/**
+ * Reopens conversations whose snooze has elapsed and returns the current read
+ * state for the account, keyed by conversation id.
+ */
+function currentReadState(
+  repo: ConversationReadStateRepository | undefined,
+): Map<string, ConversationReadStateRecord> {
+  if (!repo) return new Map();
+  repo.expireSnoozes();
+  return new Map(repo.list().map((record) => [record.conversationId, record]));
+}
+
 export function createConversationsRouter(ctx: WebContext): Router {
   const router = Router();
 
@@ -110,21 +172,32 @@ export function createConversationsRouter(ctx: WebContext): Router {
   }
 
   router.get('/conversations', async (req, res) => {
-    if (!ctx.services.chat) {
+    const chatService = ctx.services.chat;
+    if (!chatService) {
       res.status(503).json({ error: 'Chat service not initialized' });
       return;
     }
     try {
       const runtimes = ctx.services.accountRuntimes?.list().filter((runtime) => runtime.state === 'running') ?? [];
-      const conversations: OwnedActiveChat[] = runtimes.length > 0
+      const conversations: TriagedActiveChat[] = runtimes.length > 0
         ? (await Promise.all(runtimes.map(async (snapshot) => {
             const runtime = ctx.services.accountRuntimes!.get(snapshot.accountId)!;
-            return (await runtime.resources.chatService.getCachedConversations()).map((conversation) => withSellerMarker({
-              ...conversation,
-              owner: { accountId: snapshot.accountId, redditUsername: snapshot.redditUsername },
-            }, runtime.resources.conversationStateRepo));
+            const readState = currentReadState(runtime.resources.conversationReadStateRepo);
+            return (await runtime.resources.chatService.getCachedConversations()).map((conversation) => annotateTriage(
+              withSellerMarker({
+                ...conversation,
+                owner: { accountId: snapshot.accountId, redditUsername: snapshot.redditUsername },
+              }, runtime.resources.conversationStateRepo),
+              conversation.roomId ? readState.get(conversation.roomId) : undefined,
+            ));
           }))).flat()
-        : (await ctx.services.chat.getCachedConversations()).map((conversation) => withSellerMarker(conversation));
+        : await (async () => {
+            const readState = currentReadState(ctx.services.conversationReadState);
+            return (await chatService.getCachedConversations()).map((conversation) => annotateTriage(
+              withSellerMarker(conversation),
+              conversation.roomId ? readState.get(conversation.roomId) : undefined,
+            ));
+          })();
       if (req.query.includeImageMetadata !== 'true') {
         res.json(conversations.map((conversation) => ({ ...conversation, owner: conversation.owner ?? currentOwner() })));
         return;
@@ -269,8 +342,20 @@ export function createConversationsRouter(ctx: WebContext): Router {
         return { ...message, isAi: true };
       });
 
+      const readStateRepo = resolved.runtime?.resources.conversationReadStateRepo ?? ctx.services.conversationReadState;
+      readStateRepo?.expireSnoozes();
+      const triage = annotateTriage(
+        conversation,
+        conversation.roomId ? readStateRepo?.get(conversation.roomId) : undefined,
+      );
+
       res.json({
         ...conversation,
+        hasUnreadInbound: triage.hasUnreadInbound,
+        needsReply: triage.needsReply,
+        triageStatus: triage.triageStatus,
+        snoozedUntil: triage.snoozedUntil,
+        lastReadMessageTs: triage.lastReadMessageTs,
         sellerDetected: intelligence.state.sellerDetected,
         ...(annotatedMessages ? { messages: annotatedMessages } : {}),
         owner: resolved.runtime
@@ -452,6 +537,101 @@ export function createConversationsRouter(ctx: WebContext): Router {
     }
   });
 
+  /**
+   * Resolves the read-state repository that owns a conversation, plus the room
+   * id to key it by. Conversations without a Matrix room cannot carry read
+   * state — there is no stable identifier to hang it off.
+   */
+  async function resolveReadState(id: string): Promise<
+    | { repo: ConversationReadStateRepository; roomId: string; conversation: ActiveChat }
+    | { error: { status: number; message: string } }
+  > {
+    const resolved = await findConversation(decodeURIComponent(id));
+    if (!resolved) return { error: { status: 404, message: 'Conversation not found' } };
+    const repo = resolved.runtime?.resources.conversationReadStateRepo ?? ctx.services.conversationReadState;
+    if (!repo) return { error: { status: 503, message: 'Conversation read state is not initialized' } };
+    if (!resolved.conversation.roomId) {
+      return { error: { status: 409, message: 'This conversation has no Matrix room.' } };
+    }
+    return { repo, roomId: resolved.conversation.roomId, conversation: resolved.conversation };
+  }
+
+  router.post('/conversations/:id/read', async (req, res) => {
+    const parsed = MarkReadRequestSchema.safeParse(req.body ?? {});
+    if (!parsed.success) {
+      res.status(400).json({ error: 'throughTimestamp must be a non-negative integer' });
+      return;
+    }
+    const resolved = await resolveReadState(req.params.id);
+    if ('error' in resolved) {
+      res.status(resolved.error.status).json({ error: resolved.error.message });
+      return;
+    }
+    const through = parsed.data.throughTimestamp ?? resolved.conversation.lastMessageTime;
+    resolved.repo.markRead(resolved.roomId, through);
+    res.json({ success: true });
+  });
+
+  router.post('/conversations/:id/unread', async (req, res) => {
+    const parsed = MarkUnreadRequestSchema.safeParse(req.body ?? {});
+    if (!parsed.success) {
+      res.status(400).json({ error: 'beforeTimestamp must be a non-negative integer' });
+      return;
+    }
+    const resolved = await resolveReadState(req.params.id);
+    if ('error' in resolved) {
+      res.status(resolved.error.status).json({ error: resolved.error.message });
+      return;
+    }
+    const before = parsed.data.beforeTimestamp ?? resolved.conversation.lastMessageTime;
+    resolved.repo.markUnread(resolved.roomId, before);
+    res.json({ success: true });
+  });
+
+  router.post('/conversations/:id/status', async (req, res) => {
+    const parsed = SetStatusRequestSchema.safeParse(req.body ?? {});
+    if (!parsed.success) {
+      res.status(400).json({ error: 'status must be one of open, snoozed, done' });
+      return;
+    }
+    if (parsed.data.status === 'snoozed' && !parsed.data.snoozeUntil) {
+      res.status(400).json({ error: 'snoozeUntil is required when snoozing a conversation' });
+      return;
+    }
+    const resolved = await resolveReadState(req.params.id);
+    if ('error' in resolved) {
+      res.status(resolved.error.status).json({ error: resolved.error.message });
+      return;
+    }
+    resolved.repo.setStatus(resolved.roomId, parsed.data.status, parsed.data.snoozeUntil ?? null);
+    res.json({ success: true });
+  });
+
+  /**
+   * Marks many conversations read in one request so draining a filtered backlog
+   * is a single action. Individual failures are reported rather than aborting
+   * the whole batch — a stopped account should not block the rest.
+   */
+  router.post('/conversations/bulk-read', async (req, res) => {
+    const parsed = BulkMarkReadRequestSchema.safeParse(req.body ?? {});
+    if (!parsed.success) {
+      res.status(400).json({ error: 'conversationIds must be a non-empty array of at most 500 ids' });
+      return;
+    }
+    const failed: string[] = [];
+    let updated = 0;
+    for (const id of parsed.data.conversationIds) {
+      const resolved = await resolveReadState(id);
+      if ('error' in resolved) {
+        failed.push(id);
+        continue;
+      }
+      resolved.repo.markRead(resolved.roomId, resolved.conversation.lastMessageTime);
+      updated += 1;
+    }
+    res.json({ success: true, updated, failed });
+  });
+
   router.get('/conversations/:id/shareable-media', async (req, res) => {
     try {
       const resolved = await findConversation(decodeURIComponent(req.params.id));

src/web/server.ts

@@ -9,6 +9,7 @@ import type {
   MessagesRepository,
   SuccessfulMessagesRepository,
   ConversationStateRepository,
+  ConversationReadStateRepository,
   ActivityLogRepository,
 } from '../database/repositories.js';
 import type { ChatService } from '../services/chat.js';
@@ -247,6 +248,10 @@ export class WebServer {
     this.services.conversationState = conversationState;
   }
 
+  setConversationReadStateRepository(conversationReadState: ConversationReadStateRepository): void {
+    this.services.conversationReadState = conversationReadState;
+  }
+
   setCandidateQueueRepository(candidateQueue: CandidateQueueRepository): void {
     this.services.candidateQueue = candidateQueue;
   }