ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: update session notices and skill reloads

Browse source
This commit is contained in:
Dax Raad 2026-07-01 14:37:27 -04:00
commit fb884bb91e
29 changed files with 304 additions and 75 deletions
Download patch file Download diff file Expand all files Collapse all files

22
packages/core/src/filesystem/watcher.ts
View file

@ -6,6 +6,7 @@ import type ParcelWatcher from "@parcel/watcher"
import { makeLocationNode } from "../effect/app-node"
import { Cause, Context, Effect, Layer } from "effect"
import { FileSystemWatcher } from "@opencode-ai/schema/filesystem-watcher"
import os from "os"
import path from "path"
import { Config } from "../config"
import { EventV2 } from "../event"
@ -57,10 +58,14 @@ export class Service extends Context.Service<Service, Interface>()("@opencode/v2
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
if (yield* Flag.OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER) return Service.of({})
if (Flag.OPENCODE_DISABLE_FILEWATCHER) return Service.of({})
const backend = getBackend()
const location = yield* Location.Service
if (path.resolve(location.directory) === path.resolve(os.homedir())) {
yield* Effect.logInfo("watcher skipped home directory", { directory: location.directory })
return Service.of({})
}
if (!backend) {
yield* Effect.logError("watcher backend not supported", {
directory: location.directory,
@ -84,6 +89,7 @@ export const layer = Layer.effect(
)
const callback: ParcelWatcher.SubscribeCallback = (_error, updates) => {
if (_error) runFork(Effect.logError("watcher callback failed", { error: _error }))
for (const update of updates) {
if (update.type === "create") runFork(events.publish(Event.Updated, { file: update.path, event: "add" }))
if (update.type === "update") runFork(events.publish(Event.Updated, { file: update.path, event: "change" }))
@ -94,7 +100,11 @@ export const layer = Layer.effect(
const subscribe = (directory: string, ignore: string[]) => {
const pending = w.subscribe(directory, callback, { ignore, backend })
return Effect.promise(() => pending).pipe(
Effect.tap((subscription) => Effect.sync(() => subscriptions.push(subscription))),
Effect.tap((subscription) =>
Effect.sync(() => subscriptions.push(subscription)).pipe(
Effect.andThen(Effect.logInfo("watcher subscribed", { directory, backend, ignores: ignore.length })),
),
),
Effect.timeout(SUBSCRIBE_TIMEOUT_MS),
Effect.catchCause((cause) => {
pending.then((subscription) => subscription.unsubscribe()).catch(() => {})
@ -106,11 +116,9 @@ export const layer = Layer.effect(
const config = (yield* (yield* Config.Service).entries())
.filter((entry): entry is Config.Document => entry.type === "document")
.flatMap((item) => item.info.watcher?.ignore ?? [])
if (yield* Flag.OPENCODE_EXPERIMENTAL_FILEWATCHER) {
yield* Effect.forkScoped(
subscribe(location.directory, [...Ignore.PATTERNS, ...config, ...protecteds(location.directory)]),
)
}
yield* Effect.forkScoped(
subscribe(location.directory, [...Ignore.PATTERNS, ...config, ...protecteds(location.directory)]),
)
if (location.vcs?.type === "git") {
const resolved = (yield* git.repo.discover(location.directory))?.gitDirectory

7
packages/core/src/flag/flag.ts
View file

@ -32,14 +32,9 @@ export const Flag = {
OPENCODE_SERVER_PASSWORD: process.env["OPENCODE_SERVER_PASSWORD"],
OPENCODE_SERVER_USERNAME: process.env["OPENCODE_SERVER_USERNAME"],
OPENCODE_DISABLE_FFF: fff === undefined ? process.platform === "win32" : truthy("OPENCODE_DISABLE_FFF"),
OPENCODE_DISABLE_FILEWATCHER: truthy("OPENCODE_DISABLE_FILEWATCHER"),
// Experimental
OPENCODE_EXPERIMENTAL_FILEWATCHER: Config.boolean("OPENCODE_EXPERIMENTAL_FILEWATCHER").pipe(
Config.withDefault(false),
),
OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER: Config.boolean("OPENCODE_EXPERIMENTAL_DISABLE_FILEWATCHER").pipe(
Config.withDefault(false),
),
OPENCODE_EXPERIMENTAL_DISABLE_COPY_ON_SELECT:
copy === undefined ? process.platform === "win32" : truthy("OPENCODE_EXPERIMENTAL_DISABLE_COPY_ON_SELECT"),
OPENCODE_MODELS_URL: process.env["OPENCODE_MODELS_URL"],

27
packages/core/src/session.ts
View file

@ -39,6 +39,7 @@ import { Revert } from "@opencode-ai/schema/revert"
import { FSUtil } from "./fs-util"
import { SessionDurable } from "@opencode-ai/schema/durable-event-manifest"
import { SkillV2 } from "./skill"
import { Job } from "./job"
export const RevertState = Revert.State
export type RevertState = Revert.State
@ -191,9 +192,14 @@ export interface Interface {
) => Effect.Effect<void, NotFoundError | BusyError | MessageDecodeError | OperationUnavailableError>
readonly wait: (id: SessionSchema.ID) => Effect.Effect<void, NotFoundError>
readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
readonly background: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError>
readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError | SessionRunner.RunError>
readonly interrupt: (sessionID: SessionSchema.ID) => Effect.Effect<void>
readonly synthetic: (input: { sessionID: SessionSchema.ID; text: string }) => Effect.Effect<void, NotFoundError>
readonly synthetic: (input: {
sessionID: SessionSchema.ID
text: string
description?: string
}) => Effect.Effect<void, NotFoundError>
readonly revert: {
readonly stage: (input: {
sessionID: SessionSchema.ID
@ -217,6 +223,7 @@ export const layer = Layer.effect(
const execution = yield* SessionExecution.Service
const store = yield* SessionStore.Service
const locations = yield* LocationServiceMap.Service
const jobs = yield* Job.Service
const scope = yield* Scope.Scope
const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Message)
const isDurableSessionEvent = Schema.is(SessionEvent.Durable)
@ -512,6 +519,22 @@ export const layer = Layer.effect(
yield* execution.awaitIdle(sessionID)
}),
active: execution.active,
background: Effect.fn("V2Session.background")(function* (sessionID) {
yield* result.get(sessionID)
const backgrounded = yield* jobs.backgroundAll({ sessionID })
if (backgrounded.length === 0) return
yield* result.synthetic({
sessionID,
text: [
"User requested that active blocking work be moved to the background.",
"",
"Backgrounded work:",
...backgrounded.map((job) => `- ${job.type}: ${job.title && job.title.length > 0 ? job.title : job.id}`),
"",
"The backgrounded work is still unfinished. Move on to other work if you can. If there is nothing else useful to do, finish your response. Do not wait, sleep, poll, or report the backgrounded work as complete until a later completion notification is added to the conversation.",
].join("\n"),
})
}),
resume: Effect.fn("V2Session.resume")(function* (sessionID) {
yield* result.get(sessionID)
yield* execution.resume(sessionID)
@ -523,6 +546,7 @@ export const layer = Layer.effect(
messageID: SessionMessage.ID.create(),
timestamp: yield* DateTime.now,
text: input.text,
description: input.description,
})
yield* execution.resume(input.sessionID).pipe(Effect.ignore, Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid)
}),
@ -587,6 +611,7 @@ export const node = makeGlobalNode({
service: Service,
layer: layer.pipe(Layer.orDie),
deps: [
Job.node,
Database.node,
EventV2.node,
ProjectV2.node,

1
packages/core/src/session/message-updater.ts
View file

@ -153,6 +153,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) {
SessionMessage.Synthetic.make({
sessionID: event.data.sessionID,
text: event.data.text,
description: event.data.description,
id: event.data.messageID,
type: "synthetic",
time: { created: event.data.timestamp },

52
packages/core/src/skill.ts
View file

@ -2,10 +2,12 @@ export * as SkillV2 from "./skill"
import { makeLocationNode } from "./effect/app-node"
import path from "path"
import { Context, Effect, Layer, Schema, Types } from "effect"
import { Context, Effect, Layer, Schema, Stream, Types } from "effect"
import { FileSystemWatcher } from "@opencode-ai/schema/filesystem-watcher"
import { Skill } from "@opencode-ai/schema/skill"
import { AgentV2 } from "./agent"
import { ConfigMarkdown } from "./config/markdown"
import { EventV2 } from "./event"
import { FSUtil } from "./fs-util"
import { PermissionV2 } from "./permission"
import { AbsolutePath } from "./schema"
@ -27,6 +29,8 @@ export type Source = typeof Source.Type
export const Info = Skill.Info
export type Info = Skill.Info
export const Event = Skill.Event
export const available = (skills: ReadonlyArray<Info>, agent: AgentV2.Info) =>
skills.filter((skill) => PermissionV2.evaluate("skill", skill.name, agent.permissions).effect !== "deny")
@ -58,6 +62,7 @@ export const layer = Layer.effect(
Effect.gen(function* () {
const discovery = yield* SkillDiscovery.Service
const fs = yield* FSUtil.Service
const events = yield* EventV2.Service
const state = State.create<Data, Draft>({
initial: () => ({ sources: [] }),
@ -72,7 +77,15 @@ export const layer = Layer.effect(
const load = Effect.fn("SkillV2.load")(function* (source: Source) {
const skills: Info[] = []
if (source.type === "embedded") return [source.skill]
if (source.type === "embedded") {
yield* Effect.logDebug("skill source loaded", {
source: Source.key(source),
type: source.type,
directories: [],
skills: [source.skill.name],
})
return { skills: [source.skill], directories: [] }
}
const directories = source.type === "directory" ? [source.path] : yield* discovery.pull(source.url)
for (const directory of directories) {
const files = yield* fs
@ -101,19 +114,42 @@ export const layer = Layer.effect(
})
}
}
return skills
yield* Effect.logDebug("skill source loaded", {
source: Source.key(source),
type: source.type,
directories,
skills: skills.map((skill) => skill.name),
})
return { skills, directories }
})
// QUESTION(Dax): Should local skill sources invalidate on filesystem watch
// events, following the reload policy chosen for other context sources?
const cache = new Map<string, Info[]>()
const cache = new Map<string, { skills: Info[]; directories: readonly string[] }>()
const invalidate = Effect.fn("SkillV2.invalidateFromWatcher")(function* (file: string) {
const invalidated = Array.from(cache.entries()).filter(([, loaded]) =>
loaded.directories.some((directory) => FSUtil.contains(directory, file)),
)
if (invalidated.length === 0) return
for (const [key] of invalidated) cache.delete(key)
yield* Effect.logInfo("skill cache invalidated", {
file,
sources: invalidated.map(([key]) => key),
skills: invalidated.flatMap(([, loaded]) => loaded.skills.map((skill) => skill.name)),
})
yield* events.publish(Event.Updated, {}).pipe(Effect.asVoid)
})
yield* events.subscribe(FileSystemWatcher.Event.Updated).pipe(
Stream.runForEach((event) => invalidate(event.data.file)),
Effect.forkScoped({ startImmediately: true }),
)
const list = Effect.fn("SkillV2.list")(function* () {
const skills = new Map<string, Info>()
for (const source of state.get().sources) {
const key = Source.key(source)
const loaded = cache.get(key) ?? (yield* load(source))
cache.set(key, loaded)
for (const skill of loaded) skills.set(skill.name, skill)
for (const skill of loaded.skills) skills.set(skill.name, skill)
}
return Array.from(skills.values())
})
@ -131,4 +167,4 @@ export const layer = Layer.effect(
export const locationLayer = layer.pipe(Layer.provide(SkillDiscovery.defaultLayer))
export const node = makeLocationNode({ service: Service, layer, deps: [SkillDiscovery.node, FSUtil.node] })
export const node = makeLocationNode({ service: Service, layer, deps: [SkillDiscovery.node, FSUtil.node, EventV2.node] })