Skip to content

Commit c1fa943

Browse files
AceRothstein71AceRothstein71
committed
fix(session,server,tui,core): wave 1-5 reliability hardening — races, SSE correctness, projector hot paths
Adversarial-audit hardening across processor/session, server SSE, TUI, DB projector, share-next, provider, snapshot, tools, hydration and lifecycle surfaces. P0: snapshot-lock TOCTOU race corrupting tracked baselines, processor flush/coalescer races (mid-stream part loss), retry re-execution of settled tasks. Highlights: bounded event fan-out + Option decode on durable reads; worker admission cap; SSE id/gap sequence + frame cap; provider stream teardown + snowflake split; TUI reactive memoization + data test surface; projector hot paths (previous-row read gated on step-finish, SQL-filtered shell reads) + migration 20260917212851 (drops part_session_idx/session_message_time_created_idx; adds event_seq_idx/event_message_id_idx); bounded hydration tail; retry finalization settles tool calls; step-start snapshot reuse; instance-cache TTL/reaping; share-next lifecycle idempotency verified. Tests: new sse-frame, lru-cache, rpc, httpapi-global-event suites; 9-file regression battery 161 tests / 0 fail / 1 skip; typecheck green (core, opencode, tui); prettier clean. Design-blocked (documented in-repo): F-025 structuredClone in session.ts is load-bearing; F-055 GlobalBus dual-emit upstream handoff.
1 parent 857ce25 commit c1fa943

88 files changed

Lines changed: 3443 additions & 1107 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎packages/core/schema.json‎

Lines changed: 137 additions & 180 deletions
Large diffs are not rendered by default.

‎packages/core/src/background-job.ts‎

