fix(app-control): dedupe verification verdicts + harden bridge unsubscribe

VerificationRoomBridgeService now deduplicates verdict broadcasts keyed
by `${sessionId}:${verdict}` with a 10-minute TTL so a replayed
task_complete or escalation event cannot post a duplicate continuation
message into the originating chat room. Sweep is opportunistic on each
insert.

stop() now reads the stored unsubscribe locally, clears the field
first (so a retry of stop() is a no-op), runtime-checks that the value
is a function, and wraps the call in a single-purpose try/catch that
warn-logs a misbehaving coordinator without crashing service teardown.

Adds five tests:
 - dedupes duplicate verdict events for the same session
 - emits separately for the same session when verdict differs
 - re-emits a verdict after the dedupe TTL expires (fake timers)
 - stop() handles a non-function unsubscribe gracefully
 - stop() handles an unsubscribe that throws

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: Shaw

plugins/plugin-app-control/typescript/src/services/__tests__/verification-room-bridge.retry.test.ts

@@ -38,7 +38,7 @@
  */
 
 import type { IAgentRuntime } from "@elizaos/core";
-import { beforeEach, describe, expect, it, vi } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { VerificationRoomBridgeService } from "../verification-room-bridge.js";
 
 /**
@@ -172,6 +172,39 @@ function escalationEvent(args: {
 	};
 }
 
+/**
+ * Shape of the broadcast `task_complete` (pass-verdict) event from
+ * `swarm-decision-loop.ts`. Mirrors `passEvent` in the unit suite but
+ * keeps a sessionId argument so dedupe tests can pin it explicitly.
+ */
+function passEvent(args: {
+	sessionId: string;
+	originRoomId: string;
+	appName: string;
+}): SwarmEventLike {
+	return {
+		type: "task_complete",
+		sessionId: args.sessionId,
+		timestamp: Date.now(),
+		data: {
+			reasoning: "validator pass",
+			verification: {
+				source: "custom-validator",
+				verdict: "pass",
+				validator: { service: "app-verification", method: "verifyApp" },
+				params: {
+					workdir: "/tmp/wd",
+					appName: args.appName,
+					profile: "full",
+				},
+			},
+			originRoomId: args.originRoomId,
+			label: `create-app:${args.appName}`,
+			workdir: "/tmp/wd",
+		},
+	};
+}
+
 describe("VerificationRoomBridgeService — retry-loop integration", () => {
 	let coord: FakeCoordinator;
 
@@ -288,3 +321,85 @@ describe("VerificationRoomBridgeService — retry-loop integration", () => {
 		expect(memory.content.text).toContain("2/3");
 	});
 });
