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;