diff --git a/backend/__tests__/unit/services/agentMemoryService.systemExchanges.test.ts b/backend/__tests__/unit/services/agentMemoryService.systemExchanges.test.ts index 8df94b6b0..61da58409 100644 --- a/backend/__tests__/unit/services/agentMemoryService.systemExchanges.test.ts +++ b/backend/__tests__/unit/services/agentMemoryService.systemExchanges.test.ts @@ -179,6 +179,114 @@ describe('appendSystemExchange (DB-backed)', () => { expect(codex.sections.system_exchanges.entries[0].peers).toEqual(['default']); }); + // dedupeWindowMs — opt-in repeat suppression for constant-payload kinds. + // The ring is fixed at 50; a repeat writer with an unvarying takeaway evicts + // the entries the ring exists to preserve. Measured live before this landed: + // ~45 of 50 slots held one byte-identical loop-trip notice. + describe('dedupeWindowMs', () => { + const WINDOW = 6 * 60 * 60 * 1000; + const trip = { + ...baseEntry, + kind: 'agent-dm-loop-trip', + takeaway: '8 consecutive bot turns within 30 min — guard tripped', + dedupeWindowMs: WINDOW, + }; + + it('suppresses a byte-identical repeat inside the window without touching revision', async () => { + await appendSystemExchange({ ...trip, ts: new Date('2026-05-03T00:00:00Z') }); + const second = await appendSystemExchange({ ...trip, ts: new Date('2026-05-03T01:00:00Z') }); + expect(second?.deduped).toBe(true); + + const doc = await AgentMemory.findOne({ agentName: 'pixel', instanceId: 'default' }).lean(); + expect(doc.sections.system_exchanges.entries).toHaveLength(1); + // revision is the delta cursor drivers read; a suppressed append must not + // advance it, or a seat re-fetches an envelope that did not change. + expect(doc.revision).toBe(1); + }); + + it('records the recurrence once the window has passed', async () => { + await appendSystemExchange({ ...trip, ts: new Date('2026-05-03T00:00:00Z') }); + const later = await appendSystemExchange({ ...trip, ts: new Date('2026-05-03T06:00:01Z') }); + expect(later?.deduped).toBeUndefined(); + + const doc = await AgentMemory.findOne({ agentName: 'pixel', instanceId: 'default' }).lean(); + expect(doc.sections.system_exchanges.entries).toHaveLength(2); + }); + + // The control that stops this becoming "suppress everything". Only the + // NEWEST entry is compared, and only on all three of kind/pod/takeaway — + // so a different pod, a different kind, or a real per-event payload is + // never suppressed, which is what makes the option safe to leave off by + // default for every other caller. + // + // One term per case, deliberately. Chaining all three variations into a + // single four-append test proves nothing about any of them: the comparison + // only looks at the NEWEST entry, so each variant is checked against the + // previous VARIANT rather than against the base, and they differ on some + // other term anyway. Delete the kind term or the takeaway term from the + // comparison and that chained test stays green. Two appends per case — + // base, then one changed field — is what makes each term load-bearing. + it('does not suppress a different pod', async () => { + const ts = new Date('2026-05-03T00:00:00Z'); + await appendSystemExchange({ ...trip, ts }); + await appendSystemExchange({ ...trip, surfacePodId: '69f7b89aabbccddeeff00022', ts }); + + const doc = await AgentMemory.findOne({ agentName: 'pixel', instanceId: 'default' }).lean(); + expect(doc.sections.system_exchanges.entries).toHaveLength(2); + }); + + it('does not suppress a different kind', async () => { + const ts = new Date('2026-05-03T00:00:00Z'); + await appendSystemExchange({ ...trip, ts }); + await appendSystemExchange({ ...trip, kind: 'agent-dm-conclusion', ts }); + + const doc = await AgentMemory.findOne({ agentName: 'pixel', instanceId: 'default' }).lean(); + expect(doc.sections.system_exchanges.entries).toHaveLength(2); + }); + + it('does not suppress a different takeaway', async () => { + const ts = new Date('2026-05-03T00:00:00Z'); + await appendSystemExchange({ ...trip, ts }); + await appendSystemExchange({ ...trip, takeaway: 'something else', ts }); + + const doc = await AgentMemory.findOne({ agentName: 'pixel', instanceId: 'default' }).lean(); + expect(doc.sections.system_exchanges.entries).toHaveLength(2); + }); + + // Dedupe compares against the newest entry only, so an unrelated append in + // between reopens the window. Pinned as a KNOWN limit rather than left to + // be rediscovered: it is what keeps the check one indexed read instead of + // a scan, and the failure it guards is a run of consecutive repeats. + it('is defeated by an interleaved entry, by design', async () => { + await appendSystemExchange({ ...trip, ts: new Date('2026-05-03T00:00:00Z') }); + await appendSystemExchange({ ...baseEntry, ts: new Date('2026-05-03T00:30:00Z') }); + await appendSystemExchange({ ...trip, ts: new Date('2026-05-03T01:00:00Z') }); + + const doc = await AgentMemory.findOne({ agentName: 'pixel', instanceId: 'default' }).lean(); + expect(doc.sections.system_exchanges.entries).toHaveLength(3); + }); + + // An out-of-order ts yields a negative age. Treating that as "inside the + // window" would silently drop an entry that is genuinely older than the + // one it is being compared to. + it('does not suppress an entry older than the newest one', async () => { + await appendSystemExchange({ ...trip, ts: new Date('2026-05-03T02:00:00Z') }); + await appendSystemExchange({ ...trip, ts: new Date('2026-05-03T01:00:00Z') }); + + const doc = await AgentMemory.findOne({ agentName: 'pixel', instanceId: 'default' }).lean(); + expect(doc.sections.system_exchanges.entries).toHaveLength(2); + }); + + it('appends every repeat when the option is omitted — off by default', async () => { + const { dedupeWindowMs, ...noOpt } = trip; + await appendSystemExchange({ ...noOpt, ts: new Date('2026-05-03T00:00:00Z') }); + await appendSystemExchange({ ...noOpt, ts: new Date('2026-05-03T00:00:01Z') }); + + const doc = await AgentMemory.findOne({ agentName: 'pixel', instanceId: 'default' }).lean(); + expect(doc.sections.system_exchanges.entries).toHaveLength(2); + }); + }); + it('returns null on missing required identity fields', async () => { expect(await appendSystemExchange({ ...baseEntry, agentName: '' })).toBeNull(); expect(await appendSystemExchange({ ...baseEntry, instanceId: '' })).toBeNull(); diff --git a/backend/__tests__/unit/services/systemExchangeTriggers.loopTrip.test.ts b/backend/__tests__/unit/services/systemExchangeTriggers.loopTrip.test.ts new file mode 100644 index 000000000..9633b7229 --- /dev/null +++ b/backend/__tests__/unit/services/systemExchangeTriggers.loopTrip.test.ts @@ -0,0 +1,126 @@ +// @ts-nocheck +// The seam nothing else pinned: that recordAgentDmLoopTrip actually OPTS IN to +// repeat suppression. `dedupeWindowMs` is off by default on +// appendSystemExchange, so a helper-only test passes with the wiring deleted — +// the option would simply never be requested and every trip would append. +// +// This is also the only test of any kind for this trigger; grep for +// `recordAgentDmLoopTrip` before this file existed returned two source files +// and zero tests. + +const mongoose = require('mongoose'); + +const AgentMemory = require('../../../models/AgentMemory'); +const Pod = require('../../../models/Pod'); +const User = require('../../../models/User'); +const { recordAgentDmLoopTrip } = require('../../../services/systemExchangeTriggers'); +const { setupMongoDb, closeMongoDb, clearMongoDb } = require('../../utils/testUtils'); + +describe('recordAgentDmLoopTrip — repeat suppression wiring', () => { + beforeAll(async () => { await setupMongoDb(); }); + afterAll(async () => { await closeMongoDb(); }); + afterEach(async () => { await clearMongoDb(); }); + + const makeDmPod = async () => { + const bots = await User.create([ + { + username: 'openclaw-aria', + email: 'aria@example.test', + password: 'x', + isBot: true, + botMetadata: { agentName: 'openclaw', instanceId: 'aria' }, + }, + { + username: 'openclaw-pixel', + email: 'pixel@example.test', + password: 'x', + isBot: true, + botMetadata: { agentName: 'openclaw', instanceId: 'pixel' }, + }, + ]); + const pod = await Pod.create({ + name: 'aria ↔ pixel', + type: 'agent-dm', + members: bots.map((b) => b._id), + createdBy: bots[0]._id, + }); + return String(pod._id); + }; + + const entriesFor = async (instanceId) => { + const doc = await AgentMemory.findOne({ agentName: 'openclaw', instanceId }).lean(); + return doc?.sections?.system_exchanges?.entries || []; + }; + + it('writes one entry per peer on the first trip', async () => { + const podId = await makeDmPod(); + await recordAgentDmLoopTrip({ podId, ts: new Date('2026-05-03T00:00:00Z') }); + + expect(await entriesFor('aria')).toHaveLength(1); + expect((await entriesFor('pixel'))[0].kind).toBe('agent-dm-loop-trip'); + }); + + it('suppresses a second trip in the same pod inside the window, for BOTH peers', async () => { + const podId = await makeDmPod(); + await recordAgentDmLoopTrip({ podId, ts: new Date('2026-05-03T00:00:00Z') }); + await recordAgentDmLoopTrip({ podId, ts: new Date('2026-05-03T00:31:00Z') }); + await recordAgentDmLoopTrip({ podId, ts: new Date('2026-05-03T01:15:00Z') }); + + // Suppression has to hold on both envelopes: the trigger fans out per peer, + // so a check that only reads one of them would miss half a regression. + expect(await entriesFor('aria')).toHaveLength(1); + expect(await entriesFor('pixel')).toHaveLength(1); + }); + + it('records the recurrence after the 6h window', async () => { + const podId = await makeDmPod(); + await recordAgentDmLoopTrip({ podId, ts: new Date('2026-05-03T00:00:00Z') }); + await recordAgentDmLoopTrip({ podId, ts: new Date('2026-05-03T06:00:01Z') }); + + expect(await entriesFor('aria')).toHaveLength(2); + expect(await entriesFor('pixel')).toHaveLength(2); + }); + + it('does not suppress across pods', async () => { + const podA = await makeDmPod(); + await recordAgentDmLoopTrip({ podId: podA, ts: new Date('2026-05-03T00:00:00Z') }); + await clearOnlyPods(); + const podB = await makeDmPod(); + await recordAgentDmLoopTrip({ podId: podB, ts: new Date('2026-05-03T00:10:00Z') }); + + expect(await entriesFor('aria')).toHaveLength(2); + }); + + // Pods are recreated rather than cleared wholesale so the memory envelopes + // written by the first trip survive into the second — clearing everything + // would make the cross-pod case pass for the wrong reason. + async function clearOnlyPods() { + await Pod.deleteMany({}); + await User.deleteMany({}); + } + + it('ignores a non-agent-dm pod entirely', async () => { + const bot = await User.create({ + username: 'openclaw-solo', + email: 'solo@example.test', + password: 'x', + isBot: true, + botMetadata: { agentName: 'openclaw', instanceId: 'solo' }, + }); + const pod = await Pod.create({ + name: 'a channel', + type: 'chat', + members: [bot._id], + createdBy: bot._id, + }); + await recordAgentDmLoopTrip({ podId: String(pod._id) }); + expect(await entriesFor('solo')).toHaveLength(0); + }); + + it('never throws on a bad podId — the trigger is fire-and-forget', async () => { + await expect(recordAgentDmLoopTrip({ podId: 'not-an-object-id' })).resolves.toBeUndefined(); + await expect( + recordAgentDmLoopTrip({ podId: String(new mongoose.Types.ObjectId()) }), + ).resolves.toBeUndefined(); + }); +}); diff --git a/backend/services/agentMemoryService.ts b/backend/services/agentMemoryService.ts index 4a0a77ee2..c408d9319 100644 --- a/backend/services/agentMemoryService.ts +++ b/backend/services/agentMemoryService.ts @@ -462,11 +462,22 @@ export interface AppendSystemExchangeArgs { peers?: string[]; takeaway?: string; ts?: Date; + // Opt-in repeat suppression for kinds whose payload is a CONSTANT. Skip the + // append when the newest entry already carries the same + // (kind, surfacePodId, takeaway) and is younger than this many ms. + // + // This exists because `entries` is a fixed-size ring + // (SYSTEM_EXCHANGE_ENTRY_CAP = 50) and a repeat writer with an unvarying + // payload evicts precisely the entries the ring exists to preserve. Measured + // on one live envelope: 50 entries, ~45 of them byte-identical loop-trip + // notices, leaving two entries of any other kind. Off by default — a caller + // whose takeaway carries real per-event content must keep every append. + dedupeWindowMs?: number; } export async function appendSystemExchange( args: AppendSystemExchangeArgs, -): Promise<{ revision: number } | null> { +): Promise<{ revision: number; deduped?: boolean } | null> { const { agentName, instanceId, @@ -476,6 +487,7 @@ export async function appendSystemExchange( peers = [], takeaway = '', ts = new Date(), + dedupeWindowMs = 0, } = args; if (!agentName || !instanceId) { @@ -497,6 +509,36 @@ export async function appendSystemExchange( takeaway: truncateTakeaway(takeaway), }; + // Repeat suppression, when the caller opted in. Deliberately a read-then- + // decide rather than a filtered update: the update below is an UPSERT, and a + // filter that fails to match would insert a second envelope rather than do + // nothing. So this races — two simultaneous trips can both see no duplicate + // and both append. That is a bounded loss of exactly the property being + // bought (2 entries instead of 1) and it is not the failure this guards + // against, which is 45 identical entries accumulated one at a time over days. + if (dedupeWindowMs > 0) { + const existing = await AgentMemory.findOne({ agentName, instanceId }) + .select({ revision: 1, 'sections.system_exchanges.entries': { $slice: 1 } }) + .lean<{ + revision?: number; + sections?: { system_exchanges?: { entries?: ISystemExchangeEntry[] } }; + } | null>(); + const newest = existing?.sections?.system_exchanges?.entries?.[0]; + if (newest + && newest.kind === entry.kind + && String(newest.surfacePodId) === entry.surfacePodId + && String(newest.takeaway ?? '') === entry.takeaway) { + const newestTs = newest.ts instanceof Date ? newest.ts : new Date(String(newest.ts || '')); + const age = entry.ts.getTime() - newestTs.getTime(); + // `age >= 0` matters: an out-of-order `ts` (a backfill, a clock skew) + // yields a negative age, and treating that as "within the window" would + // silently drop a genuinely older entry the ring should still receive. + if (!Number.isNaN(newestTs.getTime()) && age >= 0 && age < dedupeWindowMs) { + return { revision: existing?.revision ?? 1, deduped: true }; + } + } + } + // Single atomic op — works whether the doc exists, the `sections` envelope // exists, or `sections.system_exchanges` exists. Mongo auto-creates // intermediate path nodes for `$push`. `$setOnInsert` handles the upsert diff --git a/backend/services/systemExchangeTriggers.ts b/backend/services/systemExchangeTriggers.ts index e5b16775c..ac0aecf29 100644 --- a/backend/services/systemExchangeTriggers.ts +++ b/backend/services/systemExchangeTriggers.ts @@ -285,6 +285,20 @@ export async function recordAgentDmLoopTrip(args: RecordAgentDmLoopTripArgs): Pr const surfaceLabel = surfaceLabelFor(podType, podName, podId); const takeaway = '8 consecutive bot turns within 30 min — guard tripped'; + // This is the one trigger whose takeaway is a CONSTANT — every trip writes + // the same string, so N trips in the same pod are N indistinguishable + // entries in a 50-slot ring. Measured on a live envelope before this + // landed: 50 entries, ~45 of them this exact notice, and exactly two + // entries of any other kind survived. A seat that saves durable state + // perfectly still watched it evicted by a writer it does not control. + // + // Six hours, scoped per (kind, pod): long enough that a pod stuck in a + // repeating loop contributes one entry rather than dozens, short enough + // that a genuinely separate recurrence on another day is still recorded. + // The console.warn in agentMentionService is unaffected — every trip is + // still observable there; what is suppressed is only the memory append. + const LOOP_TRIP_DEDUPE_WINDOW_MS = 6 * 60 * 60 * 1000; + const writes = agents.map(async (a) => { const peers = agents .filter((p) => !(p.agentName === a.agentName && p.instanceId === a.instanceId)) @@ -298,6 +312,7 @@ export async function recordAgentDmLoopTrip(args: RecordAgentDmLoopTripArgs): Pr peers, takeaway, ts, + dedupeWindowMs: LOOP_TRIP_DEDUPE_WINDOW_MS, }); if (result === null) { console.error(