fix(tlon): stop SSE reconnects after client close (#108168)

* fix(tlon): stop reconnecting after SSE client close

* fix(tlon): cancel SSE reconnect timers when monitoring stops

Preserve the original Tlon SSE reconnect fix while canceling both retry timers at stopReceiving(), the real monitor shutdown boundary.

Verified against an authenticated, SSRF-guarded loopback SSE server; both reconnect waits settle immediately without changing replay, close cleanup, or uninterrupted reconnection.

Co-authored-by: wahaha1223 <0668001153@xydigit.com>

* fix(tlon): prepare conflict-free reconnect refresh

Prepare a conflict-free refresh of the original Tlon SSE shutdown correction without changing its ownership or losing the contributor commit.

Co-authored-by: wahaha1223 <0668001153@xydigit.com>

* fix(tlon): cancel reconnect timers when monitoring stops

Cancel both Tlon SSE reconnect waits at stopReceiving(), the monitor-owned shutdown boundary, while preserving guarded HTTP, durable same-channel replay, and the original contributor's fix.

Verified with authenticated loopback SSE, both real retry delays, normal reconnection, all 252 Tlon tests, extension type checks, lint, and focused owner regressions.

Co-authored-by: wahaha1223 <0668001153@xydigit.com>

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>

extensions/tlon/src/urbit/sse-client.reconnect-close.test.ts

@@ -0,0 +1,234 @@
+import { createServer, type Server, type ServerResponse } from "node:http";
+import type { AddressInfo } from "node:net";
+import type { LookupFn } from "openclaw/plugin-sdk/ssrf-runtime";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { UrbitSSEClient } from "./sse-client.js";
+
+const proofCookie = "urbauth-~zod=synthetic-reconnect-proof";
+const lookupLoopback = (async () => [{ address: "127.0.0.1", family: 4 }]) as unknown as LookupFn;
+const runningServers: Server[] = [];
+
+type UrbitChannelProof = {
+  baseUrl: string;
+  requests: string[];
+  unauthorizedRequests: number;
+};
+
+async function startUrbitChannelServer(options: { holdStream?: boolean } = {}) {
+  const proof: UrbitChannelProof = {
+    baseUrl: "",
+    requests: [],
+    unauthorizedRequests: 0,
+  };
+  const heldStreams = new Set<ServerResponse>();
+  const server = createServer((request, response) => {
+    if (!(request.headers.cookie ?? "").includes(proofCookie)) {
+      proof.unauthorizedRequests += 1;
+      response.writeHead(401).end();
+      return;
+    }
+    proof.requests.push(`${request.method ?? "GET"} ${request.url ?? "/"}`);
+    if (request.method === "GET") {
+      response.writeHead(200, {
+        "Cache-Control": "no-cache",
+        "Content-Type": "text/event-stream",
+      });
+      if (options.holdStream) {
+        heldStreams.add(response);
+        response.once("close", () => heldStreams.delete(response));
+        response.write(": connected\n\n");
+      } else {
+        response.end();
+      }
+      return;
+    }
+    response.writeHead(204).end();
+  });
+  await new Promise<void>((resolve, reject) => {
+    server.once("error", reject);
+    server.listen(0, "127.0.0.1", resolve);
+  });
+  runningServers.push(server);
+  const address = server.address() as AddressInfo;
+  proof.baseUrl = `http://127.0.0.1:${address.port}`;
+  return proof;
+}
+
+async function waitFor(predicate: () => boolean, timeoutMs = 2_000): Promise<void> {
+  const deadline = Date.now() + timeoutMs;
+  while (!predicate()) {
+    if (Date.now() >= deadline) {
+      throw new Error("Timed out waiting for the real Tlon SSE reconnect lifecycle");
+    }
+    await new Promise<void>((resolve) => {
+      setTimeout(resolve, 5);
+    });
+  }
+}
+
+async function expectPromptReconnectSettlement(reconnect: Promise<void>) {
+  let deadline: ReturnType<typeof setTimeout> | undefined;
+  try {
+    const outcome = await Promise.race([
+      reconnect.then(() => "settled" as const),
+      new Promise<"timed out">((resolve) => {
+        deadline = setTimeout(() => resolve("timed out"), 250);
+      }),
+    ]);
+    expect(outcome).toBe("settled");
+  } finally {
+    if (deadline !== undefined) {
+      clearTimeout(deadline);
+    }
+  }
+}
+
+afterEach(async () => {
+  vi.restoreAllMocks();
+  await Promise.all(
+    runningServers.splice(0).map(
+      (server) =>
+        new Promise<void>((resolve) => {
+          server.close(() => resolve());
+          server.closeAllConnections();
+        }),
+    ),
+  );
+});
+
+describe("UrbitSSEClient real reconnect shutdown lifecycle", () => {
+  it("settles a real SSE reconnect when the monitor stops receiving", async () => {
+    const proof = await startUrbitChannelServer();
+    const logs: string[] = [];
+    const onReconnect = vi.fn();
+    const client = new UrbitSSEClient(proof.baseUrl, proofCookie, {
+      ship: "zod",
+      ssrfPolicy: { allowPrivateNetwork: true },
+      lookupFn: lookupLoopback,
+      reconnectDelay: 400,
+      logger: { log: (message) => logs.push(message) },
+      onReconnect,
+    });
+    const reconnectSpy = vi.spyOn(client, "attemptReconnect");
+
+    try {
+      await client.connect();
+      await waitFor(() => logs.some((message) => message.includes("in 400ms")));
+      const reconnect = reconnectSpy.mock.results[0]?.value as Promise<void> | undefined;
+      if (!reconnect) {
+        throw new Error("The real SSE stream did not enter its reconnect backoff");
+      }
+
+      const requestsBeforeStop = proof.requests.length;
+      client.stopReceiving();
+      await expectPromptReconnectSettlement(reconnect);
+
+      expect(onReconnect).not.toHaveBeenCalled();
+      expect(proof.requests).toHaveLength(requestsBeforeStop);
+      expect(proof.requests.some((request) => request.startsWith("GET /~/channel/"))).toBe(true);
+
+      const unauthorizedResponse = await fetch(`${proof.baseUrl}/unauthorized-control`);
+      expect(unauthorizedResponse.status).toBe(401);
+      expect(proof.unauthorizedRequests).toBe(1);
+    } finally {
+      await client.close();
+    }
+  });
+
+  it("settles the real ten-second retry cooldown when the monitor stops receiving", async () => {
+    const proof = await startUrbitChannelServer({ holdStream: true });
+    const logs: string[] = [];
+    const onReconnect = vi.fn();
+    const client = new UrbitSSEClient(proof.baseUrl, proofCookie, {
+      ship: "zod",
+      ssrfPolicy: { allowPrivateNetwork: true },
+      lookupFn: lookupLoopback,
+      maxReconnectAttempts: 1,
+      reconnectDelay: 20,
+      logger: { log: (message) => logs.push(message) },
+      onReconnect,
+    });
+
+    try {
+      await client.connect();
+      client.reconnectAttempts = client.maxReconnectAttempts;
+      const reconnect = client.attemptReconnect();
+      expect(logs.some((message) => message.includes("Waiting 10s"))).toBe(true);
+
+      const requestsBeforeStop = proof.requests.length;
+      client.stopReceiving();
+      await expectPromptReconnectSettlement(reconnect);
+
+      expect(onReconnect).not.toHaveBeenCalled();
+      expect(proof.requests).toHaveLength(requestsBeforeStop);
+      expect(logs.some((message) => message.includes("reset, resuming"))).toBe(false);
+    } finally {
+      await client.close();
+    }
+  });
+
+  it("settles a real pending reconnect when the public client closes", async () => {
+    const proof = await startUrbitChannelServer();
+    const logs: string[] = [];
+    const onReconnect = vi.fn();
+    const client = new UrbitSSEClient(proof.baseUrl, proofCookie, {
+      ship: "zod",
+      ssrfPolicy: { allowPrivateNetwork: true },
+      lookupFn: lookupLoopback,
+      reconnectDelay: 400,
+      logger: { log: (message) => logs.push(message) },
+      onReconnect,
+    });
+    const reconnectSpy = vi.spyOn(client, "attemptReconnect");
+
+    try {
+      await client.connect();
+      await waitFor(() => logs.some((message) => message.includes("in 400ms")));
+      const reconnect = reconnectSpy.mock.results[0]?.value as Promise<void> | undefined;
+      if (!reconnect) {
+        throw new Error("The real SSE stream did not enter its reconnect backoff");
+      }
+
+      await client.close();
+      const requestsAfterClose = proof.requests.length;
+      await expectPromptReconnectSettlement(reconnect);
+
+      expect(onReconnect).not.toHaveBeenCalled();
+      expect(proof.requests).toHaveLength(requestsAfterClose);
+      expect(proof.requests.some((request) => request.startsWith("DELETE /~/channel/"))).toBe(true);
+    } finally {
+      await client.close();
+    }
+  });
+
+  it("still reconnects an uninterrupted authenticated SSE stream", async () => {
+    const proof = await startUrbitChannelServer();
+    let reconnects = 0;
+    const client = new UrbitSSEClient(proof.baseUrl, proofCookie, {
+      ship: "zod",
+      ssrfPolicy: { allowPrivateNetwork: true },
+      lookupFn: lookupLoopback,
+      reconnectDelay: 25,
+      onReconnect: (reconnectingClient) => {
+        reconnects += 1;
+        reconnectingClient.autoReconnect = false;
+      },
+    });
+
+    try {
+      await client.connect();
+      await waitFor(() => reconnects === 1);
+      await waitFor(
+        () =>
+          proof.requests.filter((request) => request.startsWith("GET /~/channel/")).length === 2,
+      );
+      expect(reconnects).toBe(1);
+      expect(
+        proof.requests.filter((request) => request.startsWith("GET /~/channel/")),
+      ).toHaveLength(2);
+      expect(proof.unauthorizedRequests).toBe(0);
+    } finally {
+      await client.close();
+    }
+  });
+});

extensions/tlon/src/urbit/sse-client.reconnect-close.unit.test.ts

@@ -0,0 +1,43 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { UrbitSSEClient } from "./sse-client.js";
+
+afterEach(() => {
+  vi.useRealTimers();
+  vi.restoreAllMocks();
+});
+
+describe("UrbitSSEClient owned reconnect timers", () => {
+  it.each([
+    { name: "ordinary reconnect", exhausted: false, delayMs: 1_000 },
+    { name: "exhausted ten-second cooldown", exhausted: true, delayMs: 10_000 },
+  ])("clears the $name timer immediately when the monitor stops receiving", async (params) => {
+    vi.useFakeTimers();
+    const setTimeoutSpy = vi.spyOn(globalThis, "setTimeout");
+    const onReconnect = vi.fn();
+    const logger = { log: vi.fn() };
+    const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=synthetic", {
+      maxReconnectAttempts: 1,
+      reconnectDelay: 1_000,
+      onReconnect,
+      logger,
+    });
+    if (params.exhausted) {
+      client.reconnectAttempts = client.maxReconnectAttempts;
+    }
+
+    const reconnect = client.attemptReconnect();
+
+    expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), params.delayMs);
+    expect(vi.getTimerCount()).toBe(1);
+    client.stopReceiving();
+
+    await expect(reconnect).resolves.toBeUndefined();
+    expect(vi.getTimerCount()).toBe(0);
+    expect(onReconnect).not.toHaveBeenCalled();
+    if (params.exhausted) {
+      expect(logger.log).not.toHaveBeenCalledWith(
+        expect.stringContaining("reset, resuming reconnection"),
+      );
+    }
+  });
+});

