Skip to content

Commit e04ea79

Browse files
steipetesteipete
authored
feat(twitch): preserve accepted chat through local crashes (#110852)
* feat(twitch): add durable ingress queue * fix(twitch): satisfy ingress type boundaries
1 parent 736dddb commit e04ea79

11 files changed

Lines changed: 817 additions & 36 deletions

‎extensions/twitch/src/access-control.test.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ describe("checkTwitchAccessControl", () => {
1212
};
1313

1414
const mockMessage: TwitchChatMessage = {
15+
id: "message-1",
1516
username: "testuser",
1617
userId: "123456",
1718
message: "hello bot",

‎extensions/twitch/src/monitor.test.ts‎

Lines changed: 36 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,8 +4,11 @@ import type { TwitchChatMessage } from "./types.js";
44

55
const mocks = vi.hoisted(() => ({
66
checkAccess: vi.fn(async () => ({ allowed: true })),
7+
createIngress: vi.fn(),
78
getClient: vi.fn(async () => ({})),
89
getRuntime: vi.fn(),
10+
ingressStart: vi.fn(),
11+
ingressStop: vi.fn(async () => undefined),
912
onMessage: vi.fn(),
1013
runInbound: vi.fn(),
1114
sendMessage: vi.fn(),
@@ -28,6 +31,10 @@ vi.mock("./runtime.js", () => ({
2831
getTwitchRuntime: mocks.getRuntime,
2932
}));
3033

34+
vi.mock("./twitch-ingress.js", () => ({
35+
createTwitchIngress: mocks.createIngress,
36+
}));
37+
3138
import { monitorTwitchProvider } from "./monitor.js";
3239

3340
type InboundRunInput = {
@@ -47,6 +54,32 @@ describe("monitorTwitchProvider", () => {
4754
vi.clearAllMocks();
4855
mocks.getClient.mockResolvedValue({});
4956
mocks.sendMessage.mockResolvedValue({ ok: true, messageId: "message-id" });
57+
mocks.createIngress.mockImplementation(
58+
(options: {
59+
deliver: (
60+
message: TwitchChatMessage,
61+
lifecycle: {
62+
admission: "exclusive";
63+
abortSignal: AbortSignal;
64+
onAdopted: () => Promise<void>;
65+
onDeferred: () => void;
66+
onAbandoned: () => Promise<void>;
67+
},
68+
) => Promise<void>;
69+
}) => ({
70+
accept: async (message: TwitchChatMessage) => {
71+
await options.deliver(message, {
72+
admission: "exclusive",
73+
abortSignal: new AbortController().signal,
74+
onAdopted: async () => undefined,
75+
onDeferred: () => undefined,
76+
onAbandoned: async () => undefined,
77+
});
78+
},
79+
start: mocks.ingressStart,
80+
stop: mocks.ingressStop,
81+
}),
82+
);
5083
mocks.runInbound.mockImplementation(async (input: InboundRunInput) => {
5184
const ingested = input.adapter.ingest(input.raw);
5285
const turn = await input.adapter.resolveTurn(ingested);
@@ -108,6 +141,7 @@ describe("monitorTwitchProvider", () => {
108141
});
109142

110143
onMessage?.({
144+
id: "message-1",
111145
username: "viewer",
112146
userId: "viewer-1",
113147
message: "hello bot",
@@ -124,7 +158,8 @@ describe("monitorTwitchProvider", () => {
124158
);
125159
});
126160

127-
monitor.stop();
161+
await monitor.stop();
128162
expect(mocks.unregister).toHaveBeenCalledOnce();
163+
expect(mocks.ingressStop).toHaveBeenCalledOnce();
129164
});
130165
});

‎extensions/twitch/src/monitor.ts‎

Lines changed: 43 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ import { normalizeLowercaseStringOrEmpty } from "openclaw/plugin-sdk/string-coer
1313
import { checkTwitchAccessControl } from "./access-control.js";
1414
import { getOrCreateClientManager } from "./client-manager-registry.js";
1515
import { getTwitchRuntime } from "./runtime.js";
16+
import { createTwitchIngress } from "./twitch-ingress.js";
1617
import type { TwitchAccountConfig, TwitchChatMessage } from "./types.js";
1718
import { stripMarkdownForTwitch } from "./utils/markdown.js";
1819

@@ -31,10 +32,11 @@ type TwitchMonitorOptions = {
3132
};
3233

3334
type TwitchMonitorResult = {
34-
stop: () => void;
35+
stop: () => Promise<void>;
3536
};
3637

3738
type TwitchCoreRuntime = ReturnType<typeof getTwitchRuntime>;
39+
type TwitchIngressLifecycle = Parameters<Parameters<typeof createTwitchIngress>[0]["deliver"]>[1];
3840

3941
/**
4042
* Process an incoming Twitch message and dispatch to agent.
@@ -46,19 +48,22 @@ async function processTwitchMessage(params: {
4648
config: unknown;
4749
runtime: TwitchRuntimeEnv;
4850
core: TwitchCoreRuntime;
51+
turnAdoptionLifecycle: TwitchIngressLifecycle;
4952
statusSink?: (patch: { lastInboundAt?: number; lastOutboundAt?: number }) => void;
5053
}): Promise<void> {
51-
const { message, account, accountId, config, runtime, core, statusSink } = params;
54+
const { message, account, accountId, config, runtime, core, turnAdoptionLifecycle, statusSink } =
55+
params;
5256
const cfg = config as OpenClawConfig;
5357

5458
await core.channel.inbound.run({
5559
channel: "twitch",
5660
accountId,
5761
raw: message,
62+
turnAdoptionLifecycle,
5863
adapter: {
5964
ingest: (incoming) => ({
60-
id: incoming.id ?? `${incoming.channel}:${incoming.timestamp?.getTime() ?? Date.now()}`,
61-
timestamp: incoming.timestamp?.getTime(),
65+
id: incoming.id,
66+
timestamp: incoming.timestamp,
6267
rawText: incoming.message,
6368
textForAgent: incoming.message,
6469
textForCommands: incoming.message,
@@ -220,6 +225,7 @@ export async function monitorTwitchProvider(
220225

221226
const core = getTwitchRuntime();
222227
let stopped = false;
228+
let stopTask: Promise<void> | undefined;
223229

224230
const coreLogger = core.logging.getChildLogger({ module: "twitch" });
225231
const logVerboseMessage = (message: string) => {
@@ -249,12 +255,10 @@ export async function monitorTwitchProvider(
249255
throw error;
250256
}
251257

252-
const unregisterHandler = clientManager.onMessage(account, (message) => {
253-
if (stopped) {
254-
return;
255-
}
256-
257-
void (async () => {
258+
const ingress = createTwitchIngress({
259+
accountId,
260+
runtime,
261+
deliver: async (message, turnAdoptionLifecycle) => {
258262
const botUsername = normalizeLowercaseStringOrEmpty(account.username);
259263
if (normalizeLowercaseStringOrEmpty(message.username) === botUsername) {
260264
return;
@@ -266,7 +270,7 @@ export async function monitorTwitchProvider(
266270
botUsername,
267271
});
268272

269-
if (stopped || !access.allowed) {
273+
if (!access.allowed) {
270274
return;
271275
}
272276

@@ -279,19 +283,41 @@ export async function monitorTwitchProvider(
279283
config,
280284
runtime,
281285
core,
286+
turnAdoptionLifecycle,
282287
statusSink,
283288
});
284-
})().catch((err: unknown) => {
285-
runtime.error?.(`Message processing failed: ${String(err)}`);
289+
},
290+
});
291+
ingress.start();
292+
293+
const unregisterHandler = clientManager.onMessage(account, (message) => {
294+
if (stopped) {
295+
return;
296+
}
297+
298+
void ingress.accept(message).catch((err: unknown) => {
299+
runtime.error?.(`Message durable admission failed: ${String(err)}`);
286300
});
287301
});
288302

289-
const stop = () => {
290-
stopped = true;
291-
unregisterHandler();
303+
const stop = (): Promise<void> => {
304+
stopTask ??= (async () => {
305+
stopped = true;
306+
unregisterHandler();
307+
await ingress.stop();
308+
})();
309+
return stopTask;
292310
};
293311

294-
abortSignal.addEventListener("abort", stop, { once: true });
312+
abortSignal.addEventListener(
313+
"abort",
314+
() => {
315+
void stop().catch((error: unknown) => {
316+
runtime.error?.(`Twitch ingress stop failed: ${String(error)}`);
317+
});
318+
},
319+
{ once: true },
320+
);
295321

296322
return { stop };
297323
}

‎extensions/twitch/src/plugin.live.test.ts‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/**
2-
* Live Twitch IRC verification for the runStoppablePassiveMonitor lifecycle
2+
* Live Twitch IRC verification for the passive account lifecycle
33
* pattern used by the Twitch gateway.
44
*
55
* This test connects to irc.chat.twitch.tv using the same twurple stack the
@@ -19,7 +19,7 @@
1919

2020
import { StaticAuthProvider } from "@twurple/auth";
2121
import { ChatClient } from "@twurple/chat";
22-
import { runStoppablePassiveMonitor } from "openclaw/plugin-sdk/extension-shared";
22+
import { runPassiveAccountLifecycle } from "openclaw/plugin-sdk/channel-outbound";
2323
import { describe, expect, it } from "vitest";
2424

2525
const LIVE = process.env.TWITCH_LIVE_TEST === "1";
@@ -33,7 +33,7 @@ const HAS_CREDS = Boolean(
3333
const maybeDescribe = LIVE && HAS_CREDS ? describe : describe.skip;
3434

3535
maybeDescribe("twitch live IRC lifecycle (skipped unless TWITCH_LIVE_TEST=1)", () => {
36-
it("real twurple connection + runStoppablePassiveMonitor stays pending until abort, then stops cleanly", async () => {
36+
it("real twurple connection stays pending until abort, then stops cleanly", async () => {
3737
const accessTokenRaw = process.env.TWITCH_ACCESS_TOKEN!.replace(/^oauth:/, "");
3838
const clientId = process.env.TWITCH_CLIENT_ID!;
3939
const channel = process.env.TWITCH_CHANNEL!;
@@ -56,7 +56,7 @@ maybeDescribe("twitch live IRC lifecycle (skipped unless TWITCH_LIVE_TEST=1)", (
5656
let settled = false;
5757
let stopCalled = false;
5858

59-
const task = runStoppablePassiveMonitor({
59+
const task = runPassiveAccountLifecycle({
6060
abortSignal: abort.signal,
6161
start: async () => {
6262
const chat = new ChatClient({
@@ -86,6 +86,9 @@ maybeDescribe("twitch live IRC lifecycle (skipped unless TWITCH_LIVE_TEST=1)", (
8686
},
8787
};
8888
},
89+
stop: async (monitor) => {
90+
monitor.stop();
91+
},
8992
})
9093
.then(() => {
9194
settled = true;

‎extensions/twitch/src/plugin.ts‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -12,15 +12,13 @@ import {
1212
createChatChannelPlugin,
1313
stripChannelTargetPrefix,
1414
} from "openclaw/plugin-sdk/channel-core";
15+
import { runPassiveAccountLifecycle } from "openclaw/plugin-sdk/channel-outbound";
1516
import {
1617
createLoggedPairingApprovalNotifier,
1718
createPairingPrefixStripper,
1819
} from "openclaw/plugin-sdk/channel-pairing";
1920
import type { OpenClawConfig } from "openclaw/plugin-sdk/config-contracts";
20-
import {
21-
buildPassiveProbedChannelStatusSummary,
22-
runStoppablePassiveMonitor,
23-
} from "openclaw/plugin-sdk/extension-shared";
21+
import { buildPassiveProbedChannelStatusSummary } from "openclaw/plugin-sdk/extension-shared";
2422
import {
2523
createComputedAccountStatusAdapter,
2624
createDefaultChannelRuntimeState,
@@ -221,7 +219,7 @@ export const twitchPlugin: ChannelPlugin<ResolvedTwitchAccount> =
221219
// supervisor reads the settled task as `channel exited without an
222220
// error` and triggers a restart loop. See #60071.
223221
try {
224-
await runStoppablePassiveMonitor({
222+
await runPassiveAccountLifecycle({
225223
abortSignal: ctx.abortSignal,
226224
start: async () => {
227225
// Lazy import: the monitor pulls the reply pipeline; avoid ESM init cycles.
@@ -234,6 +232,9 @@ export const twitchPlugin: ChannelPlugin<ResolvedTwitchAccount> =
234232
abortSignal: ctx.abortSignal,
235233
});
236234
},
235+
stop: async (monitor) => {
236+
await monitor.stop();
237+
},
237238
});
238239
} catch (error) {
239240
ctx.setStatus?.({

‎extensions/twitch/src/twitch-client.test.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -677,11 +677,11 @@ describe("TwitchClientManager", () => {
677677
expect(capturedMessage?.displayName).toBe("TestUser");
678678
expect(capturedMessage?.userId).toBe("12345");
679679
expect(capturedMessage?.message).toBe("Hello bot!");
680-
expect(capturedMessage?.channel).toBe("testchannel");
680+
expect(capturedMessage?.channel).toBe("#testchannel");
681681
expect(capturedMessage?.chatType).toBe("group");
682682
});
683683

684-
it("should normalize channel names without # prefix", async () => {
684+
it("should preserve channel names without a # prefix", async () => {
685685
await manager.getClient(testAccount);
686686

687687
const onMessageCallback = expectDefined(messageHandlers[0], "Twitch message handler");

‎extensions/twitch/src/twitch-client.ts‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -274,21 +274,21 @@ export class TwitchClientManager {
274274
client.onMessage((channelName, _user, messageText, msg) => {
275275
const handler = this.messageHandlers.get(key);
276276
if (handler) {
277-
const normalizedChannel = channelName.startsWith("#") ? channelName.slice(1) : channelName;
278277
const from = `twitch:${msg.userInfo.userName}`;
279278
const preview = sliceUtf16Safe(messageText, 0, 100).replace(/\n/g, "\\n");
280279
this.logger.debug?.(
281-
`twitch inbound: channel=${normalizedChannel} from=${from} len=${messageText.length} preview="${preview}"`,
280+
`twitch inbound: channel=${channelName} from=${from} len=${messageText.length} preview="${preview}"`,
282281
);
283282

284283
handler({
285284
username: msg.userInfo.userName,
286285
displayName: msg.userInfo.displayName,
287286
userId: msg.userInfo.userId,
288287
message: messageText,
289-
channel: normalizedChannel,
288+
// Preserve the raw callback channel; durable dispatch normalizes it.
289+
channel: channelName,
290290
id: msg.id,
291-
timestamp: new Date(),
291+
timestamp: Date.now(),
292292
isMod: msg.userInfo.isMod,
293293
isOwner: msg.userInfo.isBroadcaster,
294294
isVip: msg.userInfo.isVip,

0 commit comments

Comments
 (0)