diff --git a/packages/session-projection/session-projection/src/index.ts b/packages/session-projection/session-projection/src/index.ts index 2f952166cc..5b7a3c1f4c 100644 --- a/packages/session-projection/session-projection/src/index.ts +++ b/packages/session-projection/session-projection/src/index.ts @@ -233,9 +233,11 @@ export class SessionProjectionRegistry extends Service { * log). This is the write side of the persisted projection cache: the * returned rows are the `(key → {stateVersion, observedSeq, state})` part * of the durable `(sessionId, key, stateVersion, observedSeq, state)` - * rows. States are the units' live references — plain JSON by the unit - * contract, treated as immutable; a durable writer snapshots them at its - * own boundary. + * rows. Every `state` is a DETACHED structured clone — never the live + * cell reference: the watermark cache is this registry's authoritative + * mutable state, and a caller reaching the live reference could corrupt + * every subsequent snapshot and frame through it (plain JSON by the unit + * contract, so the clone is total). * @param session - the session whose unit states are checkpointed. * @returns one row per registered key; empty when no unit is registered. */ @@ -246,7 +248,7 @@ export class SessionProjectionRegistry extends Service { rows[registration.def.key] = { stateVersion: registration.def.stateVersion, observedSeq: cell.observedSeq, - state: cell.state, + state: structuredClone(cell.state), } } return rows diff --git a/packages/session-projection/session-projection/tests/registry.spec.ts b/packages/session-projection/session-projection/tests/registry.spec.ts index fc167a1625..17ee5aaf84 100644 --- a/packages/session-projection/session-projection/tests/registry.spec.ts +++ b/packages/session-projection/session-projection/tests/registry.spec.ts @@ -182,6 +182,19 @@ describe('SessionProjectionRegistry drive', () => { expect(ctx.sessionProjections.checkpoint(fresh)['test/marks']).toEqual({ stateVersion: 1, observedSeq: -1, state: null }) }) + it('checkpoint states are detached clones — mutating them cannot corrupt the watermark cache', async () => { + const { ctx, session } = await harness() + ctx.sessionProjections.register(marksUnit()) + mark(session, ['a']) + const rows = ctx.sessionProjections.checkpoint(session) + // Hostile (or merely careless) consumer mutates the handed-out state. + ;(rows['test/marks']?.state as { marks: string[] }).marks.push('INJECTED') + // The registry's authoritative cell is untouched: snapshot and a fresh + // checkpoint both still serve the committed value. + expect(ctx.sessionProjections.snapshot(session).values['test/marks']).toEqual({ marks: ['a'] }) + expect(ctx.sessionProjections.checkpoint(session)['test/marks']?.state).toEqual({ marks: ['a'] }) + }) + it('restoreFloor anchors one below the lowest usable watermark and at 0 for missing or mismatched rows', async () => { const { ctx } = await harness() expect(ctx.sessionProjections.restoreFloor({})).toBeUndefined() // no unit registered