Skip to content

Commit fdb3778

Browse files
AceRothstein71AceRothstein71
committed
fix: address audit findings across diff, events, and sqlite lanes
- diffFull returns undefined on failure or disabled snapshots; summarize never publishes an empty diff over a stored one - bounded single SSE queue with drop-oldest semantics; remove the unconsumed server.desync marker - gate reads on the shared connection when read/write lanes alias (:memory:, disableWAL) to prevent dirty reads - bound the diff memo to 8 entries and return copies on both paths - log snapshot corruption recovery only after a successful retry - halve the reader connection page cache; invalidate pending delta merges on message/session lifecycle events - regenerate the SDK for FileDiff.truncated
1 parent 4551204 commit fdb3778

10 files changed

Lines changed: 87 additions & 64 deletions

File tree

‎packages/core/src/database/sqlite.bun.ts‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -64,7 +64,7 @@ const make = (options: Config) =>
6464
if (readNative !== native) {
6565
yield* Effect.addFinalizer(() => Effect.sync(() => readNative.close()))
6666
readNative.run("PRAGMA busy_timeout = 5000;")
67-
readNative.run("PRAGMA cache_size = -64000;")
67+
readNative.run("PRAGMA cache_size = -32000;")
6868
readNative.run("PRAGMA query_only = ON;")
6969
}
7070

@@ -138,14 +138,21 @@ const make = (options: Config) =>
138138