+
+describe("VerificationRoomBridgeService — verdict dedupe", () => {
+	let coord: FakeCoordinator;
+
+	beforeEach(() => {
+		coord = createFakeCoordinator();
+	});
+
+	afterEach(() => {
+		// Always restore real timers in case a test installed fakes.
+		vi.useRealTimers();
+	});
+
+	it("dedupes duplicate verdict events for the same session", async () => {
+		const { runtime, createMemory } = createRuntime(coord);
+		await VerificationRoomBridgeService.start(runtime);
+
+		const evt = passEvent({
+			sessionId: "sess-dedupe-1",
+			originRoomId: "room-dedupe",
+			appName: "notes-app",
+		});
+		coord.emit(evt);
+		coord.emit(evt);
+		await flushMicrotasks();
+
+		expect(createMemory).toHaveBeenCalledTimes(1);
+	});
+
+	it("emits separately for the same session when verdict differs", async () => {
+		const { runtime, createMemory } = createRuntime(coord);
+		await VerificationRoomBridgeService.start(runtime);
+
+		const sessionId = "sess-dedupe-mixed";
+		const room = "room-dedupe-mixed";
+		coord.emit(
+			passEvent({ sessionId, originRoomId: room, appName: "notes-app" }),
+		);
+		coord.emit(
+			escalationEvent({
+				sessionId,
+				retryCount: 3,
+				maxRetries: 3,
+				originRoomId: room,
+				appName: "notes-app",
+			}),
+		);
+		await flushMicrotasks();
+
+		expect(createMemory).toHaveBeenCalledTimes(2);
+		const verdicts = createMemory.mock.calls.map((call) => {
+			const [memory] = call as [{ content: { text: string } }, string];
+			return memory.content.text.includes("verification failure")
+				? "fail"
+				: "pass";
+		});
+		expect(verdicts.sort()).toEqual(["fail", "pass"]);
+	});
+
+	it("re-emits a verdict after the dedupe TTL expires", async () => {
+		vi.useFakeTimers();
+		const { runtime, createMemory } = createRuntime(coord);
+		await VerificationRoomBridgeService.start(runtime);
+
+		const evt = passEvent({
+			sessionId: "sess-dedupe-ttl",
+			originRoomId: "room-ttl",
+			appName: "notes-app",
+		});
+		coord.emit(evt);
+		await flushMicrotasks();
+		expect(createMemory).toHaveBeenCalledTimes(1);
+
+		// Advance past the 10-minute TTL. The next identical event should
+		// pass the dedupe check and produce a second memory.
+		vi.advanceTimersByTime(10 * 60 * 1000 + 1);
+		coord.emit(evt);
+		await flushMicrotasks();
+
+		expect(createMemory).toHaveBeenCalledTimes(2);
+	});
+});

plugins/plugin-app-control/typescript/src/services/__tests__/verification-room-bridge.test.ts

@@ -15,6 +15,7 @@
  */
 
 import type { IAgentRuntime } from "@elizaos/core";
+import { logger } from "@elizaos/core";
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import { VerificationRoomBridgeService } from "../verification-room-bridge.js";
 
@@ -272,4 +273,50 @@ describe("VerificationRoomBridgeService", () => {
 		expect(coord.listeners.length).toBe(0);
 		expect(coord.unsubscribed).toBe(1);
 	});
+
+	it("stop() handles a non-function unsubscribe gracefully", async () => {
+		const { runtime } = createRuntime(coord);
+		const service = await VerificationRoomBridgeService.start(runtime);
+		// Force the stored unsubscribe into an invalid runtime shape — what
+		// would happen if a future coordinator surface returned an object
+		// instead of the documented `() => void`.
+		(service as unknown as { unsubscribe: unknown }).unsubscribe = {
+			not: "callable",
+		};
+		const warnSpy = vi.spyOn(logger, "warn").mockImplementation(() => {});
+		await expect(service.stop()).resolves.toBeUndefined();
+		expect(warnSpy).toHaveBeenCalledWith(
+			expect.stringContaining(
+				"[VerificationRoomBridge] stored unsubscribe was not a function",
+			),
+		);
+		// Field cleared so a second stop() is a no-op.
+		expect(
+			(service as unknown as { unsubscribe: unknown }).unsubscribe,
+		).toBeNull();
+		warnSpy.mockRestore();
+	});
+
+	it("stop() handles an unsubscribe that throws", async () => {
+		const { runtime } = createRuntime(coord);
+		const service = await VerificationRoomBridgeService.start(runtime);
+		const boom = new Error("coordinator exploded");
+		(service as unknown as { unsubscribe: () => void }).unsubscribe = () => {
+			throw boom;
+		};
+		const warnSpy = vi.spyOn(logger, "warn").mockImplementation(() => {});
+		await expect(service.stop()).resolves.toBeUndefined();
+		expect(warnSpy).toHaveBeenCalledWith(
+			expect.stringContaining(
+				"[VerificationRoomBridge] unsubscribe threw during stop()",
+			),
+		);
+		expect(warnSpy).toHaveBeenCalledWith(
+			expect.stringContaining("coordinator exploded"),
+		);
+		expect(
+			(service as unknown as { unsubscribe: unknown }).unsubscribe,
+		).toBeNull();
+		warnSpy.mockRestore();
+	});
 });

