ewilliams28/openclaw
1
0
Fork 0
mirror of https://github.com/openclaw/openclaw.git synced 2026-08-12 21:53:00 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix(gateway): streamed chat text stalls mid-answer until the next agent event (#119566)

Browse Source 
* fix(gateway): flush trailing chat deltas

Keep the fixed-deadline wake-up on the existing chat run record so terminal, abort, and shutdown cleanup cancel it at the lifecycle owner. Normalize voice runs onto their unique per-turn ID instead of compensating for a stale client alias downstream.

Fixes #119557

Co-authored-by: Serghei <43180231+xyrolle@users.noreply.github.com>

* test(gateway): drop obsolete delta length assertions

Remove stale fixtures and assertions for the write-only chat delta length field deleted by the owner-boundary repair.

Co-authored-by: Serghei <43180231+xyrolle@users.noreply.github.com>

* fix(gateway): preserve chat run state type contract

Keep the optional delta length field and its cleanup semantics in the exported ChatRunState closure without restoring runtime writes.

Co-authored-by: Serghei <43180231+xyrolle@users.noreply.github.com>

* refactor(gateway): hide chat delta timer from SDK shape

Keep the trailing wake physically on each run record while exposing it only through internal state-module accessors, preserving the public ChatRunState closure.

Co-authored-by: Serghei <43180231+xyrolle@users.noreply.github.com>

* refactor(gateway): narrow chat delta timer access

Use one Gateway-internal record accessor so the run-owned timer remains outside the public ChatRunState shape without extra state maps.

Co-authored-by: Serghei <43180231+xyrolle@users.noreply.github.com>

* refactor(gateway): keep chat timer casts private

Keep the run-owned timer invisible to the generated Plugin SDK closure by using module-private casts in each owning Gateway module.

Co-authored-by: Serghei <43180231+xyrolle@users.noreply.github.com>

---------

