itsablabla · replicas-connector · May 30, 2026 · May 30, 2026
diff --git a/replicas-matrix-bridge/src/dispatch.ts b/replicas-matrix-bridge/src/dispatch.ts
@@ -43,12 +43,38 @@ export async function handleMatrixMessage(
 ): Promise<void> {
 	const key = `room:${roomId}`;
 	const seenKey = `seen:${roomId}:${eventId}`;
-	const seen = await env.MAP.get(seenKey);
-	if (seen) {
-		console.log(`[dispatch] DEDUPE skip room=${roomId} ev=${eventId}`);
-		return;
+	// Audit follow-up: dedupe via MatrixListener DO instead of KV. KV's
+	// eventual consistency let two concurrent dispatches both read
+	// "absent" and both proceed; routing through the singleton listener
+	// DO uses blockConcurrencyWhile for genuine put-if-absent semantics.
+	// The narrow race only fired when two paths hit the same event_id at
+	// once (listener /sync + /admin/recover-room replay, or /dispatch
+	// admin call racing the listener), but the cost of fixing it
+	// properly is one cross-DO RPC per dispatch and a periodic prune of
+	// the seen:* prefix (handled by the listener's existing prune cron).
+	try {
+		const listenerStub = env.LISTENER.get(env.LISTENER.idFromName("global"));
+		const claim = await listenerStub.fetch("https://listener/dedup-claim", {
+			method: "POST",
+			headers: { "Content-Type": "application/json" },
+			body: JSON.stringify({ key: seenKey }),
+		});
+		const j = (await claim.json()) as { claimed?: boolean };
+		if (!j.claimed) {
+			console.log(`[dispatch] DEDUPE skip room=${roomId} ev=${eventId}`);
+			return;
+		}
+	} catch (e) {
+		// Failsafe to legacy KV path if the DO is unreachable — better to
+		// risk a (very rare) double-dispatch than to drop the message.
+		console.log(`[dispatch] dedup DO failed, falling back to KV: ${e instanceof Error ? e.message : e}`);
+		const seen = await env.MAP.get(seenKey);
+		if (seen) {
+			console.log(`[dispatch] DEDUPE (kv fallback) skip room=${roomId} ev=${eventId}`);
+			return;
+		}
+		await env.MAP.put(seenKey, "1", { expirationTtl: 600 });
 	}
-	await env.MAP.put(seenKey, "1", { expirationTtl: 600 });
 
 	// Capture the 👀 ack reaction id so the watcher can redact it later when
 	// it swaps in the terminal emoji — otherwise both stack on the prompt.

diff --git a/replicas-matrix-bridge/src/index.ts b/replicas-matrix-bridge/src/index.ts
@@ -120,6 +120,10 @@ export default {
 			const stub = env.LISTENER.get(env.LISTENER.idFromName("global"));
 			return stub.fetch("https://listener/reset", { method: "POST" });
 		}
+		if (req.method === "POST" && url.pathname === "/admin/listener/reload-keys") {
+			const stub = env.LISTENER.get(env.LISTENER.idFromName("global"));
+			return stub.fetch("https://listener/reload-keys", { method: "POST" });
+		}
 		if (req.method === "POST" && url.pathname === "/admin/vault/reset") {
 			const stub = env.OLM_VAULT.get(env.OLM_VAULT.idFromName("global"));
 			return stub.fetch("https://vault/reset", { method: "POST" });

diff --git a/replicas-matrix-bridge/src/listener.ts b/replicas-matrix-bridge/src/listener.ts
@@ -53,6 +53,34 @@ export class MatrixListener {
 			await this.state.storage.setAlarm(Date.now() + 100);
 			return new Response("ok");
 		}
+		if (req.method === "POST" && url.pathname === "/dedup-claim") {
+			// Atomic put-if-absent for the dispatch dedupe path. KV's
+			// eventual consistency lets two concurrent reads both see
+			// "not present" and both write "1", letting the same Matrix
+			// event spawn the agent twice. Routing through this DO uses
+			// blockConcurrencyWhile so the check + write is genuinely
+			// atomic — at most one caller per event_id gets {claimed:true}.
+			const body = (await req.json()) as { key: string };
+			let claimed = false;
+			await this.state.blockConcurrencyWhile(async () => {
+				const existing = await this.state.storage.get<number>(body.key);
+				if (existing) return;
+				await this.state.storage.put(body.key, Date.now());
+				claimed = true;
+			});
+			return Response.json({ claimed });
+		}
+		if (req.method === "POST" && url.pathname === "/reload-keys") {
+			// Clear the in-memory cachedKeys so the next megolmKeys() call
+			// re-reads MATRIX_MEGOLM_KEYS_JSON from env. Lets the operator
+			// rotate Megolm session keys without a full redeploy (set the
+			// secret, then POST here to flush the cache). cachedCurve25519
+			// is keyed on the OlmVault identity which can change after a
+			// vault reset, so flush that too.
+			this.cachedKeys = undefined;
+			this.cachedCurve25519 = undefined;
+			return Response.json({ ok: true, message: "in-memory key caches cleared" });
+		}
 		if (req.method === "POST" && url.pathname === "/stop") {
 			await this.state.storage.deleteAlarm();
 			return new Response("ok");
@@ -92,6 +120,15 @@ export class MatrixListener {
 					skipped += 1;
 					continue;
 				}
+				if (!senderKey) {
+					// Audit follow-up: missing outer sender_key means our
+					// m.room_key_request can only route to `<user>:*` and
+					// the originating device may not pick it up. Surface so
+					// the operator can spot bridges that strip sender_key.
+					console.log(
+						`[recover-room] WARN missing sender_key ev=${ev.event_id} room=${body.roomId} session=${sessionId.slice(0, 16)}…`,
+					);
+				}
 				triedDecrypt += 1;
 				const sessionKey = await this.findKey(body.roomId, sessionId, megolmKeys);
 				if (!sessionKey) {
@@ -176,9 +213,67 @@ export class MatrixListener {
 		} catch (e) {
 			console.error("[listener] alarm threw", e instanceof Error ? e.message : String(e));
 		}
+		// Audit follow-up: periodic prune of pending-decrypt:* queues.
+		// Run at most every PRUNE_INTERVAL_MS so we don't burn DO storage
+		// budget on every 1s alarm tick. Drops queue entries whose
+		// origin_server_ts is more than PENDING_TTL_MS old AND deletes
+		// the whole queue if it ends up empty after pruning. Bounds
+		// long-term storage growth from sessions that were never sent
+		// (sender device went offline before forwarding the key).
+		try {
+			await this.maybePruneStaleDecryptQueues();
+		} catch (e) {
+			console.error("[listener] prune threw", e instanceof Error ? e.message : String(e));
+		}
 		await this.state.storage.setAlarm(Date.now() + ALARM_INTERVAL_MS);
 	}
 
+	private async maybePruneStaleDecryptQueues(): Promise<void> {
+		const PRUNE_INTERVAL_MS = 30 * 60 * 1000; // every 30 min
+		const PENDING_TTL_MS = 7 * 24 * 60 * 60 * 1000; // 7 days
+		// seen:* keys only need to survive Matrix's re-delivery window,
+		// which is bounded by /sync's since-token behavior. 1h is generous.
+		const SEEN_TTL_MS = 60 * 60 * 1000;
+		const lastRun = (await this.state.storage.get<number>("last-prune-at")) ?? 0;
+		const now = Date.now();
+		if (now - lastRun < PRUNE_INTERVAL_MS) return;
+		await this.state.storage.put("last-prune-at", now);
+		const cutoff = now - PENDING_TTL_MS;
+		const all = await this.state.storage.list({ prefix: "pending-decrypt:" });
+		let pruned = 0;
+		let emptied = 0;
+		for (const [key, value] of all) {
+			const list = value as Array<{ origin_server_ts?: number }> | undefined;
+			if (!Array.isArray(list)) continue;
+			const kept = list.filter((e) => (e.origin_server_ts ?? 0) >= cutoff);
+			if (kept.length === 0) {
+				await this.state.storage.delete(key);
+				emptied += 1;
+				pruned += list.length;
+			} else if (kept.length < list.length) {
+				await this.state.storage.put(key, kept);
+				pruned += list.length - kept.length;
+			}
+		}
+		if (pruned > 0) {
+			console.log(`[listener] prune: dropped ${pruned} stale pending-decrypt entries (${emptied} queues emptied)`);
+		}
+		// Prune seen:* dedup markers older than SEEN_TTL_MS.
+		const seenList = await this.state.storage.list({ prefix: "seen:" });
+		const seenCutoff = now - SEEN_TTL_MS;
+		let seenPruned = 0;
+		for (const [key, value] of seenList) {
+			const ts = typeof value === "number" ? value : 0;
+			if (ts < seenCutoff) {
+				await this.state.storage.delete(key);
+				seenPruned += 1;
+			}
+		}
+		if (seenPruned > 0) {
+			console.log(`[listener] prune: dropped ${seenPruned} stale seen:* entries`);
+		}
+	}
+
 	private async alarmInner(): Promise<void> {
 		const since = await this.state.storage.get<string>("since");
 		let resp: SyncResponse;
@@ -219,16 +314,52 @@ export class MatrixListener {
 					captured?: boolean;
 					capturedRoomId?: string;
 					capturedSessionId?: string;
+					capturedSource?: "live" | "forwarded";
 				};
 				console.log(
-					`[listener] to_device sender=${senderKey.slice(0, 12)}… type=${entry.type} ok=${j.ok} captured=${j.captured}`,
+					`[listener] to_device sender=${senderKey.slice(0, 12)}… type=${entry.type} ok=${j.ok} captured=${j.captured} src=${j.capturedSource ?? "n/a"}`,
 				);
-				// E2EE auto-recovery: when the vault captures a Megolm
-				// session key (either initial m.room_key or our requested
-				// m.forwarded_room_key response), walk the pending-decrypt
-				// queue for that (room, session) and dispatch any
-				// previously-stuck user messages.
+				// Audit follow-up: m.forwarded_room_key validation. We only
+				// accept forwarded Megolm keys for (room, session) pairs we
+				// previously asked about via our own m.room_key_request.
+				// Without this, any Olm-paired sender could plant arbitrary
+				// Megolm keys for arbitrary rooms by emitting an unsolicited
+				// m.forwarded_room_key — and the vault would store it
+				// without checking. m.room_key (the initial share) is
+				// passed through unchanged because that's how legitimate
+				// senders bootstrap a new outbound session.
 				if (j.captured && j.capturedRoomId && j.capturedSessionId) {
+					if (j.capturedSource === "forwarded") {
+						const wasRequested = await this.wasKeyRequested(
+							j.capturedRoomId,
+							j.capturedSessionId,
+						);
+						if (!wasRequested) {
+							console.log(
+								`[listener] UNSOLICITED forwarded_room_key — evicting room=${j.capturedRoomId} session=${j.capturedSessionId.slice(0, 16)}…`,
+							);
+							const evict = this.env.OLM_VAULT.get(
+								this.env.OLM_VAULT.idFromName("global"),
+							);
+							await evict
+								.fetch("https://vault/keystore-delete", {
+									method: "POST",
+									headers: { "Content-Type": "application/json" },
+									body: JSON.stringify({
+										roomId: j.capturedRoomId,
+										sessionId: j.capturedSessionId,
+										reason: "no outstanding m.room_key_request",
+									}),
+								})
+								.catch(() => {});
+							continue;
+						}
+					}
+					// E2EE auto-recovery: when the vault captures a Megolm
+					// session key (either initial m.room_key or our requested
+					// m.forwarded_room_key response), walk the pending-decrypt
+					// queue for that (room, session) and dispatch any
+					// previously-stuck user messages.
 					await this.drainPendingForSession(j.capturedRoomId, j.capturedSessionId);
 				}
 			} catch (e) {
@@ -503,6 +634,21 @@ export class MatrixListener {
 		}
 	}
 
+	/**
+	 * Audit follow-up: check whether we previously emitted an
+	 * m.room_key_request for this (room, session). Used to gate the
+	 * acceptance of m.forwarded_room_key — if we never asked, the
+	 * forwarded key is unsolicited and may be an attacker trying to
+	 * plant a Megolm key. The key-request:* prefix is owned by
+	 * maybeSendKeyRequest and is bucketed by 5-min windows, so we just
+	 * check if any bucket exists for the (room, session) pair.
+	 */
+	private async wasKeyRequested(roomId: string, sessionId: string): Promise<boolean> {
+		const prefix = `key-request:${roomId}:${sessionId}:`;
+		const list = await this.state.storage.list({ prefix, limit: 1 });
+		return list.size > 0;
+	}
+
 	/**
 	 * Called when the olm-vault decrypts a to-device event and captures
 	 * a Megolm session key. Walks the pending-decrypt queue for that

diff --git a/replicas-matrix-bridge/src/olm-vault.ts b/replicas-matrix-bridge/src/olm-vault.ts
@@ -82,6 +82,48 @@ export class OlmVault {
 				const hit = list.find((e) => e.session_id === sessionId);
 				return Response.json({ found: !!hit, session_key: hit?.session_key ?? null });
 			}
+			if (req.method === "POST" && url.pathname === "/usage-bump") {
+				// Audit follow-up: per-org usage log was being read-modify-
+				// written from the poller via KV, which loses concurrent
+				// appends across rooms (last write wins). Routing through
+				// this singleton DO with blockConcurrencyWhile makes the
+				// append atomic. Storage shape kept identical so the
+				// renderer's bucketing logic stays unchanged.
+				const body = (await req.json()) as { ts: number; cost: number; tok: number };
+				const CUTOFF_MS = 8 * 24 * 60 * 60 * 1000;
+				let log: { ts: number; cost: number; tok: number }[] = [];
+				await this.state.blockConcurrencyWhile(async () => {
+					const prior = (await this.state.storage.get<typeof log>("usage:org")) ?? [];
+					const cutoff = Date.now() - CUTOFF_MS;
+					log = prior.filter((e) => e.ts >= cutoff);
+					log.push({ ts: body.ts, cost: body.cost, tok: body.tok });
+					await this.state.storage.put("usage:org", log);
+				});
+				return Response.json({ ok: true, entries: log.length });
+			}
+			if (req.method === "GET" && url.pathname === "/usage-read") {
+				const log = (await this.state.storage.get<{ ts: number; cost: number; tok: number }[]>("usage:org")) ?? [];
+				return Response.json({ log });
+			}
+			if (req.method === "POST" && url.pathname === "/keystore-delete") {
+				// Used by the listener to evict a forwarded Megolm key that
+				// failed the "did we actually request this?" check. Without
+				// this, an Olm-paired sender could plant arbitrary Megolm
+				// keys for any room. With it, forwarded keys are restricted
+				// to (room, session) pairs we previously asked about.
+				const body = (await req.json()) as { roomId: string; sessionId: string; reason?: string };
+				const ks = (await this.state.storage.get<Record<string, MegolmKeyEntry[]>>(K_KEYSTORE)) ?? {};
+				const list = ks[body.roomId] ?? [];
+				const next = list.filter((e) => e.session_id !== body.sessionId);
+				if (next.length === list.length) return Response.json({ ok: true, evicted: 0 });
+				if (next.length === 0) delete ks[body.roomId];
+				else ks[body.roomId] = next;
+				await this.state.storage.put(K_KEYSTORE, ks);
+				console.log(
+					`[olm-vault] evicted unsolicited forwarded key room=${body.roomId} session=${body.sessionId.slice(0, 16)}… reason=${body.reason ?? "unspecified"}`,
+				);
+				return Response.json({ ok: true, evicted: list.length - next.length });
+			}
 			return new Response("not found", { status: 404 });
 		} catch (e) {
 			console.log(`[olm-vault] ${url.pathname} threw: ${e instanceof Error ? `${e.name}: ${e.message}` : String(e)}`);
@@ -382,7 +424,21 @@ export class OlmVault {
 			if (usedSessionPickle) {
 				const next = pickles.filter((p) => p !== usedSessionPickle);
 				next.push(usedSessionPickle);
-				sessionsMap[senderCurve25519] = next.slice(-8);
+				// Cap=32 (was 8). Verified + multi-device senders can rotate
+				// Olm sessions fast enough that 8 was occasionally dropping
+				// legitimate sessions whose ratchet was still alive (silent
+				// decryption failure on subsequent to-device events that
+				// referenced the evicted session). mautrix-go keeps ~64;
+				// 32 is a comfortable mid-point. Log evictions so the
+				// operator can spot pathological churn.
+				const CAP = 32;
+				if (next.length > CAP) {
+					const evicted = next.length - CAP;
+					console.log(
+						`[olm-vault] session cap evicted ${evicted} oldest session(s) sender=${senderCurve25519.slice(0, 12)}…`,
+					);
+				}
+				sessionsMap[senderCurve25519] = next.slice(-CAP);
 				await this.state.storage.put(K_SESSIONS, sessionsMap);
 			}
 
@@ -395,6 +451,7 @@ export class OlmVault {
 			let captured = false;
 			let capturedRoomId: string | undefined;
 			let capturedSessionId: string | undefined;
+			let capturedSource: "live" | "forwarded" | undefined;
 			try {
 				const inner = JSON.parse(plaintext) as {
 					type?: string;
@@ -434,6 +491,7 @@ export class OlmVault {
 						captured = true;
 						capturedRoomId = c.room_id;
 						capturedSessionId = c.session_id;
+						capturedSource = inner.type === "m.forwarded_room_key" ? "forwarded" : "live";
 						console.log(
 							`[olm-vault] captured Megolm via ${inner.type} room=${c.room_id} session=${c.session_id!.slice(0, 16)}…`,
 						);
@@ -443,7 +501,7 @@ export class OlmVault {
 				/* not JSON — return raw */
 			}
 
-			return Response.json({ ok: true, captured, plaintext, capturedRoomId, capturedSessionId });
+			return Response.json({ ok: true, captured, plaintext, capturedRoomId, capturedSessionId, capturedSource });
 		} finally {
 			account.free();
 			void Olm;