fix(threads): keep a read bot read through compactions and stops

docs/bots.md

@@ -817,7 +817,9 @@ thread:
   the catalog;
 - `nextWake`: the next enabled routine occurrence;
 - `unread`: answers the user has not read on its thread, counted since it was
-  last read;
+  last read. Only an answer or a failure counts: a compaction (the hourly one
+  before the prompt cache lapses included), a keepalive, a quiet wake or a
+  stopped turn leaves the thread as read as it was (`turnIsNews`);
 - `last`: the final text block of its last non-quiet answer, first 200
   characters;
 - `today`: its turns since local midnight, the wakes among them (`routine` and

packages/contracts/src/turns.ts

@@ -163,6 +163,18 @@ export function threadActive(status: ThreadStatus): boolean {
   return ACTIVE_THREAD_STATUSES.includes(status);
 }
 
+/**
+ * Whether a settled turn left something to read: an answer or a failure. A
+ * stop is the user's own doing or the core's restart, compacting the context
+ * and a cache keepalive are housekeeping, and a persistent agent's quiet wake
+ * had nothing to say. Only news marks a thread unread.
+ */
+export function turnIsNews(turn: Pick<Turn, 'status' | 'execution' | 'quiet'>): boolean {
+  if (turn.status !== 'done' && turn.status !== 'error') return false;
+  if (turn.execution?.operation === 'compact' || turn.execution?.operation === 'keepalive') return false;
+  return turn.status === 'error' || turn.quiet !== true;
+}
+
 /**
  * Whether a finished turn is worth a notification, for the core's Web Push and
  * a client's own toast alike. A delegated agent's result goes to its parent,
@@ -177,12 +189,8 @@ export function notifiesOnFinish(
   turn: Pick<Turn, 'status' | 'execution' | 'quiet'>,
   activeChildren: number,
 ): boolean {
-  if (turn.status !== 'done' && turn.status !== 'error') return false;
-  if (thread.parentThreadId) return false;
-  if (turn.execution?.operation === 'compact' || turn.execution?.operation === 'keepalive') return false;
-  if (turn.status === 'error') return true;
-  if (turn.quiet) return false;
-  return activeChildren === 0;
+  if (!turnIsNews(turn) || thread.parentThreadId) return false;
+  return turn.status === 'error' || activeChildren === 0;
 }
 
 export interface TurnsRpcMethods {

packages/core/src/agents/store.ts

@@ -1,6 +1,6 @@
 import { mkdirSync } from 'node:fs';
 import { join } from 'node:path';
-import { builtinSettingRefusal, isAgentPicture } from '@boite/contracts';
+import { builtinSettingRefusal, isAgentPicture, turnIsNews } from '@boite/contracts';
 import type {
   AccountId, AgentBrain, AgentBuiltin, AgentProfile, AgentRoutine, AgentSave, AgentSession, AgentStatus, AgentWatch, AgentsSnapshot, ModelInfo, RpcParams, ThreadSummary, Turn,
 } from '@boite/contracts';
@@ -551,8 +551,7 @@ export class AgentStore {
   /** A finished turn the user will read: one more unread while its thread is. */
   countUnread(turn: Turn): void {
     const thread = this.core.journal.getThread(turn.threadId);
-    const operation = turn.execution?.operation;
-    if (!thread?.unread || turn.quiet || operation === 'compact' || operation === 'keepalive' || (turn.status !== 'done' && turn.status !== 'error')) return;
+    if (!thread?.unread || !turnIsNews(turn)) return;
     this.core.journal.setSetting(`${UNREAD}${turn.threadId}`, ((this.core.journal.getSetting(`${UNREAD}${turn.threadId}`) as number | undefined) ?? 0) + 1);
   }
   #readUp(threadId: string): void {

packages/core/src/threads/turn-settlement.ts

@@ -1,4 +1,4 @@
-import type { ThreadId, ThreadSummary, Turn, TurnId } from '@boite/contracts';
+import { turnIsNews, type ThreadId, type ThreadSummary, type Turn, type TurnId } from '@boite/contracts';
 import type { Core } from '../core.ts';
 import type { TurnResult } from '../drivers/types.ts';
 import { logMessageOf } from '../log-errors.ts';
@@ -115,13 +115,12 @@ function finishThread(core: Core, { thread, finished, result }: Settlement): Com
     sameSession = (current.sessionGeneration ?? 0) === (thread.sessionGeneration ?? 0)
       && current.sessionId === thread.sessionId && current.providerId === thread.providerId && current.accountId === thread.accountId;
     const sessionId = sameSession ? result.sessionId ?? current.sessionId : current.sessionId;
-    // A keepalive is the core's housekeeping: it says nothing new, so it leaves the thread as read as it was.
-    // A quiet wake of a persistent agent had nothing to say either.
-    const silent = finished.execution?.operation === 'keepalive' || finished.quiet === true;
+    // A compaction, a keepalive, a quiet wake or a stop says nothing new: the thread stays as read as it was.
+    const news = turnIsNews(finished);
     return {
       ...current, sessionId,
       sessionResumeAt: !sameSession ? current.sessionResumeAt : current.sessionResumeAt && sessionId === current.sessionId && sessionId === thread.sessionId ? current.sessionResumeAt : null,
-      unread: silent ? current.unread : current.unread || !core.subscribers.hasSubscribers(thread.id),
+      unread: news ? current.unread || !core.subscribers.hasSubscribers(thread.id) : current.unread,
       promptCache: sameSession ? promptCacheOf(result, thread, finished.finishedAt ?? Date.now(), current.promptCache ?? null) ?? current.promptCache ?? null : current.promptCache ?? null,
     };
   });

