Compare commits

..

2 Commits

Author SHA1 Message Date
Kit Langton db9a3b6c41 refactor(core): compact question tool schema (#41772) 2026-08-11 12:09:37 -04:00
opencode-agent[bot] 961b51b509 fix(core): yield while clearing migration events (#41775)
Co-authored-by: Dax Raad <d@ironbay.co>
2026-08-11 16:02:30 +00:00
6 changed files with 68 additions and 74 deletions
+11 -2
View File
@@ -7,7 +7,7 @@ import { SessionV1 } from "@opencode-ai/schema/session-v1"
import { SessionMessage } from "../session/message"
import { SessionSchema } from "../session/schema"
import { KVTable } from "../kv/sql"
import { EventSequenceTable, EventTable } from "../event/sql"
import { EventSequenceTable } from "../event/sql"
import { eq, sql } from "drizzle-orm"
import { Global } from "@opencode-ai/util/global"
import { existsSync } from "node:fs"
@@ -161,6 +161,7 @@ type NextMessage = {
const lock = Semaphore.makeUnsafe(1)
const MIGRATION_STATE_KEY = "migration.v1-v2"
const EVENT_DELETE_BATCH_SIZE = 1_000
const decodeJson = Schema.decodeUnknownOption(Schema.UnknownFromJsonString)
const decodeMessage = Schema.decodeUnknownOption(SessionV1.Info)
const decodePart = Schema.decodeUnknownOption(SessionV1.Part)
@@ -485,7 +486,15 @@ export function run(options: Options = {}): Effect.Effect<RunResult, never, Data
yield* db
.transaction((tx) =>
Effect.gen(function* () {
yield* tx.delete(EventTable).run()
while (true) {
yield* tx.run(sql`
DELETE FROM event
WHERE rowid IN (SELECT rowid FROM event LIMIT ${EVENT_DELETE_BATCH_SIZE})
`)
const deleted = (yield* tx.get<{ value: number }>(sql`SELECT changes() AS value`))?.value ?? 0
if (deleted < EVENT_DELETE_BATCH_SIZE) break
yield* Effect.yieldNow
}
yield* tx
.insert(KVTable)
.values({ key: MIGRATION_STATE_KEY, value: { phase: "sessions" } })
+3 -1
View File
@@ -21,7 +21,9 @@ Usage notes:
- If you recommend a specific option, make that the first option in the list and add "(Recommended)" at the end of the label`
export const Input = Schema.Struct({
questions: Schema.NonEmptyArray(Question.Prompt).annotate({ description: "Questions to ask" }),
questions: Schema.Array(Question.Prompt)
.check(Schema.isNonEmpty())
.annotate({ description: "Questions to ask" }),
})
export const Output = Schema.Struct({
+24
View File
@@ -89,6 +89,30 @@ const it = testEffect(
)
describe("QuestionTool", () => {
it.effect("emits one item schema for the nonempty questions array", () =>
Effect.gen(function* () {
captured = undefined
const registry = yield* Tool.Service
const definition = (yield* toolDefinitions(registry)).find((tool) => tool.name === QuestionTool.name)
expect(definition?.inputSchema).toHaveProperty("properties.questions.type", "array")
expect(definition?.inputSchema).toHaveProperty("properties.questions.minItems", 1)
expect(definition?.inputSchema).toHaveProperty("properties.questions.items")
expect(definition?.inputSchema).not.toHaveProperty("properties.questions.prefixItems")
expect(
yield* executeTool(registry, {
sessionID,
...toolIdentity,
call: { type: "tool-call", id: "call-question-empty", name: QuestionTool.name, input: { questions: [] } },
}),
).toMatchObject({
status: "error",
error: { type: "tool.execution", message: expect.stringContaining("Invalid tool input") },
})
expect(capturedInput()).toBeUndefined()
}),
)
it.effect("omits a catalog-denied question and enforces its leaf permission", () =>
Effect.gen(function* () {
captured = undefined
+30 -1
View File
@@ -11,7 +11,7 @@ import { Project } from "@opencode-ai/core/project"
import { ProjectTable } from "@opencode-ai/core/project/sql"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { Global } from "@opencode-ai/util/global"
import { Effect, Layer, Logger, Schedule, Schema, Scope } from "effect"
import { Effect, Fiber, Layer, Logger, Schedule, Schema, Scope } from "effect"
import { eq, sql } from "drizzle-orm"
import type { SqlClient } from "effect/unstable/sql/SqlClient"
import { tmpdir } from "./fixture/tmpdir"
@@ -798,6 +798,35 @@ describe("V1Migration database workflow", () => {
)
})
test("yields while clearing stale events in batches", async () => {
await database(
Effect.gen(function* () {
const { db } = yield* Database.Service
yield* db.run(sql`INSERT INTO event_sequence (aggregate_id, seq) VALUES ('stale', 2500)`)
yield* db.run(sql`
WITH RECURSIVE rows(value) AS (
VALUES(1)
UNION ALL
SELECT value + 1 FROM rows WHERE value < 2500
)
INSERT INTO event (id, aggregate_id, seq, created, type, data)
SELECT printf('event_%04d', value), 'stale', value, 1, 'session.renamed.1', '{}'
FROM rows
`)
let yielded = false
const heartbeat = yield* Effect.yieldNow.pipe(
Effect.andThen(Effect.sync(() => (yielded = true))),
Effect.forkChild({ startImmediately: true }),
)
expect(yield* V1Migration.run()).toEqual({ status: "completed" })
expect(yielded).toBe(true)
yield* Fiber.join(heartbeat)
expect(yield* db.get<{ value: number }>(sql`SELECT COUNT(*) AS value FROM event`)).toEqual({ value: 0 })
}),
)
})
test("imports previous V2 sessions and messages as part of the migration", async () => {
await using tmp = await tmpdir()
const filename = path.join(tmp.path, "opencode-next.db")
-13
View File
@@ -12,19 +12,6 @@ export const ModelHandler = HttpApiBuilder.group(Api, "server.model", (handlers)
.handle(
"model.list",
Effect.fn(function* () {
const plugins = yield* PluginSupervisor.Service
yield* plugins.flush.pipe(
Effect.timeoutOrElse({
duration: "5 seconds",
orElse: () =>
Effect.fail(
new ServiceUnavailableError({
message: "Model catalog initialization timed out",
service: "model.catalog",
}),
),
}),
)
const catalog = yield* Catalog.Service
return yield* response(catalog.model.available())
}),
-57
View File
@@ -1,57 +0,0 @@
import fs from "node:fs/promises"
import path from "node:path"
import { expect } from "bun:test"
import { Effect } from "effect"
import { HttpServer } from "effect/unstable/http"
import { tmpdir } from "../../core/test/fixture/tmpdir"
import { it } from "../../core/test/lib/effect"
import { ServerProcess } from "../src/process"
it.live("waits for plugin initialization before listing models", () =>
Effect.acquireUseRelease(
Effect.promise(() => tmpdir("opencode-model-endpoint-")),
(tmp) =>
Effect.gen(function* () {
yield* Effect.promise(() =>
fs.writeFile(
path.join(tmp.path, "opencode.json"),
JSON.stringify({
providers: {
custom: {
package: "aisdk:@ai-sdk/openai-compatible",
settings: { apiKey: "secret" },
models: { chat: {} },
},
},
}),
),
)
const server = yield* ServerProcess.start<never, never>({
hostname: "127.0.0.1",
port: 0,
password: "secret",
app: { version: "test-version" },
database: { path: ":memory:" },
config: { directory: tmp.path },
fs: { filewatcher: false },
})
const url = new URL("/api/model", HttpServer.formatAddress(server.address))
url.searchParams.set("location[directory]", tmp.path)
const response = yield* Effect.promise(() =>
fetch(url, { headers: { authorization: `Basic ${btoa("opencode:secret")}` } }),
)
expect(response.status).toBe(200)
const body: unknown = yield* Effect.promise(() => response.json())
if (!isRecord(body) || !Array.isArray(body["data"])) throw new Error("Expected a model list response")
expect(
body["data"].some((model) => isRecord(model) && model["providerID"] === "custom" && model["id"] === "chat"),
).toBeTrue()
}),
(tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
),
)
function isRecord(value: unknown): value is Record<string, unknown> {
return typeof value === "object" && value !== null && !Array.isArray(value)
}