From ecaddb40ae1266f31d0e6b3a283db1e9f01b931b Mon Sep 17 00:00:00 2001 From: Aamer Akhter Date: Mon, 26 Oct 2026 09:04:53 -0400 Subject: [PATCH] fix(agent): cap inspection cursors per owner and session Cursors shared one global FIFO of 1,024 and were never retired on use, so one agent polling every second evicted every other caller's cursors after about 17 minutes. Each owner and session now keeps its newest 8 cursors, older ones retired first, under the same global backstop; the cursor just used stays among the newest, so a retry of the last call still works. --- .../reference/inspection-and-staging.md | 4 ++-- .../reference/inspection-and-staging.md | 4 ++-- src/agent-inspection.ts | 21 ++++++++++++++-- test/agent-inspection.test.ts | 24 ++++++++++++++++++- 4 files changed, 46 insertions(+), 7 deletions(-) diff --git a/plugins/codeman/skills/codeman/reference/inspection-and-staging.md b/plugins/codeman/skills/codeman/reference/inspection-and-staging.md index 960a1f13..630088c2 100644 --- a/plugins/codeman/skills/codeman/reference/inspection-and-staging.md +++ b/plugins/codeman/skills/codeman/reference/inspection-and-staging.md @@ -45,8 +45,8 @@ these reads do not provide authoritative current-thread tracking. Terminal resizes preserve inspection cursors. Cursors are opaque, scoped to the requesting identity, session, pane lifecycle generation and attributed transcript, expire after 15 minutes, and disappear on server -restart. At most 1024 are retained; capacity eviction can invalidate an older -cursor. A replaced/truncated/rewritten source, reassociated session, foreign cursor +restart. Each caller and session keeps its newest 8 (so retrying the last call +works), at most 1024 in all; an older cursor can be retired. A replaced/truncated/rewritten source, reassociated session, foreign cursor or unavailable native source returns `409 CONFLICT`. Inspect again without `--since` to establish a new tail baseline; this may omit older evidence. diff --git a/skills/codeman/reference/inspection-and-staging.md b/skills/codeman/reference/inspection-and-staging.md index 960a1f13..630088c2 100644 --- a/skills/codeman/reference/inspection-and-staging.md +++ b/skills/codeman/reference/inspection-and-staging.md @@ -45,8 +45,8 @@ these reads do not provide authoritative current-thread tracking. Terminal resizes preserve inspection cursors. Cursors are opaque, scoped to the requesting identity, session, pane lifecycle generation and attributed transcript, expire after 15 minutes, and disappear on server -restart. At most 1024 are retained; capacity eviction can invalidate an older -cursor. A replaced/truncated/rewritten source, reassociated session, foreign cursor +restart. Each caller and session keeps its newest 8 (so retrying the last call +works), at most 1024 in all; an older cursor can be retired. A replaced/truncated/rewritten source, reassociated session, foreign cursor or unavailable native source returns `409 CONFLICT`. Inspect again without `--since` to establish a new tail baseline; this may omit older evidence. diff --git a/src/agent-inspection.ts b/src/agent-inspection.ts index c48c90fc..eec6c159 100644 --- a/src/agent-inspection.ts +++ b/src/agent-inspection.ts @@ -18,6 +18,8 @@ export interface InspectionScope { } interface CursorState { scope: string; + /** Owner and session only: the unit the per-scope cap counts. */ + holder: string; path: string; identity: string; offset: number; @@ -47,12 +49,16 @@ async function anchor(handle: FileHandle, offset: number): Promise { return createHash('sha256').update(head).update(tail).digest('hex'); } +/** Cursors kept per owner and session; older ones are retired first. */ +export const INSPECT_CURSORS_PER_SCOPE = 8; + export class InspectionReader { private readonly cursors = new Map(); constructor( private readonly now = Date.now, private readonly capacity = 1024, - private readonly ttlMs = 900000 + private readonly ttlMs = 900000, + private readonly perScope = INSPECT_CURSORS_PER_SCOPE ) {} async read(path: string, scope: InspectionScope, maxBytes: number, since?: string) { @@ -160,10 +166,21 @@ export class InspectionReader { const cursor = randomUUID(); // No await between capacity enforcement and insertion: concurrent reads // must not all reserve the same remaining slot while hashing anchors. - for (const [key, value] of this.cursors) if (value.expiresAt <= this.now()) this.cursors.delete(key); + // Each owner and session keeps only its newest cursors, so one agent polling in + // a loop retires its own oldest ones instead of pushing every other caller's out + // of the shared map. The cursor just used stays among the newest, so a retry of + // the last call still works. The global capacity is only a backstop. + const holder = JSON.stringify([scope.owner, scope.sessionId]); + const mine: string[] = []; + for (const [key, value] of this.cursors) { + if (value.expiresAt <= this.now()) this.cursors.delete(key); + else if (value.holder === holder) mine.push(key); + } + while (mine.length >= this.perScope) this.cursors.delete(mine.shift()!); while (this.cursors.size >= this.capacity) this.cursors.delete(this.cursors.keys().next().value!); this.cursors.set(cursor, { scope: scopeKey, + holder, path, identity, offset: offset + consumed, diff --git a/test/agent-inspection.test.ts b/test/agent-inspection.test.ts index 989f005b..cadbf53f 100644 --- a/test/agent-inspection.test.ts +++ b/test/agent-inspection.test.ts @@ -3,7 +3,7 @@ import { afterEach, beforeEach, expect, it } from 'vitest'; import { mkdtemp, writeFile, appendFile, rename, rm } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { InspectionReader } from '../src/agent-inspection.js'; +import { INSPECT_CURSORS_PER_SCOPE, InspectionReader } from '../src/agent-inspection.js'; let dir: string; let file: string; let reader: InspectionReader; @@ -150,6 +150,28 @@ it('keeps cursor capacity bounded across concurrent reads', async () => { const resumed = await Promise.allSettled(results.map((result) => reader.read(file, scope, 100, result.cursor))); expect(resumed.filter((result) => result.status === 'fulfilled').length).toBeLessThanOrEqual(2); }); +it('caps cursors per owner and session, so one polling scope cannot evict another', async () => { + await writeFile(file, '{}\n'); + const other = { ...scope, owner: 'bob', sessionId: 't' }; + const bobs = await reader.read(file, other, 100); + // Alice polls every second for well past the old global FIFO of 1,024. + let last = await reader.read(file, scope, 100); + for (let i = 0; i < 1100; i++) last = await reader.read(file, scope, 100, last.cursor); + // Bob's cursor survives, and so does a retry of Alice's last call. + await expect(reader.read(file, other, 100, bobs.cursor)).resolves.toMatchObject({ evidence: '' }); + await expect(reader.read(file, scope, 100, last.cursor)).resolves.toMatchObject({ evidence: '' }); + // Alice keeps only her newest cursors: the oldest of one more than the cap is retired. + const issued: string[] = []; + for (let i = 0; i <= INSPECT_CURSORS_PER_SCOPE; i++) issued.push((await reader.read(file, scope, 100)).cursor); + await expect(reader.read(file, scope, 100, issued[0])).rejects.toThrow(/cursor/i); + await expect(reader.read(file, scope, 100, issued[issued.length - 1])).resolves.toBeDefined(); + // The cap counts owner and session: a new pane generation of Alice's session shares + // it, Bob's session does not. + const nextPane = { ...scope, generation: 'pane-2' }; + for (let i = 0; i < INSPECT_CURSORS_PER_SCOPE; i++) await reader.read(file, nextPane, 100); + await expect(reader.read(file, scope, 100, issued[issued.length - 1])).rejects.toThrow(/cursor/i); + await expect(reader.read(file, other, 100, bobs.cursor)).resolves.toBeDefined(); +}); it('keeps an incomplete UTF8 sequence pending until its JSONL record is complete', async () => { const row = Buffer.from('{"text":"😀"}\n'); const split = row.indexOf(Buffer.from('😀')) + 2;