refactor(instance): move scoped services to LayerMap (#17544)
This commit is contained in:
@@ -0,0 +1,12 @@
|
|||||||
|
const disposers = new Set<(directory: string) => Promise<void>>()
|
||||||
|
|
||||||
|
export function registerDisposer(disposer: (directory: string) => Promise<void>) {
|
||||||
|
disposers.add(disposer)
|
||||||
|
return () => {
|
||||||
|
disposers.delete(disposer)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export async function disposeInstance(directory: string) {
|
||||||
|
await Promise.allSettled([...disposers].map((disposer) => disposer(directory)))
|
||||||
|
}
|
||||||
@@ -0,0 +1,52 @@
|
|||||||
|
import { Effect, Layer, LayerMap, ServiceMap } from "effect"
|
||||||
|
import { registerDisposer } from "./instance-registry"
|
||||||
|
import { ProviderAuthService } from "@/provider/auth-service"
|
||||||
|
import { QuestionService } from "@/question/service"
|
||||||
|
import { PermissionService } from "@/permission/service"
|
||||||
|
import { Instance } from "@/project/instance"
|
||||||
|
import type { Project } from "@/project/project"
|
||||||
|
|
||||||
|
export declare namespace InstanceContext {
|
||||||
|
export interface Shape {
|
||||||
|
readonly directory: string
|
||||||
|
readonly project: Project.Info
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export class InstanceContext extends ServiceMap.Service<InstanceContext, InstanceContext.Shape>()(
|
||||||
|
"opencode/InstanceContext",
|
||||||
|
) {}
|
||||||
|
|
||||||
|
export type InstanceServices = QuestionService | PermissionService | ProviderAuthService
|
||||||
|
|
||||||
|
function lookup(directory: string) {
|
||||||
|
const project = Instance.project
|
||||||
|
const ctx = Layer.sync(InstanceContext, () => InstanceContext.of({ directory, project }))
|
||||||
|
return Layer.mergeAll(
|
||||||
|
Layer.fresh(QuestionService.layer),
|
||||||
|
Layer.fresh(PermissionService.layer),
|
||||||
|
Layer.fresh(ProviderAuthService.layer),
|
||||||
|
).pipe(Layer.provide(ctx))
|
||||||
|
}
|
||||||
|
|
||||||
|
export class Instances extends ServiceMap.Service<Instances, LayerMap.LayerMap<string, InstanceServices>>()(
|
||||||
|
"opencode/Instances",
|
||||||
|
) {
|
||||||
|
static readonly layer = Layer.effect(
|
||||||
|
Instances,
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const layerMap = yield* LayerMap.make(lookup, { idleTimeToLive: Infinity })
|
||||||
|
const unregister = registerDisposer((directory) => Effect.runPromise(layerMap.invalidate(directory)))
|
||||||
|
yield* Effect.addFinalizer(() => Effect.sync(unregister))
|
||||||
|
return Instances.of(layerMap)
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
|
||||||
|
static get(directory: string): Layer.Layer<InstanceServices, never, Instances> {
|
||||||
|
return Layer.unwrap(Instances.use((map) => Effect.succeed(map.get(directory))))
|
||||||
|
}
|
||||||
|
|
||||||
|
static invalidate(directory: string): Effect.Effect<void, never, Instances> {
|
||||||
|
return Instances.use((map) => map.invalidate(directory))
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -1,9 +1,14 @@
|
|||||||
import { Layer, ManagedRuntime } from "effect"
|
import { Effect, Layer, ManagedRuntime } from "effect"
|
||||||
import { AccountService } from "@/account/service"
|
import { AccountService } from "@/account/service"
|
||||||
import { AuthService } from "@/auth/service"
|
import { AuthService } from "@/auth/service"
|
||||||
import { PermissionService } from "@/permission/service"
|
import { Instances } from "@/effect/instances"
|
||||||
import { QuestionService } from "@/question/service"
|
import type { InstanceServices } from "@/effect/instances"
|
||||||
|
import { Instance } from "@/project/instance"
|
||||||
|
|
||||||
export const runtime = ManagedRuntime.make(
|
export const runtime = ManagedRuntime.make(
|
||||||
Layer.mergeAll(AccountService.defaultLayer, AuthService.defaultLayer, PermissionService.layer, QuestionService.layer),
|
Layer.mergeAll(AccountService.defaultLayer, Instances.layer).pipe(Layer.provideMerge(AuthService.defaultLayer)),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
export function runPromiseInstance<A, E>(effect: Effect.Effect<A, E, InstanceServices>) {
|
||||||
|
return runtime.runPromise(effect.pipe(Effect.provide(Instances.get(Instance.directory))))
|
||||||
|
}
|
||||||
|
|||||||
@@ -1,18 +1,9 @@
|
|||||||
import { runtime } from "@/effect/runtime"
|
import { runPromiseInstance } from "@/effect/runtime"
|
||||||
import { Config } from "@/config/config"
|
import { Config } from "@/config/config"
|
||||||
import { fn } from "@/util/fn"
|
import { fn } from "@/util/fn"
|
||||||
import { Wildcard } from "@/util/wildcard"
|
import { Wildcard } from "@/util/wildcard"
|
||||||
import { Effect } from "effect"
|
|
||||||
import os from "os"
|
import os from "os"
|
||||||
import * as S from "./service"
|
import * as S from "./service"
|
||||||
import type {
|
|
||||||
Action as ActionType,
|
|
||||||
PermissionError,
|
|
||||||
Reply as ReplyType,
|
|
||||||
Request as RequestType,
|
|
||||||
Rule as RuleType,
|
|
||||||
Ruleset as RulesetType,
|
|
||||||
} from "./service"
|
|
||||||
|
|
||||||
export namespace PermissionNext {
|
export namespace PermissionNext {
|
||||||
function expand(pattern: string): string {
|
function expand(pattern: string): string {
|
||||||
@@ -23,20 +14,16 @@ export namespace PermissionNext {
|
|||||||
return pattern
|
return pattern
|
||||||
}
|
}
|
||||||
|
|
||||||
function runPromise<A>(f: (service: S.PermissionService.Api) => Effect.Effect<A, PermissionError>) {
|
|
||||||
return runtime.runPromise(S.PermissionService.use(f))
|
|
||||||
}
|
|
||||||
|
|
||||||
export const Action = S.Action
|
export const Action = S.Action
|
||||||
export type Action = ActionType
|
export type Action = S.Action
|
||||||
export const Rule = S.Rule
|
export const Rule = S.Rule
|
||||||
export type Rule = RuleType
|
export type Rule = S.Rule
|
||||||
export const Ruleset = S.Ruleset
|
export const Ruleset = S.Ruleset
|
||||||
export type Ruleset = RulesetType
|
export type Ruleset = S.Ruleset
|
||||||
export const Request = S.Request
|
export const Request = S.Request
|
||||||
export type Request = RequestType
|
export type Request = S.Request
|
||||||
export const Reply = S.Reply
|
export const Reply = S.Reply
|
||||||
export type Reply = ReplyType
|
export type Reply = S.Reply
|
||||||
export const Approval = S.Approval
|
export const Approval = S.Approval
|
||||||
export const Event = S.Event
|
export const Event = S.Event
|
||||||
export const Service = S.PermissionService
|
export const Service = S.PermissionService
|
||||||
@@ -66,12 +53,16 @@ export namespace PermissionNext {
|
|||||||
return rulesets.flat()
|
return rulesets.flat()
|
||||||
}
|
}
|
||||||
|
|
||||||
export const ask = fn(S.AskInput, async (input) => runPromise((service) => service.ask(input)))
|
export const ask = fn(S.AskInput, async (input) =>
|
||||||
|
runPromiseInstance(S.PermissionService.use((service) => service.ask(input))),
|
||||||
|
)
|
||||||
|
|
||||||
export const reply = fn(S.ReplyInput, async (input) => runPromise((service) => service.reply(input)))
|
export const reply = fn(S.ReplyInput, async (input) =>
|
||||||
|
runPromiseInstance(S.PermissionService.use((service) => service.reply(input))),
|
||||||
|
)
|
||||||
|
|
||||||
export async function list() {
|
export async function list() {
|
||||||
return runPromise((service) => service.list())
|
return runPromiseInstance(S.PermissionService.use((service) => service.list()))
|
||||||
}
|
}
|
||||||
|
|
||||||
export function evaluate(permission: string, pattern: string, ...rulesets: Ruleset[]): Rule {
|
export function evaluate(permission: string, pattern: string, ...rulesets: Ruleset[]): Rule {
|
||||||
|
|||||||
@@ -1,11 +1,10 @@
|
|||||||
import { Bus } from "@/bus"
|
import { Bus } from "@/bus"
|
||||||
import { BusEvent } from "@/bus/bus-event"
|
import { BusEvent } from "@/bus/bus-event"
|
||||||
import { Instance } from "@/project/instance"
|
import { InstanceContext } from "@/effect/instances"
|
||||||
import { ProjectID } from "@/project/schema"
|
import { ProjectID } from "@/project/schema"
|
||||||
import { MessageID, SessionID } from "@/session/schema"
|
import { MessageID, SessionID } from "@/session/schema"
|
||||||
import { PermissionTable } from "@/session/session.sql"
|
import { PermissionTable } from "@/session/session.sql"
|
||||||
import { Database, eq } from "@/storage/db"
|
import { Database, eq } from "@/storage/db"
|
||||||
import { InstanceState } from "@/util/instance-state"
|
|
||||||
import { Log } from "@/util/log"
|
import { Log } from "@/util/log"
|
||||||
import { Wildcard } from "@/util/wildcard"
|
import { Wildcard } from "@/util/wildcard"
|
||||||
import { Deferred, Effect, Layer, Schema, ServiceMap } from "effect"
|
import { Deferred, Effect, Layer, Schema, ServiceMap } from "effect"
|
||||||
@@ -104,11 +103,6 @@ interface PendingEntry {
|
|||||||
deferred: Deferred.Deferred<void, RejectedError | CorrectedError>
|
deferred: Deferred.Deferred<void, RejectedError | CorrectedError>
|
||||||
}
|
}
|
||||||
|
|
||||||
type State = {
|
|
||||||
pending: Map<PermissionID, PendingEntry>
|
|
||||||
approved: Ruleset
|
|
||||||
}
|
|
||||||
|
|
||||||
export const AskInput = Request.partial({ id: true }).extend({
|
export const AskInput = Request.partial({ id: true }).extend({
|
||||||
ruleset: Ruleset,
|
ruleset: Ruleset,
|
||||||
})
|
})
|
||||||
@@ -133,25 +127,19 @@ export class PermissionService extends ServiceMap.Service<PermissionService, Per
|
|||||||
static readonly layer = Layer.effect(
|
static readonly layer = Layer.effect(
|
||||||
PermissionService,
|
PermissionService,
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
const instanceState = yield* InstanceState.make<State>(() =>
|
const { project } = yield* InstanceContext
|
||||||
Effect.sync(() => {
|
const row = Database.use((db) =>
|
||||||
const row = Database.use((db) =>
|
db.select().from(PermissionTable).where(eq(PermissionTable.project_id, project.id)).get(),
|
||||||
db.select().from(PermissionTable).where(eq(PermissionTable.project_id, Instance.project.id)).get(),
|
|
||||||
)
|
|
||||||
return {
|
|
||||||
pending: new Map<PermissionID, PendingEntry>(),
|
|
||||||
approved: row?.data ?? [],
|
|
||||||
}
|
|
||||||
}),
|
|
||||||
)
|
)
|
||||||
|
const pending = new Map<PermissionID, PendingEntry>()
|
||||||
|
const approved: Ruleset = row?.data ?? []
|
||||||
|
|
||||||
const ask = Effect.fn("PermissionService.ask")(function* (input: z.infer<typeof AskInput>) {
|
const ask = Effect.fn("PermissionService.ask")(function* (input: z.infer<typeof AskInput>) {
|
||||||
const state = yield* InstanceState.get(instanceState)
|
|
||||||
const { ruleset, ...request } = input
|
const { ruleset, ...request } = input
|
||||||
let pending = false
|
let needsAsk = false
|
||||||
|
|
||||||
for (const pattern of request.patterns) {
|
for (const pattern of request.patterns) {
|
||||||
const rule = evaluate(request.permission, pattern, ruleset, state.approved)
|
const rule = evaluate(request.permission, pattern, ruleset, approved)
|
||||||
log.info("evaluated", { permission: request.permission, pattern, action: rule })
|
log.info("evaluated", { permission: request.permission, pattern, action: rule })
|
||||||
if (rule.action === "deny") {
|
if (rule.action === "deny") {
|
||||||
return yield* new DeniedError({
|
return yield* new DeniedError({
|
||||||
@@ -159,10 +147,10 @@ export class PermissionService extends ServiceMap.Service<PermissionService, Per
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
if (rule.action === "allow") continue
|
if (rule.action === "allow") continue
|
||||||
pending = true
|
needsAsk = true
|
||||||
}
|
}
|
||||||
|
|
||||||
if (!pending) return
|
if (!needsAsk) return
|
||||||
|
|
||||||
const id = request.id ?? PermissionID.ascending()
|
const id = request.id ?? PermissionID.ascending()
|
||||||
const info: Request = {
|
const info: Request = {
|
||||||
@@ -172,22 +160,21 @@ export class PermissionService extends ServiceMap.Service<PermissionService, Per
|
|||||||
log.info("asking", { id, permission: info.permission, patterns: info.patterns })
|
log.info("asking", { id, permission: info.permission, patterns: info.patterns })
|
||||||
|
|
||||||
const deferred = yield* Deferred.make<void, RejectedError | CorrectedError>()
|
const deferred = yield* Deferred.make<void, RejectedError | CorrectedError>()
|
||||||
state.pending.set(id, { info, deferred })
|
pending.set(id, { info, deferred })
|
||||||
void Bus.publish(Event.Asked, info)
|
void Bus.publish(Event.Asked, info)
|
||||||
return yield* Effect.ensuring(
|
return yield* Effect.ensuring(
|
||||||
Deferred.await(deferred),
|
Deferred.await(deferred),
|
||||||
Effect.sync(() => {
|
Effect.sync(() => {
|
||||||
state.pending.delete(id)
|
pending.delete(id)
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
})
|
})
|
||||||
|
|
||||||
const reply = Effect.fn("PermissionService.reply")(function* (input: z.infer<typeof ReplyInput>) {
|
const reply = Effect.fn("PermissionService.reply")(function* (input: z.infer<typeof ReplyInput>) {
|
||||||
const state = yield* InstanceState.get(instanceState)
|
const existing = pending.get(input.requestID)
|
||||||
const existing = state.pending.get(input.requestID)
|
|
||||||
if (!existing) return
|
if (!existing) return
|
||||||
|
|
||||||
state.pending.delete(input.requestID)
|
pending.delete(input.requestID)
|
||||||
void Bus.publish(Event.Replied, {
|
void Bus.publish(Event.Replied, {
|
||||||
sessionID: existing.info.sessionID,
|
sessionID: existing.info.sessionID,
|
||||||
requestID: existing.info.id,
|
requestID: existing.info.id,
|
||||||
@@ -200,9 +187,9 @@ export class PermissionService extends ServiceMap.Service<PermissionService, Per
|
|||||||
input.message ? new CorrectedError({ feedback: input.message }) : new RejectedError(),
|
input.message ? new CorrectedError({ feedback: input.message }) : new RejectedError(),
|
||||||
)
|
)
|
||||||
|
|
||||||
for (const [id, item] of state.pending.entries()) {
|
for (const [id, item] of pending.entries()) {
|
||||||
if (item.info.sessionID !== existing.info.sessionID) continue
|
if (item.info.sessionID !== existing.info.sessionID) continue
|
||||||
state.pending.delete(id)
|
pending.delete(id)
|
||||||
void Bus.publish(Event.Replied, {
|
void Bus.publish(Event.Replied, {
|
||||||
sessionID: item.info.sessionID,
|
sessionID: item.info.sessionID,
|
||||||
requestID: item.info.id,
|
requestID: item.info.id,
|
||||||
@@ -217,20 +204,20 @@ export class PermissionService extends ServiceMap.Service<PermissionService, Per
|
|||||||
if (input.reply === "once") return
|
if (input.reply === "once") return
|
||||||
|
|
||||||
for (const pattern of existing.info.always) {
|
for (const pattern of existing.info.always) {
|
||||||
state.approved.push({
|
approved.push({
|
||||||
permission: existing.info.permission,
|
permission: existing.info.permission,
|
||||||
pattern,
|
pattern,
|
||||||
action: "allow",
|
action: "allow",
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
for (const [id, item] of state.pending.entries()) {
|
for (const [id, item] of pending.entries()) {
|
||||||
if (item.info.sessionID !== existing.info.sessionID) continue
|
if (item.info.sessionID !== existing.info.sessionID) continue
|
||||||
const ok = item.info.patterns.every(
|
const ok = item.info.patterns.every(
|
||||||
(pattern) => evaluate(item.info.permission, pattern, state.approved).action === "allow",
|
(pattern) => evaluate(item.info.permission, pattern, approved).action === "allow",
|
||||||
)
|
)
|
||||||
if (!ok) continue
|
if (!ok) continue
|
||||||
state.pending.delete(id)
|
pending.delete(id)
|
||||||
void Bus.publish(Event.Replied, {
|
void Bus.publish(Event.Replied, {
|
||||||
sessionID: item.info.sessionID,
|
sessionID: item.info.sessionID,
|
||||||
requestID: item.info.id,
|
requestID: item.info.id,
|
||||||
@@ -246,8 +233,7 @@ export class PermissionService extends ServiceMap.Service<PermissionService, Per
|
|||||||
})
|
})
|
||||||
|
|
||||||
const list = Effect.fn("PermissionService.list")(function* () {
|
const list = Effect.fn("PermissionService.list")(function* () {
|
||||||
const state = yield* InstanceState.get(instanceState)
|
return Array.from(pending.values(), (item) => item.info)
|
||||||
return Array.from(state.pending.values(), (item) => item.info)
|
|
||||||
})
|
})
|
||||||
|
|
||||||
return PermissionService.of({ ask, reply, list })
|
return PermissionService.of({ ask, reply, list })
|
||||||
|
|||||||
@@ -1,4 +1,3 @@
|
|||||||
import { Effect } from "effect"
|
|
||||||
import { Log } from "@/util/log"
|
import { Log } from "@/util/log"
|
||||||
import { Context } from "../util/context"
|
import { Context } from "../util/context"
|
||||||
import { Project } from "./project"
|
import { Project } from "./project"
|
||||||
@@ -6,7 +5,7 @@ import { State } from "./state"
|
|||||||
import { iife } from "@/util/iife"
|
import { iife } from "@/util/iife"
|
||||||
import { GlobalBus } from "@/bus/global"
|
import { GlobalBus } from "@/bus/global"
|
||||||
import { Filesystem } from "@/util/filesystem"
|
import { Filesystem } from "@/util/filesystem"
|
||||||
import { InstanceState } from "@/util/instance-state"
|
import { disposeInstance } from "@/effect/instance-registry"
|
||||||
|
|
||||||
interface Context {
|
interface Context {
|
||||||
directory: string
|
directory: string
|
||||||
@@ -108,17 +107,18 @@ export const Instance = {
|
|||||||
async reload(input: { directory: string; init?: () => Promise<any>; project?: Project.Info; worktree?: string }) {
|
async reload(input: { directory: string; init?: () => Promise<any>; project?: Project.Info; worktree?: string }) {
|
||||||
const directory = Filesystem.resolve(input.directory)
|
const directory = Filesystem.resolve(input.directory)
|
||||||
Log.Default.info("reloading instance", { directory })
|
Log.Default.info("reloading instance", { directory })
|
||||||
await Promise.all([State.dispose(directory), Effect.runPromise(InstanceState.dispose(directory))])
|
await Promise.all([State.dispose(directory), disposeInstance(directory)])
|
||||||
cache.delete(directory)
|
cache.delete(directory)
|
||||||
const next = track(directory, boot({ ...input, directory }))
|
const next = track(directory, boot({ ...input, directory }))
|
||||||
emit(directory)
|
emit(directory)
|
||||||
return await next
|
return await next
|
||||||
},
|
},
|
||||||
async dispose() {
|
async dispose() {
|
||||||
Log.Default.info("disposing instance", { directory: Instance.directory })
|
const directory = Instance.directory
|
||||||
await Promise.all([State.dispose(Instance.directory), Effect.runPromise(InstanceState.dispose(Instance.directory))])
|
Log.Default.info("disposing instance", { directory })
|
||||||
cache.delete(Instance.directory)
|
await Promise.all([State.dispose(directory), disposeInstance(directory)])
|
||||||
emit(Instance.directory)
|
cache.delete(directory)
|
||||||
|
emit(directory)
|
||||||
},
|
},
|
||||||
async disposeAll() {
|
async disposeAll() {
|
||||||
if (disposal.all) return disposal.all
|
if (disposal.all) return disposal.all
|
||||||
|
|||||||
@@ -1,12 +1,9 @@
|
|||||||
import { Effect, Layer, Record, ServiceMap, Struct } from "effect"
|
|
||||||
import { Instance } from "@/project/instance"
|
|
||||||
import { Plugin } from "../plugin"
|
|
||||||
import { filter, fromEntries, map, pipe } from "remeda"
|
|
||||||
import type { AuthOuathResult } from "@opencode-ai/plugin"
|
import type { AuthOuathResult } from "@opencode-ai/plugin"
|
||||||
import { NamedError } from "@opencode-ai/util/error"
|
import { NamedError } from "@opencode-ai/util/error"
|
||||||
import * as Auth from "@/auth/service"
|
import * as Auth from "@/auth/service"
|
||||||
import { InstanceState } from "@/util/instance-state"
|
|
||||||
import { ProviderID } from "./schema"
|
import { ProviderID } from "./schema"
|
||||||
|
import { Effect, Layer, Record, ServiceMap, Struct } from "effect"
|
||||||
|
import { filter, fromEntries, map, pipe } from "remeda"
|
||||||
import z from "zod"
|
import z from "zod"
|
||||||
|
|
||||||
export const Method = z
|
export const Method = z
|
||||||
@@ -54,21 +51,13 @@ export type ProviderAuthError =
|
|||||||
|
|
||||||
export namespace ProviderAuthService {
|
export namespace ProviderAuthService {
|
||||||
export interface Service {
|
export interface Service {
|
||||||
/** Get available auth methods for each provider (e.g. OAuth, API key). */
|
|
||||||
readonly methods: () => Effect.Effect<Record<string, Method[]>>
|
readonly methods: () => Effect.Effect<Record<string, Method[]>>
|
||||||
|
|
||||||
/** Start an OAuth authorization flow for a provider. Returns the URL to redirect to. */
|
|
||||||
readonly authorize: (input: { providerID: ProviderID; method: number }) => Effect.Effect<Authorization | undefined>
|
readonly authorize: (input: { providerID: ProviderID; method: number }) => Effect.Effect<Authorization | undefined>
|
||||||
|
|
||||||
/** Complete an OAuth flow after the user has authorized. Exchanges the code/callback for credentials. */
|
|
||||||
readonly callback: (input: {
|
readonly callback: (input: {
|
||||||
providerID: ProviderID
|
providerID: ProviderID
|
||||||
method: number
|
method: number
|
||||||
code?: string
|
code?: string
|
||||||
}) => Effect.Effect<void, ProviderAuthError>
|
}) => Effect.Effect<void, ProviderAuthError>
|
||||||
|
|
||||||
/** Set an API key directly for a provider (no OAuth flow). */
|
|
||||||
readonly api: (input: { providerID: ProviderID; key: string }) => Effect.Effect<void, Auth.AuthServiceError>
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -79,32 +68,29 @@ export class ProviderAuthService extends ServiceMap.Service<ProviderAuthService,
|
|||||||
ProviderAuthService,
|
ProviderAuthService,
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
const auth = yield* Auth.AuthService
|
const auth = yield* Auth.AuthService
|
||||||
const state = yield* InstanceState.make(() =>
|
const hooks = yield* Effect.promise(async () => {
|
||||||
Effect.promise(async () => {
|
const mod = await import("../plugin")
|
||||||
const methods = pipe(
|
return pipe(
|
||||||
await Plugin.list(),
|
await mod.Plugin.list(),
|
||||||
filter((x) => x.auth?.provider !== undefined),
|
filter((x) => x.auth?.provider !== undefined),
|
||||||
map((x) => [x.auth!.provider, x.auth!] as const),
|
map((x) => [x.auth!.provider, x.auth!] as const),
|
||||||
fromEntries(),
|
fromEntries(),
|
||||||
)
|
)
|
||||||
return { methods, pending: new Map<ProviderID, AuthOuathResult>() }
|
})
|
||||||
}),
|
const pending = new Map<ProviderID, AuthOuathResult>()
|
||||||
)
|
|
||||||
|
|
||||||
const methods = Effect.fn("ProviderAuthService.methods")(function* () {
|
const methods = Effect.fn("ProviderAuthService.methods")(function* () {
|
||||||
const x = yield* InstanceState.get(state)
|
return Record.map(hooks, (item) => item.methods.map((method): Method => Struct.pick(method, ["type", "label"])))
|
||||||
return Record.map(x.methods, (y) => y.methods.map((z): Method => Struct.pick(z, ["type", "label"])))
|
|
||||||
})
|
})
|
||||||
|
|
||||||
const authorize = Effect.fn("ProviderAuthService.authorize")(function* (input: {
|
const authorize = Effect.fn("ProviderAuthService.authorize")(function* (input: {
|
||||||
providerID: ProviderID
|
providerID: ProviderID
|
||||||
method: number
|
method: number
|
||||||
}) {
|
}) {
|
||||||
const s = yield* InstanceState.get(state)
|
const method = hooks[input.providerID].methods[input.method]
|
||||||
const method = s.methods[input.providerID].methods[input.method]
|
|
||||||
if (method.type !== "oauth") return
|
if (method.type !== "oauth") return
|
||||||
const result = yield* Effect.promise(() => method.authorize())
|
const result = yield* Effect.promise(() => method.authorize())
|
||||||
s.pending.set(input.providerID, result)
|
pending.set(input.providerID, result)
|
||||||
return {
|
return {
|
||||||
url: result.url,
|
url: result.url,
|
||||||
method: result.method,
|
method: result.method,
|
||||||
@@ -117,17 +103,14 @@ export class ProviderAuthService extends ServiceMap.Service<ProviderAuthService,
|
|||||||
method: number
|
method: number
|
||||||
code?: string
|
code?: string
|
||||||
}) {
|
}) {
|
||||||
const s = yield* InstanceState.get(state)
|
const match = pending.get(input.providerID)
|
||||||
const match = s.pending.get(input.providerID)
|
|
||||||
if (!match) return yield* Effect.fail(new OauthMissing({ providerID: input.providerID }))
|
if (!match) return yield* Effect.fail(new OauthMissing({ providerID: input.providerID }))
|
||||||
|
|
||||||
if (match.method === "code" && !input.code)
|
if (match.method === "code" && !input.code)
|
||||||
return yield* Effect.fail(new OauthCodeMissing({ providerID: input.providerID }))
|
return yield* Effect.fail(new OauthCodeMissing({ providerID: input.providerID }))
|
||||||
|
|
||||||
const result = yield* Effect.promise(() =>
|
const result = yield* Effect.promise(() =>
|
||||||
match.method === "code" ? match.callback(input.code!) : match.callback(),
|
match.method === "code" ? match.callback(input.code!) : match.callback(),
|
||||||
)
|
)
|
||||||
|
|
||||||
if (!result || result.type !== "success") return yield* Effect.fail(new OauthCallbackFailed({}))
|
if (!result || result.type !== "success") return yield* Effect.fail(new OauthCallbackFailed({}))
|
||||||
|
|
||||||
if ("key" in result) {
|
if ("key" in result) {
|
||||||
@@ -148,18 +131,10 @@ export class ProviderAuthService extends ServiceMap.Service<ProviderAuthService,
|
|||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
const api = Effect.fn("ProviderAuthService.api")(function* (input: { providerID: ProviderID; key: string }) {
|
|
||||||
yield* auth.set(input.providerID, {
|
|
||||||
type: "api",
|
|
||||||
key: input.key,
|
|
||||||
})
|
|
||||||
})
|
|
||||||
|
|
||||||
return ProviderAuthService.of({
|
return ProviderAuthService.of({
|
||||||
methods,
|
methods,
|
||||||
authorize,
|
authorize,
|
||||||
callback,
|
callback,
|
||||||
api,
|
|
||||||
})
|
})
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -1,25 +1,16 @@
|
|||||||
import { Effect, ManagedRuntime } from "effect"
|
|
||||||
import z from "zod"
|
import z from "zod"
|
||||||
|
|
||||||
|
import { runPromiseInstance } from "@/effect/runtime"
|
||||||
import { fn } from "@/util/fn"
|
import { fn } from "@/util/fn"
|
||||||
import * as S from "./auth-service"
|
import * as S from "./auth-service"
|
||||||
import { ProviderID } from "./schema"
|
import { ProviderID } from "./schema"
|
||||||
|
|
||||||
// Separate runtime: ProviderAuthService can't join the shared runtime because
|
|
||||||
// runtime.ts → auth-service.ts → provider/auth.ts creates a circular import.
|
|
||||||
// AuthService is stateless file I/O so the duplicate instance is harmless.
|
|
||||||
const rt = ManagedRuntime.make(S.ProviderAuthService.defaultLayer)
|
|
||||||
|
|
||||||
function runPromise<A>(f: (service: S.ProviderAuthService.Service) => Effect.Effect<A, S.ProviderAuthError>) {
|
|
||||||
return rt.runPromise(S.ProviderAuthService.use(f))
|
|
||||||
}
|
|
||||||
|
|
||||||
export namespace ProviderAuth {
|
export namespace ProviderAuth {
|
||||||
export const Method = S.Method
|
export const Method = S.Method
|
||||||
export type Method = S.Method
|
export type Method = S.Method
|
||||||
|
|
||||||
export async function methods() {
|
export async function methods() {
|
||||||
return runPromise((service) => service.methods())
|
return runPromiseInstance(S.ProviderAuthService.use((service) => service.methods()))
|
||||||
}
|
}
|
||||||
|
|
||||||
export const Authorization = S.Authorization
|
export const Authorization = S.Authorization
|
||||||
@@ -30,7 +21,8 @@ export namespace ProviderAuth {
|
|||||||
providerID: ProviderID.zod,
|
providerID: ProviderID.zod,
|
||||||
method: z.number(),
|
method: z.number(),
|
||||||
}),
|
}),
|
||||||
async (input): Promise<Authorization | undefined> => runPromise((service) => service.authorize(input)),
|
async (input): Promise<Authorization | undefined> =>
|
||||||
|
runPromiseInstance(S.ProviderAuthService.use((service) => service.authorize(input))),
|
||||||
)
|
)
|
||||||
|
|
||||||
export const callback = fn(
|
export const callback = fn(
|
||||||
@@ -39,15 +31,7 @@ export namespace ProviderAuth {
|
|||||||
method: z.number(),
|
method: z.number(),
|
||||||
code: z.string().optional(),
|
code: z.string().optional(),
|
||||||
}),
|
}),
|
||||||
async (input) => runPromise((service) => service.callback(input)),
|
async (input) => runPromiseInstance(S.ProviderAuthService.use((service) => service.callback(input))),
|
||||||
)
|
|
||||||
|
|
||||||
export const api = fn(
|
|
||||||
z.object({
|
|
||||||
providerID: ProviderID.zod,
|
|
||||||
key: z.string(),
|
|
||||||
}),
|
|
||||||
async (input) => runPromise((service) => service.api(input)),
|
|
||||||
)
|
)
|
||||||
|
|
||||||
export import OauthMissing = S.OauthMissing
|
export import OauthMissing = S.OauthMissing
|
||||||
|
|||||||
@@ -1,13 +1,8 @@
|
|||||||
import { Effect } from "effect"
|
import { runPromiseInstance } from "@/effect/runtime"
|
||||||
import { runtime } from "@/effect/runtime"
|
|
||||||
import * as S from "./service"
|
import * as S from "./service"
|
||||||
import type { QuestionID } from "./schema"
|
import type { QuestionID } from "./schema"
|
||||||
import type { SessionID, MessageID } from "@/session/schema"
|
import type { SessionID, MessageID } from "@/session/schema"
|
||||||
|
|
||||||
function runPromise<A, E>(f: (service: S.QuestionService.Service) => Effect.Effect<A, E>) {
|
|
||||||
return runtime.runPromise(S.QuestionService.use(f))
|
|
||||||
}
|
|
||||||
|
|
||||||
export namespace Question {
|
export namespace Question {
|
||||||
export const Option = S.Option
|
export const Option = S.Option
|
||||||
export type Option = S.Option
|
export type Option = S.Option
|
||||||
@@ -27,18 +22,18 @@ export namespace Question {
|
|||||||
questions: Info[]
|
questions: Info[]
|
||||||
tool?: { messageID: MessageID; callID: string }
|
tool?: { messageID: MessageID; callID: string }
|
||||||
}): Promise<Answer[]> {
|
}): Promise<Answer[]> {
|
||||||
return runPromise((service) => service.ask(input))
|
return runPromiseInstance(S.QuestionService.use((service) => service.ask(input)))
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function reply(input: { requestID: QuestionID; answers: Answer[] }): Promise<void> {
|
export async function reply(input: { requestID: QuestionID; answers: Answer[] }): Promise<void> {
|
||||||
return runPromise((service) => service.reply(input))
|
return runPromiseInstance(S.QuestionService.use((service) => service.reply(input)))
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function reject(requestID: QuestionID): Promise<void> {
|
export async function reject(requestID: QuestionID): Promise<void> {
|
||||||
return runPromise((service) => service.reject(requestID))
|
return runPromiseInstance(S.QuestionService.use((service) => service.reject(requestID)))
|
||||||
}
|
}
|
||||||
|
|
||||||
export async function list(): Promise<Request[]> {
|
export async function list(): Promise<Request[]> {
|
||||||
return runPromise((service) => service.list())
|
return runPromiseInstance(S.QuestionService.use((service) => service.list()))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -2,7 +2,6 @@ import { Deferred, Effect, Layer, Schema, ServiceMap } from "effect"
|
|||||||
import { Bus } from "@/bus"
|
import { Bus } from "@/bus"
|
||||||
import { BusEvent } from "@/bus/bus-event"
|
import { BusEvent } from "@/bus/bus-event"
|
||||||
import { SessionID, MessageID } from "@/session/schema"
|
import { SessionID, MessageID } from "@/session/schema"
|
||||||
import { InstanceState } from "@/util/instance-state"
|
|
||||||
import { Log } from "@/util/log"
|
import { Log } from "@/util/log"
|
||||||
import z from "zod"
|
import z from "zod"
|
||||||
import { QuestionID } from "./schema"
|
import { QuestionID } from "./schema"
|
||||||
@@ -104,18 +103,13 @@ export class QuestionService extends ServiceMap.Service<QuestionService, Questio
|
|||||||
static readonly layer = Layer.effect(
|
static readonly layer = Layer.effect(
|
||||||
QuestionService,
|
QuestionService,
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
const instanceState = yield* InstanceState.make<Map<QuestionID, PendingEntry>>(() =>
|
const pending = new Map<QuestionID, PendingEntry>()
|
||||||
Effect.succeed(new Map<QuestionID, PendingEntry>()),
|
|
||||||
)
|
|
||||||
|
|
||||||
const getPending = InstanceState.get(instanceState)
|
|
||||||
|
|
||||||
const ask = Effect.fn("QuestionService.ask")(function* (input: {
|
const ask = Effect.fn("QuestionService.ask")(function* (input: {
|
||||||
sessionID: SessionID
|
sessionID: SessionID
|
||||||
questions: Info[]
|
questions: Info[]
|
||||||
tool?: { messageID: MessageID; callID: string }
|
tool?: { messageID: MessageID; callID: string }
|
||||||
}) {
|
}) {
|
||||||
const pending = yield* getPending
|
|
||||||
const id = QuestionID.ascending()
|
const id = QuestionID.ascending()
|
||||||
log.info("asking", { id, questions: input.questions.length })
|
log.info("asking", { id, questions: input.questions.length })
|
||||||
|
|
||||||
@@ -138,7 +132,6 @@ export class QuestionService extends ServiceMap.Service<QuestionService, Questio
|
|||||||
})
|
})
|
||||||
|
|
||||||
const reply = Effect.fn("QuestionService.reply")(function* (input: { requestID: QuestionID; answers: Answer[] }) {
|
const reply = Effect.fn("QuestionService.reply")(function* (input: { requestID: QuestionID; answers: Answer[] }) {
|
||||||
const pending = yield* getPending
|
|
||||||
const existing = pending.get(input.requestID)
|
const existing = pending.get(input.requestID)
|
||||||
if (!existing) {
|
if (!existing) {
|
||||||
log.warn("reply for unknown request", { requestID: input.requestID })
|
log.warn("reply for unknown request", { requestID: input.requestID })
|
||||||
@@ -155,7 +148,6 @@ export class QuestionService extends ServiceMap.Service<QuestionService, Questio
|
|||||||
})
|
})
|
||||||
|
|
||||||
const reject = Effect.fn("QuestionService.reject")(function* (requestID: QuestionID) {
|
const reject = Effect.fn("QuestionService.reject")(function* (requestID: QuestionID) {
|
||||||
const pending = yield* getPending
|
|
||||||
const existing = pending.get(requestID)
|
const existing = pending.get(requestID)
|
||||||
if (!existing) {
|
if (!existing) {
|
||||||
log.warn("reject for unknown request", { requestID })
|
log.warn("reject for unknown request", { requestID })
|
||||||
@@ -171,7 +163,6 @@ export class QuestionService extends ServiceMap.Service<QuestionService, Questio
|
|||||||
})
|
})
|
||||||
|
|
||||||
const list = Effect.fn("QuestionService.list")(function* () {
|
const list = Effect.fn("QuestionService.list")(function* () {
|
||||||
const pending = yield* getPending
|
|
||||||
return Array.from(pending.values(), (x) => x.info)
|
return Array.from(pending.values(), (x) => x.info)
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|||||||
@@ -1,63 +0,0 @@
|
|||||||
import { Effect, ScopedCache, Scope } from "effect"
|
|
||||||
|
|
||||||
import { Instance } from "@/project/instance"
|
|
||||||
|
|
||||||
type Disposer = (directory: string) => Effect.Effect<void>
|
|
||||||
const disposers = new Set<Disposer>()
|
|
||||||
|
|
||||||
const TypeId = "~opencode/InstanceState"
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Effect version of `Instance.state` — lazily-initialized, per-directory
|
|
||||||
* cached state for Effect services.
|
|
||||||
*
|
|
||||||
* Values are created on first access for a given directory and cached for
|
|
||||||
* subsequent reads. Concurrent access shares a single initialization —
|
|
||||||
* no duplicate work or races. Use `Effect.acquireRelease` in `init` if
|
|
||||||
* the value needs cleanup on disposal.
|
|
||||||
*/
|
|
||||||
export interface InstanceState<A, E = never, R = never> {
|
|
||||||
readonly [TypeId]: typeof TypeId
|
|
||||||
readonly cache: ScopedCache.ScopedCache<string, A, E, R>
|
|
||||||
}
|
|
||||||
|
|
||||||
export namespace InstanceState {
|
|
||||||
/** Create a new InstanceState with the given initializer. */
|
|
||||||
export const make = <A, E = never, R = never>(
|
|
||||||
init: (directory: string) => Effect.Effect<A, E, R | Scope.Scope>,
|
|
||||||
): Effect.Effect<InstanceState<A, E, Exclude<R, Scope.Scope>>, never, R | Scope.Scope> =>
|
|
||||||
Effect.gen(function* () {
|
|
||||||
const cache = yield* ScopedCache.make<string, A, E, R>({
|
|
||||||
capacity: Number.POSITIVE_INFINITY,
|
|
||||||
lookup: init,
|
|
||||||
})
|
|
||||||
|
|
||||||
const disposer: Disposer = (directory) => ScopedCache.invalidate(cache, directory)
|
|
||||||
disposers.add(disposer)
|
|
||||||
yield* Effect.addFinalizer(() => Effect.sync(() => void disposers.delete(disposer)))
|
|
||||||
|
|
||||||
return {
|
|
||||||
[TypeId]: TypeId,
|
|
||||||
cache,
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
/** Get the cached value for the current directory, initializing it if needed. */
|
|
||||||
export const get = <A, E, R>(self: InstanceState<A, E, R>) =>
|
|
||||||
Effect.suspend(() => ScopedCache.get(self.cache, Instance.directory))
|
|
||||||
|
|
||||||
/** Check whether a value exists for the current directory. */
|
|
||||||
export const has = <A, E, R>(self: InstanceState<A, E, R>) =>
|
|
||||||
Effect.suspend(() => ScopedCache.has(self.cache, Instance.directory))
|
|
||||||
|
|
||||||
/** Invalidate the cached value for the current directory. */
|
|
||||||
export const invalidate = <A, E, R>(self: InstanceState<A, E, R>) =>
|
|
||||||
Effect.suspend(() => ScopedCache.invalidate(self.cache, Instance.directory))
|
|
||||||
|
|
||||||
/** Invalidate the given directory across all InstanceState caches. */
|
|
||||||
export const dispose = (directory: string) =>
|
|
||||||
Effect.all(
|
|
||||||
[...disposers].map((disposer) => disposer(directory)),
|
|
||||||
{ concurrency: "unbounded" },
|
|
||||||
)
|
|
||||||
}
|
|
||||||
@@ -1,7 +1,9 @@
|
|||||||
import { test, expect } from "bun:test"
|
import { afterEach, test, expect } from "bun:test"
|
||||||
import os from "os"
|
import os from "os"
|
||||||
|
import { Effect } from "effect"
|
||||||
import { Bus } from "../../src/bus"
|
import { Bus } from "../../src/bus"
|
||||||
import { runtime } from "../../src/effect/runtime"
|
import { runtime } from "../../src/effect/runtime"
|
||||||
|
import { Instances } from "../../src/effect/instances"
|
||||||
import { PermissionNext } from "../../src/permission/next"
|
import { PermissionNext } from "../../src/permission/next"
|
||||||
import * as S from "../../src/permission/service"
|
import * as S from "../../src/permission/service"
|
||||||
import { PermissionID } from "../../src/permission/schema"
|
import { PermissionID } from "../../src/permission/schema"
|
||||||
@@ -9,6 +11,10 @@ import { Instance } from "../../src/project/instance"
|
|||||||
import { tmpdir } from "../fixture/fixture"
|
import { tmpdir } from "../fixture/fixture"
|
||||||
import { MessageID, SessionID } from "../../src/session/schema"
|
import { MessageID, SessionID } from "../../src/session/schema"
|
||||||
|
|
||||||
|
afterEach(async () => {
|
||||||
|
await Instance.disposeAll()
|
||||||
|
})
|
||||||
|
|
||||||
async function rejectAll(message?: string) {
|
async function rejectAll(message?: string) {
|
||||||
for (const req of await PermissionNext.list()) {
|
for (const req of await PermissionNext.list()) {
|
||||||
await PermissionNext.reply({
|
await PermissionNext.reply({
|
||||||
@@ -1020,7 +1026,7 @@ test("ask - abort should clear pending request", async () => {
|
|||||||
always: [],
|
always: [],
|
||||||
ruleset: [{ permission: "bash", pattern: "*", action: "ask" }],
|
ruleset: [{ permission: "bash", pattern: "*", action: "ask" }],
|
||||||
}),
|
}),
|
||||||
),
|
).pipe(Effect.provide(Instances.get(Instance.directory))),
|
||||||
{ signal: ctl.signal },
|
{ signal: ctl.signal },
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -1,20 +0,0 @@
|
|||||||
import { afterEach, expect, test } from "bun:test"
|
|
||||||
import { Auth } from "../../src/auth"
|
|
||||||
import { ProviderAuth } from "../../src/provider/auth"
|
|
||||||
import { ProviderID } from "../../src/provider/schema"
|
|
||||||
|
|
||||||
afterEach(async () => {
|
|
||||||
await Auth.remove("test-provider-auth")
|
|
||||||
})
|
|
||||||
|
|
||||||
test("ProviderAuth.api persists auth via AuthService", async () => {
|
|
||||||
await ProviderAuth.api({
|
|
||||||
providerID: ProviderID.make("test-provider-auth"),
|
|
||||||
key: "sk-test",
|
|
||||||
})
|
|
||||||
|
|
||||||
expect(await Auth.get("test-provider-auth")).toEqual({
|
|
||||||
type: "api",
|
|
||||||
key: "sk-test",
|
|
||||||
})
|
|
||||||
})
|
|
||||||
@@ -1,10 +1,14 @@
|
|||||||
import { test, expect } from "bun:test"
|
import { afterEach, test, expect } from "bun:test"
|
||||||
import { Question } from "../../src/question"
|
import { Question } from "../../src/question"
|
||||||
import { Instance } from "../../src/project/instance"
|
import { Instance } from "../../src/project/instance"
|
||||||
import { QuestionID } from "../../src/question/schema"
|
import { QuestionID } from "../../src/question/schema"
|
||||||
import { tmpdir } from "../fixture/fixture"
|
import { tmpdir } from "../fixture/fixture"
|
||||||
import { SessionID } from "../../src/session/schema"
|
import { SessionID } from "../../src/session/schema"
|
||||||
|
|
||||||
|
afterEach(async () => {
|
||||||
|
await Instance.disposeAll()
|
||||||
|
})
|
||||||
|
|
||||||
/** Reject all pending questions so dangling Deferred fibers don't hang the test. */
|
/** Reject all pending questions so dangling Deferred fibers don't hang the test. */
|
||||||
async function rejectAll() {
|
async function rejectAll() {
|
||||||
const pending = await Question.list()
|
const pending = await Question.list()
|
||||||
|
|||||||
@@ -1,261 +0,0 @@
|
|||||||
import { afterEach, expect, test } from "bun:test"
|
|
||||||
import { Duration, Effect, Layer, ManagedRuntime, ServiceMap } from "effect"
|
|
||||||
|
|
||||||
import { Instance } from "../../src/project/instance"
|
|
||||||
import { InstanceState } from "../../src/util/instance-state"
|
|
||||||
import { tmpdir } from "../fixture/fixture"
|
|
||||||
|
|
||||||
async function access<A, E>(state: InstanceState<A, E>, dir: string) {
|
|
||||||
return Instance.provide({
|
|
||||||
directory: dir,
|
|
||||||
fn: () => Effect.runPromise(InstanceState.get(state)),
|
|
||||||
})
|
|
||||||
}
|
|
||||||
|
|
||||||
afterEach(async () => {
|
|
||||||
await Instance.disposeAll()
|
|
||||||
})
|
|
||||||
|
|
||||||
test("InstanceState caches values for the same instance", async () => {
|
|
||||||
await using tmp = await tmpdir()
|
|
||||||
let n = 0
|
|
||||||
|
|
||||||
await Effect.runPromise(
|
|
||||||
Effect.scoped(
|
|
||||||
Effect.gen(function* () {
|
|
||||||
const state = yield* InstanceState.make(() => Effect.sync(() => ({ n: ++n })))
|
|
||||||
|
|
||||||
const a = yield* Effect.promise(() => access(state, tmp.path))
|
|
||||||
const b = yield* Effect.promise(() => access(state, tmp.path))
|
|
||||||
|
|
||||||
expect(a).toBe(b)
|
|
||||||
expect(n).toBe(1)
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
})
|
|
||||||
|
|
||||||
test("InstanceState isolates values by directory", async () => {
|
|
||||||
await using a = await tmpdir()
|
|
||||||
await using b = await tmpdir()
|
|
||||||
let n = 0
|
|
||||||
|
|
||||||
await Effect.runPromise(
|
|
||||||
Effect.scoped(
|
|
||||||
Effect.gen(function* () {
|
|
||||||
const state = yield* InstanceState.make((dir) => Effect.sync(() => ({ dir, n: ++n })))
|
|
||||||
|
|
||||||
const x = yield* Effect.promise(() => access(state, a.path))
|
|
||||||
const y = yield* Effect.promise(() => access(state, b.path))
|
|
||||||
const z = yield* Effect.promise(() => access(state, a.path))
|
|
||||||
|
|
||||||
expect(x).toBe(z)
|
|
||||||
expect(x).not.toBe(y)
|
|
||||||
expect(n).toBe(2)
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
})
|
|
||||||
|
|
||||||
test("InstanceState is disposed on instance reload", async () => {
|
|
||||||
await using tmp = await tmpdir()
|
|
||||||
const seen: string[] = []
|
|
||||||
let n = 0
|
|
||||||
|
|
||||||
await Effect.runPromise(
|
|
||||||
Effect.scoped(
|
|
||||||
Effect.gen(function* () {
|
|
||||||
const state = yield* InstanceState.make(() =>
|
|
||||||
Effect.acquireRelease(
|
|
||||||
Effect.sync(() => ({ n: ++n })),
|
|
||||||
(value) =>
|
|
||||||
Effect.sync(() => {
|
|
||||||
seen.push(String(value.n))
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
|
|
||||||
const a = yield* Effect.promise(() => access(state, tmp.path))
|
|
||||||
yield* Effect.promise(() => Instance.reload({ directory: tmp.path }))
|
|
||||||
const b = yield* Effect.promise(() => access(state, tmp.path))
|
|
||||||
|
|
||||||
expect(a).not.toBe(b)
|
|
||||||
expect(seen).toEqual(["1"])
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
})
|
|
||||||
|
|
||||||
test("InstanceState is disposed on disposeAll", async () => {
|
|
||||||
await using a = await tmpdir()
|
|
||||||
await using b = await tmpdir()
|
|
||||||
const seen: string[] = []
|
|
||||||
|
|
||||||
await Effect.runPromise(
|
|
||||||
Effect.scoped(
|
|
||||||
Effect.gen(function* () {
|
|
||||||
const state = yield* InstanceState.make((dir) =>
|
|
||||||
Effect.acquireRelease(
|
|
||||||
Effect.sync(() => ({ dir })),
|
|
||||||
(value) =>
|
|
||||||
Effect.sync(() => {
|
|
||||||
seen.push(value.dir)
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
|
|
||||||
yield* Effect.promise(() => access(state, a.path))
|
|
||||||
yield* Effect.promise(() => access(state, b.path))
|
|
||||||
yield* Effect.promise(() => Instance.disposeAll())
|
|
||||||
|
|
||||||
expect(seen.sort()).toEqual([a.path, b.path].sort())
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
})
|
|
||||||
|
|
||||||
test("InstanceState.get reads correct directory per-evaluation (not captured once)", async () => {
|
|
||||||
await using a = await tmpdir()
|
|
||||||
await using b = await tmpdir()
|
|
||||||
|
|
||||||
// Regression: InstanceState.get must be lazy (Effect.suspend) so the
|
|
||||||
// directory is read per-evaluation, not captured once at the call site.
|
|
||||||
// Without this, a service built inside a ManagedRuntime Layer would
|
|
||||||
// freeze to whichever directory triggered the first layer build.
|
|
||||||
|
|
||||||
interface TestApi {
|
|
||||||
readonly getDir: () => Effect.Effect<string>
|
|
||||||
}
|
|
||||||
|
|
||||||
class TestService extends ServiceMap.Service<TestService, TestApi>()("@test/ALS-lazy") {
|
|
||||||
static readonly layer = Layer.effect(
|
|
||||||
TestService,
|
|
||||||
Effect.gen(function* () {
|
|
||||||
const state = yield* InstanceState.make((dir) => Effect.sync(() => dir))
|
|
||||||
// `get` is created once during layer build — must be lazy
|
|
||||||
const get = InstanceState.get(state)
|
|
||||||
|
|
||||||
const getDir = Effect.fn("TestService.getDir")(function* () {
|
|
||||||
return yield* get
|
|
||||||
})
|
|
||||||
|
|
||||||
return TestService.of({ getDir })
|
|
||||||
}),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
const rt = ManagedRuntime.make(TestService.layer)
|
|
||||||
|
|
||||||
try {
|
|
||||||
const resultA = await Instance.provide({
|
|
||||||
directory: a.path,
|
|
||||||
fn: () => rt.runPromise(TestService.use((s) => s.getDir())),
|
|
||||||
})
|
|
||||||
expect(resultA).toBe(a.path)
|
|
||||||
|
|
||||||
// Second call with different directory must NOT return A's directory
|
|
||||||
const resultB = await Instance.provide({
|
|
||||||
directory: b.path,
|
|
||||||
fn: () => rt.runPromise(TestService.use((s) => s.getDir())),
|
|
||||||
})
|
|
||||||
expect(resultB).toBe(b.path)
|
|
||||||
} finally {
|
|
||||||
await rt.dispose()
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
test("InstanceState.get isolates concurrent fibers across real delays, yields, and timer callbacks", async () => {
|
|
||||||
await using a = await tmpdir()
|
|
||||||
await using b = await tmpdir()
|
|
||||||
await using c = await tmpdir()
|
|
||||||
|
|
||||||
// Adversarial: concurrent fibers with real timer delays (macrotask
|
|
||||||
// boundaries via setTimeout/Bun.sleep), explicit scheduler yields,
|
|
||||||
// and many async steps. If ALS context leaks or gets lost at any
|
|
||||||
// point, a fiber will see the wrong directory.
|
|
||||||
|
|
||||||
interface TestApi {
|
|
||||||
readonly getDir: () => Effect.Effect<string>
|
|
||||||
}
|
|
||||||
|
|
||||||
class TestService extends ServiceMap.Service<TestService, TestApi>()("@test/ALS-adversarial") {
|
|
||||||
static readonly layer = Layer.effect(
|
|
||||||
TestService,
|
|
||||||
Effect.gen(function* () {
|
|
||||||
const state = yield* InstanceState.make((dir) => Effect.sync(() => dir))
|
|
||||||
|
|
||||||
const getDir = Effect.fn("TestService.getDir")(function* () {
|
|
||||||
// Mix of async boundary types to maximise interleaving:
|
|
||||||
// 1. Real timer delay (macrotask — setTimeout under the hood)
|
|
||||||
yield* Effect.promise(() => Bun.sleep(1))
|
|
||||||
// 2. Effect.sleep (Effect's own timer, uses its internal scheduler)
|
|
||||||
yield* Effect.sleep(Duration.millis(1))
|
|
||||||
// 3. Explicit scheduler yields
|
|
||||||
for (let i = 0; i < 100; i++) {
|
|
||||||
yield* Effect.yieldNow
|
|
||||||
}
|
|
||||||
// 4. Microtask boundaries
|
|
||||||
for (let i = 0; i < 100; i++) {
|
|
||||||
yield* Effect.promise(() => Promise.resolve())
|
|
||||||
}
|
|
||||||
// 5. Another Effect.sleep
|
|
||||||
yield* Effect.sleep(Duration.millis(2))
|
|
||||||
// 6. Another real timer to force a second macrotask hop
|
|
||||||
yield* Effect.promise(() => Bun.sleep(1))
|
|
||||||
// NOW read the directory — ALS must still be correct
|
|
||||||
return yield* InstanceState.get(state)
|
|
||||||
})
|
|
||||||
|
|
||||||
return TestService.of({ getDir })
|
|
||||||
}),
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
const rt = ManagedRuntime.make(TestService.layer)
|
|
||||||
|
|
||||||
try {
|
|
||||||
const [resultA, resultB, resultC] = await Promise.all([
|
|
||||||
Instance.provide({
|
|
||||||
directory: a.path,
|
|
||||||
fn: () => rt.runPromise(TestService.use((s) => s.getDir())),
|
|
||||||
}),
|
|
||||||
Instance.provide({
|
|
||||||
directory: b.path,
|
|
||||||
fn: () => rt.runPromise(TestService.use((s) => s.getDir())),
|
|
||||||
}),
|
|
||||||
Instance.provide({
|
|
||||||
directory: c.path,
|
|
||||||
fn: () => rt.runPromise(TestService.use((s) => s.getDir())),
|
|
||||||
}),
|
|
||||||
])
|
|
||||||
|
|
||||||
expect(resultA).toBe(a.path)
|
|
||||||
expect(resultB).toBe(b.path)
|
|
||||||
expect(resultC).toBe(c.path)
|
|
||||||
} finally {
|
|
||||||
await rt.dispose()
|
|
||||||
}
|
|
||||||
})
|
|
||||||
|
|
||||||
test("InstanceState dedupes concurrent lookups for the same directory", async () => {
|
|
||||||
await using tmp = await tmpdir()
|
|
||||||
let n = 0
|
|
||||||
|
|
||||||
await Effect.runPromise(
|
|
||||||
Effect.scoped(
|
|
||||||
Effect.gen(function* () {
|
|
||||||
const state = yield* InstanceState.make(() =>
|
|
||||||
Effect.promise(async () => {
|
|
||||||
n += 1
|
|
||||||
await Bun.sleep(10)
|
|
||||||
return { n }
|
|
||||||
}),
|
|
||||||
)
|
|
||||||
|
|
||||||
const [a, b] = yield* Effect.promise(() => Promise.all([access(state, tmp.path), access(state, tmp.path)]))
|
|
||||||
expect(a).toBe(b)
|
|
||||||
expect(n).toBe(1)
|
|
||||||
}),
|
|
||||||
),
|
|
||||||
)
|
|
||||||
})
|
|
||||||
Reference in New Issue
Block a user