139139
const connection = identity<SqliteConnection>({
140140
execute(query, params, transformRows) {
141-
const effect = READS.test(query) ? run(readNative, query, params) : gated(run(native, query, params))
141+
const effect =
142+
READS.test(query) && readNative !== native
143+
? run(readNative, query, params)
144+
: gated(run(native, query, params))
142145
return transformRows ? Effect.map(effect, transformRows) : effect
143146
},
144147
executeRaw(query, params) {
145-
return READS.test(query) ? run(readNative, query, params) : gated(run(native, query, params))
148+
return READS.test(query) && readNative !== native
149+
? run(readNative, query, params)
150+
: gated(run(native, query, params))
146151
},
147152
executeValues(query, params) {
148-
return READS.test(query) ? runValues(readNative, query, params) : gated(runValues(native, query, params))
153+
return READS.test(query) && readNative !== native
154+
? runValues(readNative, query, params)
155+
: gated(runValues(native, query, params))
149156
},
150157
executeUnprepared(query, params, transformRows) {
151158
return this.execute(query, params, transformRows)

‎packages/core/src/database/sqlite.node.ts‎

Lines changed: 11 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -68,7 +68,7 @@ const make = (options: Config) =>
6868
if (readNative !== native) {
6969
yield* Effect.addFinalizer(() => Effect.sync(() => readNative.close()))
7070
readNative.exec("PRAGMA busy_timeout = 5000;")
71-
readNative.exec("PRAGMA cache_size = -64000;")
71+
readNative.exec("PRAGMA cache_size = -32000;")
7272
readNative.exec("PRAGMA query_only = ON;")
7373
}
7474

@@ -136,14 +136,21 @@ const make = (options: Config) =>
136136

137137
const connection = identity<SqliteConnection>({
138138
execute(query, params, transformRows) {
139-
const result = READS.test(query) ? run(readNative, query, params) : gated(run(native, query, params))
139+
const result =
140+
READS.test(query) && readNative !== native
141+
? run(readNative, query, params)
142+
: gated(run(native, query, params))
140143
return transformRows ? Effect.map(result, transformRows) : result
141144
},
142145
executeRaw(query, params) {
143-
return READS.test(query) ? run(readNative, query, params) : gated(run(native, query, params))
146+
return READS.test(query) && readNative !== native
147+
? run(readNative, query, params)
148+
: gated(run(native, query, params))
144149
},
145150
executeValues(query, params) {
146-
return READS.test(query) ? runValues(readNative, query, params) : gated(runValues(native, query, params))
151+
return READS.test(query) && readNative !== native
152+
? runValues(readNative, query, params)
153+
: gated(runValues(native, query, params))
147154
},
148155
executeUnprepared(query, params, transformRows) {
149156
return this.execute(query, params, transformRows)

‎packages/opencode/src/cli/tui/event-batch.ts‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,8 @@ function deltaKey(event: GlobalEvent, properties: DeltaProperties): string {
5252
function partReplacementID(event: GlobalEvent): string | undefined {
5353
const payload = event.payload
5454
if (!payload || typeof payload !== "object") return
55+
if (payload.type === "message.updated" || payload.type === "message.removed" || payload.type === "session.deleted")
56+
return "*"
5557
if (payload.type !== "message.part.updated" && payload.type !== "message.part.removed") return
5658
const properties = payload.properties
5759
if (!properties || typeof properties !== "object") return
@@ -82,7 +84,8 @@ export function collapseEventBatch(events: GlobalEvent[]): GlobalEvent[] {
8284
return state
8385
}
8486
const partID = partReplacementID(event)
85-
if (partID) {
87+
if (partID === "*") state.keys.clear()
88+
else if (partID) {
8689
for (const [key, index] of state.keys) {
8790
const pending = state.collapsed[index]
8891
if (pending && deltaProperties(pending)?.partID === partID) state.keys.delete(key)

‎packages/opencode/src/server/routes/instance/httpapi/handlers/event.ts‎

Lines changed: 4 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -28,20 +28,15 @@ function eventResponse(events: EventV2.Interface) {
2828
const workspaceID = yield* InstanceState.workspaceID
2929
// Listener registration is eager, so events published after this point cannot
3030
// be lost while the HTTP body fiber is starting or emitting server.connected.
31-
// The buffer is bounded: when it overflows a server.desync marker is sent so
32-
// clients resynchronize instead of silently diverging.
33-
const queue = yield* Queue.dropping<{ id: string; type: string; properties: unknown }>(1024)
34-
let dropped = false
31+
// The buffer is bounded: a slow consumer drops the oldest events instead of
32+
// growing memory for every event in the process.
33+
const queue = yield* Queue.sliding<{ id: string; type: string; properties: unknown }>(8192)
3534
const unsubscribe = yield* events.listen((event) =>
3635
Effect.sync(() => {
3736
// Filter before enqueue so foreign directories and workspaces never occupy buffer space.
3837
if (event.location?.directory !== instance.directory) return
3938
if (event.location.workspaceID !== undefined && event.location.workspaceID !== workspaceID) return
40-
if (dropped) {
41-
if (!Queue.offerUnsafe(queue, { id: eventID(), type: "server.desync", properties: {} })) return
42-
dropped = false
43-
}
44-
if (!Queue.offerUnsafe(queue, { id: event.id, type: event.type, properties: event.data })) dropped = true
39+
Queue.offerUnsafe(queue, { id: event.id, type: event.type, properties: event.data })
4540
}),
4641
)
4742
yield* Effect.addFinalizer(() => unsubscribe)

‎packages/opencode/src/session/revert.ts‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,7 @@ const layer = Layer.effect(
7979
if (rev.snapshot) rev.diff = yield* snap.diff(rev.snapshot)
8080
const index = all.findIndex((msg) => msg.info.id === rev.messageID)
8181
const range = index < 0 ? [] : all.slice(index)
82-
const diffs = yield* summary.computeDiff({ messages: range })
82+
const diffs = (yield* summary.computeDiff({ messages: range })) ?? []
8383
yield* storage.write(["session_diff", input.sessionID], diffs).pipe(Effect.ignore)
8484
yield* events.publish(Session.Event.Diff, { sessionID: input.sessionID, diff: diffs })
8585
yield* sessions.setRevert({
@@ -100,7 +100,9 @@ const layer = Layer.effect(
100100
const session = yield* sessions.get(input.sessionID).pipe(Effect.orDie)
101101
if (!session.revert) return session
102102
if (session.revert.snapshot && !(yield* snap.restore(session.revert.snapshot))) {
103-
yield* Effect.logError("session unrevert aborted because snapshot restore failed", { sessionID: input.sessionID })
103+
yield* Effect.logError("session unrevert aborted because snapshot restore failed", {
104+
sessionID: input.sessionID,
105+
})
104106
return session
105107
}
106108
yield* sessions.clearRevert(input.sessionID)

‎packages/opencode/src/session/summary.ts‎

Lines changed: 6 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -78,7 +78,7 @@ export interface Interface {
7878
messages: SessionV1.WithParts[]
7979
sessionID?: SessionID
8080
messageID?: MessageID
81-
}) => Effect.Effect<Snapshot.FileDiff[]>
81+
}) => Effect.Effect<Snapshot.FileDiff[] | undefined>
8282
}
8383

8484
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionSummary") {}
@@ -176,16 +176,16 @@ const layer = Layer.effect(
176176
if (cached && cached.from === from && cached.to === to) return cached.diffs.map((item) => ({ ...item }))
177177
}
178178
const diffs = yield* snapshot.diffFull(from, to)
179-
// A failed or disabled snapshot read surfaces as an empty diff; caching it
180-
// would freeze that transient result as the message's real diff.
179+
// A failed or disabled snapshot read returns undefined instead of an empty diff.
180+
if (diffs === undefined) return undefined
181181
if (cacheKey && diffs.length) {
182-
if (diffCache.size >= 32) {
182+
if (diffCache.size >= 8) {
183183
const oldest = diffCache.keys().next()
184184
if (!oldest.done) diffCache.delete(oldest.value)
185185
}
186186
diffCache.set(cacheKey, { from, to, diffs })
187187
}
188-
return diffs
188+
return diffs.map((item) => ({ ...item }))
189189
})
190190

191191
const runSummarize = Effect.fn("SessionSummary.summarize")(function* (input: {
@@ -203,6 +203,7 @@ const layer = Layer.effect(
203203
if (!target || target.info.role !== "user") return
204204
yield* Effect.yieldNow
205205
const msgDiffs = yield* computeDiff({ sessionID: input.sessionID, messageID: input.messageID, messages })
206+
if (msgDiffs === undefined) return
206207
const dedicated = yield* database.db
207208
.select({ message_id: MessageDiffTable.message_id })
208209
.from(MessageDiffTable)

‎packages/opencode/src/snapshot/index.ts‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -49,7 +49,7 @@ export interface Interface {
4949
readonly restore: (snapshot: string) => Effect.Effect<boolean>
5050
readonly revert: (patches: Patch[]) => Effect.Effect<boolean>
5151
readonly diff: (hash: string) => Effect.Effect<string>
52-
readonly diffFull: (from: string, to: string) => Effect.Effect<FileDiff[]>
52+
readonly diffFull: (from: string, to: string) => Effect.Effect<FileDiff[] | undefined>
5353
}
5454

5555
export class Service extends Context.Service<Service, Interface>()("@opencode/Snapshot") {}
@@ -198,10 +198,12 @@ const layer: Layer.Layer<Service, never, FSUtil.Service | AppProcess.Service | C
198198
)
199199
if (!removed) {
200200
yield* Effect.logWarning("snapshot_index_corruption_detected", { gitdir: state.gitdir })
201-
return yield* settle(execute)
202201
}
203-
yield* Effect.logWarning("snapshot_index_corruption_recovered", { gitdir: state.gitdir })
204-
return yield* settle(execute)
202+
const result = yield* settle(execute)
203+
if (removed && result.code === 0) {
204+
yield* Effect.logWarning("snapshot_index_corruption_recovered", { gitdir: state.gitdir })
205+
}
206+
return result
205207
})
206208
}),
207209
)
@@ -862,12 +864,14 @@ const layer: Layer.Layer<Service, never, FSUtil.Service | AppProcess.Service | C
862864
)
863865
})
864866

867+
// Failure and disabled state return undefined so callers never publish an
868+
// empty diff over a stored one.
865869
const diffFull = Effect.fnUntraced(function* (from: string, to: string) {
866870
return yield* safeLocked(
867871
"diffFull",
868-
[],
872+
undefined,
869873
Effect.gen(function* () {
870-
if (!(yield* enabled())) return []
874+
if (!(yield* enabled())) return undefined
871875
type Row = {
872876
file: string
873877
status: "added" | "deleted" | "modified"

‎packages/opencode/test/server/session-message-diff-events.test.ts‎

Lines changed: 21 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -140,12 +140,13 @@ describe("session message diff events", () => {
140140
cost: 0,
141141
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
142142
})
143-
const expectedChangedDiffs = yield* summary.computeDiff({
144-
messages: (yield* Session.use.messages({ sessionID: session.id })).filter(
145-
(item) =>
146-
item.info.id === messageID || (item.info.role === "assistant" && item.info.parentID === messageID),
147-
),
148-
})
143+
const expectedChangedDiffs =
144+
(yield* summary.computeDiff({
145+
messages: (yield* Session.use.messages({ sessionID: session.id })).filter(
146+
(item) =>
147+
item.info.id === messageID || (item.info.role === "assistant" && item.info.parentID === messageID),
148+
),
149+
})) ?? []
149150
yield* summary.summarize({ sessionID: session.id, messageID })
150151
const changedDiffs = (yield* Session.use.messages({ sessionID: session.id })).find(
151152
(item) => item.info.id === messageID,
@@ -332,12 +333,13 @@ describe("session message diff events", () => {
332333
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
333334
})
334335
const summary = yield* SessionSummary.Service
335-
const expected = yield* summary.computeDiff({
336-
messages: (yield* Session.use.messages({ sessionID: session.id })).filter(
337-
(item) =>
338-
item.info.id === messageID || (item.info.role === "assistant" && item.info.parentID === messageID),
339-
),
340-
})
336+
const expected =
337+
(yield* summary.computeDiff({
338+
messages: (yield* Session.use.messages({ sessionID: session.id })).filter(
339+
(item) =>
340+
item.info.id === messageID || (item.info.role === "assistant" && item.info.parentID === messageID),
341+
),
342+
})) ?? []
341343
expect(expected.length).toBeGreaterThan(0)
342344
expect(JSON.stringify(expected)).toContain("imported patch")
343345
yield* summary.summarize({ sessionID: session.id, messageID })
@@ -471,12 +473,13 @@ describe("session message diff events", () => {
471473
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
472474
})
473475
const summary = yield* SessionSummary.Service
474-
const expected = yield* summary.computeDiff({
475-
messages: (yield* Session.use.messages({ sessionID: session.id })).filter(
476-
(item) =>
477-
item.info.id === messageID || (item.info.role === "assistant" && item.info.parentID === messageID),
478-
),
479-
})
476+
const expected =
477+
(yield* summary.computeDiff({
478+
messages: (yield* Session.use.messages({ sessionID: session.id })).filter(
479+
(item) =>
480+
item.info.id === messageID || (item.info.role === "assistant" && item.info.parentID === messageID),
481+
),
482+
})) ?? []
480483
expect(expected.length).toBeGreaterThan(0)
481484
yield* summary.summarize({ sessionID: session.id, messageID })
482485
const live = (yield* Session.use.messages({ sessionID: session.id })).find(

0 commit comments

Comments
 (0)