Lines changed: 78 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -20,14 +20,16 @@ export type Info = {
2020

2121
type Active = {
2222
info: Info
23-
done: Deferred.Deferred<Info>
23+
// Settled entries drop `promoted`, which for a promoted job pins a full
24+
// result snapshot that only a running `waitForPromotion` can still observe.
25+
done?: Deferred.Deferred<Info>
2426
scope: Scope.Closeable
2527
token: object
2628
pending: number
2729
next: number
2830
output?: { sequence: number; text: string }
2931
tail: Deferred.Deferred<void>
30-
promoted: Deferred.Deferred<Info>
32+
promoted?: Deferred.Deferred<Info>
3133
onPromote?: Effect.Effect<void>
3234
}
3335

@@ -110,12 +112,36 @@ function errorText(error: unknown) {
110112
return String(error)
111113
}
112114

115+
/** Terminal entries kept for late `get`/`list` observation; running jobs are never pruned. */
116+
export const RETAINED_TERMINAL_JOBS = 32
117+
118+
/** Retained `Info.output` cap; the full body still reaches `done` exactly once. */
119+
const MAX_RETAINED_OUTPUT = 30_000
120+
121+
function previewOutput(text: string) {
122+
if (text.length <= MAX_RETAINED_OUTPUT) return text
123+
return `...\n\n${text.slice(-MAX_RETAINED_OUTPUT)}`
124+
}
125+
126+
/** Drops the oldest terminal entries once the retained ring is full. Call under the registry lock. */
127+
function pruneTerminalJobs(jobs: Map<string, Active>) {
128+
const terminal = Array.from(jobs.values()).filter((job) => job.info.status !== "running")
129+
if (terminal.length <= RETAINED_TERMINAL_JOBS) return
130+
terminal
131+
.sort((a, b) => (a.info.completed_at ?? a.info.started_at) - (b.info.completed_at ?? b.info.started_at))
132+
.slice(0, terminal.length - RETAINED_TERMINAL_JOBS)
133+
.forEach((job) => jobs.delete(job.info.id))
134+
}
135+
113136
/**
114137
* Makes one scoped, process-local registry. Entries are intentionally not
115138
* durable: process restart or owner-scope closure loses status and interrupts
116139
* live work. Persisted observation, restart recovery, and remote workers need a
117140
* separate durable ownership slice rather than pretending this registry has
118-
* those semantics.
141+
* those semantics. Retention is bounded: running entries live until they settle,
142+
* then settled entries are pruned to the most recent `RETAINED_TERMINAL_JOBS`,
143+
* and each retained entry keeps only a capped `Info.output` preview while the
144+
* lossless body is delivered once through the `done` deferred.
119145
*/
120146
export const make = Effect.gen(function* () {
121147
const state: State = {
@@ -141,27 +167,31 @@ export const make = Effect.gen(function* () {
141167
? { sequence, text: exit.value }
142168
: job.output
143169
if (Exit.isSuccess(exit) && pending > 0) {
144-
return [{}, new Map(jobs).set(id, { ...job, pending, output })]
170+
jobs.set(id, { ...job, pending, output })
171+
return [{}, jobs]
145172
}
146173
const status: Exclude<Status, "running"> = Exit.isSuccess(exit)
147174
? "completed"
148175
: Cause.hasInterruptsOnly(exit.cause)
149176
? "cancelled"
150177
: "error"
151-
const next = {
178+
const info: Info = {
179+
...job.info,
180+
status,
181+
completed_at,
182+
...(output ? { output: output.text } : {}),
183+
...(Exit.isFailure(exit) ? { error: errorText(Cause.squash(exit.cause)) } : {}),
184+
}
185+
jobs.set(id, {
152186
...job,
187+
promoted: undefined,
153188
onPromote: undefined,
154189
pending: 0,
155-
output,
156-
info: {
157-
...job.info,
158-
status,
159-
completed_at,
160-
...(output ? { output: output.text } : {}),
161-
...(Exit.isFailure(exit) ? { error: errorText(Cause.squash(exit.cause)) } : {}),
162-
},
163-
}
164-
return [{ info: snapshot(next), done: job.done, scope: job.scope }, new Map(jobs).set(id, next)]
190+
output: undefined,
191+
info: { ...info, ...(output ? { output: previewOutput(output.text) } : {}) },
192+
})
193+
pruneTerminalJobs(jobs)
194+
return [{ info: snapshot({ ...job, info }), done: job.done, scope: job.scope }, jobs]
165195
})
166196
if (result.info && result.done) yield* Deferred.succeed(result.done, result.info).pipe(Effect.ignore)
167197
if (result.scope) {
@@ -234,10 +264,8 @@ export const make = Effect.gen(function* () {
234264
promoted,
235265
onPromote: input.onPromote,
236266
}
237-
return [{ info: snapshot(job), scope, token }, new Map(jobs).set(id, job)] as readonly [
238-
StartResult,
239-
Map<string, Active>,
240-
]
267+
jobs.set(id, job)
268+
return [{ info: snapshot(job), scope, token }, jobs] as readonly [StartResult, Map<string, Active>]
241269
}),
242270
)
243271
if ("scope" in result)
@@ -262,14 +290,15 @@ export const make = Effect.gen(function* () {
262290
(jobs): readonly [ExtendResult, Map<string, Active>] => {
263291
const job = jobs.get(input.id)
264292
if (!job || job.info.status !== "running") return [{ extended: false }, jobs]
293+
jobs.set(input.id, {
294+
...job,
295+
pending: job.pending + 1,
296+
next: job.next + 1,
297+
tail,
298+
})
265299
return [
266300
{ extended: true, previous: job.tail, scope: job.scope, tail, token: job.token, sequence: job.next },
267-
new Map(jobs).set(input.id, {
268-
...job,
269-
pending: job.pending + 1,
270-
next: job.next + 1,
271-
tail,
272-
}),
301+
jobs,
273302
]
274303
},
275304
)
@@ -292,7 +321,12 @@ export const make = Effect.gen(function* () {
292321
const wait: Interface["wait"] = Effect.fn("BackgroundJob.wait")(function* (input) {
293322
const job = (yield* SynchronizedRef.get(state.jobs)).get(input.id)
294323
if (!job) return { timedOut: false }
295-
if (job.info.status !== "running") return { info: snapshot(job), timedOut: false }
324+
if (job.info.status !== "running") {
325+
// `done` carries the lossless result; the retained snapshot is capped.
326+
if (job.done) return { info: yield* Deferred.await(job.done), timedOut: false }
327+
return { info: snapshot(job), timedOut: false }
328+
}
329+
if (!job.done) return { info: snapshot(job), timedOut: false }
296330
if (input.timeout === undefined) return { info: yield* Deferred.await(job.done), timedOut: false }
297331
if (input.timeout <= 0) return { info: snapshot(job), timedOut: true }
298332
const info = yield* Deferred.await(job.done).pipe(Effect.timeoutOption(input.timeout))
@@ -302,7 +336,7 @@ export const make = Effect.gen(function* () {
302336

303337
const waitForPromotion: Interface["waitForPromotion"] = Effect.fn("BackgroundJob.waitForPromotion")(function* (id) {
304338
const job = (yield* SynchronizedRef.get(state.jobs)).get(id)
305-
if (!job || job.info.status !== "running") return yield* Effect.never
339+
if (!job || job.info.status !== "running" || !job.promoted) return yield* Effect.never
306340
if (job.info.metadata?.background === true) return snapshot(job)
307341
return yield* Deferred.await(job.promoted)
308342
})
@@ -323,10 +357,11 @@ export const make = Effect.gen(function* () {
323357
metadata: { ...job.info.metadata, background: true },
324358
},
325359
}
326-
return [
327-
{ info: snapshot(next), onPromote: job.onPromote, promoted: job.promoted },
328-
new Map(jobs).set(id, next),
329-
] as readonly [PromoteResult, Map<string, Active>]
360+
jobs.set(id, next)
361+
return [{ info: snapshot(next), onPromote: job.onPromote, promoted: job.promoted }, jobs] as readonly [
362+
PromoteResult,
363+
Map<string, Active>,
364+
]
330365
}),
331366
)
332367
if (result.info && result.promoted) yield* Deferred.succeed(result.promoted, result.info).pipe(Effect.ignore)
@@ -340,17 +375,21 @@ export const make = Effect.gen(function* () {
340375
const job = jobs.get(id)
341376
if (!job) return [{}, jobs]
342377
if (job.info.status !== "running") return [{ info: snapshot(job) }, jobs]
343-
const next = {
378+
const info: Info = {
379+
...job.info,
380+
status: "cancelled" as const,
381+
completed_at,
382+
}
383+
jobs.set(id, {
344384
...job,
385+
promoted: undefined,
345386
onPromote: undefined,
346387
pending: 0,
347-
info: {
348-
...job.info,
349-
status: "cancelled" as const,
350-
completed_at,
351-
},
352-
}
353-
return [{ info: snapshot(next), done: job.done, scope: job.scope }, new Map(jobs).set(id, next)]
388+
output: undefined,
389+
info,
390+
})
391+
pruneTerminalJobs(jobs)
392+
return [{ info: snapshot({ ...job, info }), done: job.done, scope: job.scope }, jobs]
354393
})
355394
if (result.info && result.done) yield* Deferred.succeed(result.done, result.info).pipe(Effect.ignore)
356395
if (result.scope) yield* Scope.close(result.scope, Exit.void)

‎packages/core/src/database/migration.gen.ts‎

Lines changed: 1 addition & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎packages/core/src/database/migration/20260917212851_db_projector_hot_paths.ts‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
import { Effect } from "effect"
2+
import type { DatabaseMigration } from "../migration"
3+
4+
export default {
5+
id: "20260917212851_db_projector_hot_paths",
6+
up(tx) {
7+
return Effect.gen(function* () {
8+
yield* tx.run(`DROP INDEX IF EXISTS \`part_session_idx\`;`)
9+
yield* tx.run(`DROP INDEX IF EXISTS \`session_message_time_created_idx\`;`)
10+
yield* tx.run(`CREATE INDEX \`event_seq_idx\` ON \`event\` (\`seq\`);`)
11+
yield* tx.run(`CREATE INDEX \`event_message_id_idx\` ON \`event\` (json_extract("data", '$.messageID'));`)
12+
yield* tx.run(
13+
`CREATE INDEX \`session_project_time_updated_idx\` ON \`session\` (\`project_id\`,\`time_updated\`,\`id\`);`,
14+
)
15+
yield* tx.run(`CREATE INDEX \`session_time_updated_idx\` ON \`session\` (\`time_updated\`,\`id\`);`)
16+
yield* tx.run(
17+
`CREATE INDEX \`session_directory_time_updated_idx\` ON \`session\` (\`directory\`,\`time_updated\`,\`id\`);`,
18+
)
19+
})
20+
},
21+
} satisfies DatabaseMigration.Migration

‎packages/core/src/database/schema.gen.ts‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -247,6 +247,8 @@ export default {
247247
`)
248248
yield* tx.run(`CREATE UNIQUE INDEX \`event_aggregate_seq_idx\` ON \`event\` (\`aggregate_id\`,\`seq\`);`)
249249
yield* tx.run(`CREATE INDEX \`event_aggregate_type_seq_idx\` ON \`event\` (\`aggregate_id\`,\`type\`,\`seq\`);`)
250+
yield* tx.run(`CREATE INDEX \`event_seq_idx\` ON \`event\` (\`seq\`);`)
251+
yield* tx.run(`CREATE INDEX \`event_message_id_idx\` ON \`event\` (json_extract("data", '$.messageID'));`)
250252
yield* tx.run(
251253
`CREATE UNIQUE INDEX \`permission_project_action_resource_idx\` ON \`permission\` (\`project_id\`,\`action\`,\`resource\`);`,
252254
)
@@ -255,7 +257,6 @@ export default {
255257
`CREATE INDEX \`message_session_time_created_id_idx\` ON \`message\` (\`session_id\`,\`time_created\`,\`id\`);`,
256258
)
257259
yield* tx.run(`CREATE INDEX \`part_message_id_id_idx\` ON \`part\` (\`message_id\`,\`id\`);`)
258-
yield* tx.run(`CREATE INDEX \`part_session_idx\` ON \`part\` (\`session_id\`);`)
259260
yield* tx.run(
260261
`CREATE INDEX \`session_input_session_pending_delivery_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`,\`delivery\`,\`admitted_seq\`);`,
261262
)
@@ -274,10 +275,16 @@ export default {
274275
yield* tx.run(
275276
`CREATE INDEX \`session_message_session_time_created_id_idx\` ON \`session_message\` (\`session_id\`,\`time_created\`,\`id\`);`,
276277
)
277-
yield* tx.run(`CREATE INDEX \`session_message_time_created_idx\` ON \`session_message\` (\`time_created\`);`)
278278
yield* tx.run(`CREATE INDEX \`session_project_idx\` ON \`session\` (\`project_id\`);`)
279279
yield* tx.run(`CREATE INDEX \`session_workspace_idx\` ON \`session\` (\`workspace_id\`);`)
280280
yield* tx.run(`CREATE INDEX \`session_parent_idx\` ON \`session\` (\`parent_id\`);`)
281+
yield* tx.run(
282+
`CREATE INDEX \`session_project_time_updated_idx\` ON \`session\` (\`project_id\`,\`time_updated\`,\`id\`);`,
283+
)
284+
yield* tx.run(`CREATE INDEX \`session_time_updated_idx\` ON \`session\` (\`time_updated\`,\`id\`);`)
285+
yield* tx.run(
286+
`CREATE INDEX \`session_directory_time_updated_idx\` ON \`session\` (\`directory\`,\`time_updated\`,\`id\`);`,
287+
)
281288
yield* tx.run(`CREATE INDEX \`todo_session_idx\` ON \`todo\` (\`session_id\`);`)
282289
})
283290
},

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

Lines changed: 25 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import { DatabaseSync, type SQLInputValue } from "node:sqlite"
1+
import { DatabaseSync, type SQLInputValue, type StatementSync } from "node:sqlite"
22
import { drizzle } from "drizzle-orm/node-sqlite"
33
import * as Context from "effect/Context"
44
import * as Effect from "effect/Effect"
@@ -23,6 +23,27 @@ const READS = /^\s*(?:select|explain)\b/i
2323
const TypeId = "~@opencode-ai/core/database/SqliteNode" as const
2424
type TypeId = typeof TypeId
2525

26+
// node:sqlite has no statement cache, so re-preparing on every call costs a
27+
// parse+plan per query. Cache per connection, evicting least-recently-used.
28+
const STATEMENT_CACHE_LIMIT = 256
29+
const statementCacheByTarget = new WeakMap<object, Map<string, StatementSync>>()
30+
31+
function cachedStatement(target: DatabaseSync, mode: "all" | "values", query: string) {
32+
const cache = statementCacheByTarget.get(target) ?? new Map<string, StatementSync>()
33+
if (!statementCacheByTarget.has(target)) statementCacheByTarget.set(target, cache)
34+
const key = `${mode}:${query}`
35+
const cached = cache.get(key)
36+
if (cached) {
37+
cache.delete(key)
38+
cache.set(key, cached)
39+
return cached
40+
}
41+
const statement = target.prepare(query)
42+
if (cache.size >= STATEMENT_CACHE_LIMIT) cache.delete(cache.keys().next().value!)
43+
cache.set(key, statement)
44+
return statement
45+
}
46+
2647
interface SqliteClient extends Client.SqlClient {
2748
readonly [TypeId]: TypeId
2849
readonly config: Config
@@ -74,7 +95,7 @@ const make = (options: Config) =>
7495

7596
const run = (target: DatabaseSync, query: string, params: ReadonlyArray<unknown> = []) =>
7697
Effect.withFiber<Array<Record<string, unknown>>, SqlError>((fiber) => {
77-
const statement = target.prepare(query)
98+
const statement = cachedStatement(target, "all", query)
7899
statement.setReadBigInts(Context.get(fiber.context, Client.SafeIntegers))
79100
try {
80101
return Effect.succeed(statement.all(...(params as SQLInputValue[])) as Array<Record<string, unknown>>)
@@ -89,7 +110,7 @@ const make = (options: Config) =>
89110

90111
const runValues = (target: DatabaseSync, query: string, params: ReadonlyArray<unknown> = []) =>
91112
Effect.withFiber<ReadonlyArray<ReadonlyArray<unknown>>, SqlError>((fiber) => {
92-
const statement = target.prepare(query)
113+
const statement = cachedStatement(target, "values", query)
93114
statement.setReadBigInts(Context.get(fiber.context, Client.SafeIntegers))
94115
statement.setReturnArrays(true)
95116
try {
@@ -204,6 +225,7 @@ const nativeLayer = (config: Config) =>
204225
open: true,
205226
})
206227
yield* Effect.addFinalizer(() => Effect.sync(() => native.close()))
228+
if (config.readonly !== true) native.exec("PRAGMA busy_timeout = 5000;")
207229
if (config.disableWAL !== true && config.readonly !== true) native.exec("PRAGMA journal_mode = WAL;")
208230
return native
209231
}),

0 commit comments

Comments
 (0)