plugins/plugin-app-control/typescript/src/services/verification-room-bridge.ts

@@ -42,6 +42,16 @@ const APP_VERIFICATION_SERVICE = "app-verification";
 const VERIFY_APP_METHOD = "verifyApp";
 const VERIFY_PLUGIN_METHOD = "verifyPlugin";
 
+/**
+ * Dedupe TTL for verdict events keyed by `${sessionId}:${verdict}`.
+ *
+ * The broadcast bus may replay events under network blips, supervisor
+ * retries, or a future multi-listener architecture. A real verdict for
+ * a given session lands once, within seconds; 10 minutes is well past
+ * the window where a duplicate is anything other than a replay.
+ */
+const VERDICT_DEDUPE_TTL_MS = 10 * 60 * 1000;
+
 /**
  * Minimal shape of the SwarmCoordinator service surface this bridge
  * depends on. We only need `subscribe`; declared locally so we don't
@@ -181,6 +191,15 @@ export class VerificationRoomBridgeService extends Service {
 
 	private unsubscribe: (() => void) | null = null;
 
+	/**
+	 * Dedupe map: `${sessionId}:${verdict}` -> expiresAt epoch ms. Drops
+	 * replayed verdict events that would otherwise post duplicate chat
+	 * memories. Entries age out via `VERDICT_DEDUPE_TTL_MS`; we sweep
+	 * opportunistically on each insert (single-digit concurrent verdicts
+	 * in practice).
+	 */
+	private readonly verdictDedupe: Map<string, number> = new Map();
+
 	static override async start(
 		runtime: IAgentRuntime,
 	): Promise<VerificationRoomBridgeService> {
@@ -190,9 +209,25 @@ export class VerificationRoomBridgeService extends Service {
 	}
 
 	override async stop(): Promise<void> {
-		if (this.unsubscribe) {
-			this.unsubscribe();
-			this.unsubscribe = null;
+		const unsub = this.unsubscribe;
+		// Always clear the field first so a retry of stop() can't double-call.
+		this.unsubscribe = null;
+		if (unsub === null) return;
+		if (typeof unsub !== "function") {
+			logger.warn(
+				"[VerificationRoomBridge] stored unsubscribe was not a function; skipping",
+			);
+			return;
+		}
+		// Single-purpose catch: a misbehaving coordinator must not crash
+		// service teardown. Translate the failure into a structured warn
+		// log and continue.
+		try {
+			unsub();
+		} catch (err) {
+			logger.warn(
+				`[VerificationRoomBridge] unsubscribe threw during stop(): ${err instanceof Error ? err.message : String(err)}`,
+			);
 		}
 	}
 
@@ -226,6 +261,18 @@ export class VerificationRoomBridgeService extends Service {
 		const payload = decodeEvent(event);
 		if (!payload) return;
 
+		const dedupeKey = `${event.sessionId}:${payload.verdict}`;
+		const now = Date.now();
+		this.sweepExpiredDedupe(now);
+		const existingExpiry = this.verdictDedupe.get(dedupeKey);
+		if (existingExpiry !== undefined && existingExpiry > now) {
+			logger.debug(
+				`[VerificationRoomBridge] dedupe drop sessionId=${event.sessionId} verdict=${payload.verdict}`,
+			);
+			return;
+		}
+		this.verdictDedupe.set(dedupeKey, now + VERDICT_DEDUPE_TTL_MS);
+
 		const text =
 			payload.verdict === "pass"
 				? buildPassMessage(payload)
@@ -248,6 +295,12 @@ export class VerificationRoomBridgeService extends Service {
 			`[VerificationRoomBridge] posted ${payload.verdict} verdict for ${payload.targetName} into room=${payload.originRoomId}`,
 		);
 	}
+
+	private sweepExpiredDedupe(now: number): void {
+		for (const [key, expiresAt] of this.verdictDedupe) {
+			if (expiresAt <= now) this.verdictDedupe.delete(key);
+		}
+	}
 }
 
 export default VerificationRoomBridgeService;