diff --git a/packages/daemon/src/daemon.ts b/packages/daemon/src/daemon.ts index e0548e26e..71a3578d4 100644 --- a/packages/daemon/src/daemon.ts +++ b/packages/daemon/src/daemon.ts @@ -12246,6 +12246,10 @@ export class Daemon { * not yet terminal). Guards startup replay from re-admitting a row whose entry is already * live in the gate (idempotency — a duplicate/in-flight id is not double-processed). */ private liveInboxIds = new Set() + /** Loop scope → the `trippedAt` epoch this member has already enforced against its own live + * work. A peer owns the trip's warning and its own backlog, so each member stops its turns + * once per latch, on the first admission it refuses, not on every subsequent message. */ + private enforcedLoopScopes = new Map() private isSessionMuted(key: string): boolean { return this.store.isSessionMuted(key) @@ -12478,6 +12482,7 @@ export class Daemon { } const wasOpen = this.store.isLoopGuardOpen(scope) this.store.resetLoopGuard(scope) + this.enforcedLoopScopes.delete(scope) const wasMuted = this.isSessionMuted(key) if (wasMuted) this.setSessionMuted(key, false) if (wasOpen || wasMuted) { @@ -13881,7 +13886,9 @@ export class Daemon { /** Inbox rows intentionally keep the complete normalized message instead of a * duplicated conversation column. A loop trip is rare, so scanning this small, * already-bounded backlog keeps the persisted schema compatible while purging every - * agent/session in the affected conversation. */ + * agent/session in the affected conversation THIS member serves. On a shared store the + * scan sees the whole install's backlog, and a peer's queued row is its holder's to + * discard — deleting it here would destroy work the peer is still about to run. */ private purgeLoopScopeInbox(scope: string): number { let removed = 0 try { @@ -13890,6 +13897,7 @@ export class Daemon { // reports are an unacknowledged outbox. The interrupt/replay paths below // terminalize the former and the CP ACK releases the latter. if (row.hookContext || row.terminalReport) continue + if (!this.servesAgent(row.agentId)) continue let msg: NormalizedMessage try { msg = JSON.parse(row.msg) as NormalizedMessage @@ -13911,15 +13919,10 @@ export class Daemon { return (usesLoopGuard(msg) || includeHook) && this.store.isLoopGuardOpen(loopGuardScope(msg)) } - /** First-open side effects for a durable conversation circuit: purge restart work, - * drop every matching serial queue, cancel live ACP turns across ALL agents sharing - * the conversation, and emit exactly one operator-facing warning. */ - private onLoopGuardTripped( - scope: string, - reason: string, - trigger: { agentId: string; msg: NormalizedMessage; integrationId?: string } - ): void { - const purged = this.purgeLoopScopeInbox(scope) + /** Stop this member's own live work in an open loop scope: drop every matching serial + * queue and cancel live ACP turns across all agents it holds in that conversation. A + * peer's turns are unreachable from here and are its own to stop when it next refuses. */ + private interruptLoopScopeTurns(scope: string): number { const targets = new Map() for (const [key, entry] of this.activeGateEntries) { if (loopGuardScope(entry.msg) !== scope) continue @@ -13938,9 +13941,33 @@ export class Daemon { for (const [key, target] of targets) { this.interruptTurn(target.agentId, key, 'loop protection', target.acpSessionId, { dropQueued: true }) } + const trippedAt = this.store.getLoopGuard(scope)?.trippedAt + this.enforcedLoopScopes.set(scope, trippedAt ?? this.clock.now()) + return targets.size + } + + /** A circuit a peer latched still has to stop this member's live turns, or the loop simply + * keeps running here. Runs once per latch, on the first admission this member refuses. */ + private enforceLatchedLoopScope(scope: string): void { + const trippedAt = this.store.getLoopGuard(scope)?.trippedAt + if (trippedAt === null || trippedAt === undefined) return + if (this.enforcedLoopScopes.get(scope) === trippedAt) return + const interrupted = this.interruptLoopScopeTurns(scope) + if (interrupted > 0) this.log.warn(`loop guard: OPEN ${scope} elsewhere; interrupted=${interrupted} here`) + } + + /** First-open side effects for a durable conversation circuit: purge the restart work this + * member owns, stop its live turns, and emit exactly one operator-facing warning. */ + private onLoopGuardTripped( + scope: string, + reason: string, + trigger: { agentId: string; msg: NormalizedMessage; integrationId?: string } + ): void { + const purged = this.purgeLoopScopeInbox(scope) + const interrupted = this.interruptLoopScopeTurns(scope) this.log.warn( - `loop guard: OPEN ${scope} reason=${reason}; interrupted=${targets.size}, purgedInbox=${purged}; explicit !resume required` + `loop guard: OPEN ${scope} reason=${reason}; interrupted=${interrupted}, purgedInbox=${purged}; explicit !resume required` ) if (trigger.msg.headless) return const conn = this.replyConnFor(trigger.agentId, trigger.integrationId) @@ -13975,9 +14002,12 @@ export class Daemon { ? this.store.recordLoopGuardTurnForInbox(inboxReplayId, scope, this.clock.now(), !isTrustedHumanTurn(msg), limits) : this.store.recordLoopGuardTurn(scope, this.clock.now(), !isTrustedHumanTurn(msg), limits) if (verdict.allowed) return true - if (verdict.trippedNow) + if (verdict.trippedNow) { this.onLoopGuardTripped(scope, verdict.reason ?? 'turn_burst', { agentId, msg, integrationId }) - else this.purgeLoopScopeInbox(scope) + } else { + this.purgeLoopScopeInbox(scope) + this.enforceLatchedLoopScope(scope) + } return false } @@ -14171,6 +14201,9 @@ export class Daemon { // predate that purge; fresh spam is not yet in inbox, so avoid an O(inbox) scan // on every message while the durable latch is open. if (opts?.fromInboxReplay) this.purgeLoopScopeInbox(loopScope) + // The trip may have been owned by a peer, whose interrupts could not reach this + // member's turns. Stop them here once, or the loop keeps running on this member. + this.enforceLatchedLoopScope(loopScope) this.log.warn(`dispatch: skipped ${msg.msgId}; loop guard is open for ${loopScope}`) settleAdmission({ accepted: false, reason: 'loop_protection' }) resolve(null) @@ -14185,13 +14218,18 @@ export class Daemon { // lost its thread. Non-DM wrappers may each have a synthetic outer ts, so // drop them without creating an unbounded set of permanent latches. const verdict = this.store.tripLoopGuard(loopScope, this.clock.now(), 'malformed_platform_event') - if (verdict.trippedNow) + if (verdict.trippedNow) { this.onLoopGuardTripped(loopScope, verdict.reason ?? 'malformed_platform_event', { agentId, msg, integrationId }) - else this.purgeLoopScopeInbox(loopScope) + } else { + // The CAS elected a peer (or the circuit was already open): this member still + // owns stopping its own turns, exactly like the counter-latch path below. + this.purgeLoopScopeInbox(loopScope) + this.enforceLatchedLoopScope(loopScope) + } } else { this.log.warn(`dispatch: dropped malformed Slack platform event ${msg.msgId}`) } diff --git a/packages/daemon/src/store/local-store.ts b/packages/daemon/src/store/local-store.ts index d5a1767e0..3f91e4c42 100644 --- a/packages/daemon/src/store/local-store.ts +++ b/packages/daemon/src/store/local-store.ts @@ -4187,67 +4187,69 @@ export class LocalStore { WHERE trippedAt IS NULL AND windowStartedAt <= @cutoff AND automaticWindowStartedAt <= @cutoff` ) .run({ cutoff: now - limits.windowMs }) - const current = this.getLoopGuard(scopeKey) - if (current?.trippedAt !== null && current?.trippedAt !== undefined) { - return { - allowed: false, - trippedNow: false, - totalCount: current.totalCount, - automaticCount: current.automaticCount, - ...(current.reason ? { reason: current.reason } : {}) - } - } + // The charge is one relative, window-aware statement, never a JS read-modify-write: + // pool members sharing this store charge the same conversation concurrently, and an + // absolute upsert would let the faster loop lose exactly the increments that matter. + // The DO UPDATE guard makes a latched circuit skip the charge and return no row. + const charged = this.db + .prepare( + `INSERT INTO loop_guard + (scopeKey, windowStartedAt, totalCount, automaticWindowStartedAt, automaticCount, trippedAt, reason) + VALUES (@scopeKey, @now, 1, @now, @automatic, NULL, NULL) + ON CONFLICT(scopeKey) DO UPDATE SET + windowStartedAt=CASE WHEN @now - loop_guard.windowStartedAt >= @windowMs + THEN @now ELSE loop_guard.windowStartedAt END, + totalCount=CASE WHEN @now - loop_guard.windowStartedAt >= @windowMs + THEN 1 ELSE loop_guard.totalCount + 1 END, + automaticWindowStartedAt=CASE WHEN @automatic = 0 + OR @now - loop_guard.automaticWindowStartedAt >= @windowMs + THEN @now ELSE loop_guard.automaticWindowStartedAt END, + automaticCount=CASE WHEN @automatic = 0 THEN 0 + WHEN @now - loop_guard.automaticWindowStartedAt >= @windowMs + THEN 1 ELSE loop_guard.automaticCount + 1 END + WHERE loop_guard.trippedAt IS NULL + RETURNING totalCount, automaticCount` + ) + .get({ scopeKey, now, automatic: automatic ? 1 : 0, windowMs: limits.windowMs }) as + { totalCount: number; automaticCount: number } | undefined + if (!charged) return this.latchedLoopGuardVerdict(scopeKey) - const totalWindowExpired = !current || now - current.windowStartedAt >= limits.windowMs - const automaticWindowExpired = !current || now - current.automaticWindowStartedAt >= limits.windowMs - const windowStartedAt = totalWindowExpired ? now : current.windowStartedAt - const totalCount = totalWindowExpired ? 1 : current.totalCount + 1 - const automaticWindowStartedAt = automaticWindowExpired || !automatic ? now : current.automaticWindowStartedAt - const automaticCount = automatic ? (automaticWindowExpired ? 1 : current.automaticCount + 1) : 0 + const totalCount = Number(charged.totalCount) + const automaticCount = Number(charged.automaticCount) const reason = automaticCount > limits.maxAutomatic ? 'automatic_turn_burst' : totalCount > limits.maxTotal ? 'turn_rate_burst' : undefined - const trippedAt = reason ? now : null - - this.db - .prepare( - `INSERT INTO loop_guard - (scopeKey, windowStartedAt, totalCount, automaticWindowStartedAt, automaticCount, trippedAt, reason) - VALUES (@scopeKey, @windowStartedAt, @totalCount, @automaticWindowStartedAt, @automaticCount, @trippedAt, @reason) - ON CONFLICT(scopeKey) DO UPDATE SET - windowStartedAt=excluded.windowStartedAt, - totalCount=excluded.totalCount, - automaticWindowStartedAt=excluded.automaticWindowStartedAt, - automaticCount=excluded.automaticCount, - trippedAt=excluded.trippedAt, - reason=excluded.reason` - ) - .run({ - scopeKey, - windowStartedAt, - totalCount, - automaticWindowStartedAt, - automaticCount, - trippedAt, - reason: reason ?? null - }) + if (!reason) return { allowed: true, trippedNow: false, totalCount, automaticCount } + // The verdict is computed from what was actually stored, so the latch is a CAS: a + // member that loses it still refuses the turn but runs no duplicate side effects. + const latched = + this.db + .prepare('UPDATE loop_guard SET trippedAt=@now, reason=@reason WHERE scopeKey=@scopeKey AND trippedAt IS NULL') + .run({ scopeKey, now, reason }).changes === 1 + return { allowed: false, trippedNow: latched, totalCount, automaticCount, reason } + } + /** The verdict for a scope another writer already latched: refuse, own no side effects. */ + private latchedLoopGuardVerdict(scopeKey: string): LoopGuardVerdict { + const current = this.getLoopGuard(scopeKey) return { - allowed: reason === undefined, - trippedNow: reason !== undefined, - totalCount, - automaticCount, - ...(reason ? { reason } : {}) + allowed: false, + trippedNow: false, + totalCount: Number(current?.totalCount ?? 0), + automaticCount: Number(current?.automaticCount ?? 0), + ...(current?.reason ? { reason: current.reason } : {}) } } - /** Charge a migrated inbox delivery and advance its marker in one SQLite transaction. - * A crash can therefore neither lose the charge nor charge the same retained row again - * after ownership moves. A tripping delivery is intentionally left at marker 0: the - * newly-open durable circuit makes its whole scope terminal and replay purges it. */ + /** Charge a migrated inbox delivery and advance its marker in one transaction, so a crash + * can neither lose the charge nor charge the same retained row again after ownership moves. + * The transaction buys atomicity only — the charge itself is a relative SQL statement and + * never depends on an exclusive writer, which a shared PostgreSQL store cannot give it. + * A tripping delivery is intentionally left at marker 0: the newly-open durable circuit + * makes its whole scope terminal and replay purges it. */ recordLoopGuardTurnForInbox( inboxId: string, scopeKey: string, @@ -4270,38 +4272,25 @@ export class LocalStore { } } - /** Open a loop circuit immediately for a structurally-invalid platform event. */ + /** Open a loop circuit immediately for a structurally-invalid platform event. The latch + * is a single guarded statement, so concurrent members elect exactly one side-effect owner. */ tripLoopGuard(scopeKey: string, now: number, reason: string): LoopGuardVerdict { - const current = this.getLoopGuard(scopeKey) - if (current?.trippedAt !== null && current?.trippedAt !== undefined) { - return { - allowed: false, - trippedNow: false, - totalCount: current.totalCount, - automaticCount: current.automaticCount, - ...(current.reason ? { reason: current.reason } : {}) - } - } - const row = current ?? { - scopeKey, - windowStartedAt: now, - totalCount: 0, - automaticWindowStartedAt: now, - automaticCount: 0 - } - this.db + const latched = this.db .prepare( `INSERT INTO loop_guard (scopeKey, windowStartedAt, totalCount, automaticWindowStartedAt, automaticCount, trippedAt, reason) - VALUES (@scopeKey, @windowStartedAt, @totalCount, @automaticWindowStartedAt, @automaticCount, @trippedAt, @reason) - ON CONFLICT(scopeKey) DO UPDATE SET trippedAt=excluded.trippedAt, reason=excluded.reason` + VALUES (@scopeKey, @now, 0, @now, 0, @now, @reason) + ON CONFLICT(scopeKey) DO UPDATE SET trippedAt=@now, reason=@reason + WHERE loop_guard.trippedAt IS NULL + RETURNING totalCount, automaticCount` ) - .run({ ...row, trippedAt: now, reason }) + .get({ scopeKey, now, reason }) as { totalCount: number; automaticCount: number } | undefined + if (!latched) return this.latchedLoopGuardVerdict(scopeKey) return { allowed: false, trippedNow: true, - totalCount: row.totalCount, - automaticCount: row.automaticCount, + totalCount: Number(latched.totalCount), + automaticCount: Number(latched.automaticCount), reason } } diff --git a/packages/daemon/src/store/postgres-store-worker.js b/packages/daemon/src/store/postgres-store-worker.js index 8ddac610a..9e69db348 100644 --- a/packages/daemon/src/store/postgres-store-worker.js +++ b/packages/daemon/src/store/postgres-store-worker.js @@ -8,6 +8,8 @@ let client function rewrite(sql) { let out = sql + // PostgreSQL has no SQLite exclusive-writer transaction, so IMMEDIATE is dropped: a + // shared-store statement must be a CAS or a relative write, never a read-then-write. .replace(/BEGIN\s+IMMEDIATE/gi, 'BEGIN') .replace(/INTEGER\s+PRIMARY\s+KEY\s+AUTOINCREMENT/gi, 'BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY') .replace(/\bINTEGER\b/gi, 'BIGINT') diff --git a/packages/daemon/test/daemon-loop-guard-pool.test.ts b/packages/daemon/test/daemon-loop-guard-pool.test.ts new file mode 100644 index 000000000..0aa73f466 --- /dev/null +++ b/packages/daemon/test/daemon-loop-guard-pool.test.ts @@ -0,0 +1,210 @@ +import { describe, it, expect, vi } from 'vitest' +import { mkdtempSync, mkdirSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { DatabaseSync } from 'node:sqlite' +import { Daemon } from '../src/daemon.js' +import { LocalStore } from '../src/store/local-store.js' +import { statePath } from '../src/paths.js' +import { FakeClock } from './cp/fake-clock.js' + +/** + * #1038 — the loop guard's durable backlog is one inbox table for the whole install, but a + * trip can only interrupt the turns living in the memory of the member that ran it. So the + * trip acts on what this member serves: a peer's queued row is skipped, never destroyed, + * and the peer stops its own turns on the first admission its latched circuit refuses. + */ + +const AGENT_A = 'bot-a' +const AGENT_B = 'bot-b' +const GROUP_A = '11111111-1111-4111-8111-111111111111' +const GROUP_B = '22222222-2222-4222-8222-222222222222' +const SCOPE = 'slack:C1:T1' + +function scaffold(): string { + const root = mkdtempSync(join(tmpdir(), 'ac-loop-guard-pool-')) + writeFileSync( + join(root, 'config.json'), + JSON.stringify({ + version: 1, + controlPlane: { enabled: false }, + runtimes: { claude: { command: 'node', args: ['unused'] } } + }) + ) + for (const id of [AGENT_A, AGENT_B]) { + const adir = join(root, 'agents', id) + mkdirSync(adir, { recursive: true }) + writeFileSync( + join(adir, 'agent.json'), + JSON.stringify({ + id, + name: id, + status: 'active', + runtime: 'claude', + workspace: { mode: 'from-scratch', path: join(adir, 'workspace') }, + integrations: [], + output: { mode: 'low' } + }) + ) + } + return root +} + +async function boot(root: string, daemonId: string, scope: 'frame' | 'legacy') { + const daemon = new Daemon({ root, hostFactory: () => ({}) as any, clock: new FakeClock() }) + await daemon.start() + const inner = daemon as any + inner.cfg.daemonId = daemonId + inner.cpClient = { + organizationScope: () => scope, + state: 'READY', + stop: async () => {}, + releaseDuties: vi.fn(async () => {}), + reportDutiesNow: vi.fn(() => {}), + fetchDutyAgent: vi.fn() + } + return { daemon, inner } +} + +/** Two members over ONE store: the same root, so both open the same database. */ +async function bootPool() { + const root = scaffold() + const a = await boot(root, 'daemon-a', 'frame') + const b = await boot(root, 'daemon-b', 'frame') + const path = statePath(root) + const locals: LocalStore[] = [a.inner.store, b.inner.store] + a.inner.store = new LocalStore({ database: new DatabaseSync(path), shared: true, ownerId: 'daemon-a' }) + b.inner.store = new LocalStore({ database: new DatabaseSync(path), shared: true, ownerId: 'daemon-b' }) + const stop = async () => { + await Promise.all([a.daemon.stop(), b.daemon.stop()]) + for (const local of locals) local.close() + } + return { a, b, shared: a.inner.store as LocalStore, stop } +} + +const grant = (groupId: string, agentId: string) => ({ + groupId, + orgId: 'org-1', + term: '1', + members: [{ kind: 'agent' as const, refId: agentId }] +}) +const hold = (inner: any, groupId: string, agentId: string) => inner.duties.applyGrant([grant(groupId, agentId)]) + +/** A plain Slack reply inside thread T1 — the coordinates that key SCOPE. */ +function scopedMessage(agentId: string, ts: string) { + return { + msgId: `slack:C1:${ts}`, + platform: 'slack', + channel: 'C1', + thread: 'T1', + isDm: false, + source: 'user', + text: 'loop', + sender: { id: `U-${agentId}`, name: agentId, isBot: true }, + headless: true + } +} + +function seedInbox(store: LocalStore, id: string, agentId: string, ts: string): void { + store.appendInbox({ + id, + sessionKey: `slack:C1:T1:${agentId}`, + agentId, + msg: JSON.stringify(scopedMessage(agentId, ts)), + enqueuedAt: `0000000000000000000${ts.at(-1)}` + }) +} + +/** The Slack poison shape — an anonymous, empty, attachment-less user turn — in a DM. */ +function malformedDm(ts: string) { + return { + msgId: `slack:D1:${ts}`, + platform: 'slack', + channel: 'D1', + thread: ts, + isDm: true, + source: 'user', + text: '', + attachments: [], + sender: { id: 'unknown', name: 'unknown', isBot: false }, + headless: true + } +} + +/** A live turn in the member's serial gate: state only that member can cancel. */ +function liveTurn(inner: any, key: string, agentId: string, msg?: any): any { + const entry = { + agentId, + msg: msg ?? scopedMessage(agentId, '900.1'), + initAbort: new AbortController(), + resolve: () => {}, + reject: () => {} + } + inner.activeGateEntries.set(key, entry) + return entry +} + +describe('loop guard on a daemon pool acts only on what the member serves (#1038)', () => { + it("a trip neither deletes a peer's durable rows nor pretends to stop its turns", async () => { + const { a, b, shared, stop } = await bootPool() + hold(a.inner, GROUP_A, AGENT_A) + hold(b.inner, GROUP_B, AGENT_B) + seedInbox(shared, 'inbox-a', AGENT_A, '100.1') + seedInbox(shared, 'inbox-b', AGENT_B, '200.1') + const turnA = liveTurn(a.inner, 'key-a', AGENT_A) + const turnB = liveTurn(b.inner, 'key-b', AGENT_B) + + shared.tripLoopGuard(SCOPE, 1_000, 'turn_rate_burst') + a.inner.onLoopGuardTripped(SCOPE, 'turn_rate_burst', { agentId: AGENT_A, msg: scopedMessage(AGENT_A, '100.1') }) + + // A's own backlog and turn are terminal; B's row is left for B, which is still running it. + expect(shared.listInboxBySessionKeyFifo().map((row: any) => row.id)).toEqual(['inbox-b']) + expect(turnA.cancelledReason).toBe('loop protection') + expect(turnB.cancelledReason).toBeUndefined() + + // B enforces the latch on its own work the first time the open circuit refuses a turn. + b.inner.enforceLatchedLoopScope(SCOPE) + expect(turnB.cancelledReason).toBe('loop protection') + expect(b.inner.purgeLoopScopeInbox(SCOPE)).toBe(1) + expect(shared.listInboxBySessionKeyFifo()).toEqual([]) + + // The latch is enforced once per member, not on every subsequent refusal. + const turnBAgain = liveTurn(b.inner, 'key-b2', AGENT_B) + b.inner.enforceLatchedLoopScope(SCOPE) + expect(turnBAgain.cancelledReason).toBeUndefined() + // These heads have no dispatch behind them, so release them before shutdown drains the gate. + for (const member of [a, b]) member.inner.activeGateEntries.clear() + await stop() + }, 15_000) + + it('a member that loses the structural trip still stops its own turns', async () => { + const { a, b, shared, stop } = await bootPool() + hold(a.inner, GROUP_A, AGENT_A) + hold(b.inner, GROUP_B, AGENT_B) + const dmScope = 'slack:D1:dm' + const turnA = liveTurn(a.inner, 'key-dm-a', AGENT_A, malformedDm('900.1')) + + // B latches the DM circuit first; A read it as closed a moment earlier, which is the + // exact race the trip's CAS resolves — A's own trip returns trippedNow: false. + expect(shared.tripLoopGuard(dmScope, 2_000, 'malformed_platform_event').trippedNow).toBe(true) + a.inner.store.isLoopGuardOpen = () => false + await a.inner.dispatch(AGENT_A, malformedDm('300.1')) + + expect(turnA.cancelledReason).toBe('loop protection') + for (const member of [a, b]) member.inner.activeGateEntries.clear() + await stop() + }, 15_000) + + it('a single local daemon still purges the whole conversation backlog', async () => { + const root = scaffold() + const { daemon, inner } = await boot(root, 'daemon-solo', 'legacy') + const store: LocalStore = inner.store + seedInbox(store, 'inbox-a', AGENT_A, '100.1') + seedInbox(store, 'inbox-b', AGENT_B, '200.1') + + // No duty enforcement: every agent is served here, so nothing is left behind. + expect(inner.purgeLoopScopeInbox(SCOPE)).toBe(2) + expect(store.listInboxBySessionKeyFifo()).toEqual([]) + await daemon.stop() + }, 15_000) +}) diff --git a/packages/daemon/test/loop-guard.test.ts b/packages/daemon/test/loop-guard.test.ts index c5c214f29..985b73127 100644 --- a/packages/daemon/test/loop-guard.test.ts +++ b/packages/daemon/test/loop-guard.test.ts @@ -2,12 +2,46 @@ import { describe, expect, it } from 'vitest' import { mkdtempSync } from 'node:fs' import { tmpdir } from 'node:os' import { join } from 'node:path' +import { DatabaseSync } from 'node:sqlite' import { LocalStore } from '../src/store/local-store.js' function dbPath(): string { return join(mkdtempSync(join(tmpdir(), 'ac-loop-guard-')), 'local.sqlite') } +/** Two pool members over ONE store, exactly as an install-wide shared database presents it. */ +function pool(): { a: LocalStore; b: LocalStore; close: () => void } { + const path = dbPath() + const seed = new LocalStore(path) + seed.close() + const a = new LocalStore({ database: new DatabaseSync(path), shared: true, ownerId: 'daemon-a' }) + const b = new LocalStore({ database: new DatabaseSync(path), shared: true, ownerId: 'daemon-b' }) + return { a, b, close: () => [a, b].forEach((store) => store.close()) } +} + +/** Run `concurrent` at the moment the member's next loop-guard write is about to execute — + * the exact window in which a read-modify-write member loses a peer's increment. */ +function interleaveLoopGuardWrite(store: LocalStore, concurrent: () => void): void { + const db = (store as unknown as { db: { prepare: (sql: string) => unknown } }).db + const prepare = db.prepare.bind(db) + let fired = false + db.prepare = (sql: string) => { + const statement = prepare(sql) as Record unknown> + if (!/^\s*INSERT INTO loop_guard/i.test(sql)) return statement + for (const method of ['run', 'get'] as const) { + const original = statement[method]!.bind(statement) + statement[method] = (...args: unknown[]) => { + if (!fired) { + fired = true + concurrent() + } + return original(...args) + } + } + return statement + } +} + describe('LocalStore loop guard', () => { it('tracks consecutive automatic turns, with a trusted human turn resetting only that streak', () => { const s = new LocalStore(dbPath()) @@ -100,3 +134,65 @@ describe('LocalStore loop guard', () => { s.close() }) }) + +describe('LocalStore loop guard on a shared pool store (#1038)', () => { + const limits = { windowMs: 10_000, maxTotal: 100, maxAutomatic: 100 } + + it('sums every member’s charge exactly, including a write that lands mid-charge', () => { + const { a, b, close } = pool() + // A charges while B charges: an absolute upsert derived from a prior read drops B's turn. + interleaveLoopGuardWrite(a, () => { + b.recordLoopGuardTurn('slack:C1:T1', 100, true, limits) + }) + expect(a.recordLoopGuardTurn('slack:C1:T1', 100, true, limits)).toMatchObject({ + allowed: true, + totalCount: 2, + automaticCount: 2 + }) + for (let i = 0; i < 8; i++) (i % 2 === 0 ? a : b).recordLoopGuardTurn('slack:C1:T1', 200 + i, true, limits) + expect(a.getLoopGuard('slack:C1:T1')).toMatchObject({ totalCount: 10, automaticCount: 10 }) + close() + }) + + it('elects exactly one member to own the trip’s side effects', () => { + const { a, b, close } = pool() + const burst = { windowMs: 10_000, maxTotal: 1, maxAutomatic: 100 } + a.recordLoopGuardTurn('slack:C1:T1', 100, false, burst) + // Both members exceed the budget in the same window; only the winner runs the trip. + interleaveLoopGuardWrite(a, () => { + expect(b.recordLoopGuardTurn('slack:C1:T1', 100, false, burst)).toMatchObject({ + allowed: false, + trippedNow: true, + reason: 'turn_rate_burst' + }) + }) + expect(a.recordLoopGuardTurn('slack:C1:T1', 100, false, burst)).toMatchObject({ + allowed: false, + trippedNow: false, + reason: 'turn_rate_burst' + }) + expect(a.getLoopGuard('slack:C1:T1')).toMatchObject({ totalCount: 2, trippedAt: 100 }) + // A concurrent structural trip is elected the same way. + expect(a.tripLoopGuard('slack:C2:T2', 7, 'malformed_platform_event').trippedNow).toBe(true) + expect(b.tripLoopGuard('slack:C2:T2', 9, 'malformed_platform_event')).toMatchObject({ + trippedNow: false, + reason: 'malformed_platform_event' + }) + expect(b.getLoopGuard('slack:C2:T2')).toMatchObject({ trippedAt: 7 }) + close() + }) + + it('never recharges a latched scope, whichever member reads it', () => { + const { a, b, close } = pool() + a.recordLoopGuardTurn('slack:C1:T1', 100, true, limits) + a.tripLoopGuard('slack:C1:T1', 150, 'incident') + expect(b.recordLoopGuardTurn('slack:C1:T1', 200, true, limits)).toMatchObject({ + allowed: false, + trippedNow: false, + totalCount: 1, + reason: 'incident' + }) + expect(a.getLoopGuard('slack:C1:T1')).toMatchObject({ totalCount: 1, trippedAt: 150, reason: 'incident' }) + close() + }) +})