export * as State from "./state.js" import { Cause, Context, Effect, Exit, Fiber, Scope } from "effect" /** * A synchronous, replayable edit to the current domain state. * * Domain editors expose readable and writable state while preserving concise * plugin/config code. Transforms synchronously rebuild derived state. */ type TransformCallback = (editor: Editor) => void export type MakeEditor = (state: State) => Editor export interface Registration { readonly dispose: Effect.Effect } /** * Registers a scoped transform. Reads rebuild by applying every registered transform in order. * Closing the owning Scope removes the transform and invalidates the current value. */ export type Transform = ( transform: TransformCallback, ) => Effect.Effect /** Invalidates the current value after captured inputs change and notifies like a registration would. */ export type Reload = () => Effect.Effect export interface Transformable { readonly transform: Transform readonly reload: Reload } export interface Failure { readonly state: string readonly cause: unknown } type GroupedRegistration = { readonly remove: () => boolean readonly notify: Effect.Effect } type RegistrationGroup = { failed: boolean readonly registrations: Set readonly report: (failure: Failure, refresh: Effect.Effect) => void } const CurrentGroup = Context.Reference("@opencode/State/CurrentGroup", { defaultValue: () => undefined, }) /** * Groups registrations without coupling State to plugin identity or asynchronous cleanup. * A failed group is detached synchronously; its supervisor must run refresh and close its scope. */ export function group(report: RegistrationGroup["report"]) { const group: RegistrationGroup = { failed: false, registrations: new Set(), report } return (effect: Effect.Effect) => Effect.provideService(effect, CurrentGroup, group) } function disable(group: RegistrationGroup, failure: Failure) { if (group.failed) return group.failed = true const notifications = new Set>() for (const registration of group.registrations) { registration.remove() notifications.add(registration.notify) } group.report( failure, Effect.forEach(notifications, (notify) => notify, { discard: true }), ) } type Batch = { active: boolean readonly shutdown: boolean readonly notifications: Set> } const CurrentBatch = Context.Reference("@opencode/State/CurrentBatch", { defaultValue: () => undefined, }) /** Coalesces notifications until the effect completes. Reads inside stay fresh; nothing is rolled back. */ export function batch(effect: Effect.Effect) { return run(effect, false) } /** Runs the effect as shutdown: States changed inside it close permanently and never notify again. */ export function shutdown(effect: Effect.Effect) { return run(effect, true) } function run(effect: Effect.Effect, shutdown: boolean) { return Effect.uninterruptibleMask((restore) => Effect.gen(function* () { const current = yield* CurrentBatch if (current?.active && !shutdown) return yield* restore(effect) const batch: Batch = { active: true, shutdown, notifications: new Set() } const exit = yield* restore(effect.pipe(Effect.provideService(CurrentBatch, batch))).pipe(Effect.exit) batch.active = false // A shutdown batch never collects notifications: changed() closes the State instead. const notifications = yield* Effect.forEach(batch.notifications, (notify) => restore(notify).pipe(Effect.exit)) // Aggregate ordinary failures across domains, while allowing cancellation to stop observer work. yield* Exit.asVoidAll([exit, ...notifications]) return yield* exit }), ) } /** * A `notify` body that runs resource reconciliation in the owning layer's FiberSet and awaits it, so work * queued behind the layer's locks is interrupted with the layer. That interruption is not a failure. */ export function reconcile( root: Scope.Scope, fork: (effect: Effect.Effect) => Fiber.Fiber, work: () => Effect.Effect, ): Effect.Effect { return Effect.gen(function* () { const exit = yield* Fiber.await(fork(work())) if (Exit.isFailure(exit) && root.state._tag === "Closed" && Cause.hasInterruptsOnly(exit.cause)) return yield* exit }) } export const inherit = Effect.fnUntraced(function* () { const batch = yield* CurrentBatch return (effect: Effect.Effect) => Effect.provideService(effect, CurrentBatch, batch) }) export interface Options { readonly name?: string /** Creates the empty base value for every rebuild. */ readonly initial: () => State /** Wraps mutable state in a domain-specific editor API. */ readonly editor: MakeEditor /** * Observes the freshly rebuilt value outside the read path. Every registration, disposal, or * reload notifies once it is applied; a batch coalesces them into one notification at its end. * Resource reconciliation owns its execution scope and coordination. */ readonly notify?: (state: State) => Effect.Effect } export interface Interface extends Transformable { /** * Rebuilds synchronously when transforms changed since the last read. Each rebuild produces a new * value and never touches earlier ones, so callers may retain what they read. */ readonly get: () => State /** Marks a changed read-time dependency without scheduling another notification. */ readonly invalidate: () => void /** Changes synchronously, including batched edits and grouped-registration removal. */ readonly revision: () => number } export function create(options: Options): Interface { let state = options.initial() const transforms = new Set<{ run: TransformCallback; group: RegistrationGroup | undefined }>() let dirty = false let closed = false let version = 0 const invalidate = () => { dirty = true version++ } const get = () => { if (closed || !dirty) return state while (true) { const started = version const next = options.initial() const editor = options.editor(next) for (const transform of transforms) { try { transform.run(editor) } catch (cause) { if (!transform.group) throw cause disable(transform.group, { state: options.name ?? "anonymous", cause }) } // A nested read can disable a group that already contributed to this candidate. if (version !== started) break } if (version !== started) continue // Ungrouped failures still propagate; grouped failures restart from a fresh candidate. state = next dirty = false return state } } // One stable value per State, so a batch's notification Set holds it at most once. const notify: Effect.Effect = Effect.gen(function* () { if (closed) return const value = get() if (options.notify) yield* options.notify(value) }).pipe(Effect.withSpan("State.notify")) const changed = Effect.uninterruptibleMask((restore) => Effect.gen(function* () { if (closed) return invalidate() const batch = yield* CurrentBatch if (batch?.active) { if (batch.shutdown) { closed = true return } batch.notifications.add(notify) return } yield* restore(notify) }), ) return { get, invalidate, revision: () => version, transform: Effect.fn("State.transform")(function* (update) { yield* Effect.annotateCurrentSpan("state", options.name ?? "anonymous") const scope = yield* Scope.Scope const group = yield* CurrentGroup if (group?.failed) return { dispose: Effect.void } return yield* Effect.uninterruptible( Effect.gen(function* () { const transform = { run: update, group } const registration: GroupedRegistration = { remove: () => { if (!transforms.delete(transform)) return false group?.registrations.delete(registration) invalidate() return true }, notify: changed, } const dispose = Effect.uninterruptible(Effect.suspend(() => (registration.remove() ? changed : Effect.void))) transforms.add(transform) group?.registrations.add(registration) yield* Scope.addFinalizer(scope, dispose) yield* changed return { dispose } }), ) }), reload: () => changed, } }