mirror of
https://github.com/anomalyco/opencode.git
synced 2026-11-14 20:21:39 +03:00
fix(core): publish catalog updates only when they change (#56267)
This commit is contained in:
committed by
GitHub
Unverified
parent
eadc7397d4
commit
54157a7a0b
+22
-20
@@ -50,6 +50,24 @@ const layer = Layer.effect(
|
||||
Service,
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const selectable = (agent: Info | undefined) =>
|
||||
agent && agent.mode !== "subagent" && !agent.hidden ? agent : undefined
|
||||
const selectedDefault = (data: Data) => {
|
||||
const configured = data.default ? selectable(data.agents.get(data.default)) : undefined
|
||||
if (configured) return configured
|
||||
const build = selectable(data.agents.get(ID.make("build")))
|
||||
if (build) return build
|
||||
for (const agent of data.agents.values()) {
|
||||
const fallback = selectable(agent)
|
||||
if (fallback) return fallback
|
||||
}
|
||||
}
|
||||
const list = (data: Data) => {
|
||||
const agents = Array.fromIterable(data.agents.values())
|
||||
const selected = selectedDefault(data)
|
||||
if (!selected) return agents
|
||||
return [selected, ...agents.filter((agent) => agent.id !== selected.id)]
|
||||
}
|
||||
const state = State.create<Data, Editor>({
|
||||
name: "agent",
|
||||
initial: () => ({ agents: new Map() }),
|
||||
@@ -69,21 +87,8 @@ const layer = Layer.effect(
|
||||
editor.agents.delete(id)
|
||||
},
|
||||
}),
|
||||
notify: () => bus.publish(Agent.Event.Updated, {}).pipe(Effect.asVoid),
|
||||
notify: State.changes(list, () => bus.publish(Agent.Event.Updated, {}).pipe(Effect.asVoid)),
|
||||
})
|
||||
const selectable = (agent: Info | undefined) =>
|
||||
agent && agent.mode !== "subagent" && !agent.hidden ? agent : undefined
|
||||
const selectedDefault = () => {
|
||||
const data = state.get()
|
||||
const configured = data.default ? selectable(data.agents.get(data.default)) : undefined
|
||||
if (configured) return configured
|
||||
const build = selectable(data.agents.get(ID.make("build")))
|
||||
if (build) return build
|
||||
for (const agent of data.agents.values()) {
|
||||
const fallback = selectable(agent)
|
||||
if (fallback) return fallback
|
||||
}
|
||||
}
|
||||
|
||||
return Service.of({
|
||||
transform: state.transform,
|
||||
@@ -93,21 +98,18 @@ const layer = Layer.effect(
|
||||
}),
|
||||
resolve: Effect.fnUntraced(function* (id) {
|
||||
if (id !== undefined) return state.get().agents.get(ID.make(id))
|
||||
return selectedDefault()
|
||||
return selectedDefault(state.get())
|
||||
}),
|
||||
select: Effect.fn("Agent.select")(function* (id) {
|
||||
if (id !== undefined) {
|
||||
const selected = ID.make(id)
|
||||
return { id: selected, info: state.get().agents.get(selected) }
|
||||
}
|
||||
const info = selectedDefault()
|
||||
const info = selectedDefault(state.get())
|
||||
return { id: info?.id ?? defaultID, info }
|
||||
}),
|
||||
list: Effect.fn("Agent.list")(function* () {
|
||||
const agents = Array.fromIterable(state.get().agents.values())
|
||||
const selected = selectedDefault()
|
||||
if (!selected) return agents
|
||||
return [selected, ...agents.filter((agent) => agent.id !== selected.id)]
|
||||
return list(state.get())
|
||||
}),
|
||||
})
|
||||
}),
|
||||
|
||||
@@ -54,19 +54,22 @@ export const layer = Layer.effect(
|
||||
Service,
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const info = (definition: Definition) =>
|
||||
Info.make({
|
||||
name: definition.name,
|
||||
description: definition.description,
|
||||
})
|
||||
const state = State.create<Map<string, Definition>, Editor>({
|
||||
name: "command",
|
||||
initial: () => new Map(),
|
||||
editor: (editor) => ({
|
||||
add: (definition) => editor.set(definition.name, definition),
|
||||
}),
|
||||
notify: () => bus.publish(Command.Event.Updated, {}).pipe(Effect.asVoid),
|
||||
notify: State.changes(
|
||||
(commands) => Array.from(commands.values(), info),
|
||||
() => bus.publish(Command.Event.Updated, {}).pipe(Effect.asVoid),
|
||||
),
|
||||
})
|
||||
const info = (definition: Definition) =>
|
||||
Info.make({
|
||||
name: definition.name,
|
||||
description: definition.description,
|
||||
})
|
||||
|
||||
return Service.of({
|
||||
reload: state.reload,
|
||||
|
||||
@@ -379,7 +379,13 @@ const layer = Layer.effect(
|
||||
},
|
||||
},
|
||||
}),
|
||||
notify: () => bus.publish(Integration.Event.Updated, {}).pipe(Effect.asVoid),
|
||||
notify: State.changes(
|
||||
(data) =>
|
||||
Array.from(data.integrations.values(), (entry) => project(entry, resolveConnections(entry, []))).toSorted(
|
||||
(a, b) => a.name.localeCompare(b.name),
|
||||
),
|
||||
() => bus.publish(Integration.Event.Updated, {}).pipe(Effect.asVoid),
|
||||
),
|
||||
})
|
||||
|
||||
const createCredential = Effect.fnUntraced(function* (input: Parameters<Credential.Interface["create"]>[0]) {
|
||||
|
||||
@@ -40,6 +40,10 @@ const layer = Layer.effect(
|
||||
const pendingFailures = yield* Queue.unbounded<PendingFailure>()
|
||||
let discovered: readonly Failure[] = []
|
||||
let inventory: Plugin.Info[] = []
|
||||
const publish = State.changes(
|
||||
(items: readonly Plugin.Info[]) => items,
|
||||
() => bus.publish(Plugin.Event.Updated, {}).pipe(Effect.asVoid),
|
||||
)
|
||||
const list = Effect.fn("Plugin.list")(function* () {
|
||||
return inventory
|
||||
})
|
||||
@@ -115,10 +119,8 @@ const layer = Layer.effect(
|
||||
if (entry) active.set(definition.id, { ...entry, plugin: definition })
|
||||
}
|
||||
if (prefix === definitions.length && active.size === definitions.length) {
|
||||
const nextInventory = [...Array.from(active.values()).map(slotInfo), ...failures]
|
||||
if (JSON.stringify(inventory) === JSON.stringify(nextInventory)) return
|
||||
inventory = nextInventory
|
||||
yield* bus.publish(Plugin.Event.Updated, {})
|
||||
inventory = [...Array.from(active.values()).map(slotInfo), ...failures]
|
||||
yield* publish(inventory)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -172,7 +174,7 @@ const layer = Layer.effect(
|
||||
inventory = [...Array.from(active.values()).map(slotInfo), ...failures]
|
||||
}),
|
||||
)
|
||||
yield* bus.publish(Plugin.Event.Updated, {})
|
||||
yield* publish(inventory)
|
||||
}),
|
||||
),
|
||||
(release) => release,
|
||||
@@ -195,7 +197,7 @@ const layer = Layer.effect(
|
||||
// cannot disable a replacement and teardown need not wait for this worker.
|
||||
inventory = [...Array.from(active.values()).map(slotInfo), ...discovered]
|
||||
const refreshed = yield* State.batch(item.refresh).pipe(Effect.exit)
|
||||
yield* bus.publish(Plugin.Event.Updated, {})
|
||||
yield* publish(inventory)
|
||||
if (Exit.isFailure(refreshed))
|
||||
yield* Effect.logWarning("failed to refresh state after disabling plugin", {
|
||||
"plugin.id": item.plugin.id,
|
||||
|
||||
@@ -48,8 +48,8 @@ const layer = Layer.effect(
|
||||
const bus = yield* Bus.Service
|
||||
const cache = yield* RepositoryCache.Service
|
||||
const scope = yield* Scope.Scope
|
||||
const list = (): Info[] =>
|
||||
Array.from(state.get().sources).flatMap(([name, source]) => {
|
||||
const list = (data: Data = state.get()): Info[] =>
|
||||
Array.from(data.sources).flatMap(([name, source]) => {
|
||||
const info = {
|
||||
name,
|
||||
source,
|
||||
@@ -101,11 +101,12 @@ const layer = Layer.effect(
|
||||
list: () => Array.from(editor.sources),
|
||||
get: (name) => editor.sources.get(name),
|
||||
}),
|
||||
notify: () =>
|
||||
notify: State.changes(list, () =>
|
||||
Effect.gen(function* () {
|
||||
yield* refresh().pipe(Effect.forkIn(scope))
|
||||
yield* bus.publish(Reference.Event.Updated, {})
|
||||
}),
|
||||
),
|
||||
})
|
||||
|
||||
// Check independently of session activity; the shared cache throttles Git work daily.
|
||||
|
||||
@@ -116,7 +116,10 @@ const layer = Layer.effect(
|
||||
editor.skills.delete(ID.make(id))
|
||||
},
|
||||
}),
|
||||
notify: () => bus.publish(Skill.Event.Updated, {}).pipe(Effect.asVoid),
|
||||
notify: State.changes(
|
||||
(data) => Array.from(data.skills.values()),
|
||||
() => bus.publish(Skill.Event.Updated, {}).pipe(Effect.asVoid),
|
||||
),
|
||||
})
|
||||
|
||||
return Service.of({
|
||||
|
||||
@@ -126,6 +126,25 @@ export function reconcile(
|
||||
})
|
||||
}
|
||||
|
||||
/**
|
||||
* Runs `notify` only when the serialized projection of `project(state)` differs from the last
|
||||
* notified value. The first notification always runs so a freshly booted Location announces its
|
||||
* current catalog even when empty.
|
||||
*/
|
||||
export function changes<State>(
|
||||
project: (state: State) => unknown,
|
||||
notify: (state: State) => Effect.Effect<void>,
|
||||
): (state: State) => Effect.Effect<void> {
|
||||
let published: string | undefined
|
||||
return (state) =>
|
||||
Effect.suspend(() => {
|
||||
const next = JSON.stringify(project(state)) ?? ""
|
||||
if (next === published) return Effect.void
|
||||
published = next
|
||||
return notify(state)
|
||||
})
|
||||
}
|
||||
|
||||
export const inherit = Effect.fnUntraced(function* () {
|
||||
const batch = yield* CurrentBatch
|
||||
return <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.provideService(effect, CurrentBatch, batch)
|
||||
|
||||
@@ -99,6 +99,12 @@ const layer = Layer.effect(
|
||||
),
|
||||
Effect.forkScoped({ startImmediately: true }),
|
||||
)
|
||||
const providers = (data: Data) =>
|
||||
Array.from(data.providers.values(), (provider) => ({
|
||||
id: provider.id,
|
||||
name: provider.name,
|
||||
})).toSorted((a, b) => a.name.localeCompare(b.name))
|
||||
|
||||
const state = State.create<Data, Editor>({
|
||||
initial: () => ({ providers: new Map() }),
|
||||
editor: (editor) => ({
|
||||
@@ -108,7 +114,7 @@ const layer = Layer.effect(
|
||||
set: (selection) => (editor.selection = selection),
|
||||
},
|
||||
}),
|
||||
notify: () => bus.publish(WebSearch.Event.Updated, {}).pipe(Effect.asVoid),
|
||||
notify: State.changes(providers, () => bus.publish(WebSearch.Event.Updated, {}).pipe(Effect.asVoid)),
|
||||
})
|
||||
|
||||
const requireProvider = (providers: Map<ID, ProviderImplementation>, providerID: ID) => {
|
||||
@@ -162,10 +168,7 @@ const layer = Layer.effect(
|
||||
transform: state.transform,
|
||||
reload: state.reload,
|
||||
providers: Effect.fn("WebSearch.providers")(function* () {
|
||||
return Array.from(state.get().providers.values(), (provider) => ({
|
||||
id: provider.id,
|
||||
name: provider.name,
|
||||
})).toSorted((a, b) => a.name.localeCompare(b.name))
|
||||
return providers(state.get())
|
||||
}),
|
||||
default: Effect.fn("WebSearch.defaultInfo")(function* () {
|
||||
const provider = yield* defaultProvider(yield* selection())
|
||||
|
||||
@@ -8,10 +8,12 @@ import { Command } from "@opencode/core/command"
|
||||
import { Config } from "@opencode/core/config"
|
||||
import { ConfigAgentPlugin } from "@opencode/core/config/plugin/agent"
|
||||
import { ConfigCommandPlugin } from "@opencode/core/config/plugin/command"
|
||||
import { ConfigPolicyPlugin } from "@opencode/core/config/plugin/policy"
|
||||
import { ConfigProviderPlugin } from "@opencode/core/config/plugin/provider"
|
||||
import { ConfigReferencePlugin } from "@opencode/core/config/plugin/reference"
|
||||
import { ConfigCompatibilityPlugin } from "@opencode/core/config/plugin/compatibility"
|
||||
import { ConfigSkillPlugin } from "@opencode/core/config/plugin/skill"
|
||||
import { ConfigWebSearchPlugin } from "@opencode/core/config/plugin/websearch"
|
||||
import { Bus } from "@opencode/core/bus"
|
||||
import { Integration } from "@opencode/core/integration"
|
||||
import { Plugin } from "@opencode/core/plugin"
|
||||
@@ -20,6 +22,8 @@ import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder"
|
||||
import { Provider } from "@opencode/core/provider"
|
||||
import { Reference } from "@opencode/core/reference"
|
||||
import { Skill } from "@opencode/core/skill"
|
||||
import { State } from "@opencode/core/state"
|
||||
import { WebSearch } from "@opencode/core/websearch"
|
||||
import { ShellSelect } from "@opencode/core/shell/select"
|
||||
import { Job } from "@opencode/core/job"
|
||||
import { SubagentJob } from "@opencode/core/session/subagent-job"
|
||||
@@ -323,6 +327,161 @@ describe("config plugin reloads", () => {
|
||||
Effect.provideService(Global.Service, Global.Service.of(Global.make())),
|
||||
),
|
||||
)
|
||||
|
||||
it.live("publishes catalog updates on boot and real changes but skips unchanged reloads", () =>
|
||||
Effect.gen(function* () {
|
||||
const bus = yield* Bus.Service
|
||||
const integrations = yield* Integration.Service
|
||||
const plugins = yield* Plugin.Service
|
||||
const websearch = yield* WebSearch.Service
|
||||
const baseHost = yield* PluginHost.make(plugins)
|
||||
const test = yield* Config.Test
|
||||
const tracked = new Set<string>([
|
||||
Agent.Event.Updated.type,
|
||||
Command.Event.Updated.type,
|
||||
Skill.Event.Updated.type,
|
||||
Reference.Event.Updated.type,
|
||||
WebSearch.Event.Updated.type,
|
||||
Integration.Event.Updated.type,
|
||||
Credential.Event.Updated.type,
|
||||
Credential.Event.Switched.type,
|
||||
])
|
||||
const counts = new Map<string, number>()
|
||||
const reloads = { agent: 0, command: 0, skill: 0, reference: 0, websearch: 0, integration: 0 }
|
||||
const count = (type: string) => counts.get(type) ?? 0
|
||||
yield* Effect.acquireRelease(
|
||||
bus.listen((event) =>
|
||||
Effect.sync(() => {
|
||||
if (!tracked.has(event.type)) return
|
||||
counts.set(event.type, count(event.type) + 1)
|
||||
}),
|
||||
),
|
||||
(unsubscribe) => unsubscribe,
|
||||
)
|
||||
const host: typeof baseHost = {
|
||||
...baseHost,
|
||||
agent: {
|
||||
...baseHost.agent,
|
||||
reload: () => baseHost.agent.reload().pipe(Effect.tap(() => Effect.sync(() => void reloads.agent++))),
|
||||
},
|
||||
command: {
|
||||
...baseHost.command,
|
||||
reload: () => baseHost.command.reload().pipe(Effect.tap(() => Effect.sync(() => void reloads.command++))),
|
||||
},
|
||||
skill: {
|
||||
...baseHost.skill,
|
||||
reload: () => baseHost.skill.reload().pipe(Effect.tap(() => Effect.sync(() => void reloads.skill++))),
|
||||
},
|
||||
reference: {
|
||||
...baseHost.reference,
|
||||
reload: () => baseHost.reference.reload().pipe(Effect.tap(() => Effect.sync(() => void reloads.reference++))),
|
||||
},
|
||||
websearch: {
|
||||
...baseHost.websearch,
|
||||
reload: () => baseHost.websearch.reload().pipe(Effect.tap(() => Effect.sync(() => void reloads.websearch++))),
|
||||
},
|
||||
integration: {
|
||||
...baseHost.integration,
|
||||
reload: () =>
|
||||
baseHost.integration.reload().pipe(Effect.tap(() => Effect.sync(() => void reloads.integration++))),
|
||||
},
|
||||
}
|
||||
|
||||
let providerName = "First search"
|
||||
yield* State.batch(
|
||||
Effect.gen(function* () {
|
||||
yield* websearch.transform((editor) =>
|
||||
editor.add({ id: WebSearch.ID.make("search"), name: providerName, execute: () => Effect.succeed([]) }),
|
||||
)
|
||||
yield* ConfigAgentPlugin.Plugin.effect(host)
|
||||
yield* ConfigCommandPlugin.Plugin.effect(host)
|
||||
yield* ConfigSkillPlugin.Plugin.effect(host)
|
||||
yield* ConfigPolicyPlugin.Plugin.effect(host)
|
||||
yield* ConfigReferencePlugin.Plugin.effect(host)
|
||||
yield* ConfigWebSearchPlugin.Plugin.effect(host)
|
||||
yield* ConfigProviderPlugin.Plugin.effect(host)
|
||||
}),
|
||||
)
|
||||
expect(Object.fromEntries(counts)).toEqual({
|
||||
[Agent.Event.Updated.type]: 1,
|
||||
[Command.Event.Updated.type]: 1,
|
||||
[Skill.Event.Updated.type]: 1,
|
||||
[Reference.Event.Updated.type]: 1,
|
||||
[WebSearch.Event.Updated.type]: 1,
|
||||
[Integration.Event.Updated.type]: 1,
|
||||
})
|
||||
|
||||
counts.clear()
|
||||
yield* Effect.yieldNow
|
||||
yield* bus.publish(Event.Updated, {})
|
||||
yield* waitUntil(
|
||||
Effect.sync(
|
||||
() =>
|
||||
reloads.agent === 1 &&
|
||||
reloads.command === 1 &&
|
||||
reloads.skill === 2 &&
|
||||
reloads.reference === 1 &&
|
||||
reloads.websearch === 1 &&
|
||||
reloads.integration === 1,
|
||||
),
|
||||
)
|
||||
expect(Object.fromEntries(counts)).toEqual({})
|
||||
|
||||
yield* integrations.connection.key({
|
||||
integrationID: Integration.ID.make("first"),
|
||||
key: "sk-test",
|
||||
label: "Work",
|
||||
})
|
||||
const connected = yield* integrations.get(Integration.ID.make("first"))
|
||||
expect(connected?.connections).toMatchObject([{ type: "credential", label: "Work", method: "key" }])
|
||||
expect(Object.fromEntries(counts)).toEqual({
|
||||
[Credential.Event.Updated.type]: 1,
|
||||
[Credential.Event.Switched.type]: 1,
|
||||
})
|
||||
|
||||
counts.clear()
|
||||
yield* bus.publish(Event.Updated, {})
|
||||
yield* waitUntil(
|
||||
Effect.sync(
|
||||
() =>
|
||||
reloads.agent === 2 &&
|
||||
reloads.command === 2 &&
|
||||
reloads.skill === 4 &&
|
||||
reloads.reference === 2 &&
|
||||
reloads.websearch === 2 &&
|
||||
reloads.integration === 2,
|
||||
),
|
||||
)
|
||||
expect(Object.fromEntries(counts)).toEqual({})
|
||||
|
||||
yield* test.setEntries([config("second")])
|
||||
providerName = "Second search"
|
||||
yield* host.websearch.reload()
|
||||
yield* bus.publish(Event.Updated, {})
|
||||
yield* waitUntil(
|
||||
Effect.sync(
|
||||
() =>
|
||||
reloads.agent === 3 &&
|
||||
reloads.command === 3 &&
|
||||
reloads.skill === 6 &&
|
||||
reloads.reference === 3 &&
|
||||
reloads.websearch === 4 &&
|
||||
reloads.integration === 3,
|
||||
),
|
||||
)
|
||||
expect(Object.fromEntries(counts)).toEqual({
|
||||
[Agent.Event.Updated.type]: 1,
|
||||
[Command.Event.Updated.type]: 1,
|
||||
[Skill.Event.Updated.type]: 1,
|
||||
[Reference.Event.Updated.type]: 1,
|
||||
[WebSearch.Event.Updated.type]: 1,
|
||||
[Integration.Event.Updated.type]: 1,
|
||||
})
|
||||
}).pipe(
|
||||
Effect.provide(Config.testLayer([config("first")])),
|
||||
Effect.provideService(Global.Service, Global.Service.of(Global.make())),
|
||||
),
|
||||
)
|
||||
})
|
||||
|
||||
function liveConfig(directory: string, native?: Watcher.NativeInterface, options: Config.Options = { global: false }) {
|
||||
|
||||
@@ -416,7 +416,8 @@ describe("ConfigSkillPlugin.Plugin", () => {
|
||||
expect(yield* watcher.subscriptions()).toEqual([{ path: path.join(tmp.path, "generated"), type: "file" }])
|
||||
|
||||
yield* Effect.promise(() => fs.mkdir(path.join(tmp.path, "generated")))
|
||||
yield* emitAndWait({ type: "create", path: path.join(tmp.path, "generated") })
|
||||
yield* watcher.emit({ type: "create", path: path.join(tmp.path, "generated") })
|
||||
while ((yield* watcher.subscriptions()).at(-1)?.path !== source) yield* Effect.sleep("10 millis")
|
||||
yield* Effect.promise(async () => {
|
||||
await fs.mkdir(path.join(source, "deploy"), { recursive: true })
|
||||
await write(source, "deploy", "Deploy")
|
||||
|
||||
@@ -202,6 +202,37 @@ describe("State", () => {
|
||||
expect(finalized).toBe(2)
|
||||
}),
|
||||
)
|
||||
|
||||
it.effect("publishes the initial projection and skips unchanged reloads", () =>
|
||||
Effect.gen(function* () {
|
||||
let notifications = 0
|
||||
let label = "first"
|
||||
const state = State.create({
|
||||
initial: () => ({ items: [] as { name: string; execute: () => void }[] }),
|
||||
editor: (editor) => ({
|
||||
add: (name: string) => editor.items.push({ name, execute: () => {} }),
|
||||
}),
|
||||
notify: State.changes(
|
||||
(data) => data.items.map((item) => ({ name: item.name })),
|
||||
() => Effect.sync(() => void notifications++),
|
||||
),
|
||||
})
|
||||
|
||||
yield* state.transform(() => {})
|
||||
expect(notifications).toBe(1)
|
||||
yield* state.reload()
|
||||
expect(notifications).toBe(1)
|
||||
|
||||
yield* state.transform((editor) => editor.add(label))
|
||||
expect(notifications).toBe(2)
|
||||
yield* state.reload()
|
||||
expect(notifications).toBe(2)
|
||||
|
||||
label = "second"
|
||||
yield* state.reload()
|
||||
expect(notifications).toBe(3)
|
||||
}),
|
||||
)
|
||||
})
|
||||
|
||||
describe("State rebuild", () => {
|
||||
|
||||
Reference in New Issue
Block a user