mirrors/opencode
1
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-11-09 22:25:02 +03:00
Code Issues Packages Projects Releases Wiki Activity

fix(core): count Console config as applied only once its reloads finish

Browse Source
This commit is contained in:
Adam
2026-11-04 08:13:35 -06:00
Unverified
parent 5d1ee4908a
commit bd5957ea9e
2 changed files with 158 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/core/src/plugin/provider/opencode.ts
+42 -26
View File
@@ -203,6 +203,9 @@ export const OpencodePlugin = define<
// a new one. Unset while no Console answer is in place: signed out, an external credential, or a failed load.
workspace: string | undefined
} = { remote: undefined, connection: undefined, organization: undefined, workspace: undefined }
// The snapshot's workspace once `apply` has finished rebuilding everything it feeds, which `snapshot` alone runs
// ahead of while the reloads that read it are still running.
let appliedWorkspace: string | undefined
let instructions: InstructionDiscovery.File[] = []
// Tagged with the account and workspace they were downloaded for, so a switch hides them right away.
let librarySkills: { owner: string; items: LibrarySkill[] } = { owner: "", items: [] }
@@ -452,6 +455,7 @@ export const OpencodePlugin = define<
? initial.next
: fallback(initial?.connection ?? (yield* ctx.integration.connection.active("opencode")))
yield* publish(snapshot)
appliedWorkspace = snapshot.workspace
yield* ctx.provider.transform((providers) => {
if (integrated) ConfigProviderPlugin.integrate(providers, integrated)
})
@@ -497,6 +501,8 @@ export const OpencodePlugin = define<
})
const apply = Effect.fn("OpencodePlugin.apply")(function* (next: typeof snapshot) {
if (appliedWorkspace !== next.workspace) appliedWorkspace = undefined
// Set first: the transforms the reloads below run read it.
snapshot = next
yield* publish(next)
yield* guard(next)
@@ -504,8 +510,24 @@ export const OpencodePlugin = define<
[ctx.provider.reload(), ctx.websearch.reload(), ctx.mcp.reload(), discovery.reload(), skills.reload()],
{ concurrency: 5, discard: true },
)
appliedWorkspace = next.workspace
})
type Outcome = Effect.Success<ReturnType<typeof load>>
// Annotated: `apply` re-registers the hooks that call `check`, which would otherwise make inference circular.
const settle = (outcome: Outcome): Effect.Effect<void> =>
loading.withPermit(
Effect.gen(function* () {
// A switch or a key replaced while this load ran makes it stale; the change's own check applies it.
const loaded = outcome._tag === "Loaded" ? outcome.next.connection : outcome.connection
const active = yield* ctx.integration.connection.active("opencode")
if (IntegrationConnection.key(loaded) !== IntegrationConnection.key(active)) return
const workspace = outcome._tag === "Loaded" ? outcome.next.workspace : undefined
if (workspace && (!active || workspace !== (yield* stored(active)))) return
const next = outcome._tag === "Loaded" ? outcome.next : fallback(outcome.connection)
// An apply that didn't finish runs again.
if (!Equal.equals(snapshot, next) || appliedWorkspace !== next.workspace) yield* apply(next)
}),
)
// Loads run outside `loading`, so a slow one never holds up another check. `wait` bounds the load: one still
// running carries on in `SharedCache`, retries included, for the next check, while the current snapshot stays.
const check = (maxAge?: Duration.Input, wait?: Duration.Input): Effect.Effect<void> =>
@@ -514,19 +536,7 @@ export const OpencodePlugin = define<
wait === undefined
? yield* load(maxAge)
: Option.getOrUndefined(yield* load(maxAge).pipe(Effect.timeoutOption(wait)))
if (!outcome) return
yield* loading.withPermit(
Effect.gen(function* () {
// A switch or a key replaced while this load ran makes it stale; the change's own check applies it.
const loaded = outcome._tag === "Loaded" ? outcome.next.connection : outcome.connection
const active = yield* ctx.integration.connection.active("opencode")
if (IntegrationConnection.key(loaded) !== IntegrationConnection.key(active)) return
const workspace = outcome._tag === "Loaded" ? outcome.next.workspace : undefined
if (workspace && (!active || workspace !== (yield* stored(active)))) return
const next = outcome._tag === "Loaded" ? outcome.next : fallback(outcome.connection)
if (!Equal.equals(snapshot, next)) yield* apply(next)
}),
)
if (outcome) yield* settle(outcome)
})
// Console inference rejects a missing SSO sign-in with a generic 401/403. Re-check the Console config, which
@@ -565,28 +575,34 @@ export const OpencodePlugin = define<
// With the `console.required` server option, no session model request is built on another sign-in's config or
// on none: each waits until the active credential's config is loaded and applied, or fails. Until then every
// request loads again rather than reusing a failure, waiting up to 10 seconds. Once applied, the first request
// each minute refreshes stale config, waiting up to 3 seconds, and the config stays in effect if the Console
// doesn't answer; a sign-in the Console rejects is never applied. A load that outlasts its wait still applies.
const applied = Effect.gen(function* () {
// request loads again rather than reusing a failure, waiting up to 10 seconds for the Console. Once applied, the
// first request each minute refreshes stale config, waiting up to 3 seconds, and the config stays in effect if
// the Console doesn't answer; a sign-in the Console rejects is never applied. Only the Console is waited on with a
// bound: applying its config, such as connecting MCP servers, is waited out, and a load that outlasts its wait
// still applies.
const status = Effect.gen(function* () {
const active = yield* ctx.integration.connection.active("opencode")
if (!active || !snapshot.workspace || reportedFor(active)?.status === "needs_auth") return false
return (
IntegrationConnection.key(active) === IntegrationConnection.key(snapshot.connection) &&
snapshot.workspace === (yield* stored(active))
)
if (!active || reportedFor(active)?.status === "needs_auth") return "unapplied"
if (IntegrationConnection.key(active) !== IntegrationConnection.key(snapshot.connection)) return "unapplied"
if (!snapshot.workspace || snapshot.workspace !== (yield* stored(active))) return "unapplied"
return appliedWorkspace === snapshot.workspace ? "applied" : "applying"
})
let refreshed = Number.NEGATIVE_INFINITY
if (settings.required) {
yield* gate.register(
Effect.gen(function* () {
// The active sign-in's config is being applied: wait for that rather than loading it again.
if ((yield* status) === "applying") yield* loading.withPermit(Effect.void)
const now = yield* Clock.currentTimeMillis
const current = yield* applied
const current = (yield* status) === "applied"
if (current && now - refreshed < Duration.toMillis(Duration.minutes(1))) return
refreshed = now
const refresh = yield* check(current ? Duration.minutes(1) : Duration.zero).pipe(Effect.forkIn(scope))
yield* Fiber.join(refresh).pipe(Effect.timeoutOption(current ? Duration.seconds(3) : Duration.seconds(10)))
if (yield* applied) return
const loaded = yield* load(current ? Duration.minutes(1) : Duration.zero).pipe(Effect.forkIn(scope))
const outcome = Option.getOrUndefined(
yield* Fiber.join(loaded).pipe(Effect.timeoutOption(current ? Duration.seconds(3) : Duration.seconds(10))),
)
yield* outcome ? settle(outcome) : Fiber.join(loaded).pipe(Effect.flatMap(settle), Effect.forkIn(scope))
if ((yield* status) === "applied") return
const active = yield* ctx.integration.connection.active("opencode")
// A status the Console reported, such as an expired SSO sign-in, names the fix.
const reason = !active
packages/core/test/plugin/provider-opencode.test.ts
+116 -1
View File
@@ -4,7 +4,7 @@ import { LLMClient, RequestExecutor } from "@opencode/ai/route"
import { Document, Event, Info } from "@opencode/schema/config"
import { ConfigPolicy } from "@opencode/schema/config/policy"
import { Money } from "@opencode/schema/money"
import { Cause, Clock, Effect, Exit, Fiber, Layer, Schema, Stream } from "effect"
import { Cause, Clock, Deferred, Effect, Exit, Fiber, Layer, Schema, Stream } from "effect"
import { TestClock } from "effect/testing"
import { HttpClient, HttpClientRequest, HttpClientResponse } from "effect/http"
import { Agent } from "@opencode/core/agent"
@@ -124,6 +124,29 @@ const recordingMcp = (() => {
return { configured, layer: Layer.succeed(Mcp.Service, service) }
})()
// An MCP service whose reloads wait for the test once it blocks them, standing in for slow server handshakes.
const blockingMcp = (() => {
const state: { started?: Deferred.Deferred<void>; release?: Deferred.Deferred<void> } = {}
const reload = () =>
Effect.suspend(() =>
state.started && state.release
? Deferred.succeed(state.started, undefined).pipe(Effect.andThen(Deferred.await(state.release)))
: Effect.void,
)
const block = Effect.gen(function* () {
const started = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
Object.assign(state, { started, release })
return {
started: Deferred.await(started),
release: Effect.sync(() => Object.assign(state, { started: undefined, release: undefined })).pipe(
Effect.andThen(Deferred.succeed(release, undefined)),
),
}
})
return { block, layer: Layer.succeed(Mcp.Service, Mcp.Service.of({ ...emptyMcp, reload })) }
})()
function consoleServer(orgID: string | null | undefined, unavailable = false) {
const config: { authorization: string | null; orgID: string | null }[] = []
const requests: string[] = []
@@ -1171,7 +1194,14 @@ describe("OpencodePlugin", () => {
const next = yield* settled(yield* request())
expect(Exit.isSuccess(next.exit)).toBe(true)
expect(next.elapsed).toBe(0)
// The refresh that outlasted its wait applies once the Console answers.
api.state.answer.set("one", () =>
Response.json({ providers: {}, library: { instructions: [{ id: "one", content: "Use one, refreshed" }] } }),
)
api.release()
yield* drain
expect(yield* instructions).toEqual(["Use one, refreshed"])
}),
)
@@ -1254,6 +1284,91 @@ describe("OpencodePlugin", () => {
}),
)
const slow = testEffect(makePluginTestLayer(undefined, undefined, blockingMcp.layer))
slow.effect("holds a request that arrives while the config is being applied until it is", () =>
Effect.gen(function* () {
const api = consoleApi()
const reload = yield* blockingMcp.block
yield* connect(api)
yield* reload.started
const held = yield* request()
// 4 seconds, past the 3 a refresh of an applied config waits.
yield* drain
yield* drain
expect(held.pollUnsafe()).toBeUndefined()
yield* reload.release
const result = yield* Fiber.join(held)
expect(Exit.isSuccess(result.exit)).toBe(true)
expect(result.instructions).toEqual(["Use one"])
// It waited for the apply in progress instead of loading again.
expect(api.state.requests).toEqual(["one"])
}),
)
slow.effect("holds a request within a minute of the last while a switch's config is being applied", () =>
Effect.gen(function* () {
const credentials = yield* Credential.Service
const api = consoleApi()
yield* connect(api)
expect(Exit.isSuccess((yield* settled(yield* request())).exit)).toBe(true)
const reload = yield* blockingMcp.block
yield* credentials.create({ integrationID: Integration.ID.make("opencode"), value: key("two") })
yield* reload.started
const held = yield* request()
yield* drain
expect(held.pollUnsafe()).toBeUndefined()
yield* reload.release
const result = yield* Fiber.join(held)
expect(Exit.isSuccess(result.exit)).toBe(true)
expect(result.instructions).toEqual(["Use two"])
expect(api.state.requests).toEqual(["one", "two"])
}),
)
slow.effect("waits out a slow apply after the Console answers instead of failing", () =>
Effect.gen(function* () {
const credentials = yield* Credential.Service
const api = consoleApi()
const credential = yield* connect(api)
expect(Exit.isSuccess((yield* settled(yield* request())).exit)).toBe(true)
const reload = yield* blockingMcp.block
yield* credentials.update(credential.id, { value: key("rotated") })
const held = yield* request()
yield* reload.started
// 12 seconds, past the 10 the Console is waited for.
yield* Effect.forEach([1, 2, 3, 4, 5, 6], () => drain, { discard: true })
expect(held.pollUnsafe()).toBeUndefined()
yield* reload.release
const result = yield* Fiber.join(held)
expect(Exit.isSuccess(result.exit)).toBe(true)
expect(result.instructions).toEqual(["Use rotated"])
}),
)
slow.effect("applies a config again when a request is cancelled while applying it", () =>
Effect.gen(function* () {
const credentials = yield* Credential.Service
const api = consoleApi()
const credential = yield* connect(api)
expect(Exit.isSuccess((yield* settled(yield* request())).exit)).toBe(true)
const reload = yield* blockingMcp.block
yield* credentials.update(credential.id, { value: key("rotated") })
const cancelled = yield* request()
yield* reload.started
yield* Fiber.interrupt(cancelled)
yield* reload.release
const result = yield* settled(yield* request())
expect(Exit.isSuccess(result.exit)).toBe(true)
expect(result.instructions).toEqual(["Use rotated"])
}),
)
it.effect("holds no request without the option", () =>
Effect.gen(function* () {
const api = consoleApi()