extensions/tlon/src/urbit/sse-client.ts

@@ -3,6 +3,7 @@ import { randomUUID } from "node:crypto";
 import { Readable } from "node:stream";
 import { resolveTimerTimeoutMs } from "openclaw/plugin-sdk/number-runtime";
 import { readResponseTextLimited } from "openclaw/plugin-sdk/provider-http";
+import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env";
 import type { LookupFn, SsrFPolicy } from "openclaw/plugin-sdk/ssrf-runtime";
 import { ensureUrbitChannelOpen, pokeUrbitChannel, scryUrbitPath } from "./channel-ops.js";
 import { getUrbitContext, normalizeUrbitCookie } from "./context.js";
@@ -89,6 +90,9 @@ export class UrbitSSEClient {
   private lastHeardEventId = -1;
   private lastAcknowledgedEventId = -1;
   private readonly ackThreshold = 20;
+  // The monitor stops ingress before it closes the channel; cancel both retry
+  // delays at that first shutdown edge so stale work cannot outlive its owner.
+  private readonly reconnectAbortController = new AbortController();
 
   constructor(url: string, cookie: string, options: UrbitSseOptions = {}) {
     const ctx = getUrbitContext(url, options.ship);
@@ -472,9 +476,9 @@ export class UrbitSSEClient {
       );
       // Wait 10 seconds before resetting and trying again
       const extendedBackoff = 10000; // 10 seconds
-      await new Promise((resolve) => {
-        setTimeout(resolve, extendedBackoff);
-      });
+      if (!(await this.waitForReconnectDelay(extendedBackoff)) || this.aborted) {
+        return;
+      }
       this.reconnectAttempts = 0; // Reset counter to continue trying
       this.logger.log?.("[SSE] Reconnection attempts reset, resuming reconnection...");
     }
@@ -489,11 +493,7 @@ export class UrbitSSEClient {
       `[SSE] Reconnection attempt ${this.reconnectAttempts}/${this.maxReconnectAttempts} in ${delay}ms...`,
     );
 
-    await new Promise((resolve) => {
-      setTimeout(resolve, delay);
-    });
-
-    if (this.aborted || !this.autoReconnect) {
+    if (!(await this.waitForReconnectDelay(delay)) || this.aborted || !this.autoReconnect) {
       return;
     }
 
@@ -528,6 +528,7 @@ export class UrbitSSEClient {
   stopReceiving(): void {
     this.aborted = true;
     this.isConnected = false;
+    this.reconnectAbortController.abort();
     this.streamController?.abort();
   }
 
@@ -586,6 +587,18 @@ export class UrbitSSEClient {
     }
   }
 
+  private async waitForReconnectDelay(delayMs: number): Promise<boolean> {
+    try {
+      await sleepWithAbort(delayMs, this.reconnectAbortController.signal);
+      return true;
+    } catch (error) {
+      if (this.reconnectAbortController.signal.aborted) {
+        return false;
+      }
+      throw error;
+    }
+  }
+
   private async putChannelPayload(
     payload: unknown,
     params: { timeoutMs: number; auditContext: string },