fix(tail): reap closed follow-sessions instead of leaking them

The module-level sessions Map dropped an entry only on an explicit
ssh_tail_stop. A follow-start whose stream closed (remote ended,
connection dropped) but was never explicitly stopped left its state
in the Map for the server's lifetime. The 1 MB ring buffer bounds
per-session memory, but closed sessions were never reaped.

Track readOffset (highest delivered offset) and closedAt per session.
reapClosedSessions() drops a closed session once its buffer is fully
read, or once it has been closed past a grace period even if unread
(the stream is dead). It runs on every follow-start and follow-read.

handleSshTailRead's opening sweep excludes the session it is about to
serve so the caller still gets one final closed:true read; that
session is then reaped at the end of the call.

Tests: a closed + fully-read session is reaped on the next read with
no explicit stop (fails without the fix), a closed session past the
grace window is reaped even if never read, and an open session
survives any sweep.

Author: hunchom

src/tools/tail-tools.js

@@ -21,6 +21,8 @@ const DEFAULT_LINES = 10;
 const DEFAULT_TIMEOUT_MS = 15_000;
 const DEFAULT_MAX_LEN = 10_000;
 const RING_BUFFER_CAP = 1_000_000;
+// Grace period a closed-but-unread session lingers before a sweep reaps it.
+const CLOSED_SESSION_GRACE_MS = 5 * 60_000;
 
 /** Session registry -- module-level Map so tools across calls share it. */
 const sessions = new Map();
@@ -37,13 +39,41 @@ function rememberStopped(id) {
   }
 }
 
+/**
+ * Reap closed follow-sessions so a stream that ended without an explicit
+ * ssh_tail_stop never lingers for the server's lifetime. A closed session
+ * drops when its buffer is fully read, or when it closed past the grace
+ * period (stream dead -- a stale buffer helps nobody).
+ *
+ * `excludeId` is left intact even if reapable: handleSshTailRead passes the
+ * session it is mid-serving so the caller still gets one final closed:true
+ * read before that session is dropped at the end of the call.
+ */
+function reapClosedSessions(now = Date.now(), excludeId = null) {
+  for (const [id, st] of sessions) {
+    if (id === excludeId) continue;
+    if (!st.closed) continue;
+    const fullyRead = st.readOffset >= st.totalBytes;
+    const expired = st.closedAt != null && (now - st.closedAt) >= CLOSED_SESSION_GRACE_MS;
+    if (fullyRead || expired) {
+      sessions.delete(id);
+      rememberStopped(id);
+    }
+  }
+}
+
 /** Exposed for tests to introspect internal state. */
 export function _sessionsForTest() {
   return sessions;
 }
 export function _stoppedIdsForTest() {
   return stoppedIds;
 }
+/** Exposed for tests: drive the closed-session sweep directly. */
+export function _reapClosedSessionsForTest(now) {
+  return reapClosedSessions(now);
+}
+export const _CLOSED_SESSION_GRACE_MS = CLOSED_SESSION_GRACE_MS;
 
 /** Coerce numeric arg to a positive integer, with safe fallback. */
 function safeLines(n, fallback = DEFAULT_LINES) {
@@ -151,7 +181,9 @@ export async function handleSshTailStart({ getConnection, args }) {
         createdAt: Date.now(),
         buffer: '',
         totalBytes: 0,   // lifetime bytes appended (monotonic, pre-truncation)
+        readOffset: 0,   // highest offset a follow-read has delivered to a caller
         closed: false,
+        closedAt: null,  // ms timestamp the stream closed (null while open)
         stream,
       };
 
@@ -164,11 +196,17 @@ export async function handleSshTailStart({ getConnection, args }) {
         }
       }
 
+      function markClosed() {
+        if (state.closed) return;
+        state.closed = true;
+        state.closedAt = Date.now();
+      }
       stream.on('data', (d) => append(stripAnsi(d.toString('utf8'))));
       stream.stderr && stream.stderr.on('data', (d) => append(stripAnsi(d.toString('utf8'))));
-      stream.on('close', () => { state.closed = true; });
-      stream.on('error', () => { state.closed = true; });
+      stream.on('close', markClosed);
+      stream.on('error', markClosed);
 
+      reapClosedSessions();   // sweep stale sessions before adding a new one
       sessions.set(id, state);
       resolve(state);
     });
