fix(tlon): stop monitor tracking state from growing forever (#103658)

* fix(tlon): bound monitor tracking state

* fix(tlon): fence invite snapshot generations

* fix(tlon): simplify bounded monitor tracking

Co-authored-by: mikasa0818 <0668001030@xydigit.com>

* fix(tlon): keep thread limit internal

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

extensions/tlon/src/monitor/index.ts

@@ -36,6 +36,7 @@ import {
   mergeUniqueStrings,
   shouldMigrateTlonSetting,
 } from "./settings-helpers.js";
+import { createActiveSnapshotTracker, createParticipatedThreadTracker } from "./tracking.js";
 import { asRecord, formatErrorMessage, readString } from "./utils.js";
 import {
   extractMessageText,
@@ -160,8 +161,8 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
   let pendingApprovals: PendingApproval[] = [];
   let currentSettings: TlonSettingsStore = {};
 
-  // Track threads we've participated in (by parentId) - respond without mention requirement
-  const participatedThreads = new Set<string>();
+  // Track recent threads we've participated in so replies can omit a mention.
+  const participatedThreads = createParticipatedThreadTracker();
 
   // Track DM senders per session to detect shared sessions (security warning)
   const dmSendersBySession = new Map<string, Set<string>>();
@@ -894,8 +895,8 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
   };
 
   // Firehose handler for all DM messages (/v3)
-  // Track which DM invites we've already processed to avoid duplicate accepts
-  const processedDmInvites = new Set<string>();
+  // Track processed DM invites only while they remain in the active /v3 snapshot.
+  const processedDmInvites = createActiveSnapshotTracker();
 
   const handleChatFirehose = async (
     event: unknown,
@@ -904,9 +905,16 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
     try {
       // Handle DM invite lists (arrays)
       if (Array.isArray(event)) {
-        for (const invite of event as DmInvite[]) {
-          const ship = normalizeShip(invite.ship || "");
-          if (!ship || processedDmInvites.has(ship)) {
+        // UrbitSSEClient awaits each handler before reading and acking the next fact,
+        // so snapshot replacement and invite side effects cannot overlap.
+        // The /v3 invite array is the active snapshot. Forget ships that left it
+        // instead of retaining every invite seen during the monitor lifetime.
+        const ships = processedDmInvites.beginSnapshot(
+          (event as DmInvite[]).map((invite) => normalizeShip(invite.ship || "")).filter(Boolean),
+        );
+
+        for (const ship of ships) {
+          if (processedDmInvites.has(ship)) {
             continue;
           }
 
@@ -926,11 +934,9 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
             continue;
           }
 
+          const allowed = await isDmAllowedWithIngress(ship, effectiveDmAllowlist);
           // Auto-accept if on allowlist and auto-accept is enabled
-          if (
-            effectiveAutoAcceptDmInvites &&
-            (await isDmAllowedWithIngress(ship, effectiveDmAllowlist))
-          ) {
+          if (effectiveAutoAcceptDmInvites && allowed) {
             try {
               await api.poke({
                 app: "chat",
@@ -946,14 +952,14 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise<v
           }
 
           // If owner is configured and ship is not on allowlist, queue approval
-          if (effectiveOwnerShip && !(await isDmAllowedWithIngress(ship, effectiveDmAllowlist))) {
+          if (effectiveOwnerShip && !allowed) {
             const approval = createPendingApproval({
               type: "dm",
               requestingShip: ship,
               messagePreview: "(DM invite - no message yet)",
             });
             await queueApprovalRequest(approval);
-            processedDmInvites.add(ship); // Mark as processed to avoid duplicate notifications
+            processedDmInvites.add(ship);
           }
         }
         return;

extensions/tlon/src/monitor/tracking.test.ts

@@ -0,0 +1,47 @@
+// Tlon monitor tracking tests cover thread eviction and snapshot lifecycle.
+import { describe, expect, it } from "vitest";
+import { createActiveSnapshotTracker, createParticipatedThreadTracker } from "./tracking.js";
+
+describe("createParticipatedThreadTracker", () => {
+  it("evicts the least recently used thread at the configured limit", () => {
+    const tracker = createParticipatedThreadTracker(3);
+    tracker.add("oldest");
+    tracker.add("refreshed");
+    tracker.add("recent");
+
+    expect(tracker.has("refreshed")).toBe(true);
+    tracker.add("newest");
+
+    expect(tracker.has("oldest")).toBe(false);
+    expect(tracker.has("refreshed")).toBe(true);
+    expect(tracker.has("recent")).toBe(true);
+    expect(tracker.has("newest")).toBe(true);
+  });
+});
+
+describe("createActiveSnapshotTracker", () => {
+  it("forgets processed keys after they leave the active snapshot", () => {
+    const tracker = createActiveSnapshotTracker();
+    expect(tracker.beginSnapshot(["active", "removed"])).toEqual(new Set(["active", "removed"]));
+    tracker.add("active");
+    tracker.add("removed");
+
+    tracker.beginSnapshot(["active"]);
+    expect(tracker.has("active")).toBe(true);
+    expect(tracker.has("removed")).toBe(false);
+
+    tracker.beginSnapshot(["active", "removed"]);
+    expect(tracker.has("removed")).toBe(false);
+  });
+
+  it("does not impose a count cap on the authoritative active snapshot", () => {
+    const tracker = createActiveSnapshotTracker();
+    const keys = Array.from({ length: 2_001 }, (_, index) => `invite-${index}`);
+    const active = tracker.beginSnapshot(keys);
+    for (const key of active) {
+      tracker.add(key);
+    }
+
+    expect(keys.every((key) => tracker.has(key))).toBe(true);
+  });
+});

extensions/tlon/src/monitor/tracking.ts

@@ -0,0 +1,40 @@
+// Tlon monitor module owns bounded and snapshot-scoped identifier tracking.
+import { createDedupeCache } from "../../runtime-api.js";
+
+const TLON_PARTICIPATED_THREAD_LIMIT = 2_000;
+
+export function createParticipatedThreadTracker(limit = TLON_PARTICIPATED_THREAD_LIMIT) {
+  const cache = createDedupeCache({ ttlMs: 0, maxSize: limit });
+
+  return {
+    add: (parentId: string) => {
+      cache.check(parentId);
+    },
+    has: (parentId: string) => {
+      if (!cache.peek(parentId)) {
+        return false;
+      }
+      // Mention-free replies refresh recency before older participation is evicted.
+      cache.check(parentId);
+      return true;
+    },
+  };
+}
+
+export function createActiveSnapshotTracker() {
+  const processed = new Set<string>();
+
+  return {
+    beginSnapshot: (keys: Iterable<string>): ReadonlySet<string> => {
+      const active = new Set(keys);
+      for (const key of processed) {
+        if (!active.has(key)) {
+          processed.delete(key);
+        }
+      }
+      return active;
+    },
+    has: (key: string) => processed.has(key),
+    add: (key: string) => processed.add(key),
+  };
+}