From 162231126f9f420057b42c6d7cf1396225f39d8d Mon Sep 17 00:00:00 2001 From: zfy0701 <1646270+zfy0701@users.noreply.github.com> Date: Sun, 16 Aug 2026 11:07:17 +0800 Subject: [PATCH 1/2] fix(daemon): scope the loop-guard trip to the member and make its counters atomic MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On a daemon pool the loop guard was global in its destruction and local in its enforcement, and its counters lost increments under concurrent members. `purgeLoopScopeInbox` scanned the whole install-wide inbox and deleted every row in the scope, including rows queued on peers whose in-memory state this process cannot see; the interruption half that followed only walked this process's own maps. So a trip on one member destroyed a peer's durable backlog without ever stopping the peer's live turns. The purge now skips rows whose agent this member does not serve, exactly like the sweeps and the replay path, and a member stops its own turns once per latch on the first admission its open circuit refuses. The counters were a JS read-modify-write followed by an absolute upsert, so two members charging the same conversation both read n and both wrote n + 1 — the undercount is worst exactly when the loop is fastest. Both the charge and the latch are now single relative, window-aware statements with `RETURNING`, so the verdict is computed from what was actually stored and the latch is a CAS that elects exactly one owner for the trip's side effects. That also removes the path's dependence on an exclusive writer, which the shared store cannot give it: the Postgres facade rewrites `BEGIN IMMEDIATE` to a plain `BEGIN`. The remaining transaction buys atomicity between the charge and the inbox marker only, and the rewrite now says why a shared-store statement has to be a CAS or a relative write. Local single-daemon behavior is unchanged: with no duty enforcement every agent is served here, so the purge still clears the whole conversation backlog. Co-Authored-By: Claude Fable 5 --- packages/daemon/src/daemon.ts | 59 ++++-- packages/daemon/src/store/local-store.ts | 135 ++++++-------- .../daemon/src/store/postgres-store-worker.js | 2 + .../test/daemon-loop-guard-pool.test.ts | 176 ++++++++++++++++++ packages/daemon/test/loop-guard.test.ts | 96 ++++++++++ 5 files changed, 382 insertions(+), 86 deletions(-) create mode 100644 packages/daemon/test/daemon-loop-guard-pool.test.ts diff --git a/packages/daemon/src/daemon.ts b/packages/daemon/src/daemon.ts index e0548e26e..953948072 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) 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..5445bf88d --- /dev/null +++ b/packages/daemon/test/daemon-loop-guard-pool.test.ts @@ -0,0 +1,176 @@ +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)}` + }) +} + +/** A live turn in the member's serial gate: state only that member can cancel. */ +function liveTurn(inner: any, key: string, agentId: string): any { + const entry = { + agentId, + 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 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() + }) +}) From 34f9187c657b966dbd11ba26218becc355bd1bea Mon Sep 17 00:00:00 2001 From: zfy0701 <1646270+zfy0701@users.noreply.github.com> Date: Sun, 16 Aug 2026 11:20:36 +0800 Subject: [PATCH 2/2] fix(daemon): enforce the open latch on the structural-trip loser too The trip's CAS gives overlapping callers on losing members `trippedNow: false`, and the malformed-DM branch answered that with a member-scoped purge alone. So a member that lost the race kept its live ACP turns running until some later admission happened to hit the open latch. It now enforces the latch there as well, exactly like the counter-latch path. Co-Authored-By: Claude Fable 5 --- packages/daemon/src/daemon.ts | 9 ++++- .../test/daemon-loop-guard-pool.test.ts | 38 ++++++++++++++++++- 2 files changed, 43 insertions(+), 4 deletions(-) diff --git a/packages/daemon/src/daemon.ts b/packages/daemon/src/daemon.ts index 953948072..71a3578d4 100644 --- a/packages/daemon/src/daemon.ts +++ b/packages/daemon/src/daemon.ts @@ -14218,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/test/daemon-loop-guard-pool.test.ts b/packages/daemon/test/daemon-loop-guard-pool.test.ts index 5445bf88d..0aa73f466 100644 --- a/packages/daemon/test/daemon-loop-guard-pool.test.ts +++ b/packages/daemon/test/daemon-loop-guard-pool.test.ts @@ -115,11 +115,27 @@ function seedInbox(store: LocalStore, id: string, agentId: string, ts: string): }) } +/** 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): any { +function liveTurn(inner: any, key: string, agentId: string, msg?: any): any { const entry = { agentId, - msg: scopedMessage(agentId, '900.1'), + msg: msg ?? scopedMessage(agentId, '900.1'), initAbort: new AbortController(), resolve: () => {}, reject: () => {} @@ -161,6 +177,24 @@ describe('loop guard on a daemon pool acts only on what the member serves (#1038 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')