@@ -1,4 +1,5 @@
11import { randomUUID } from "node:crypto";
2+import { EventHub } from "./event-hub.js";
23import { normalizeGatewayEvent } from "./normalize.js";
34import { GatewayClientTransport, isConnectableTransport } from "./transport.js";
45import type {
@@ -15,6 +16,10 @@ import type {
1516SessionTarget,
1617} from "./types.js";
171819+const MAX_REPLAY_RUNS = 100;
20+const MAX_REPLAY_EVENTS_PER_RUN = 500;
21+const MAX_NORMALIZED_REPLAY_EVENTS = 2000;
22+1823export type OpenClawOptions = {
1924gateway?: "auto" | (string & {});
2025url?: string;
@@ -36,17 +41,32 @@ function resolveGatewayUrl(options: OpenClawOptions): string | undefined {
36413742function runStatusFromWaitPayload(payload: unknown): RunResult["status"] {
3843const record =
39-typeof payload === "object" && payload !== null ? (payload as { status?: unknown }) : {};
40-const status = typeof record.status === "string" ? record.status : undefined;
44+typeof payload === "object" && payload !== null
45+ ? (payload as { aborted?: unknown; status?: unknown; stopReason?: unknown })
46+ : {};
47+const status = typeof record.status === "string" ? record.status.toLowerCase() : undefined;
48+const stopReason = typeof record.stopReason === "string" ? record.stopReason.toLowerCase() : "";
49+if (
50+status === "aborted" ||
51+status === "cancelled" ||
52+status === "canceled" ||
53+status === "killed" ||
54+stopReason === "aborted" ||
55+stopReason === "cancelled" ||
56+stopReason === "canceled" ||
57+stopReason === "killed" ||
58+stopReason === "rpc" ||
59+stopReason === "user" ||
60+(record.aborted === true && stopReason === "stop")
61+) {
62+return "cancelled";
63+}
4164if (status === "ok" || status === "completed" || status === "succeeded") {
4265return "completed";
4366}
4467if (status === "timeout" || status === "timed_out") {
4568return "timed_out";
4669}
47-if (status === "cancelled" || status === "canceled") {
48-return "cancelled";
49-}
5070if (status === "accepted") {
5171return "accepted";
5272}
@@ -149,7 +169,13 @@ export class OpenClaw {
149169readonly environments: EnvironmentsNamespace;
150170151171private readonly transport: OpenClawTransport;
172+private readonly normalizedEvents = new EventHub<OpenClawEvent>({
173+replayLimit: MAX_NORMALIZED_REPLAY_EVENTS,
174+});
175+private readonly replayByRunId = new Map<string, OpenClawEvent[]>();
152176private connected = false;
177+private eventPumpPromise: Promise<void> | null = null;
178+private eventPumpReady: Promise<void> | null = null;
153179154180constructor(options: OpenClawOptions = {}) {
155181this.transport =
@@ -173,16 +199,22 @@ export class OpenClaw {
173199174200async connect(): Promise<void> {
175201if (this.connected) {
202+await this.startEventPump();
176203return;
177204}
178205if (isConnectableTransport(this.transport)) {
179206await this.transport.connect();
180207}
181208this.connected = true;
209+await this.startEventPump();
182210}
183211184212async close(): Promise<void> {
185213await this.transport.close?.();
214+await this.eventPumpPromise?.catch(() => {});
215+this.normalizedEvents.close();
216+this.eventPumpPromise = null;
217+this.eventPumpReady = null;
186218this.connected = false;
187219}
188220@@ -196,20 +228,134 @@ export class OpenClaw {
196228}
197229198230events(filter?: (event: OpenClawEvent) => boolean): AsyncIterable<OpenClawEvent> {
199-const source = this.transport.events();
200-async function* iterate(): AsyncIterable<OpenClawEvent> {
201-for await (const event of source) {
202-const normalized = normalizeGatewayEvent(event);
203-if (!filter || filter(normalized)) {
204-yield normalized;
231+return this.iterateEvents(filter);
232+}
233+234+runEvents(
235+runId: string,
236+filter?: (event: OpenClawEvent) => boolean,
237+): AsyncIterable<OpenClawEvent> {
238+return this.iterateRunEvents(runId, filter);
239+}
240+241+rawEvents(filter?: (event: GatewayEvent) => boolean): AsyncIterable<GatewayEvent> {
242+return this.transport.events(filter);
243+}
244+245+private async *iterateEvents(
246+filter?: (event: OpenClawEvent) => boolean,
247+): AsyncIterable<OpenClawEvent> {
248+await this.connect();
249+for await (const event of this.normalizedEvents.stream(filter)) {
250+yield event;
251+}
252+}
253+254+private async *iterateRunEvents(
255+runId: string,
256+filter?: (event: OpenClawEvent) => boolean,
257+): AsyncIterable<OpenClawEvent> {
258+await this.connect();
259+const matches = (event: OpenClawEvent) => {
260+if (event.runId !== runId) {
261+return false;
262+}
263+return filter ? filter(event) : true;
264+};
265+const liveSource = this.normalizedEvents.stream(matches, { replay: true });
266+const live = liveSource[Symbol.asyncIterator]();
267+let nextLive = live.next();
268+const seen = new Set<string>();
269+try {
270+for (const event of this.replaySnapshot(runId)) {
271+if (!matches(event) || seen.has(event.id)) {
272+continue;
205273}
274+seen.add(event.id);
275+yield event;
206276}
277+while (true) {
278+const next = await nextLive;
279+if (next.done) {
280+break;
281+}
282+nextLive = live.next();
283+if (seen.has(next.value.id)) {
284+continue;
285+}
286+seen.add(next.value.id);
287+yield next.value;
288+}
289+} finally {
290+await live.return?.();
207291}
208-return iterate();
209292}
210293211-rawEvents(filter?: (event: GatewayEvent) => boolean): AsyncIterable<GatewayEvent> {
212-return this.transport.events(filter);
294+private startEventPump(): Promise<void> {
295+if (this.eventPumpReady) {
296+return this.eventPumpReady;
297+}
298+let markReady = () => {};
299+let ready = false;
300+this.eventPumpReady = new Promise<void>((resolve) => {
301+markReady = () => {
302+if (ready) {
303+return;
304+}
305+ready = true;
306+resolve();
307+};
308+});
309+this.eventPumpPromise = (async () => {
310+const iterator = this.transport.events()[Symbol.asyncIterator]();
311+try {
312+while (true) {
313+const next = iterator.next();
314+await Promise.resolve();
315+markReady();
316+const result = await next;
317+if (result.done) {
318+break;
319+}
320+const normalized = normalizeGatewayEvent(result.value);
321+this.recordReplayEvent(normalized);
322+this.normalizedEvents.publish(normalized);
323+}
324+} finally {
325+markReady();
326+await iterator.return?.();
327+this.normalizedEvents.close();
328+}
329+})().catch(() => {
330+markReady();
331+this.normalizedEvents.close();
332+});
333+return this.eventPumpReady;
334+}
335+336+private recordReplayEvent(event: OpenClawEvent): void {
337+if (!event.runId) {
338+return;
339+}
340+let events = this.replayByRunId.get(event.runId);
341+if (!events) {
342+if (this.replayByRunId.size >= MAX_REPLAY_RUNS) {
343+const oldestRunId = this.replayByRunId.keys().next().value;
344+if (oldestRunId) {
345+this.replayByRunId.delete(oldestRunId);
346+}
347+}
348+events = [];
349+this.replayByRunId.set(event.runId, events);
350+}
351+events.push(event);
352+if (events.length > MAX_REPLAY_EVENTS_PER_RUN) {
353+events.splice(0, events.length - MAX_REPLAY_EVENTS_PER_RUN);
354+}
355+}
356+357+private replaySnapshot(runId: string): OpenClawEvent[] {
358+return [...(this.replayByRunId.get(runId) ?? [])];
213359}
214360}
215361@@ -241,12 +387,7 @@ export class Run {
241387) {}
242388243389events(filter?: (event: OpenClawEvent) => boolean): AsyncIterable<OpenClawEvent> {
244-return this.client.events((event) => {
245-if (event.runId !== this.id) {
246-return false;
247-}
248-return filter ? filter(event) : true;
249-});
390+return this.client.runEvents(this.id, filter);
250391}
251392252393async wait(options?: { timeoutMs?: number }): Promise<RunResult> {
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。