packages/core/test/compact.test.ts

@@ -218,6 +218,30 @@ describe('automatic compaction', () => {
     client.close();
   });
 
+  test('a thread read before its compaction or a stop stays read: neither is news', async () => {
+    // Read bots lit up again an hour later, when their prompt cache lapsed and the core compacted them.
+    const client = await h.connect();
+    const { threadId } = await quiet(client);
+    await client.call('settings.set', { autoCompact: { tokens: null, moments: ['cache-expiry'] } });
+    AUTO_COMPACT.cacheMarginMs = 300_000 - 200;
+    const compacted = client.next('turn.finished', t => t.threadId === threadId && t.execution?.operation === 'compact');
+    await turn(client, threadId, 'hello');
+    expect((await client.call('threads.get', { threadId })).unread).toBe(true);
+    await client.call('threads.markRead', { threadId });
+    expect((await compacted).status).toBe('done');
+    expect((await client.call('threads.get', { threadId })).unread).toBe(false);
+
+    // The next cache moment minutes away again: only the stop settles here.
+    AUTO_COMPACT.cacheMarginMs = saved.cacheMarginMs;
+    const stopped = client.next('turn.finished', t => t.threadId === threadId);
+    await client.call('turns.start', { threadId, prompt: '[sleep:5000]' });
+    await waitFor(() => h.core.threads.require(threadId).status === 'running');
+    await client.call('turns.stop', { threadId });
+    expect((await stopped).status).toBe('stopped');
+    expect((await client.call('threads.get', { threadId })).unread).toBe(false);
+    client.close();
+  });
+
   test('a thread left above a limit set later compacts once when a client shows it, not at its next prompt', async () => {
     // A thread idle at 212k before a 200k limit existed made its next prompt wait a minute for the agent's own compaction.
     // Its turn armed the cache's moment minutes ahead: the lower limit must not wait for that timer.

packages/ui/src/lib/fake-client/turns.ts

@@ -1,5 +1,5 @@
 /** The turn engine: start, stream and stop a turn, the way the echo driver answers. */
-import { effectiveCompaction, previewPrompt, RpcErrorCode, threadActive, type AgentLetter, type Attachment, type Message, type MessagePart, type PreviewReference, type Thread, type ThreadId, type Turn, type TurnInFlightData, type Usage } from '@boite/contracts';
+import { effectiveCompaction, previewPrompt, RpcErrorCode, threadActive, turnIsNews, type AgentLetter, type Attachment, type Message, type MessagePart, type PreviewReference, type Thread, type ThreadId, type Turn, type TurnInFlightData, type Usage } from '@boite/contracts';
 import { decodedBytes } from '../attachments';
 import { RpcFailure } from '../client';
 import { DIFF_NEW, DIFF_OLD, DIFF_PATH, DOC_TEXT, DOC_TITLE, IMAGE_BASE64, ASYNC_QUESTION_OPTIONS, ASYNC_QUESTION_TEXT, SETTING_MARKERS, SPAWN_MARKER, AGENT_SETTING_MARKER, AGENT_SETTING_REASON, AGENT_MODEL_MARKER, AGENT_MODEL_REASON } from './conversation';
@@ -383,7 +383,8 @@ function finishTurn(ctx: FakeContext, thread: Thread, turn: Turn, message: Messa
   ctx.inFlight.delete(thread.id);
   thread.status = 'idle';
   thread.progress = null;
-  thread.unread = !ctx.bus.subscribed.has(thread.id);
+  // As the core's settlement: a compaction or a stop leaves the thread as read as it was.
+  if (turnIsNews(turn)) thread.unread = !ctx.bus.subscribed.has(thread.id);
   // The context meter grows with every turn, the way a real session's does.
   thread.context = {
     tokens: prompt === '[compact]' && !record.cancelled ? compactAfter : (thread.context?.tokens ?? FAKE_CONTEXT_FLOOR) + FAKE_CONTEXT_PER_TURN + modelPrompt.length * 4,