From 138d2d071c4bbc05f42fc407e2f234305625eecc Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Sat, 18 Jul 2026 20:19:04 +0100 Subject: [PATCH] feat(tlon): add durable ingress (#110910) --- extensions/tlon/src/monitor/index.ts | 446 +++++++++--------- extensions/tlon/src/monitor/ingress.test.ts | 375 +++++++++++++++ extensions/tlon/src/monitor/ingress.ts | 366 ++++++++++++++ .../src/monitor/processed-messages.test.ts | 59 --- .../tlon/src/monitor/processed-messages.ts | 90 ---- extensions/tlon/src/urbit/channel-ops.ts | 11 +- extensions/tlon/src/urbit/sse-client.test.ts | 171 ++++++- extensions/tlon/src/urbit/sse-client.ts | 167 ++++--- 8 files changed, 1250 insertions(+), 435 deletions(-) create mode 100644 extensions/tlon/src/monitor/ingress.test.ts create mode 100644 extensions/tlon/src/monitor/ingress.ts delete mode 100644 extensions/tlon/src/monitor/processed-messages.test.ts delete mode 100644 extensions/tlon/src/monitor/processed-messages.ts diff --git a/extensions/tlon/src/monitor/index.ts b/extensions/tlon/src/monitor/index.ts index 06907d7044dc..db7f91f53a1b 100644 --- a/extensions/tlon/src/monitor/index.ts +++ b/extensions/tlon/src/monitor/index.ts @@ -1,5 +1,6 @@ import { resolveHumanDelayConfig } from "openclaw/plugin-sdk/agent-runtime"; import { createChannelInboundEnvelopeBuilder } from "openclaw/plugin-sdk/channel-inbound"; +import { bindIngressLifecycleToReplyOptions } from "openclaw/plugin-sdk/channel-outbound"; import type { ReplyPayload } from "openclaw/plugin-sdk/reply-runtime"; import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime"; import { sleepWithAbort } from "openclaw/plugin-sdk/runtime-env"; @@ -27,11 +28,8 @@ import { resolveChannelAuthorization } from "./authorization.js"; import { createTlonCitationResolver } from "./cites.js"; import { fetchAllChannels, fetchInitData } from "./discovery.js"; import { cacheMessage, fetchThreadHistory, getChannelHistory } from "./history.js"; +import { createTlonIngressMonitor, type TlonIngressLifecycle } from "./ingress.js"; import { downloadMessageImages } from "./media.js"; -import { - createProcessedMessageTracker, - runWithProcessedMessageClaim, -} from "./processed-messages.js"; import { applyTlonSettingsOverrides, buildTlonSettingsMigrations, @@ -140,7 +138,6 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise { const { messageId, @@ -320,6 +318,7 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise { runtime.error?.(`[tlon] failed updating session meta: ${String(err)}`); @@ -739,7 +741,10 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise { + const handleChannelsFirehose = async ( + event: unknown, + turnAdoptionLifecycle?: TlonIngressLifecycle, + ) => { try { const eventRecord = asRecord(event); const nest = readString(eventRecord, "nest"); @@ -780,119 +785,111 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise { - const senderShip = normalizeShip(readString(content, "author") ?? ""); - if (!senderShip || senderShip === botShipName) { - return; - } - - const rawText = extractMessageText(content.content); - if (!rawText.trim()) { - return; - } - - const contentBody = content.content; - const sentAt = readNumber(content, "sent") ?? Date.now(); - - cacheMessage(nest, { - author: senderShip, - content: rawText, - timestamp: sentAt, - id: messageId, - }); - - // Get thread info early for participation check - const seal = isThreadReply ? asRecord(replySet?.seal) : asRecord(set?.seal); - const parentId = readString(seal, "parent-id") ?? readString(seal, "parent") ?? null; - - // Check if we should respond: - // 1. Direct mention always triggers response - // 2. Thread replies where we've participated - respond if relevant (let agent decide) - const mentioned = isBotMentioned(rawText, botShipName, botNickname ?? undefined); - const inParticipatedThread = - isThreadReply && parentId && participatedThreads.has(parentId); - const mentionDecision = resolveTlonGroupMentionDecision({ - cfg, - accountId: account.accountId, - wasMentioned: mentioned, - botParticipatedInThread: Boolean(inParticipatedThread), - }); - - if (mentionDecision.shouldSkip) { - return; - } - - // Log why we're responding - if (mentionDecision.implicitMention && !mentioned) { - runtime.log?.( - `[tlon] Responding to thread we participated in (no mention): ${parentId}`, - ); - } - - // Owner is always allowed - if (isOwner(senderShip)) { - runtime.log?.(`[tlon] Owner ${senderShip} is always allowed in channels`); - } else { - const { mode, allowedShips } = resolveChannelAuthorization(cfg, nest, currentSettings); - if (mode === "restricted") { - const normalizedAllowed = allowedShips.map(normalizeShip); - if (!normalizedAllowed.includes(senderShip)) { - // If owner is configured, queue approval request - if (effectiveOwnerShip) { - const approval = createPendingApproval({ - type: "channel", - requestingShip: senderShip, - channelNest: nest, - messagePreview: sliceUtf16Safe(rawText, 0, 100), - originalMessage: { - messageId: messageId ?? "", - messageText: rawText, - messageContent: contentBody, - timestamp: sentAt, - parentId: parentId ?? undefined, - isThreadReply, - }, - }); - await queueApprovalRequest(approval); - } else { - runtime.log?.( - `[tlon] Access denied: ${senderShip} in ${nest} (allowed: ${allowedShips.join(", ")})`, - ); - } - return; - } - } - } - - const messageText = await resolveAuthorizedMessageText({ - rawText, - content: contentBody, - authorizedForCites: true, - resolveAllCites, - }); - - const parsed = parseChannelNest(nest); - await processMessage({ - messageId: messageId ?? "", - senderShip, - messageText, - messageContent: contentBody, // Pass raw content for media extraction - isGroup: true, - channelNest: nest, - hostShip: parsed?.hostShip, - channelName: parsed?.channelName, - timestamp: sentAt, - parentId, - isThreadReply, - }); - }, }); - void processed; + + // Get thread info early for participation check + const seal = isThreadReply ? asRecord(replySet?.seal) : asRecord(set?.seal); + const parentId = readString(seal, "parent-id") ?? readString(seal, "parent") ?? null; + + // Check if we should respond: + // 1. Direct mention always triggers response + // 2. Thread replies where we've participated - respond if relevant (let agent decide) + const mentioned = isBotMentioned(rawText, botShipName, botNickname ?? undefined); + const inParticipatedThread = isThreadReply && parentId && participatedThreads.has(parentId); + const mentionDecision = resolveTlonGroupMentionDecision({ + cfg, + accountId: account.accountId, + wasMentioned: mentioned, + botParticipatedInThread: Boolean(inParticipatedThread), + }); + + if (mentionDecision.shouldSkip) { + return; + } + + // Log why we're responding + if (mentionDecision.implicitMention && !mentioned) { + runtime.log?.(`[tlon] Responding to thread we participated in (no mention): ${parentId}`); + } + + // Owner is always allowed + if (isOwner(senderShip)) { + runtime.log?.(`[tlon] Owner ${senderShip} is always allowed in channels`); + } else { + const { mode, allowedShips } = resolveChannelAuthorization(cfg, nest, currentSettings); + if (mode === "restricted") { + const normalizedAllowed = allowedShips.map(normalizeShip); + if (!normalizedAllowed.includes(senderShip)) { + // If owner is configured, queue approval request + if (effectiveOwnerShip) { + const approval = createPendingApproval({ + type: "channel", + requestingShip: senderShip, + channelNest: nest, + messagePreview: sliceUtf16Safe(rawText, 0, 100), + originalMessage: { + messageId: messageId ?? "", + messageText: rawText, + messageContent: contentBody, + timestamp: sentAt, + parentId: parentId ?? undefined, + isThreadReply, + }, + }); + await queueApprovalRequest(approval); + } else { + runtime.log?.( + `[tlon] Access denied: ${senderShip} in ${nest} (allowed: ${allowedShips.join(", ")})`, + ); + } + return; + } + } + } + + const messageText = await resolveAuthorizedMessageText({ + rawText, + content: contentBody, + authorizedForCites: true, + resolveAllCites, + }); + + const parsed = parseChannelNest(nest); + await processMessage({ + messageId: messageId ?? "", + senderShip, + messageText, + messageContent: contentBody, // Pass raw content for media extraction + isGroup: true, + channelNest: nest, + hostShip: parsed?.hostShip, + channelName: parsed?.channelName, + timestamp: sentAt, + parentId, + isThreadReply, + turnAdoptionLifecycle, + }); } catch (error: unknown) { runtime.error?.(`[tlon] Error handling channel firehose event: ${formatErrorMessage(error)}`); + throw error; } }; @@ -900,7 +897,10 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise(); - const handleChatFirehose = async (event: unknown) => { + const handleChatFirehose = async ( + event: unknown, + turnAdoptionLifecycle?: TlonIngressLifecycle, + ) => { try { // Handle DM invite lists (arrays) if (Array.isArray(event)) { @@ -976,116 +976,125 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise { - const authorShip = normalizeShip(readString(essay, "author") ?? ""); - const partnerShip = extractDmPartnerShip(whom); - const senderShip = partnerShip || authorShip; + const authorShip = normalizeShip(readString(essay, "author") ?? ""); + const partnerShip = extractDmPartnerShip(whom); + const senderShip = partnerShip || authorShip; - // Ignore the bot's own outbound DM events. - if (authorShip === botShipName) { - return; - } - if (!senderShip || senderShip === botShipName) { - return; - } + // Ignore the bot's own outbound DM events. + if (authorShip === botShipName) { + return; + } + if (!senderShip || senderShip === botShipName) { + return; + } - // Log mismatch between author and partner for debugging - if (authorShip && partnerShip && authorShip !== partnerShip) { - runtime.log?.( - `[tlon] DM ship mismatch (author=${authorShip}, partner=${partnerShip}) - routing to partner`, - ); - } + // Log mismatch between author and partner for debugging + if (authorShip && partnerShip && authorShip !== partnerShip) { + runtime.log?.( + `[tlon] DM ship mismatch (author=${authorShip}, partner=${partnerShip}) - routing to partner`, + ); + } - const rawText = extractMessageText(essay.content); - if (!rawText.trim()) { - return; - } + const rawText = extractMessageText(essay.content); + if (!rawText.trim()) { + return; + } - // Check if this is the owner sending an approval response - const messageText = rawText; - if (isOwner(senderShip) && isApprovalResponse(messageText)) { - const handled = await handleApprovalResponse(messageText); - if (handled) { - runtime.log?.(`[tlon] Processed approval response from owner: ${messageText}`); - return; - } - } + // Check if this is the owner sending an approval response + const messageText = rawText; + if (isOwner(senderShip) && isApprovalResponse(messageText)) { + const handled = await handleApprovalResponse(messageText); + if (handled) { + runtime.log?.(`[tlon] Processed approval response from owner: ${messageText}`); + return; + } + } - // Check if this is the owner sending an admin command - if (isOwner(senderShip) && isAdminCommand(messageText)) { - const handled = await handleAdminCommand(messageText); - if (handled) { - runtime.log?.(`[tlon] Processed admin command from owner: ${messageText}`); - return; - } - } + // Check if this is the owner sending an admin command + if (isOwner(senderShip) && isAdminCommand(messageText)) { + const handled = await handleAdminCommand(messageText); + if (handled) { + runtime.log?.(`[tlon] Processed admin command from owner: ${messageText}`); + return; + } + } - // Owner is always allowed to DM (bypass allowlist) - if (isOwner(senderShip)) { - const resolvedMessageText = await resolveAuthorizedMessageText({ - rawText, - content: essay.content, - authorizedForCites: true, - resolveAllCites, - }); - runtime.log?.(`[tlon] Processing DM from owner ${senderShip}`); - await processMessage({ + // Owner is always allowed to DM (bypass allowlist) + if (isOwner(senderShip)) { + const resolvedMessageText = await resolveAuthorizedMessageText({ + rawText, + content: essay.content, + authorizedForCites: true, + resolveAllCites, + }); + runtime.log?.(`[tlon] Processing DM from owner ${senderShip}`); + await processMessage({ + messageId: messageId ?? "", + senderShip, + messageText: resolvedMessageText, + messageContent: essay.content, + isGroup: false, + timestamp: readNumber(essay, "sent") ?? Date.now(), + turnAdoptionLifecycle, + }); + return; + } + + // For DMs from others, check allowlist + if (!(await isDmAllowedWithIngress(senderShip, effectiveDmAllowlist))) { + // If owner is configured, queue approval request + if (effectiveOwnerShip) { + const approval = createPendingApproval({ + type: "dm", + requestingShip: senderShip, + messagePreview: sliceUtf16Safe(messageText, 0, 100), + originalMessage: { messageId: messageId ?? "", - senderShip, - messageText: resolvedMessageText, + messageText, messageContent: essay.content, - isGroup: false, timestamp: readNumber(essay, "sent") ?? Date.now(), - }); - return; - } - - // For DMs from others, check allowlist - if (!(await isDmAllowedWithIngress(senderShip, effectiveDmAllowlist))) { - // If owner is configured, queue approval request - if (effectiveOwnerShip) { - const approval = createPendingApproval({ - type: "dm", - requestingShip: senderShip, - messagePreview: sliceUtf16Safe(messageText, 0, 100), - originalMessage: { - messageId: messageId ?? "", - messageText, - messageContent: essay.content, - timestamp: readNumber(essay, "sent") ?? Date.now(), - }, - }); - await queueApprovalRequest(approval); - } else { - runtime.log?.(`[tlon] Blocked DM from ${senderShip}: not in allowlist`); - } - return; - } - - await processMessage({ - messageText: await resolveAuthorizedMessageText({ - rawText, - content: essay.content, - authorizedForCites: true, - resolveAllCites, - }), - messageId: messageId ?? "", - senderShip, - messageContent: essay.content, // Pass raw content for media extraction - isGroup: false, - timestamp: readNumber(essay, "sent") ?? Date.now(), + }, }); - }, + await queueApprovalRequest(approval); + } else { + runtime.log?.(`[tlon] Blocked DM from ${senderShip}: not in allowlist`); + } + return; + } + + await processMessage({ + messageText: await resolveAuthorizedMessageText({ + rawText, + content: essay.content, + authorizedForCites: true, + resolveAllCites, + }), + messageId: messageId ?? "", + senderShip, + messageContent: essay.content, // Pass raw content for media extraction + isGroup: false, + timestamp: readNumber(essay, "sent") ?? Date.now(), + turnAdoptionLifecycle, }); - void processed; } catch (error: unknown) { runtime.error?.(`[tlon] Error handling chat firehose event: ${formatErrorMessage(error)}`); + throw error; } }; + const ingress = createTlonIngressMonitor({ + accountId: account.accountId, + runtime, + abortSignal: opts.abortSignal, + dispatch: async (source, event, turnAdoptionLifecycle) => { + if (source === "channels") { + await handleChannelsFirehose(event, turnAdoptionLifecycle); + return; + } + await handleChatFirehose(event, turnAdoptionLifecycle); + }, + }); + try { runtime.log?.("[tlon] Subscribing to firehose updates..."); @@ -1093,8 +1102,11 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise { - void handleChannelsFirehose(event); + event: async (event) => { + const result = await ingress.receive({ source: "channels", event }); + if (result.kind === "ignored") { + await handleChannelsFirehose(event); + } }, err: (error) => { runtime.error?.(`[tlon] Channels firehose error: ${String(error)}`); @@ -1109,8 +1121,11 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise { - void handleChatFirehose(event); + event: async (event) => { + const result = await ingress.receive({ source: "chat", event }); + if (result.kind === "ignored") { + await handleChatFirehose(event); + } }, err: (error) => { runtime.error?.(`[tlon] Chat firehose error: ${String(error)}`); @@ -1474,6 +1489,7 @@ export async function monitorTlonProvider(opts: MonitorTlonOpts = {}): Promise {}); } } finally { + api?.stopReceiving(); + await ingress.stop(); try { await api?.close(); } catch (error: unknown) { diff --git a/extensions/tlon/src/monitor/ingress.test.ts b/extensions/tlon/src/monitor/ingress.test.ts new file mode 100644 index 000000000000..0a30602c8f9e --- /dev/null +++ b/extensions/tlon/src/monitor/ingress.test.ts @@ -0,0 +1,375 @@ +// Tlon durable ingress tests cover append, recovery, tombstones, and guard parity. +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { + closeOpenClawStateDatabaseForTest, + createChannelIngressQueueForTests, +} from "openclaw/plugin-sdk/plugin-state-test-runtime"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { UrbitHttpError } from "../urbit/errors.js"; +import { createTlonIngressMonitor } from "./ingress.js"; + +type TlonIngressQueue = NonNullable[0]["queue"]>; +type TlonIngressPayload = Parameters[1]; +type TlonIngressDispatch = Parameters[0]["dispatch"]; + +function channelEvent(params?: { id?: string; nest?: string; text?: string }) { + return { + nest: params?.nest ?? "chat/~zod/general", + response: { + post: { + id: params?.id ?? "message-1", + "r-post": { + set: { + essay: { + author: "~nec", + content: [{ text: params?.text ?? "hello" }], + sent: 1_700_000_000_000, + }, + }, + }, + }, + }, + }; +} + +function channelReplyEvent(params?: { id?: string; nest?: string; text?: string }) { + return { + nest: params?.nest ?? "chat/~zod/general", + response: { + post: { + id: "parent-post", + "r-post": { + reply: { + id: params?.id ?? "reply-1", + "r-reply": { + set: { + memo: { + author: "~nec", + content: [{ text: params?.text ?? "hello" }], + sent: 1_700_000_000_000, + }, + }, + }, + }, + }, + }, + }, + }; +} + +function chatEvent(params?: { id?: string; peer?: string; text?: string }) { + return { + whom: params?.peer ?? "~nec", + id: params?.id ?? "dm-1", + response: { + add: { + essay: { + author: params?.peer ?? "~nec", + content: [{ text: params?.text ?? "hello" }], + sent: 1_700_000_000_000, + }, + }, + }, + }; +} + +function createQueue(stateDir: string, accountId = "default"): TlonIngressQueue { + return createChannelIngressQueueForTests({ + channelId: "tlon", + accountId, + stateDir, + }); +} + +async function withQueue( + fn: (queue: TlonIngressQueue, stateDir: string) => Promise, +): Promise { + const created = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-tlon-ingress-")); + const stateDir = await fs.realpath(created); + try { + return await fn(createQueue(stateDir), stateDir); + } finally { + closeOpenClawStateDatabaseForTest(); + await fs.rm(stateDir, { recursive: true, force: true }); + } +} + +function startMonitor(queue: TlonIngressQueue, dispatch: TlonIngressDispatch) { + const monitor = createTlonIngressMonitor({ + accountId: "default", + queue, + dispatch, + runtime: { error: vi.fn(), log: vi.fn() }, + pollIntervalMs: 60_000, + adoptionStallTimeoutMs: 5_000, + }); + monitor.start(); + return monitor; +} + +function deferred() { + let resolve = () => {}; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +afterEach(() => { + closeOpenClawStateDatabaseForTest(); + vi.restoreAllMocks(); +}); + +describe("Tlon durable ingress", () => { + it("propagates durable append failure before dispatch", async () => { + await withQueue(async (queue) => { + const appendError = new Error("sqlite unavailable"); + const failingQueue = { + ...queue, + enqueue: vi.fn().mockRejectedValue(appendError), + } satisfies TlonIngressQueue; + const dispatch = vi.fn(); + const monitor = startMonitor(failingQueue, dispatch); + try { + await expect(monitor.receive({ source: "channels", event: channelEvent() })).rejects.toBe( + appendError, + ); + expect(dispatch).not.toHaveBeenCalled(); + } finally { + await monitor.stop(); + } + }); + }); + + it("recovers an uncompleted message with a fresh drain exactly once", async () => { + await withQueue(async (queue, stateDir) => { + const interruptedDispatch = vi.fn(async () => ({ kind: "deferred" }) as const); + const interrupted = startMonitor(queue, interruptedDispatch); + await interrupted.receive({ + source: "channels", + event: channelEvent({ id: "message-restart" }), + }); + await interrupted.waitForIdle(); + expect(await queue.listClaims()).toHaveLength(1); + await interrupted.stop(); + closeOpenClawStateDatabaseForTest(); + + const recoveredDispatch = vi.fn(async (_source, _event, lifecycle) => { + await lifecycle.onAdopted(); + }); + const recovered = startMonitor(createQueue(stateDir), recoveredDispatch); + try { + await recovered.waitForIdle(); + expect(recoveredDispatch).toHaveBeenCalledTimes(1); + } finally { + await recovered.stop(); + } + }); + }); + + it("retains completion so a duplicate message id cannot dispatch twice", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async (_source, _event, lifecycle) => { + await lifecycle.onAdopted(); + }); + const monitor = startMonitor(queue, dispatch); + try { + const event = chatEvent({ id: "dm-completed" }); + await monitor.receive({ source: "chat", event }); + await monitor.waitForIdle(); + await monitor.receive({ source: "chat", event }); + await monitor.waitForIdle(); + expect(dispatch).toHaveBeenCalledTimes(1); + } finally { + await monitor.stop(); + } + }); + }); + + it("preserves the retired guard for reissued or edited message envelopes", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async (_source, _event, lifecycle) => { + await lifecycle.onAdopted(); + }); + const monitor = startMonitor(queue, dispatch); + try { + await monitor.receive({ + source: "channels", + event: channelEvent({ id: "message-guard", text: "first delivery" }), + }); + await monitor.waitForIdle(); + await monitor.receive({ + source: "channels", + event: channelEvent({ id: "message-guard", text: "edited redelivery" }), + }); + await monitor.waitForIdle(); + expect(dispatch).toHaveBeenCalledTimes(1); + } finally { + await monitor.stop(); + } + }); + }); + + it("retains completed logical ids by count rather than age", async () => { + await withQueue(async (queue) => { + let now = 1_700_000_000_000; + vi.spyOn(Date, "now").mockImplementation(() => now); + const dispatch = vi.fn(async (_source, _event, lifecycle) => { + await lifecycle.onAdopted(); + }); + const monitor = startMonitor(queue, dispatch); + try { + const original = channelEvent({ id: "message-aged" }); + await monitor.receive({ source: "channels", event: original }); + await monitor.waitForIdle(); + + now += 31 * 24 * 60 * 60 * 1_000; + await monitor.receive({ + source: "channels", + event: channelEvent({ id: "message-prune-trigger" }), + }); + await monitor.waitForIdle(); + await monitor.receive({ source: "channels", event: original }); + await monitor.waitForIdle(); + + expect(dispatch).toHaveBeenCalledTimes(2); + } finally { + await monitor.stop(); + } + }); + }); + + it("stores raw envelopes in per-conversation lanes", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async () => ({ kind: "deferred" }) as const); + const monitor = startMonitor(queue, dispatch); + const group = channelEvent({ id: "message-raw", nest: "chat/~zod/ops" }); + try { + await monitor.receive({ source: "channels", event: group }); + await monitor.waitForIdle(); + expect(await queue.listClaims()).toEqual([ + expect.objectContaining({ + id: "message-raw", + laneKey: "group:chat/~zod/ops", + payload: expect.objectContaining({ rawEvent: JSON.stringify(group) }), + }), + ]); + } finally { + await monitor.stop(); + } + }); + }); + + it("uses the nested reply id rather than the parent post id", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async () => ({ kind: "deferred" }) as const); + const monitor = startMonitor(queue, dispatch); + const reply = channelReplyEvent({ id: "reply-stable", nest: "chat/~zod/ops" }); + try { + await monitor.receive({ source: "channels", event: reply }); + await monitor.waitForIdle(); + expect(await queue.listClaims()).toEqual([ + expect.objectContaining({ id: "reply-stable", laneKey: "group:chat/~zod/ops" }), + ]); + } finally { + await monitor.stop(); + } + }); + }); + + it("dead-letters malformed persisted payloads and authentication failures", async () => { + await withQueue(async (queue) => { + await queue.enqueue( + "message-malformed", + { version: 1, receivedAt: 1, source: "channels", rawEvent: "{" }, + { receivedAt: 1, laneKey: "group:chat/~zod/general" }, + ); + const dispatch = vi.fn(async () => { + throw new UrbitHttpError({ operation: "Poke", status: 401 }); + }); + const monitor = startMonitor(queue, dispatch); + try { + await monitor.waitForIdle(); + expect((await queue.enqueue("message-malformed", {} as TlonIngressPayload)).kind).toBe( + "failed", + ); + + await monitor.receive({ + source: "chat", + event: chatEvent({ id: "message-auth" }), + }); + await monitor.waitForIdle(); + expect((await queue.enqueue("message-auth", {} as TlonIngressPayload)).kind).toBe("failed"); + } finally { + await monitor.stop(); + } + }); + }); + + it("keeps unrelated authentication failures retryable", async () => { + await withQueue(async (queue) => { + const dispatch = vi.fn(async () => { + throw Object.assign(new Error("model credentials expired"), { status: 401 }); + }); + const monitor = startMonitor(queue, dispatch); + try { + await monitor.receive({ + source: "chat", + event: chatEvent({ id: "message-provider-auth" }), + }); + await monitor.waitForIdle(); + expect((await queue.listPending({ limit: "all" })).map((record) => record.id)).toEqual([ + "message-provider-auth", + ]); + } finally { + await monitor.stop(); + } + }); + }); + + it("waits for admitted work and leaves it pending on repeated stop", async () => { + await withQueue(async (queue) => { + const stored = deferred(); + const release = deferred(); + const enqueue = queue.enqueue.bind(queue); + queue.enqueue = async (...args) => { + const result = await enqueue(...args); + stored.resolve(); + await release.promise; + return result; + }; + const dispatch = vi.fn(); + const monitor = startMonitor(queue, dispatch); + const admission = monitor.receive({ + source: "channels", + event: channelEvent({ id: "message-stop-1" }), + }); + await stored.promise; + const queuedAdmission = monitor.receive({ + source: "channels", + event: channelEvent({ id: "message-stop-2" }), + }); + + let stopSettled = false; + const stopping = monitor.stop().then(() => { + stopSettled = true; + }); + const stoppingAgain = monitor.stop(); + await Promise.resolve(); + expect(stopSettled).toBe(false); + await expect( + monitor.receive({ source: "channels", event: channelEvent({ id: "message-too-late" }) }), + ).rejects.toThrow("stopped before dispatch adoption"); + + release.resolve(); + await Promise.all([admission, queuedAdmission, stopping, stoppingAgain]); + expect(dispatch).not.toHaveBeenCalled(); + expect((await queue.listPending({ limit: "all" })).map((record) => record.id)).toEqual([ + "message-stop-1", + "message-stop-2", + ]); + }); + }); +}); diff --git a/extensions/tlon/src/monitor/ingress.ts b/extensions/tlon/src/monitor/ingress.ts new file mode 100644 index 000000000000..eb9460968319 --- /dev/null +++ b/extensions/tlon/src/monitor/ingress.ts @@ -0,0 +1,366 @@ +// Tlon plugin module owns raw Urbit firehose durable ingress mapping and draining. +import { + createChannelIngressDrain, + DEFAULT_INGRESS_ADOPTION_STALL_MS, + DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS, + DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, + type ChannelIngressQueue, +} from "openclaw/plugin-sdk/channel-outbound"; +import { collectErrorGraphCandidates, formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; +import type { RuntimeEnv } from "openclaw/plugin-sdk/runtime-env"; +import { getTlonRuntime } from "../runtime.js"; +import { UrbitAuthError, UrbitHttpError } from "../urbit/errors.js"; + +const TLON_INGRESS_PAYLOAD_VERSION = 1; +const TLON_INGRESS_POLL_INTERVAL_MS = 1_000; +const TLON_INGRESS_PRUNE_INTERVAL_MS = 60 * 60 * 1_000; +const TLON_INGRESS_FAILED_TTL_MS = 30 * 24 * 60 * 60 * 1_000; +// Preserve the retired process-local guard's full 2,000-message key window. +const TLON_INGRESS_TOMBSTONE_MAX_ENTRIES = 2_000; + +export type TlonIngressLifecycle = { + abortSignal: AbortSignal; + onAdopted: () => void | Promise; + onDeferred: () => void; + onAdoptionFinalizing: () => void; + onAbandoned: () => void | Promise; +}; + +type TlonIngressSource = "channels" | "chat"; + +type TlonIngressPayload = { + version: 1; + receivedAt: number; + source: TlonIngressSource; + rawEvent: string; +}; + +type TlonIngressDrain = { + drainOnce: (options?: { shouldStop?: () => boolean }) => Promise<{ started: number }>; + waitForIdle: () => Promise; + dispose: () => void; +}; + +type TlonIngressDispatchResult = + | { kind: "completed" } + | { kind: "deferred" } + | { kind: "failed-retryable"; error: unknown }; + +type TlonIngressDispatch = ( + source: TlonIngressSource, + event: unknown, + lifecycle: TlonIngressLifecycle, +) => Promise | TlonIngressDispatchResult | void; + +class TlonIngressPermanentError extends Error { + constructor( + readonly reason: "invalid-event" | "tlon-auth", + message: string, + options?: ErrorOptions, + ) { + super(message, options); + this.name = "TlonIngressPermanentError"; + } +} + +class TlonIngressShutdownError extends Error { + constructor() { + super("Tlon ingress stopped before dispatch adoption."); + this.name = "TlonIngressShutdownError"; + } +} + +function isRecord(value: unknown): value is Record { + return typeof value === "object" && value !== null && !Array.isArray(value); +} + +function nonEmptyString(value: unknown): string | null { + return typeof value === "string" && value.trim() ? value.trim() : null; +} + +function inspectChannelsEvent(event: unknown): { eventId: string; laneKey: string } | null { + const envelope = isRecord(event) ? event : null; + const nest = nonEmptyString(envelope?.nest); + const response = isRecord(envelope?.response) ? envelope.response : null; + const post = isRecord(response?.post) ? response.post : null; + const rPost = isRecord(post?.["r-post"]) ? post["r-post"] : null; + const set = isRecord(rPost?.set) ? rPost.set : null; + const reply = isRecord(rPost?.reply) ? rPost.reply : null; + const rReply = isRecord(reply?.["r-reply"]) ? reply["r-reply"] : null; + const replySet = isRecord(rReply?.set) ? rReply.set : null; + if (!nest || (!isRecord(set?.essay) && !isRecord(replySet?.memo))) { + return null; + } + const eventId = nonEmptyString(isRecord(replySet?.memo) ? reply?.id : post?.id); + return eventId ? { eventId, laneKey: `group:${nest}` } : null; +} + +function inspectChatEvent(event: unknown): { eventId: string; laneKey: string } | null { + const envelope = isRecord(event) ? event : null; + const response = isRecord(envelope?.response) ? envelope.response : null; + const add = isRecord(response?.add) ? response.add : null; + const essay = isRecord(add?.essay) ? add.essay : null; + const eventId = nonEmptyString(envelope?.id); + if (!essay || !eventId) { + return null; + } + const whom = isRecord(envelope?.whom) ? nonEmptyString(envelope.whom.ship) : null; + const peer = nonEmptyString(envelope?.whom) ?? whom ?? nonEmptyString(essay.author); + return { eventId, laneKey: peer ? `direct:${peer}` : `event:${eventId}` }; +} + +function inspectTlonIngressEvent( + source: TlonIngressSource, + event: unknown, +): { eventId: string; laneKey: string } | null { + // Urbit SSE ids belong to a disposable HTTP channel. The message id inside + // each firehose envelope survives resubscription and preserves the retired guard key. + return source === "channels" ? inspectChannelsEvent(event) : inspectChatEvent(event); +} + +function parseClaimedEvent(payload: TlonIngressPayload, claimedId: string): unknown { + if ( + payload.version !== TLON_INGRESS_PAYLOAD_VERSION || + (payload.source !== "channels" && payload.source !== "chat") || + typeof payload.rawEvent !== "string" + ) { + throw new TlonIngressPermanentError( + "invalid-event", + `Tlon ingress row ${claimedId} has an invalid payload.`, + ); + } + let event: unknown; + try { + event = JSON.parse(payload.rawEvent); + } catch (error) { + throw new TlonIngressPermanentError( + "invalid-event", + `Tlon ingress row ${claimedId} contains invalid JSON.`, + { cause: error }, + ); + } + const facts = inspectTlonIngressEvent(payload.source, event); + if (!facts || facts.eventId !== claimedId) { + throw new TlonIngressPermanentError( + "invalid-event", + `Tlon ingress row ${claimedId} has invalid message identity.`, + ); + } + return event; +} + +function resolveTlonIngressNonRetryableFailure(error: unknown) { + if (error instanceof TlonIngressPermanentError) { + return { reason: error.reason, message: error.message }; + } + for (const candidate of collectErrorGraphCandidates(error, (current) => [current.cause])) { + if ( + candidate instanceof UrbitAuthError || + (candidate instanceof UrbitHttpError && + (candidate.status === 401 || candidate.status === 403)) + ) { + return { reason: "tlon-auth", message: formatErrorMessage(candidate) }; + } + } + return null; +} + +type TlonIngressMonitor = { + receive: (params: { + source: TlonIngressSource; + event: unknown; + }) => Promise<{ kind: "accepted" } | { kind: "ignored" }>; + start: () => void; + stop: () => Promise; + waitForIdle: () => Promise; +}; + +export function createTlonIngressMonitor(options: { + accountId: string; + queue?: ChannelIngressQueue; + dispatch: TlonIngressDispatch; + runtime: Pick; + pollIntervalMs?: number; + adoptionStallTimeoutMs?: number; + abortSignal?: AbortSignal; +}): TlonIngressMonitor { + let queue = options.queue; + let drain: TlonIngressDrain | undefined; + let accepting = true; + let running = false; + let stopped = false; + let requested = false; + let pumping: Promise | undefined; + let pollTimer: ReturnType | undefined; + let lastPrunedAt = 0; + let stopPromise: Promise | undefined; + + const getQueue = (): ChannelIngressQueue => { + queue ??= getTlonRuntime().state.openChannelIngressQueue({ + accountId: options.accountId, + }); + return queue; + }; + + const getDrain = (): TlonIngressDrain => { + drain ??= createChannelIngressDrain({ + queue: getQueue(), + adoptionStallTimeoutMs: options.adoptionStallTimeoutMs ?? DEFAULT_INGRESS_ADOPTION_STALL_MS, + retryPolicy: { + maxAttempts: DEFAULT_INGRESS_RETRY_MAX_ATTEMPTS, + deadLetterMinAgeMs: DEFAULT_INGRESS_RETRY_DEAD_LETTER_MIN_AGE_MS, + }, + resolveNonRetryableFailure: resolveTlonIngressNonRetryableFailure, + ...(options.abortSignal ? { abortSignal: options.abortSignal } : {}), + onLog: (message) => options.runtime.log?.(`tlon ${message}`), + dispatchClaimedEvent: async (record, lifecycle) => { + if (!running || lifecycle.abortSignal.aborted || options.abortSignal?.aborted) { + return { kind: "failed-retryable", error: new TlonIngressShutdownError() }; + } + const event = parseClaimedEvent(record.payload, record.id); + try { + const result = await options.dispatch(record.payload.source, event, lifecycle); + return !running || options.abortSignal?.aborted + ? { kind: "failed-retryable", error: new TlonIngressShutdownError() } + : result; + } catch (error) { + if (!running || options.abortSignal?.aborted) { + return { kind: "failed-retryable", error }; + } + throw error; + } + }, + }); + return drain; + }; + + const pruneIfDue = async (): Promise => { + const now = Date.now(); + if (now - lastPrunedAt < TLON_INGRESS_PRUNE_INTERVAL_MS) { + return; + } + await getQueue().prune({ + completedMaxEntries: TLON_INGRESS_TOMBSTONE_MAX_ENTRIES, + failedTtlMs: TLON_INGRESS_FAILED_TTL_MS, + failedMaxEntries: TLON_INGRESS_TOMBSTONE_MAX_ENTRIES, + now, + }); + lastPrunedAt = now; + }; + + const runPump = async (): Promise => { + try { + for (;;) { + requested = false; + await pruneIfDue(); + // stop() may race the async prune; never create a live drain afterward. + if (!running) { + break; + } + const activeDrain = getDrain(); + const { started } = await activeDrain.drainOnce({ shouldStop: () => !running }); + await activeDrain.waitForIdle(); + if (!running || (!requested && started === 0)) { + break; + } + } + } catch (error) { + options.runtime.error?.(`tlon ingress drain failed: ${formatErrorMessage(error)}`); + } finally { + pumping = undefined; + if (running && requested) { + requestDrain(); + } + } + }; + + const requestDrain = (): void => { + requested = true; + if (!running || pumping) { + return; + } + pumping = runPump(); + }; + + // Stream callbacks are awaited, but serialize direct test/caller admissions too. + let admissionTail: Promise = Promise.resolve(); + + const admitOnce = async (source: TlonIngressSource, event: unknown): Promise => { + const facts = inspectTlonIngressEvent(source, event); + if (!facts) { + return false; + } + const receivedAt = Date.now(); + await getQueue().enqueue( + facts.eventId, + { + version: TLON_INGRESS_PAYLOAD_VERSION, + receivedAt, + source, + rawEvent: JSON.stringify(event), + }, + { receivedAt, laneKey: facts.laneKey }, + ); + requestDrain(); + return true; + }; + + return { + receive: async ({ source, event }) => { + if (!accepting) { + throw new TlonIngressShutdownError(); + } + let accepted = false; + const admission = admissionTail.then(async () => { + accepted = await admitOnce(source, event); + }); + admissionTail = admission.catch(() => undefined); + await admission; + return { kind: accepted ? "accepted" : "ignored" }; + }, + start: () => { + if (running || stopped) { + return; + } + running = true; + pollTimer = setInterval( + requestDrain, + options.pollIntervalMs ?? TLON_INGRESS_POLL_INTERVAL_MS, + ); + pollTimer.unref?.(); + requestDrain(); + }, + stop: async () => { + if (stopPromise) { + await stopPromise; + return; + } + accepting = false; + running = false; + stopped = true; + if (pollTimer) { + clearInterval(pollTimer); + pollTimer = undefined; + } + stopPromise = (async () => { + await admissionTail; + drain?.dispose(); + await pumping; + // A pump may have lazily created the drain before observing running=false. + drain?.dispose(); + await drain?.waitForIdle(); + })(); + await stopPromise; + }, + waitForIdle: async () => { + for (;;) { + const activePump = pumping; + if (!activePump) { + break; + } + await activePump; + } + await drain?.waitForIdle(); + }, + }; +} diff --git a/extensions/tlon/src/monitor/processed-messages.test.ts b/extensions/tlon/src/monitor/processed-messages.test.ts deleted file mode 100644 index 560779bffe1c..000000000000 --- a/extensions/tlon/src/monitor/processed-messages.test.ts +++ /dev/null @@ -1,59 +0,0 @@ -// Tlon tests cover processed messages plugin behavior. -import { describe, expect, it } from "vitest"; -import { - createProcessedMessageTracker, - runWithProcessedMessageClaim, -} from "./processed-messages.js"; - -describe("createProcessedMessageTracker", () => { - it("dedupes and evicts oldest entries", () => { - const tracker = createProcessedMessageTracker(3); - - expect(tracker.mark("a")).toBe(true); - expect(tracker.mark("a")).toBe(false); - expect(tracker.has("a")).toBe(true); - - tracker.mark("b"); - tracker.mark("c"); - expect(tracker.size()).toBe(3); - - tracker.mark("d"); - expect(tracker.size()).toBe(3); - expect(tracker.has("a")).toBe(false); - expect(tracker.has("b")).toBe(true); - expect(tracker.has("c")).toBe(true); - expect(tracker.has("d")).toBe(true); - }); - - it("releases failed claims so retries can run again", async () => { - const tracker = createProcessedMessageTracker(); - - await expect( - runWithProcessedMessageClaim({ - tracker, - id: "evt-1", - task: async () => { - throw new Error("boom"); - }, - }), - ).rejects.toThrow("boom"); - - expect(tracker.has("evt-1")).toBe(false); - expect(tracker.claim("evt-1")).toEqual({ kind: "claimed" }); - }); - - it("keeps successful claims deduped", async () => { - const tracker = createProcessedMessageTracker(); - - await expect( - runWithProcessedMessageClaim({ - tracker, - id: "evt-2", - task: async () => undefined, - }), - ).resolves.toEqual({ kind: "processed", value: undefined }); - - expect(tracker.has("evt-2")).toBe(true); - expect(tracker.claim("evt-2")).toEqual({ kind: "duplicate" }); - }); -}); diff --git a/extensions/tlon/src/monitor/processed-messages.ts b/extensions/tlon/src/monitor/processed-messages.ts deleted file mode 100644 index 19997247e946..000000000000 --- a/extensions/tlon/src/monitor/processed-messages.ts +++ /dev/null @@ -1,90 +0,0 @@ -// Tlon plugin module implements processed messages behavior. -import { createDedupeCache } from "../../runtime-api.js"; - -type ProcessedMessageTracker = { - claim: (id?: string | null) => { kind: "claimed" } | { kind: "duplicate" }; - commit: (id?: string | null) => void; - release: (id?: string | null) => void; - mark: (id?: string | null) => boolean; - has: (id?: string | null) => boolean; - size: () => number; -}; - -export function createProcessedMessageTracker(limit = 2000): ProcessedMessageTracker { - const dedupe = createDedupeCache({ ttlMs: 0, maxSize: limit }); - const inFlight = new Set(); - - const claim = (id?: string | null) => { - const trimmed = id?.trim(); - if (!trimmed) { - return { kind: "claimed" } as const; - } - if (inFlight.has(trimmed) || dedupe.peek(trimmed)) { - return { kind: "duplicate" } as const; - } - inFlight.add(trimmed); - return { kind: "claimed" } as const; - }; - - const commit = (id?: string | null) => { - const trimmed = id?.trim(); - if (!trimmed) { - return; - } - inFlight.delete(trimmed); - dedupe.check(trimmed); - }; - - const release = (id?: string | null) => { - const trimmed = id?.trim(); - if (!trimmed) { - return; - } - inFlight.delete(trimmed); - }; - - const mark = (id?: string | null) => { - const claimed = claim(id); - if (claimed.kind === "duplicate") { - return false; - } - commit(id); - return true; - }; - - const has = (id?: string | null) => { - const trimmed = id?.trim(); - if (!trimmed) { - return false; - } - return dedupe.peek(trimmed); - }; - - return { - claim, - commit, - release, - mark, - has, - size: () => dedupe.size(), - }; -} - -export async function runWithProcessedMessageClaim(params: { - tracker: ProcessedMessageTracker; - id?: string | null; - task: () => Promise; -}): Promise<{ kind: "processed"; value: T } | { kind: "duplicate" }> { - const claim = params.tracker.claim(params.id); - if (claim.kind === "duplicate") { - return claim; - } - try { - const value = await params.task(); - params.tracker.commit(params.id); - return { kind: "processed", value }; - } catch (error) { - params.tracker.release(params.id); - throw error; - } -} diff --git a/extensions/tlon/src/urbit/channel-ops.ts b/extensions/tlon/src/urbit/channel-ops.ts index a9117e49ada8..6a1cf8279845 100644 --- a/extensions/tlon/src/urbit/channel-ops.ts +++ b/extensions/tlon/src/urbit/channel-ops.ts @@ -66,7 +66,11 @@ export async function pokeUrbitChannel( const errorText = await readResponseTextLimited(response, TLON_ERROR_BODY_LIMIT_BYTES).catch( () => "", ); - throw new Error(`Poke failed: ${response.status}${errorText ? ` - ${errorText}` : ""}`); + throw new UrbitHttpError({ + operation: "Poke", + status: response.status, + bodyText: errorText || undefined, + }); } return pokeId; } finally { @@ -95,7 +99,10 @@ export async function scryUrbitPath( try { if (!response.ok) { - throw new Error(`Scry failed: ${response.status} for path ${params.path}`); + throw new UrbitHttpError({ + operation: `Scry for path ${params.path}`, + status: response.status, + }); } // Successful scry bodies come from a remote Urbit and have no protocol size bound. // Keep the shared JSON ceiling while retaining the path needed to identify the endpoint. diff --git a/extensions/tlon/src/urbit/sse-client.test.ts b/extensions/tlon/src/urbit/sse-client.test.ts index 4c16c9153255..d3efa0838e20 100644 --- a/extensions/tlon/src/urbit/sse-client.test.ts +++ b/extensions/tlon/src/urbit/sse-client.test.ts @@ -2,6 +2,7 @@ import { Readable } from "node:stream"; import { MAX_TIMER_TIMEOUT_MS } from "openclaw/plugin-sdk/number-runtime"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { ensureUrbitChannelOpen } from "./channel-ops.js"; import { urbitFetch } from "./fetch.js"; import { UrbitSSEClient } from "./sse-client.js"; @@ -218,23 +219,88 @@ describe("UrbitSSEClient", () => { expect(client.reconnectAttempts).toBe(0); }); + + it("reopens the same HTTP channel so unacked events can replay", async () => { + vi.useFakeTimers(); + const mockUrbitFetch = vi.mocked(urbitFetch); + mockUrbitFetch.mockResolvedValue({ + response: { ok: true, status: 200, body: null } as unknown as Response, + finalUrl: "https://example.com", + release: vi.fn().mockResolvedValue(undefined), + }); + const onReconnect = vi.fn(); + const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123", { + reconnectDelay: 1, + maxReconnectDelay: 1, + onReconnect, + }); + const channelId = client.channelId; + + const reconnecting = client.attemptReconnect(); + await vi.advanceTimersByTimeAsync(1); + await reconnecting; + + expect(client.channelId).toBe(channelId); + expect(onReconnect).toHaveBeenCalledOnce(); + const callArgs = requireFirstMockCall(mockUrbitFetch.mock.calls, "stream reconnect")[0] as + | Parameters[0] + | undefined; + expect(callArgs?.path).toBe(`/~/channel/${channelId}`); + expect(callArgs?.init?.method).toBe("GET"); + expect(client.reconnectAttempts).toBe(0); + }); + + it("replaces a server-deleted HTTP channel and restores subscriptions", async () => { + vi.useFakeTimers(); + const mockUrbitFetch = vi.mocked(urbitFetch); + mockUrbitFetch + .mockResolvedValueOnce({ + response: { ok: false, status: 404, body: null } as unknown as Response, + finalUrl: "https://example.com", + release: vi.fn().mockResolvedValue(undefined), + }) + .mockResolvedValueOnce({ + response: { ok: true, status: 200, body: null } as unknown as Response, + finalUrl: "https://example.com", + release: vi.fn().mockResolvedValue(undefined), + }); + const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123", { + reconnectDelay: 1, + maxReconnectDelay: 1, + }); + await client.subscribe({ app: "chat", path: "/dm/~zod", event: () => {} }); + const deletedChannelId = client.channelId; + + const reconnecting = client.attemptReconnect(); + await vi.advanceTimersByTimeAsync(1); + await reconnecting; + + expect(client.channelId).not.toBe(deletedChannelId); + expect(ensureUrbitChannelOpen).toHaveBeenCalledWith( + expect.objectContaining({ channelId: client.channelId }), + expect.objectContaining({ createBody: client.subscriptions }), + ); + expect(mockUrbitFetch).toHaveBeenCalledTimes(2); + expect(mockUrbitFetch.mock.calls[1]?.[0].path).toBe(`/~/channel/${client.channelId}`); + expect(client.reconnectAttempts).toBe(0); + }); }); describe("event acking", () => { - it("logs malformed SSE JSON with an owned parser error", () => { + it("logs malformed SSE JSON with an owned parser error", async () => { const logger = { error: vi.fn() }; const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123", { logger, }); - client.processEvent("id: 1\ndata: {not json"); + await client.processEvent("id: 1\ndata: {not json"); expect(logger.error).toHaveBeenCalledWith( "Error parsing SSE event: Error: Tlon Urbit SSE event was malformed JSON", ); }); - it("guards JSON.parse against oversized SSE payload to prevent OOM", () => { + it("guards JSON.parse against oversized SSE payload to prevent OOM", async () => { const errors: string[] = []; const logger = { error: (msg: string) => errors.push(msg) }; const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123", { logger }); @@ -247,7 +313,7 @@ describe("UrbitSSEClient", () => { const padLen = cap + 1024 - jsonOverhead; const hugeJson = prefix + "A".repeat(padLen) + suffix; - client.processEvent(`id: 1\ndata: ${hugeJson}`); + await client.processEvent(`id: 1\ndata: ${hugeJson}`); expect(errors).toHaveLength(1); expect(errors[0]).toBe( @@ -255,7 +321,7 @@ describe("UrbitSSEClient", () => { ); }); - it("accepts SSE payload at the 16 MiB boundary", () => { + it("accepts SSE payload at the 16 MiB boundary", async () => { const cap = 16 * 1024 * 1024; // Allocate valid JSON whose byteLength exactly equals cap. const prefix = '{"json":{"ok":true,"x":"'; @@ -268,7 +334,7 @@ describe("UrbitSSEClient", () => { const handler = vi.fn(); client.eventHandlers.set(1, { event: handler }); - client.processEvent(`id: 1\ndata: ${hugeJson}`); + await client.processEvent(`id: 1\ndata: ${hugeJson}`); expect(handler).toHaveBeenCalledTimes(1); const payload = handler.mock.calls[0]?.[0] as { ok?: boolean; x?: string } | undefined; expect(payload?.ok).toBe(true); @@ -372,13 +438,102 @@ describe("UrbitSSEClient", () => { }); const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123"); - client.processEvent('id: 25abc\ndata: {"json":{"ok":true}}'); - await Promise.resolve(); + await client.processEvent('id: 25abc\ndata: {"json":{"ok":true}}'); expect(mockUrbitFetch).not.toHaveBeenCalled(); expect((client as unknown as { lastHeardEventId: number }).lastHeardEventId).toBe(-1); }); + it("waits for durable admission before acknowledging the transport event", async () => { + const mockUrbitFetch = vi.mocked(urbitFetch); + mockUrbitFetch.mockResolvedValue({ + response: { ok: true, status: 200 } as unknown as Response, + finalUrl: "https://example.com", + release: vi.fn().mockResolvedValue(undefined), + }); + let releaseAdmission = () => {}; + const admission = new Promise((resolve) => { + releaseAdmission = resolve; + }); + const handler = vi.fn(async () => { + await admission; + }); + const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123"); + client.eventHandlers.set(1, { event: handler }); + + const processing = client.processEvent('id: 20\ndata: {"id":1,"json":{"ok":true}}'); + await vi.waitFor(() => expect(handler).toHaveBeenCalledTimes(1)); + expect(mockUrbitFetch).not.toHaveBeenCalled(); + + releaseAdmission(); + await processing; + expect(mockUrbitFetch).toHaveBeenCalledTimes(1); + const body = mockUrbitFetch.mock.calls[0]?.[0].init?.body; + if (typeof body !== "string") { + throw new Error("Expected string ACK request body"); + } + expect(JSON.parse(body)).toEqual([{ id: expect.any(Number), action: "ack", "event-id": 20 }]); + }); + + it("does not acknowledge a failed durable admission", async () => { + const mockUrbitFetch = vi.mocked(urbitFetch); + const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123"); + client.eventHandlers.set(1, { + event: async () => { + throw new Error("sqlite unavailable"); + }, + }); + + await expect( + client.processEvent('id: 20\ndata: {"id":1,"json":{"ok":true}}'), + ).rejects.toThrow("sqlite unavailable"); + expect(mockUrbitFetch).not.toHaveBeenCalled(); + }); + + it("does not advance the ack watermark when the ack request fails", async () => { + const mockUrbitFetch = vi.mocked(urbitFetch); + mockUrbitFetch.mockResolvedValue({ + response: { ok: false, status: 503 } as unknown as Response, + finalUrl: "https://example.com", + release: vi.fn().mockResolvedValue(undefined), + }); + const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123"); + client.eventHandlers.set(1, { event: vi.fn() }); + + await expect( + client.processEvent('id: 20\ndata: {"id":1,"json":{"ok":true}}'), + ).rejects.toThrow("Ack failed with status 503"); + expect( + (client as unknown as { lastAcknowledgedEventId: number }).lastAcknowledgedEventId, + ).toBe(-1); + }); + + it("retries a failed ack when the unacknowledged event replays", async () => { + const mockUrbitFetch = vi.mocked(urbitFetch); + mockUrbitFetch + .mockResolvedValueOnce({ + response: { ok: false, status: 503 } as unknown as Response, + finalUrl: "https://example.com", + release: vi.fn().mockResolvedValue(undefined), + }) + .mockResolvedValueOnce({ + response: { ok: true, status: 204 } as unknown as Response, + finalUrl: "https://example.com", + release: vi.fn().mockResolvedValue(undefined), + }); + const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123"); + client.eventHandlers.set(1, { event: vi.fn() }); + const event = 'id: 20\ndata: {"id":1,"json":{"ok":true}}'; + + await expect(client.processEvent(event)).rejects.toThrow("Ack failed with status 503"); + await expect(client.processEvent(event)).resolves.toBeUndefined(); + + expect(mockUrbitFetch).toHaveBeenCalledTimes(2); + expect( + (client as unknown as { lastAcknowledgedEventId: number }).lastAcknowledgedEventId, + ).toBe(20); + }); + it("tracks lastHeardEventId and ackThreshold", () => { const client = new UrbitSSEClient("https://example.com", "urbauth-~zod=123"); diff --git a/extensions/tlon/src/urbit/sse-client.ts b/extensions/tlon/src/urbit/sse-client.ts index 1b7c2a915f8e..744f448cc8b4 100644 --- a/extensions/tlon/src/urbit/sse-client.ts +++ b/extensions/tlon/src/urbit/sse-client.ts @@ -6,6 +6,7 @@ import { readResponseTextLimited } from "openclaw/plugin-sdk/provider-http"; import type { LookupFn, SsrFPolicy } from "openclaw/plugin-sdk/ssrf-runtime"; import { ensureUrbitChannelOpen, pokeUrbitChannel, scryUrbitPath } from "./channel-ops.js"; import { getUrbitContext, normalizeUrbitCookie } from "./context.js"; +import { UrbitHttpError } from "./errors.js"; import { urbitFetch } from "./fetch.js"; type UrbitSseLogger = { @@ -63,7 +64,11 @@ export class UrbitSSEClient { }> = []; eventHandlers = new Map< number, - { event?: (data: unknown) => void; err?: (error: unknown) => void; quit?: () => void } + { + event?: (data: unknown) => Promise | void; + err?: (error: unknown) => void; + quit?: () => void; + } >(); aborted = false; streamController: AbortController | null = null; @@ -80,7 +85,7 @@ export class UrbitSSEClient { fetchImpl?: (input: RequestInfo | URL, init?: RequestInit) => Promise; streamRelease: (() => Promise) | null = null; - // Event ack tracking - must ack every ~50 events to keep channel healthy + // Event ack tracking keeps each HTTP channel's delivered window bounded. private lastHeardEventId = -1; private lastAcknowledgedEventId = -1; private readonly ackThreshold = 20; @@ -115,10 +120,24 @@ export class UrbitSSEClient { }; } + private resetChannelIdentity(): void { + this.channelId = `${Math.floor(Date.now() / 1000)}-${randomUUID()}`; + this.channelUrl = new URL(`/~/channel/${this.channelId}`, this.url).toString(); + this.lastHeardEventId = -1; + this.lastAcknowledgedEventId = -1; + } + + private async createCurrentChannel(): Promise { + await ensureUrbitChannelOpen(this.channelRequestContext(), { + createBody: this.subscriptions, + createAuditContext: "tlon-urbit-channel-create", + }); + } + async subscribe(params: { app: string; path: string; - event?: (data: unknown) => void; + event?: (data: unknown) => Promise | void; err?: (error: unknown) => void; quit?: () => void; }) { @@ -170,10 +189,10 @@ export class UrbitSSEClient { } async connect() { - await ensureUrbitChannelOpen(this.channelRequestContext(), { - createBody: this.subscriptions, - createAuditContext: "tlon-urbit-channel-create", - }); + // A fresh HTTP channel owns an independent event-id and ack sequence. + this.lastHeardEventId = -1; + this.lastAcknowledgedEventId = -1; + await this.createCurrentChannel(); await this.openStream(); this.isConnected = true; @@ -218,7 +237,7 @@ export class UrbitSSEClient { if (!response.ok) { this.streamRelease = null; await release(); - throw new Error(`Stream connection failed: ${response.status}`); + throw new UrbitHttpError({ operation: "Stream connection", status: response.status }); } this.processStream(response.body).catch((error: unknown) => { @@ -263,12 +282,12 @@ export class UrbitSSEClient { buffer += text; bufferBytes = nextBytes; }; - const consumeText = (text: string) => { + const consumeText = async (text: string) => { let offset = 0; if (pendingDelimiterNewline && text.length > 0) { pendingDelimiterNewline = false; if (text.startsWith("\n")) { - this.processEvent(buffer); + await this.processEvent(buffer); buffer = ""; bufferBytes = 0; offset = 1; @@ -287,7 +306,7 @@ export class UrbitSSEClient { return; } appendPending(text.slice(offset, eventEnd)); - this.processEvent(buffer); + await this.processEvent(buffer); buffer = ""; bufferBytes = 0; offset = eventEnd + 2; @@ -300,13 +319,13 @@ export class UrbitSSEClient { break; } if (typeof chunk === "string") { - consumeText(decoder.decode()); - consumeText(chunk); + await consumeText(decoder.decode()); + await consumeText(chunk); } else { - consumeText(decoder.decode(chunk as Uint8Array, { stream: true })); + await consumeText(decoder.decode(chunk as Uint8Array, { stream: true })); } } - consumeText(decoder.decode()); + await consumeText(decoder.decode()); } finally { if (this.streamRelease) { const release = this.streamRelease; @@ -322,7 +341,7 @@ export class UrbitSSEClient { } } - processEvent(eventData: string) { + async processEvent(eventData: string): Promise { const lines = eventData.split("\n"); let data: string | null = null; let eventId: number | null = null; @@ -340,49 +359,54 @@ export class UrbitSSEClient { return; } - // Track event ID and send ack if needed - if (eventId !== null && !Number.isNaN(eventId)) { - if (eventId > this.lastHeardEventId) { - this.lastHeardEventId = eventId; - if (eventId - this.lastAcknowledgedEventId > this.ackThreshold) { - this.logger.log?.( - `[SSE] Acking event ${eventId} (last acked: ${this.lastAcknowledgedEventId})`, - ); - this.ack(eventId).catch((err: unknown) => { - this.logger.error?.(`Failed to ack event ${eventId}: ${String(err)}`); - }); - } - } - } - + let parsed: ReturnType; try { - const parsed = parseUrbitSsePayload(data); - - if (parsed.response === "quit") { - if (parsed.id) { - const handlers = this.eventHandlers.get(parsed.id); - if (handlers?.quit) { - handlers.quit(); - } - } - return; - } - - if (parsed.id && this.eventHandlers.has(parsed.id)) { - const { event } = this.eventHandlers.get(parsed.id) ?? {}; - if (event && parsed.json) { - event(parsed.json); - } - } else if (parsed.json) { - for (const { event } of this.eventHandlers.values()) { - if (event) { - event(parsed.json); - } - } - } + parsed = parseUrbitSsePayload(data); } catch (error) { + // Malformed transport payloads are permanent. Count them handled so one + // poison event cannot pin the Urbit channel forever. this.logger.error?.(`Error parsing SSE event: ${String(error)}`); + await this.acknowledgeHandledEventIfNeeded(eventId); + return; } + + if (parsed.response === "quit") { + if (parsed.id) { + const handlers = this.eventHandlers.get(parsed.id); + if (handlers?.quit) { + handlers.quit(); + } + } + } else if (parsed.id && this.eventHandlers.has(parsed.id)) { + const { event } = this.eventHandlers.get(parsed.id) ?? {}; + if (event && parsed.json) { + await event(parsed.json); + } + } else if (parsed.json) { + for (const { event } of this.eventHandlers.values()) { + if (event) { + await event(parsed.json); + } + } + } + // Handler failures propagate without ack. Durable callbacks resolve only after append. + await this.acknowledgeHandledEventIfNeeded(eventId); + } + + private async acknowledgeHandledEventIfNeeded(eventId: number | null): Promise { + if (eventId === null || eventId <= this.lastAcknowledgedEventId) { + return; + } + this.lastHeardEventId = Math.max(this.lastHeardEventId, eventId); + if (this.lastHeardEventId - this.lastAcknowledgedEventId <= this.ackThreshold) { + return; + } + this.logger.log?.( + `[SSE] Acking event ${this.lastHeardEventId} (last acked: ${this.lastAcknowledgedEventId})`, + ); + // The acknowledged watermark advances only after PUT succeeds. A replay + // therefore retries a failed ack instead of leaving the subscription clogged. + await this.ack(this.lastHeardEventId); } async poke(params: { app: string; mark: string; json: unknown }) { @@ -414,8 +438,6 @@ export class UrbitSSEClient { } private async ack(eventId: number): Promise { - this.lastAcknowledgedEventId = eventId; - const ackData = { id: Date.now(), action: "ack", @@ -431,6 +453,7 @@ export class UrbitSSEClient { if (!response.ok) { throw new Error(`Ack failed with status ${response.status}`); } + this.lastAcknowledgedEventId = eventId; } finally { await release(); } @@ -470,15 +493,31 @@ export class UrbitSSEClient { setTimeout(resolve, delay); }); - try { - this.channelId = `${Math.floor(Date.now() / 1000)}-${randomUUID()}`; - this.channelUrl = new URL(`/~/channel/${this.channelId}`, this.url).toString(); + if (this.aborted || !this.autoReconnect) { + return; + } + try { if (this.onReconnect) { await this.onReconnect(this); } - await this.connect(); + try { + // Reopen the same Eyre channel. Its queue retains every unacked event; + // switching ids here would discard the cursor and strand failed admission. + await this.openStream(); + } catch (error) { + if (!(error instanceof UrbitHttpError) || error.status !== 404) { + throw error; + } + // Eyre deletes idle channels. Only a definitive missing-channel response + // permits losing the old cursor and rebuilding every subscription. + this.resetChannelIdentity(); + await this.createCurrentChannel(); + await this.openStream(); + } + this.isConnected = true; + this.reconnectAttempts = 0; this.logger.log?.("[SSE] Reconnection successful!"); } catch (error) { this.logger.error?.(`[SSE] Reconnection failed: ${String(error)}`); @@ -486,10 +525,14 @@ export class UrbitSSEClient { } } - async close() { + stopReceiving(): void { this.aborted = true; this.isConnected = false; this.streamController?.abort(); + } + + async close() { + this.stopReceiving(); try { const unsubscribes = this.subscriptions.map((sub) => ({