Co-authored-by: Peter Steinberger <steipete@gmail.com>
This commit is contained in:
Serghei
2026-08-09 14:53:16 +03:00
committed by GitHub
parent f83ba2b326
commit d7054d4daf
7 changed files with 305 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/e2e/qa-lab/runtime/agent-session-streaming.e2e.test.ts
+82 -3
View File
@@ -10,7 +10,7 @@ import {
const TEST_TIMEOUT_MS = 120_000;
const REQUEST_TIMEOUT_MS = 20_000;
const STREAM_INTERVAL_MS = 225;
const STREAM_INTERVAL_MS = 20;
const MODEL_REF = "mock-openai/gpt-5.6-luna";
const SESSION_KEY = "agent:qa:qa:session-streaming";
const IDEMPOTENCY_KEY = "qa-session-streaming";
@@ -27,6 +27,12 @@ type GatewayEvent = {
event: string;
payload?: unknown;
};
type ChatEventPayload = {
runId?: string;
sessionKey?: string;
state?: string;
deltaText?: string;
};
type AgentEvent = {
runId?: string;
sessionKey?: string;
@@ -41,6 +47,14 @@ type AgentEvent = {
const cleanups: Array<() => Promise<void>> = [];
function createDeferred() {
let resolve = () => {};
const promise = new Promise<void>((settle) => {
resolve = settle;
});
return { promise, resolve };
}
afterEach(async () => {
const errors: unknown[] = [];
for (const cleanup of cleanups.splice(0).toReversed()) {
@@ -62,7 +76,11 @@ function writeEvent(response: ServerResponse, event: unknown): void {
response.write(`data: ${JSON.stringify(event)}\n\n`);
}
async function writeStreamingResponse(response: ServerResponse): Promise<void> {
async function writeStreamingResponse(
response: ServerResponse,
deltasSent: ReturnType<typeof createDeferred>,
terminalRelease: ReturnType<typeof createDeferred>,
): Promise<void> {
const message = {
type: "message",
id: "qa-session-streaming-message",
@@ -91,6 +109,8 @@ async function writeStreamingResponse(response: ServerResponse): Promise<void> {
});
await delay(STREAM_INTERVAL_MS);
}
deltasSent.resolve();
await terminalRelease.promise;
writeEvent(response, {
type: "response.output_text.done",
item_id: message.id,
@@ -114,6 +134,13 @@ async function writeStreamingResponse(response: ServerResponse): Promise<void> {
async function startStreamingProvider() {
const providerRequests: Array<Record<string, unknown>> = [];
const transportRequests: string[] = [];
const deltasSent = createDeferred();
const terminalRelease = createDeferred();
let terminalReleased = false;
const releaseTerminal = () => {
terminalReleased = true;
terminalRelease.resolve();
};
const server = createServer((request, response) => {
void (async () => {
if (request.method === "GET" && request.url === "/v1/models") {
@@ -133,7 +160,7 @@ async function startStreamingProvider() {
providerRequests.push(
JSON.parse(Buffer.concat(chunks).toString("utf8")) as Record<string, unknown>,
);
await writeStreamingResponse(response);
await writeStreamingResponse(response, deltasSent, terminalRelease);
return;
}
transportRequests.push(`${request.method ?? "UNKNOWN"} ${request.url ?? ""}`);
@@ -153,9 +180,13 @@ async function startStreamingProvider() {
}
return {
baseUrl: `http://127.0.0.1:${address.port}`,
deltasSent: deltasSent.promise,
isTerminalReleased: () => terminalReleased,
releaseTerminal,
providerRequests,
transportRequests,
stop: async () => {
releaseTerminal();
server.closeAllConnections();
await new Promise<void>((resolve, reject) => {
server.close((error) => (error ? reject(error) : resolve()));
@@ -216,6 +247,12 @@ function asAgentEvent(event: GatewayEvent): AgentEvent | undefined {
: undefined;
}
function asChatEvent(event: GatewayEvent): ChatEventPayload | undefined {
return event.event === "chat" && event.payload && typeof event.payload === "object"
? (event.payload as ChatEventPayload)
: undefined;
}
function messageRole(message: unknown): string | undefined {
const role = message && typeof message === "object" ? (message as { role?: unknown }).role : null;
return typeof role === "string" ? role : undefined;
@@ -275,6 +312,7 @@ describe("agent session streaming", () => {
const gatewayEvents: GatewayEvent[] = [];
const client = await connectOperator(gateway, gatewayEvents);
cleanups.push(() => client.stopAndWait({ timeoutMs: 1_000 }));
await client.request("sessions.messages.subscribe", { key: SESSION_KEY });
const accepted = await client.request<AgentResult>("agent", {
sessionKey: SESSION_KEY,
message: REQUEST_MESSAGE,
@@ -286,6 +324,21 @@ describe("agent session streaming", () => {
runId: IDEMPOTENCY_KEY,
});
await provider.deltasSent;
await vi.waitFor(
() => {
expect(provider.isTerminalReleased()).toBe(false);
const streamedChatText = gatewayEvents
.map(asChatEvent)
.filter((event) => event?.runId === IDEMPOTENCY_KEY && event.state === "delta")
.map((event) => event?.deltaText ?? "")
.join("");
expect(streamedChatText).toBe(TERMINAL_TEXT);
},
{ interval: 20, timeout: REQUEST_TIMEOUT_MS },
);
provider.releaseTerminal();
const terminal = await client.request<AgentResult>(
"agent.wait",
{ runId: IDEMPOTENCY_KEY, timeoutMs: 30_000 },
@@ -345,6 +398,32 @@ describe("agent session streaming", () => {
});
expect(terminalEvents[0]?.seq).toBeGreaterThan(deltaSeqs.at(-1) ?? 0);
await vi.waitFor(
() => {
const runChatEvents = gatewayEvents
.map(asChatEvent)
.filter((event) => event?.runId === IDEMPOTENCY_KEY);
expect(runChatEvents.filter((event) => event?.state === "final")).toHaveLength(1);
},
{ interval: 20, timeout: REQUEST_TIMEOUT_MS },
);
const runChatEventsAtTerminal = gatewayEvents
.map(asChatEvent)
.filter((event) => event?.runId === IDEMPOTENCY_KEY);
const finalIndex = runChatEventsAtTerminal.findIndex((event) => event?.state === "final");
expect(finalIndex).toBeGreaterThan(-1);
expect(
runChatEventsAtTerminal.slice(finalIndex + 1).some((event) => event?.state === "delta"),
).toBe(false);
await delay(250);
expect(
gatewayEvents
.map(asChatEvent)
.filter((event) => event?.runId === IDEMPOTENCY_KEY)
.slice(finalIndex + 1)
.some((event) => event?.state === "delta"),
).toBe(false);
const streamedText = assistantDeltas.map((event) => event.data?.delta ?? "").join("");
expect(streamedText).toBe(TERMINAL_TEXT);