@@ -198,6 +236,10 @@ export async function handleSshTailStart({ getConnection, args }) {
 export async function handleSshTailRead({ args }) {
   const { session_id, since_offset, format = 'markdown' } = args || {};
 
+  // Sweep other closed/dead sessions; never the one we are about to serve --
+  // it gets one final closed:true read, then is reaped at the end of the call.
+  reapClosedSessions(Date.now(), session_id);
+
   const state = sessions.get(session_id);
   if (!state) {
     return toMcp(fail('ssh_tail_read', `unknown session_id: ${session_id}`), { format });
@@ -241,6 +283,12 @@ export async function handleSshTailRead({ args }) {
     elided_bytes: elided,
   };
 
+  // Caller has now seen everything up to `total`. Advance the read watermark
+  // and reap: a closed + fully-read session drops here -- this very response
+  // already carried closed:true, so the caller knows the stream ended.
+  state.readOffset = Math.max(state.readOffset, total);
+  reapClosedSessions();
+
   return toMcp(
     ok('ssh_tail_read', data, { server: state.server }),
     { format }

tests/test-tail-tools.js

@@ -14,6 +14,8 @@ import {
   buildTailCommand,
   _sessionsForTest,
   _stoppedIdsForTest,
+  _reapClosedSessionsForTest,
+  _CLOSED_SESSION_GRACE_MS,
 } from '../src/tools/tail-tools.js';
 import { unwrapTimeout } from './util-timeout-unwrap.js';
 
@@ -372,6 +374,78 @@ await test('handleSshTailStop: unknown id (never seen) -> structured fail', asyn
   assert.strictEqual(r.isError, true);
 });
 
+// --------------------------------------------------------------------------
+// Closed-session reaping -- a follow-start whose stream closes but is never
+// explicitly ssh_tail_stop'd must not linger in the registry forever.
+// --------------------------------------------------------------------------
+await test('closed + fully-read session is reaped on the next read (no explicit stop)', async () => {
+  const client = new FollowClient();
+  const started = await handleSshTailStart({
+    getConnection: async () => client,
+    args: { server: 's', file: '/f', format: 'json' },
+  });
+  const { session_id } = JSON.parse(started.content[0].text).data;
+  assert(_sessionsForTest().has(session_id), 'session registered');
+
+  client.feed('done\n');
+  await sleep(5);
+  // First read consumes the whole buffer.
+  const r1 = JSON.parse((await handleSshTailRead({ args: { session_id, format: 'json' } })).content[0].text);
+  assert.strictEqual(r1.data.chunk, 'done\n');
+  assert(_sessionsForTest().has(session_id), 'still alive while stream open');
+
+  // Stream closes (remote ended) -- no ssh_tail_stop is ever issued.
+  client.lastStream().emit('close', 0);
+  await sleep(5);
+
+  // Next read sees closed:true AND reaps the now-dead, fully-read session.
+  const r2 = JSON.parse((await handleSshTailRead({ args: { session_id, format: 'json' } })).content[0].text);
+  assert.strictEqual(r2.data.closed, true, 'caller is told the stream closed');
+  assert.strictEqual(_sessionsForTest().has(session_id), false,
+    'closed + fully-read session must be reaped, not leaked');
+  assert(_stoppedIdsForTest().has(session_id),
+    'reaped id is remembered so a later stop is a clean no-op');
+});
+
+await test('closed session past the grace period is reaped by the sweep even if unread', async () => {
+  const client = new FollowClient();
+  const started = await handleSshTailStart({
+    getConnection: async () => client,
+    args: { server: 's', file: '/f', format: 'json' },
+  });
+  const { session_id } = JSON.parse(started.content[0].text).data;
+
+  // Stream closes with buffered-but-never-read data.
+  client.feed('unread bytes\n');
+  client.lastStream().emit('close', 0);
+  await sleep(5);
+  assert(_sessionsForTest().has(session_id), 'closed-but-recent session still lingers');
+
+  // A fresh sweep does nothing yet (within grace).
+  _reapClosedSessionsForTest(Date.now());
+  assert(_sessionsForTest().has(session_id), 'not reaped within grace window');
+
+  // Backdate closedAt past the grace period; the sweep now reaps it.
+  _sessionsForTest().get(session_id).closedAt =
+    Date.now() - _CLOSED_SESSION_GRACE_MS - 1000;
+  _reapClosedSessionsForTest(Date.now());
+  assert.strictEqual(_sessionsForTest().has(session_id), false,
+    'closed session past grace must be reaped even though never read');
+});
+
+await test('an OPEN session is never reaped, no matter how old', async () => {
+  const client = new FollowClient();
+  const started = await handleSshTailStart({
+    getConnection: async () => client,
+    args: { server: 's', file: '/f', format: 'json' },
+  });
+  const { session_id } = JSON.parse(started.content[0].text).data;
+  // Stream stays open. Sweep with a far-future clock.
+  _reapClosedSessionsForTest(Date.now() + _CLOSED_SESSION_GRACE_MS * 100);
+  assert(_sessionsForTest().has(session_id), 'open session must survive any sweep');
+  await handleSshTailStop({ args: { session_id } });
+});
+
 // --------------------------------------------------------------------------
 // Summary
 // --------------------------------------------------------------------------