Visitar URL original
Apply PR #34042: fix(app): reconcile session pages with concurrent ev… · anomalyco/opencode@e145637 · GitHub
Skip to content

Commit e145637

Browse files
opencode-agent[bot]opencode-agent[bot]
committed
Apply PR #34042: fix(app): reconcile session pages with concurrent events
2 parents b2cf978 + e37a4e9 commit e145637

4 files changed

Lines changed: 1734 additions & 123 deletions

File tree

‎packages/app/src/context/server-sdk.test.ts‎

Lines changed: 116 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import { describe, expect, test } from "bun:test"
2-
import { coalesceServerEvents, resumeStreamAfterPageShow } from "./server-sdk"
2+
import { coalesceServerEvents, enqueueServerEvent, resumeStreamAfterPageShow } from "./server-sdk"
33
import type { Event } from "@opencode-ai/sdk/v2/client"
44

55
describe("resumeStreamAfterPageShow", () => {
@@ -15,19 +15,23 @@ describe("resumeStreamAfterPageShow", () => {
1515
})
1616

1717
describe("coalesceServerEvents", () => {
18-
const delta = (value: string, field = "text") => ({
18+
const delta = (value: string, field = "text", partID = "part") => ({
1919
directory: "/repo",
2020
payload: {
2121
type: "message.part.delta",
22-
properties: { messageID: "msg", partID: "part", field, delta: value },
22+
properties: { messageID: "msg", partID, field, delta: value },
2323
} as Event,
2424
})
2525

2626
test("merges adjacent deltas for the same field", () => {
27-
const result = coalesceServerEvents([delta("hello "), delta("world")])
27+
const first = delta("hello ")
28+
const second = delta("world")
29+
first.payload.id = "first"
30+
second.payload.id = "second"
31+
const result = coalesceServerEvents([first, second])
2832

2933
expect(result).toHaveLength(1)
30-
expect(result[0]?.payload).toMatchObject({ properties: { delta: "hello world" } })
34+
expect(result[0]?.payload).toMatchObject({ id: "second", properties: { delta: "hello world" } })
3135
})
3236

3337
test("preserves event boundaries and distinct fields", () => {
@@ -45,9 +49,113 @@ describe("coalesceServerEvents", () => {
4549
])
4650
})
4751

48-
test("drops stale deltas", () => {
49-
const result = coalesceServerEvents([delta("stale")], new Set(["/repo:msg:part"]))
52+
test("preserves event ID order across interleaved deltas", () => {
53+
const first = delta("a")
54+
const other = delta("b", "text", "other")
55+
const last = delta("c")
56+
first.payload.id = "1"
57+
other.payload.id = "2"
58+
last.payload.id = "3"
5059

51-
expect(result).toEqual([])
60+
const result = coalesceServerEvents([first, other, last])
61+
62+
expect(result.map((event) => event.payload.id)).toEqual(["1", "2", "3"])
63+
})
64+
65+
})
66+
67+
describe("enqueueServerEvent", () => {
68+
const partUpdated = (text: string) =>
69+
({
70+
type: "message.part.updated",
71+
properties: {
72+
sessionID: "session",
73+
part: { id: "part", sessionID: "session", messageID: "message", type: "text", text },
74+
},
75+
}) as Event
76+
77+
test("preserves part updates across message remove and re-add barriers", () => {
78+
const events: Array<{ directory: string; payload: Event }> = []
79+
const enqueue = (payload: Event) => enqueueServerEvent(events, { directory: "/repo", payload })
80+
81+
enqueue(partUpdated("old"))
82+
enqueue({ type: "message.removed", properties: { sessionID: "session", messageID: "message" } } as Event)
83+
enqueue({
84+
type: "message.updated",
85+
properties: {
86+
sessionID: "session",
87+
info: {
88+
id: "message",
89+
sessionID: "session",
90+
role: "user",
91+
time: { created: 1 },
92+
agent: "build",
93+
model: { providerID: "provider", modelID: "model" },
94+
},
95+
},
96+
} as Event)
97+
enqueue(partUpdated("new"))
98+
99+
expect(events.map((event) => event.payload.type)).toEqual([
100+
"message.part.updated",
101+
"message.removed",
102+
"message.updated",
103+
"message.part.updated",
104+
])
105+
})
106+
107+
test("preserves deltas after a replacement snapshot", () => {
108+
const events: Array<{ directory: string; payload: Event }> = []
109+
const enqueue = (payload: Event) => enqueueServerEvent(events, { directory: "/repo", payload })
110+
111+
enqueue(partUpdated("a"))
112+
enqueue(partUpdated("ab"))
113+
enqueue({
114+
type: "message.part.delta",
115+
properties: { sessionID: "session", messageID: "message", partID: "part", field: "text", delta: "c" },
116+
} as Event)
117+
118+
const result = coalesceServerEvents(events)
119+
expect(result.map((event) => event.payload.type)).toEqual(["message.part.updated", "message.part.delta"])
120+
expect(result[0]?.payload).toMatchObject({ properties: { part: { text: "ab" } } })
121+
expect(result[1]?.payload).toMatchObject({ properties: { delta: "c" } })
122+
})
123+
124+
test("preserves updates after session deletion", () => {
125+
const events: Array<{ directory: string; payload: Event }> = []
126+
const enqueue = (payload: Event) => enqueueServerEvent(events, { directory: "/repo", payload })
127+
128+
enqueue(partUpdated("old"))
129+
enqueue({
130+
type: "session.deleted",
131+
properties: { sessionID: "session", info: { id: "session" } },
132+
} as Event)
133+
enqueue(partUpdated("new"))
134+
135+
expect(events.map((event) => event.payload.type)).toEqual([
136+
"message.part.updated",
137+
"session.deleted",
138+
"message.part.updated",
139+
])
140+
})
141+
142+
test("does not coalesce edge-triggered session statuses", () => {
143+
const events: Array<{ directory: string; payload: Event }> = []
144+
const enqueue = (status: "retry" | "busy") =>
145+
enqueueServerEvent(events, {
146+
directory: "/repo",
147+
payload: {
148+
type: "session.status",
149+
properties: {
150+
sessionID: "session",
151+
status: status === "retry" ? { type: "retry", attempt: 1, message: "retry", next: 1 } : { type: "busy" },
152+
},
153+
} as Event,
154+
})
155+
156+
enqueue("retry")
157+
enqueue("busy")
158+
159+
expect(events).toHaveLength(2)
52160
})
53161
})

‎packages/app/src/context/server-sdk.tsx‎

Lines changed: 44 additions & 50 deletions
Original file line numberDiff line numberDiff line change
@@ -17,34 +17,59 @@ const isAbortError = (error: unknown) =>
1717
const isStreamClosed = (error: unknown, signal?: AbortSignal) => isAbortError(error) || signal?.aborted === true
1818
type QueuedServerEvent = { directory: string; payload: Event }
1919

20-
const deltaKey = (directory: string, messageID: string, partID: string) => `${directory}:${messageID}:${partID}`
20+
const coalescedKey = (event: QueuedServerEvent) => {
21+
if (event.payload.type === "lsp.updated") return `lsp.updated:${event.directory}`
22+
if (event.payload.type === "message.part.updated") {
23+
const part = event.payload.properties.part
24+
return `message.part.updated:${event.directory}:${part.messageID}:${part.id}`
25+
}
26+
return undefined
27+
}
28+
29+
export function enqueueServerEvent(
30+
queue: QueuedServerEvent[],
31+
event: QueuedServerEvent,
32+
) {
33+
const key = coalescedKey(event)
34+
const previous = queue[queue.length - 1]
35+
if (key && previous && coalescedKey(previous) === key) {
36+
queue[queue.length - 1] = event
37+
return false
38+
}
39+
queue.push(event)
40+
return true
41+
}
2142

22-
export function coalesceServerEvents(events: QueuedServerEvent[], stale?: Set<string>) {
43+
export function coalesceServerEvents(events: QueuedServerEvent[]) {
2344
const output: QueuedServerEvent[] = []
24-
const deltas = new Map<string, number>()
2545
events.forEach((event) => {
26-
if (stale && event.payload.type === "message.part.delta") {
27-
const props = event.payload.properties
28-
if (stale.has(deltaKey(event.directory, props.messageID, props.partID))) return
29-
}
3046
if (event.payload.type !== "message.part.delta") {
31-
deltas.clear()
3247
output.push(event)
3348
return
3449
}
3550
const props = event.payload.properties
36-
const id = `${deltaKey(event.directory, props.messageID, props.partID)}:${props.field}`
37-
const index = deltas.get(id)
38-
const existing = index === undefined ? undefined : output[index]
39-
if (!existing || existing.payload.type !== "message.part.delta") {
40-
deltas.set(id, output.length)
51+
const previous = output[output.length - 1]
52+
if (
53+
!previous ||
54+
previous.payload.type !== "message.part.delta" ||
55+
previous.directory !== event.directory ||
56+
previous.payload.properties.messageID !== props.messageID ||
57+
previous.payload.properties.partID !== props.partID ||
58+
previous.payload.properties.field !== props.field
59+
) {
4160
output.push({
4261
directory: event.directory,
4362
payload: { ...event.payload, properties: { ...props } },
4463
})
4564
return
4665
}
47-
existing.payload.properties.delta += props.delta
66+
output[output.length - 1] = {
67+
directory: event.directory,
68+
payload: {
69+
...event.payload,
70+
properties: { ...props, delta: previous.payload.properties.delta + props.delta },
71+
},
72+
}
4873
})
4974
return output
5075
}
@@ -85,36 +110,22 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
85110

86111
let queue: Queued[] = []
87112
let buffer: Queued[] = []
88-
const coalesced = new Map<string, number>()
89-
const staleDeltas = new Set<string>()
90113
let timer: ReturnType<typeof setTimeout> | undefined
91114
let last = 0
92115

93-
const key = (directory: string, payload: Event) => {
94-
if (payload.type === "session.status") return `session.status:${directory}:${payload.properties.sessionID}`
95-
if (payload.type === "lsp.updated") return `lsp.updated:${directory}`
96-
if (payload.type === "message.part.updated") {
97-
const part = payload.properties.part
98-
return `message.part.updated:${directory}:${part.messageID}:${part.id}`
99-
}
100-
}
101-
102116
const flush = () => {
103117
if (timer) clearTimeout(timer)
104118
timer = undefined
105119

106120
if (queue.length === 0) return
107121

108122
const events = queue
109-
const skip = staleDeltas.size > 0 ? new Set(staleDeltas) : undefined
110123
queue = buffer
111124
buffer = events
112125
queue.length = 0
113-
coalesced.clear()
114-
staleDeltas.clear()
115126

116127
last = Date.now()
117-
const output = coalesceServerEvents(events, skip)
128+
const output = coalesceServerEvents(events)
118129
batch(() => {
119130
output.forEach((event) => emitter.emit(event.directory, event.payload))
120131
})
@@ -184,28 +195,11 @@ function createServerSdkContextBase(server: ServerConnection.Any, scope: ServerS
184195
for await (const event of events.stream) {
185196
resetHeartbeat()
186197
streamErrorLogged = false
187-
const directory = event.directory ?? "global"
188-
if (event.payload.type === "sync") {
189-
continue
190-
}
191-
192-
const payload = event.payload as Event
193-
194-
const k = key(directory, payload)
195-
if (k) {
196-
const i = coalesced.get(k)
197-
if (i !== undefined) {
198-
queue[i] = { directory, payload }
199-
if (payload.type === "message.part.updated") {
200-
const part = payload.properties.part
201-
staleDeltas.add(deltaKey(directory, part.messageID, part.id))
202-
}
203-
continue
204-
}
205-
coalesced.set(k, queue.length)
198+
if (event.payload.type !== "sync") {
199+
const directory = event.directory ?? "global"
200+
const payload = event.payload as Event
201+
if (enqueueServerEvent(queue, { directory, payload })) schedule()
206202
}
207-
queue.push({ directory, payload })
208-
schedule()
209203

210204
if (Date.now() - yielded < STREAM_YIELD_MS) continue
211205
yielded = Date.now()

0 commit comments

Comments
 (0)