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

refactor(core): simplify integration credentials (#31968)

Browse source
This commit is contained in:
Dax 2026-06-12 02:15:25 -04:00 • committed by GitHub
commit 30aec297d8
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
112 changed files with 2467 additions and 47728 deletions
Download patch file Download diff file Expand all files Collapse all files

12
packages/core/src/catalog.ts
View file

@ -1,6 +1,6 @@
export * as Catalog from "./catalog"
import { Context, Effect, Layer, Option, Order, pipe, Schema, Array, Scope, Stream } from "effect"
import { Array, Context, Effect, Layer, Option, Order, pipe, Schema, Scope, Stream } from "effect"
import { castDraft, enableMapSet, type Draft } from "immer"
import { ModelV2 } from "./model"
import { ModelRequest } from "./model-request"
@ -11,7 +11,7 @@ import { EventV2 } from "./event"
import { Policy } from "./policy"
import { State } from "./state"
import { Credential } from "./credential"
import { ConnectorSchema } from "./connector/schema"
import { IntegrationSchema } from "./integration/schema"
export type ProviderRecord = {
provider: ProviderV2.Info
@ -99,8 +99,8 @@ export const layer = Layer.effect(
const credentials = yield* Credential.Service
const scope = yield* Scope.Scope
const project = (provider: ProviderV2.Info, active: Map<ConnectorSchema.ID, Credential.Info>) => {
const credential = active.get(ConnectorSchema.ID.make(provider.id))
const project = (provider: ProviderV2.Info, active: Map<IntegrationSchema.ID, Credential.Stored>) => {
const credential = active.get(IntegrationSchema.ID.make(provider.id))
if (!credential) return provider
const body = { ...provider.request.body }
if (credential.value.type === "key") {
@ -211,7 +211,9 @@ export const layer = Layer.effect(
}
}),
})
const active = () => credentials.activeAll().pipe(Effect.orDie)
const active = Effect.fn("CatalogV2.active")(function* () {
return new Map((yield* credentials.all()).map((credential) => [credential.integrationID, credential]))
})
yield* events.subscribe(PluginV2.Event.Added).pipe(
// Plugin registries are location scoped even though the event bus is process scoped.

492
packages/core/src/connector.ts
View file

@ -1,492 +0,0 @@
export * as Connector from "./connector"
import { Cause, Clock, Context, Duration, Effect, Exit, Layer, Schedule, Schema, Scope, SynchronizedRef } from "effect"
import { castDraft, enableMapSet, type Draft } from "immer"
import { Credential } from "./credential"
import { ConnectorSchema } from "./connector/schema"
import { withStatics } from "./schema"
import { State } from "./state"
import { Identifier } from "./util/identifier"
import { KeyedMutex } from "./effect/keyed-mutex"
import { EventV2 } from "./event"
export const ID = ConnectorSchema.ID
export type ID = ConnectorSchema.ID
export const MethodID = ConnectorSchema.MethodID
export type MethodID = ConnectorSchema.MethodID
export const AttemptID = Schema.String.pipe(
Schema.brand("Connector.AttemptID"),
withStatics((schema) => ({ create: () => schema.make("con_" + Identifier.ascending()) })),
)
export type AttemptID = typeof AttemptID.Type
export const When = Schema.Struct({
key: Schema.String,
op: Schema.Literals(["eq", "neq"]),
value: Schema.String,
}).annotate({ identifier: "Connector.When" })
export type When = typeof When.Type
export class TextPrompt extends Schema.Class<TextPrompt>("Connector.TextPrompt")({
type: Schema.Literal("text"),
key: Schema.String,
message: Schema.String,
placeholder: Schema.optional(Schema.String),
when: Schema.optional(When),
}) {}
export class SelectPrompt extends Schema.Class<SelectPrompt>("Connector.SelectPrompt")({
type: Schema.Literal("select"),
key: Schema.String,
message: Schema.String,
options: Schema.Array(
Schema.Struct({
label: Schema.String,
value: Schema.String,
hint: Schema.optional(Schema.String),
}),
),
when: Schema.optional(When),
}) {}
export const Prompt = Schema.Union([TextPrompt, SelectPrompt]).pipe(Schema.toTaggedUnion("type"))
export type Prompt = typeof Prompt.Type
export class OAuthMethod extends Schema.Class<OAuthMethod>("Connector.OAuthMethod")({
id: MethodID,
type: Schema.Literal("oauth"),
label: Schema.String,
prompts: Schema.optional(Schema.Array(Prompt)),
}) {}
export class KeyMethod extends Schema.Class<KeyMethod>("Connector.KeyMethod")({
id: MethodID,
type: Schema.Literal("key"),
label: Schema.String,
prompts: Schema.optional(Schema.Array(Prompt)),
}) {}
export const Method = Schema.Union([OAuthMethod, KeyMethod]).pipe(Schema.toTaggedUnion("type"))
export type Method = typeof Method.Type
export class Info extends Schema.Class<Info>("Connector.Info")({
id: ID,
name: Schema.String,
methods: Schema.Array(Method),
}) {}
export type Inputs = Readonly<{ [key: string]: string }>
export type OAuthAuthorization = {
readonly url: string
readonly instructions: string
} & (
| {
readonly mode: "auto"
readonly callback: Effect.Effect<Credential.Value, unknown>
}
| {
readonly mode: "code"
readonly callback: (code: string) => Effect.Effect<Credential.Value, unknown>
}
)
export interface OAuthImplementation {
readonly connectorID: ID
readonly method: OAuthMethod
readonly authorize: (inputs: Inputs) => Effect.Effect<OAuthAuthorization, unknown, Scope.Scope>
readonly refresh?: (credential: Credential.OAuth) => Effect.Effect<Credential.OAuth, unknown>
}
export interface KeyImplementation {
readonly connectorID: ID
readonly method: KeyMethod
readonly authorize: (key: string, inputs: Inputs) => Effect.Effect<Credential.Key, unknown>
}
export type Implementation = OAuthImplementation | KeyImplementation
function isKeyImplementation(implementation: Implementation): implementation is KeyImplementation {
return implementation.method.type === "key"
}
function isOAuthImplementation(implementation: Implementation): implementation is OAuthImplementation {
return implementation.method.type === "oauth"
}
export class Attempt extends Schema.Class<Attempt>("Connector.Attempt")({
attemptID: AttemptID,
url: Schema.String,
instructions: Schema.String,
mode: Schema.Literals(["auto", "code"]),
time: Schema.Struct({
created: Schema.Number,
expires: Schema.Number,
}),
}) {}
const Time = Schema.Struct({
created: Schema.Number,
expires: Schema.Number,
})
export const AttemptStatus = Schema.Union([
Schema.Struct({ status: Schema.Literal("pending"), time: Time }),
Schema.Struct({ status: Schema.Literal("complete"), time: Time }),
Schema.Struct({ status: Schema.Literal("failed"), message: Schema.String, time: Time }),
Schema.Struct({ status: Schema.Literal("expired"), time: Time }),
]).pipe(Schema.toTaggedUnion("status"))
export type AttemptStatus = typeof AttemptStatus.Type
export class CodeRequiredError extends Schema.TaggedErrorClass<CodeRequiredError>()("Connector.CodeRequired", {
attemptID: AttemptID,
}) {}
export class AuthorizationError extends Schema.TaggedErrorClass<AuthorizationError>()("Connector.Authorization", {
cause: Schema.Defect,
}) {}
export type Error = CodeRequiredError | AuthorizationError
export const Event = {
Updated: EventV2.define({
type: "connector.updated",
schema: {},
}),
}
type Entry = {
connector: Info
implementations: Map<MethodID, Implementation>
}
type Data = {
connectors: Map<ID, Entry>
}
export type Editor = {
list: () => readonly Info[]
get: (id: ID) => Info | undefined
update: (id: ID, update: (connector: Draft<Omit<Info, "methods">>) => void) => void
remove: (id: ID) => void
method: {
update: (implementation: Implementation) => void
remove: (connectorID: ID, methodID: MethodID) => void
}
}
export interface Interface {
/** Registers a scoped transform over the connector registry. */
readonly transform: State.Interface<Data, Editor>["transform"]
/** Registers and immediately applies a scoped connector registry update. */
readonly update: State.Interface<Data, Editor>["update"]
/** Returns one connector with its serializable login methods. */
readonly get: (id: ID) => Effect.Effect<Info | undefined>
/** Returns all connectors with their serializable login methods. */
readonly list: () => Effect.Effect<Info[]>
/** Refreshes an OAuth credential with its originating method. */
readonly refresh: (credentialID: Credential.ID) => Effect.Effect<void, AuthorizationError>
readonly connect: {
/** Runs a key method and stores the resulting credential. */
readonly key: (input: {
/** Connector receiving the credential. */
readonly connectorID: ID
/** Key method selected by the caller. */
readonly methodID: MethodID
/** Secret entered by the user. */
readonly key: string
/** Answers to the method's optional prompts. */
readonly inputs: Inputs
/** User-facing label for the stored credential. */
readonly label?: string
}) => Effect.Effect<void, AuthorizationError>
readonly oauth: {
/** Starts a stateful OAuth attempt. */
readonly begin: (input: {
/** Connector being authenticated. */
readonly connectorID: ID
/** OAuth method selected by the caller. */
readonly methodID: MethodID
/** Answers to the method's optional prompts. */
readonly inputs: Inputs
/** User-facing label for the credential created on completion. */
readonly label?: string
}) => Effect.Effect<Attempt, AuthorizationError>
/** Returns the current state of an OAuth attempt. */
readonly status: (attemptID: AttemptID) => Effect.Effect<AttemptStatus>
/** Completes the attempt and stores its credential. */
readonly complete: (input: {
/** Opaque handle returned by `begin`. */
readonly attemptID: AttemptID
/** Authorization code required by attempts in code mode. */
readonly code?: string
}) => Effect.Effect<void, CodeRequiredError | AuthorizationError>
/** Cancels an attempt and releases its resources. */
readonly cancel: (attemptID: AttemptID) => Effect.Effect<void>
}
}
}
export class Service extends Context.Service<Service, Interface>()("@opencode/v2/Connector") {}
enableMapSet()
const attemptLifetime = Duration.toMillis(Duration.minutes(10))
const terminalRetention = Duration.toMillis(Duration.minutes(1))
const scrubInterval = Duration.seconds(30)
type AttemptTime = { created: number; expires: number }
type PendingAttempt = {
status: "pending"
completing: boolean
authorization: OAuthAuthorization
connectorID: ID
methodID: MethodID
label?: string
scope: Scope.Closeable
time: AttemptTime
}
type TerminalAttempt = {
status: "complete" | "failed" | "expired"
message?: string
removeAt: number
time: AttemptTime
}
type AttemptEntry = PendingAttempt | TerminalAttempt
export const locationLayer = Layer.effect(
Service,
Effect.gen(function* () {
const credentials = yield* Credential.Service
const events = yield* EventV2.Service
const scope = yield* Scope.Scope
const attempts = SynchronizedRef.makeUnsafe(new Map<AttemptID, AttemptEntry>())
const refreshLocks = KeyedMutex.makeUnsafe<Credential.ID>()
const state = State.create<Data, Editor>({
initial: () => ({ connectors: new Map<ID, Entry>() }),
editor: (draft) => ({
list: () => Array.from(draft.connectors.values(), (entry) => entry.connector) as Info[],
get: (id) => draft.connectors.get(id)?.connector as Info | undefined,
update: (id, update) => {
const current =
draft.connectors.get(id) ??
castDraft({ connector: new Info({ id, name: id, methods: [] }), implementations: new Map() })
if (!draft.connectors.has(id)) draft.connectors.set(id, current)
update(current.connector)
current.connector.id = id
},
remove: (id) => draft.connectors.delete(id),
method: {
update: (implementation) => {
const current =
draft.connectors.get(implementation.connectorID) ??
castDraft({
connector: new Info({ id: implementation.connectorID, name: implementation.connectorID, methods: [] }),
implementations: new Map<MethodID, Implementation>(),
})
if (!draft.connectors.has(implementation.connectorID)) {
draft.connectors.set(implementation.connectorID, current)
}
const index = current.connector.methods.findIndex((method) => method.id === implementation.method.id)
if (index === -1) current.connector.methods.push(castDraft(implementation.method))
else current.connector.methods[index] = castDraft(implementation.method)
current.implementations.set(implementation.method.id, castDraft(implementation))
},
remove: (connectorID, methodID) => {
const current = draft.connectors.get(connectorID)
if (!current) return
const index = current.connector.methods.findIndex((method) => method.id === methodID)
if (index !== -1) current.connector.methods.splice(index, 1)
current.implementations.delete(methodID)
},
},
}),
finalize: () => events.publish(Event.Updated, {}).pipe(Effect.asVoid),
})
const authorize = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
effect.pipe(Effect.mapError((cause) => new AuthorizationError({ cause })))
const close = (attemptScope: Scope.Closeable) =>
Scope.close(attemptScope, Exit.void).pipe(Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid)
const message = (cause: Cause.Cause<unknown>) => {
const error = Cause.squash(cause)
return error instanceof Error ? error.message : String(error)
}
const settle = Effect.fnUntraced(function* (attemptID: AttemptID, exit: Exit.Exit<Credential.Value, unknown>) {
const now = yield* Clock.currentTimeMillis
const result = yield* SynchronizedRef.modify(attempts, (current) => {
const attempt = current.get(attemptID)
if (!attempt || attempt.status !== "pending") return [undefined, current]
const terminal: TerminalAttempt = Exit.isSuccess(exit)
? { status: "complete", time: attempt.time, removeAt: now + terminalRetention }
: { status: "failed", message: message(exit.cause), time: attempt.time, removeAt: now + terminalRetention }
return [attempt, new Map(current).set(attemptID, terminal)]
})
if (!result) return
if (Exit.isSuccess(exit)) {
yield* credentials.create({
connectorID: result.connectorID,
methodID: result.methodID,
label: result.label,
value: exit.value,
})
}
yield* close(result.scope)
})
const scrub = Effect.fnUntraced(function* () {
const now = yield* Clock.currentTimeMillis
const expired = yield* SynchronizedRef.modify(attempts, (current) => {
const next = new Map(current)
const scopes: Scope.Closeable[] = []
for (const [id, attempt] of current) {
if (attempt.status === "pending" && attempt.time.expires <= now) {
scopes.push(attempt.scope)
next.set(id, { status: "expired", time: attempt.time, removeAt: now + terminalRetention })
continue
}
if (attempt.status !== "pending" && attempt.removeAt <= now) next.delete(id)
}
return [scopes, next]
})
yield* Effect.forEach(expired, close, { discard: true })
})
yield* scrub().pipe(Effect.repeat(Schedule.spaced(scrubInterval)), Effect.forkIn(scope))
return Service.of({
transform: state.transform,
update: state.update,
get: Effect.fn("Connector.get")(function* (id) {
return state.get().connectors.get(id)?.connector
}),
list: Effect.fn("Connector.list")(function* () {
return Array.from(state.get().connectors.values(), (record) => record.connector).toSorted((a, b) =>
a.name.localeCompare(b.name),
)
}),
refresh: Effect.fn("Connector.refresh")(function* (credentialID) {
yield* refreshLocks.withLock(credentialID)(
Effect.gen(function* () {
const credential = yield* credentials.get(credentialID)
if (!credential || credential.value.type !== "oauth") {
return yield* Effect.die(`OAuth credential not found: ${credentialID}`)
}
const implementation = state
.get()
.connectors.get(credential.connectorID)
?.implementations.get(credential.methodID)
if (!implementation || !isOAuthImplementation(implementation) || !implementation.refresh) {
return yield* Effect.die(
`OAuth refresh method not found: ${credential.connectorID}/${credential.methodID}`,
)
}
const value = yield* authorize(implementation.refresh(credential.value))
yield* credentials.update(credential.id, { value })
}),
)
}),
connect: {
key: Effect.fn("Connector.connect.key")(function* (input) {
const method = state.get().connectors.get(input.connectorID)?.implementations.get(input.methodID)
if (!method || !isKeyImplementation(method)) {
return yield* Effect.die(`Key method not found: ${input.connectorID}/${input.methodID}`)
}
const value = yield* authorize(method.authorize(input.key, input.inputs))
yield* credentials.create({
connectorID: input.connectorID,
methodID: input.methodID,
label: input.label,
value,
})
}),
oauth: {
begin: Effect.fn("Connector.connect.oauth.begin")(function* (input) {
const method = state.get().connectors.get(input.connectorID)?.implementations.get(input.methodID)
if (!method || !isOAuthImplementation(method)) {
return yield* Effect.die(`OAuth method not found: ${input.connectorID}/${input.methodID}`)
}
const attemptScope = yield* Scope.fork(scope)
const authorization = yield* authorize(method.authorize(input.inputs)).pipe(
Scope.provide(attemptScope),
Effect.onExit((exit) => (Exit.isFailure(exit) ? Scope.close(attemptScope, exit) : Effect.void)),
)
const id = AttemptID.create()
const created = yield* Clock.currentTimeMillis
const time = { created, expires: created + attemptLifetime }
yield* SynchronizedRef.update(attempts, (current) =>
new Map(current).set(id, {
status: "pending",
completing: authorization.mode === "auto",
authorization,
connectorID: input.connectorID,
methodID: input.methodID,
label: input.label,
scope: attemptScope,
time,
}),
)
if (authorization.mode === "auto") {
yield* authorization.callback.pipe(
Effect.exit,
Effect.flatMap((exit) => settle(id, exit)),
Effect.forkIn(attemptScope, { startImmediately: true }),
)
}
return new Attempt({
attemptID: id,
url: authorization.url,
instructions: authorization.instructions,
mode: authorization.mode,
time,
})
}),
status: Effect.fn("Connector.connect.oauth.status")(function* (attemptID) {
const attempt = (yield* SynchronizedRef.get(attempts)).get(attemptID)
if (!attempt) return yield* Effect.die(`OAuth attempt not found: ${attemptID}`)
if (attempt.status === "failed") {
return { status: attempt.status, message: attempt.message ?? "Authorization failed", time: attempt.time }
}
return { status: attempt.status, time: attempt.time }
}),
complete: Effect.fn("Connector.connect.oauth.complete")(function* (input) {
const attempt = yield* SynchronizedRef.modify(attempts, (current) => {
const match = current.get(input.attemptID)
if (!match || match.status !== "pending" || match.completing) return [match, current]
if (match.authorization.mode === "code" && input.code === undefined) return [match, current]
return [match, new Map(current).set(input.attemptID, { ...match, completing: true })]
})
if (!attempt) return yield* Effect.die(`OAuth attempt not found: ${input.attemptID}`)
if (attempt.status !== "pending") return
if (attempt.authorization.mode === "code" && input.code === undefined) {
return yield* new CodeRequiredError({ attemptID: input.attemptID })
}
if (attempt.completing) return yield* Effect.die(`OAuth attempt already completing: ${input.attemptID}`)
const callback =
attempt.authorization.mode === "auto"
? attempt.authorization.callback
: attempt.authorization.callback(input.code as string)
const exit = yield* authorize(callback).pipe(Effect.exit)
yield* settle(input.attemptID, exit)
if (Exit.isFailure(exit)) return yield* exit
}),
cancel: Effect.fn("Connector.connect.oauth.cancel")(function* (attemptID) {
const attempt = yield* SynchronizedRef.modify(attempts, (current) => {
const match = current.get(attemptID)
if (!match || match.status !== "pending") return [undefined, current]
const next = new Map(current)
next.delete(attemptID)
return [match, next]
})
if (attempt) yield* Scope.close(attempt.scope, Exit.void)
}),
},
},
})
}),
)

9
packages/core/src/connector/schema.ts
View file

@ -1,9 +0,0 @@
export * as ConnectorSchema from "./schema"
import { Schema } from "effect"
export const ID = Schema.String.pipe(Schema.brand("Connector.ID"))
export type ID = typeof ID.Type
export const MethodID = Schema.String.pipe(Schema.brand("Connector.MethodID"))
export type MethodID = typeof MethodID.Type

285
packages/core/src/credential.ts
View file

@ -1,17 +1,12 @@
export * as Credential from "./credential"
import { and, asc, eq, ne } from "drizzle-orm"
import { Context, Effect, Layer, Option, Schema } from "effect"
import { asc, eq } from "drizzle-orm"
import { Context, Effect, Layer, Schema } from "effect"
import { Database } from "./database/database"
import { ConnectorSchema } from "./connector/schema"
import { EventV2 } from "./event"
import { IntegrationSchema } from "./integration/schema"
import { NonNegativeInt, withStatics } from "./schema"
import { CredentialTable } from "./credential/sql"
import { Identifier } from "./util/identifier"
import { FSUtil } from "./fs-util"
import { Global } from "./global"
import { DataMigrationTable } from "./data-migration.sql"
import path from "path"
import { CredentialTable } from "./credential/sql"
export const ID = Schema.String.pipe(
Schema.brand("Credential.ID"),
@ -21,6 +16,7 @@ export type ID = typeof ID.Type
export class OAuth extends Schema.Class<OAuth>("Credential.OAuth")({
type: Schema.Literal("oauth"),
methodID: IntegrationSchema.MethodID,
refresh: Schema.String,
access: Schema.String,
expires: NonNegativeInt,
@ -33,255 +29,94 @@ export class Key extends Schema.Class<Key>("Credential.Key")({
metadata: Schema.optional(Schema.Record(Schema.String, Schema.String)),
}) {}
export const Value = Schema.Union([OAuth, Key])
export const Info = Schema.Union([OAuth, Key])
.pipe(Schema.toTaggedUnion("type"))
.annotate({ identifier: "Credential.Value" })
export type Value = Schema.Schema.Type<typeof Value>
.annotate({ identifier: "Credential.Info" })
export type Info = Schema.Schema.Type<typeof Info>
const LegacyOAuth = Schema.Struct({
type: Schema.Literal("oauth"),
refresh: Schema.String,
access: Schema.String,
expires: NonNegativeInt,
accountId: Schema.optional(Schema.String),
enterpriseUrl: Schema.optional(Schema.String),
})
const LegacyKey = Schema.Struct({
type: Schema.Literal("api"),
key: Schema.String,
metadata: Schema.optional(Schema.Record(Schema.String, Schema.String)),
})
const LegacyValue = Schema.Union([LegacyOAuth, LegacyKey])
export class Info extends Schema.Class<Info>("Credential.Info")({
export class Stored extends Schema.Class<Stored>("Credential.Stored")({
id: ID,
connectorID: ConnectorSchema.ID,
methodID: ConnectorSchema.MethodID,
integrationID: IntegrationSchema.ID,
label: Schema.String,
value: Value,
value: Info,
}) {}
export const Event = {
Added: EventV2.define({
type: "credential.added",
schema: { credential: Info },
}),
Removed: EventV2.define({
type: "credential.removed",
schema: { credential: Info },
}),
Switched: EventV2.define({
type: "credential.switched",
schema: {
connectorID: ConnectorSchema.ID,
from: Schema.optional(ID),
to: Schema.optional(ID),
},
}),
}
export interface Interface {
readonly get: (id: ID) => Effect.Effect<Info | undefined>
readonly all: () => Effect.Effect<Info[]>
/** Returns every stored credential. */
readonly all: () => Effect.Effect<Stored[]>
/** Returns stored credentials belonging to one integration. */
readonly list: (integrationID: IntegrationSchema.ID) => Effect.Effect<Stored[]>
/** Replaces any credential for an integration and returns the new record. */
readonly create: (input: {
connectorID: ConnectorSchema.ID
methodID: ConnectorSchema.MethodID
value: Value
label?: string
}) => Effect.Effect<Info>
readonly update: (id: ID, updates: Partial<Pick<Info, "label" | "value">>) => Effect.Effect<void>
readonly integrationID: IntegrationSchema.ID
readonly value: Info
readonly label?: string
}) => Effect.Effect<Stored>
/** Updates the label or secret value of a stored credential. */
readonly update: (id: ID, updates: Partial<Pick<Stored, "label" | "value">>) => Effect.Effect<void>
/** Removes a stored credential. */
readonly remove: (id: ID) => Effect.Effect<void>
readonly activate: (id: ID) => Effect.Effect<void>
readonly active: (connectorID: ConnectorSchema.ID) => Effect.Effect<Info | undefined>
readonly activeAll: () => Effect.Effect<Map<ConnectorSchema.ID, Info>>
readonly forConnector: (connectorID: ConnectorSchema.ID) => Effect.Effect<Info[]>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/v2/Credential") {}
export const legacyImportLayer = Layer.effectDiscard(
Effect.gen(function* () {
const { db } = yield* Database.Service
const fs = yield* FSUtil.Service
const global = yield* Global.Service
const name = "credential.auth-json"
if (yield* db.select().from(DataMigrationTable).where(eq(DataMigrationTable.name, name)).get()) return
const raw = yield* fs.readJson(path.join(global.data, "auth.json")).pipe(Effect.option)
if (Option.isNone(raw) || typeof raw.value !== "object" || raw.value === null || Array.isArray(raw.value)) return
const decode = Schema.decodeUnknownOption(LegacyValue)
const values = Object.entries(raw.value).flatMap(([connectorID, value]) => {
const decoded = decode(value)
if (Option.isNone(decoded)) return []
const credential = decoded.value
const id = ID.create()
const connector = ConnectorSchema.ID.make(connectorID.replace(/\/+$/, ""))
const methodID = ConnectorSchema.MethodID.make(
credential.type === "api"
? "api-key"
: connector === ConnectorSchema.ID.make("openai")
? "chatgpt-browser"
: "oauth",
)
const next: Value =
credential.type === "api"
? new Key({ type: "key", key: credential.key, metadata: credential.metadata })
: new OAuth({
type: "oauth",
refresh: credential.refresh,
access: credential.access,
expires: credential.expires,
metadata: {
...(credential.accountId ? { accountID: credential.accountId } : {}),
...(credential.enterpriseUrl ? { enterpriseURL: credential.enterpriseUrl } : {}),
},
})
return [{ id, connectorID: connector, methodID, value: next }]
})
yield* db.transaction((tx) =>
Effect.gen(function* () {
for (const item of values) {
if (
yield* tx
.select({ id: CredentialTable.id })
.from(CredentialTable)
.where(eq(CredentialTable.connector_id, item.connectorID))
.get()
)
continue
yield* tx.insert(CredentialTable).values({
id: item.id,
connector_id: item.connectorID,
method_id: item.methodID,
label: "Imported",
value: item.value,
active: true,
})
}
yield* tx.insert(DataMigrationTable).values({ name, time_completed: Date.now() }).onConflictDoNothing().run()
}),
)
}).pipe(Effect.orDie),
)
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const { db } = yield* Database.Service
const events = yield* EventV2.Service
const decodeValue = Schema.decodeUnknownSync(Value)
const info = (row: typeof CredentialTable.$inferSelect) =>
new Info({
const decode = Schema.decodeUnknownSync(Info)
const stored = (row: typeof CredentialTable.$inferSelect) =>
new Stored({
id: row.id,
connectorID: row.connector_id,
methodID: row.method_id,
integrationID: row.integration_id,
label: row.label,
value: decodeValue(row.value),
value: decode(row.value),
})
const activate = Effect.fn("Credential.activate")(function* (id: ID) {
const switched = yield* db
.transaction((tx) =>
Effect.gen(function* () {
const credential = yield* tx.select().from(CredentialTable).where(eq(CredentialTable.id, id)).get()
if (!credential || credential.active) return
const current = yield* tx
.select({ id: CredentialTable.id })
.from(CredentialTable)
.where(and(eq(CredentialTable.connector_id, credential.connector_id), eq(CredentialTable.active, true)))
.get()
yield* tx
.update(CredentialTable)
.set({ active: false })
.where(eq(CredentialTable.connector_id, credential.connector_id))
.run()
yield* tx.update(CredentialTable).set({ active: true }).where(eq(CredentialTable.id, id)).run()
return { connectorID: credential.connector_id, from: current?.id, to: id }
}),
)
.pipe(Effect.orDie)
if (switched) yield* events.publish(Event.Switched, switched)
})
return Service.of({
get: Effect.fn("Credential.get")(function* (id) {
const row = yield* db.select().from(CredentialTable).where(eq(CredentialTable.id, id)).get().pipe(Effect.orDie)
return row ? info(row) : undefined
}),
all: Effect.fn("Credential.all")(function* () {
return (yield* db
.select()
.from(CredentialTable)
.orderBy(asc(CredentialTable.time_created))
.all()
.pipe(Effect.orDie)).map(info)
.pipe(Effect.orDie)).map(stored)
}),
active: Effect.fn("Credential.active")(function* (connectorID) {
const row = yield* db
.select()
.from(CredentialTable)
.where(and(eq(CredentialTable.connector_id, connectorID), eq(CredentialTable.active, true)))
.get()
.pipe(Effect.orDie)
return row ? info(row) : undefined
}),
activeAll: Effect.fn("Credential.activeAll")(function* () {
const rows = yield* db
.select()
.from(CredentialTable)
.where(eq(CredentialTable.active, true))
.all()
.pipe(Effect.orDie)
return new Map(rows.map((row) => [row.connector_id, info(row)]))
}),
forConnector: Effect.fn("Credential.forConnector")(function* (connectorID) {
list: Effect.fn("Credential.list")(function* (integrationID) {
return (yield* db
.select()
.from(CredentialTable)
.where(eq(CredentialTable.connector_id, connectorID))
.where(eq(CredentialTable.integration_id, integrationID))
.orderBy(asc(CredentialTable.time_created))
.all()
.pipe(Effect.orDie)).map(info)
.pipe(Effect.orDie)).map(stored)
}),
create: Effect.fn("Credential.create")(function* (input) {
const credential = new Info({
const credential = new Stored({
id: ID.create(),
connectorID: input.connectorID,
methodID: input.methodID,
integrationID: input.integrationID,
label: input.label ?? "default",
value: input.value,
})
const from = yield* db
yield* db
.transaction((tx) =>
Effect.gen(function* () {
const current = yield* tx
.select({ id: CredentialTable.id })
.from(CredentialTable)
.where(and(eq(CredentialTable.connector_id, input.connectorID), eq(CredentialTable.active, true)))
.get()
yield* tx
.update(CredentialTable)
.set({ active: false })
.where(eq(CredentialTable.connector_id, input.connectorID))
.delete(CredentialTable)
.where(eq(CredentialTable.integration_id, credential.integrationID))
.run()
yield* tx
.insert(CredentialTable)
.values({
id: credential.id,
connector_id: credential.connectorID,
method_id: credential.methodID,
integration_id: credential.integrationID,
label: credential.label,
value: credential.value,
active: true,
})
.run()
return current?.id
}),
)
.pipe(Effect.orDie)
yield* events.publish(Event.Added, { credential })
yield* events.publish(Event.Switched, { connectorID: credential.connectorID, from, to: credential.id })
return credential
}),
update: Effect.fn("Credential.update")(function* (id, updates) {
@ -294,50 +129,10 @@ export const layer = Layer.effect(
.pipe(Effect.orDie)
}),
remove: Effect.fn("Credential.remove")(function* (id) {
const removed = yield* db
.transaction((tx) =>
Effect.gen(function* () {
const row = yield* tx.select().from(CredentialTable).where(eq(CredentialTable.id, id)).get()
if (!row) return
yield* tx.delete(CredentialTable).where(eq(CredentialTable.id, id)).run()
if (!row.active) return { credential: info(row) }
const replacement = yield* tx
.select()
.from(CredentialTable)
.where(and(eq(CredentialTable.connector_id, row.connector_id), ne(CredentialTable.id, id)))
.orderBy(asc(CredentialTable.time_created))
.get()
if (replacement) {
yield* tx
.update(CredentialTable)
.set({ active: true })
.where(eq(CredentialTable.id, replacement.id))
.run()
}
return {
credential: info(row),
switched: { connectorID: row.connector_id, from: id, to: replacement?.id },
}
}),
)
.pipe(Effect.orDie)
if (!removed) return
yield* events.publish(Event.Removed, { credential: removed.credential })
if (removed.switched) yield* events.publish(Event.Switched, removed.switched)
yield* db.delete(CredentialTable).where(eq(CredentialTable.id, id)).run().pipe(Effect.orDie)
}),
activate,
})
}),
)
export const defaultLayer = layer.pipe(
Layer.provide(Database.defaultLayer),
Layer.provide(EventV2.defaultLayer),
Layer.provideMerge(
legacyImportLayer.pipe(
Layer.provide(Database.defaultLayer),
Layer.provide(FSUtil.defaultLayer),
Layer.provide(Global.defaultLayer),
),
),
)
export const defaultLayer = layer.pipe(Layer.provide(Database.defaultLayer))

29
packages/core/src/credential/sql.ts
View file

@ -1,23 +1,12 @@
import { sql } from "drizzle-orm"
import { integer, sqliteTable, text, uniqueIndex } from "drizzle-orm/sqlite-core"
import { sqliteTable, text } from "drizzle-orm/sqlite-core"
import { Timestamps } from "../database/schema.sql"
import type { ConnectorSchema } from "../connector/schema"
import type { IntegrationSchema } from "../integration/schema"
import type { Credential } from "../credential"
export const CredentialTable = sqliteTable(
"credential",
{
id: text().$type<Credential.ID>().primaryKey(),
connector_id: text().$type<ConnectorSchema.ID>().notNull(),
method_id: text().$type<ConnectorSchema.MethodID>().notNull(),
label: text().notNull(),
value: text({ mode: "json" }).$type<Credential.Value>().notNull(),
active: integer({ mode: "boolean" }).notNull().default(false),
...Timestamps,
},
(table) => [
uniqueIndex("credential_connector_active_idx")
.on(table.connector_id)
.where(sql`${table.active} = 1`),
],
)
export const CredentialTable = sqliteTable("credential", {
id: text().$type<Credential.ID>().primaryKey(),
integration_id: text().$type<IntegrationSchema.ID>().notNull(),
label: text().notNull(),
value: text({ mode: "json" }).$type<Credential.Info>().notNull(),
...Timestamps,
})

1
packages/core/src/database/migration.gen.ts generated
View file

@ -35,5 +35,6 @@ export const migrations = (
import("./migration/20260605003541_add_session_context_snapshot"),
import("./migration/20260605042240_add_context_epoch_agent"),
import("./migration/20260611035744_credential"),
import("./migration/20260611192811_lush_chimera"),
])
).map((module) => module.default) satisfies DatabaseMigration.Migration[]

26
packages/core/src/database/migration.ts
View file

@ -4,6 +4,7 @@ import { sql } from "drizzle-orm"
import { Effect, Semaphore } from "effect"
import type { EffectDrizzleSqlite } from "@opencode-ai/effect-drizzle-sqlite"
import { migrations } from "./migration.gen"
import schema from "./schema.gen"
type Database = EffectDrizzleSqlite.EffectSQLiteDatabase
type Transaction = Parameters<Parameters<Database["transaction"]>[0]>[0]
@ -15,7 +16,28 @@ export type Migration = {
}
export function apply(db: Database) {
return lock.withPermit(applyOnly(db, migrations))
return lock.withPermit(
Effect.gen(function* () {
const tables = yield* db.all<{ name: string }>(
sql`SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'`,
)
if (tables.some((table) => table.name === "session")) return yield* applyOnly(db, migrations)
if (tables.length > 0) return yield* Effect.die("Database is not empty and has no session table")
yield* db.transaction((tx) =>
Effect.gen(function* () {
yield* schema.up(tx)
yield* tx.run(
sql`CREATE TABLE ${sql.identifier("migration")} (id TEXT PRIMARY KEY, time_completed INTEGER NOT NULL)`,
)
yield* Effect.forEach(migrations, (migration) =>
tx.run(
sql`INSERT INTO ${sql.identifier("migration")} (id, time_completed) VALUES (${migration.id}, ${Date.now()})`,
),
)
}),
)
}),
)
}
export function applyOnly(db: Database, input: Migration[]) {
@ -48,7 +70,7 @@ export function applyOnly(db: Database, input: Migration[]) {
if (completed.has(migration.id)) continue
yield* db.transaction((tx) =>
Effect.gen(function* () {
if (!process.env.OPENCODE_SKIP_MIGRATIONS) yield* migration.up(tx)
yield* migration.up(tx)
yield* tx.run(
sql`INSERT INTO ${sql.identifier("migration")} (id, time_completed) VALUES (${migration.id}, ${Date.now()})`,
)

15
packages/core/src/database/migration/20260611192811_lush_chimera.ts Normal file
View file

@ -0,0 +1,15 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "../migration"
export default {
id: "20260611192811_lush_chimera",
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`ALTER TABLE \`credential\` ADD \`integration_id\` text NOT NULL;`)
yield* tx.run(`DROP INDEX IF EXISTS \`credential_connector_active_idx\`;`)
yield* tx.run(`ALTER TABLE \`credential\` DROP COLUMN \`connector_id\`;`)
yield* tx.run(`ALTER TABLE \`credential\` DROP COLUMN \`method_id\`;`)
yield* tx.run(`ALTER TABLE \`credential\` DROP COLUMN \`active\`;`)
})
},
} satisfies DatabaseMigration.Migration

257
packages/core/src/database/schema.gen.ts Normal file
View file

@ -0,0 +1,257 @@
import { Effect } from "effect"
import type { DatabaseMigration } from "./migration"
export default {
up(tx) {
return Effect.gen(function* () {
yield* tx.run(`
CREATE TABLE \`workspace\` (
\`id\` text PRIMARY KEY,
\`type\` text NOT NULL,
\`name\` text DEFAULT '' NOT NULL,
\`branch\` text,
\`directory\` text,
\`extra\` text,
\`project_id\` text NOT NULL,
\`time_used\` integer NOT NULL,
CONSTRAINT \`fk_workspace_project_id_project_id_fk\` FOREIGN KEY (\`project_id\`) REFERENCES \`project\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`data_migration\` (
\`name\` text PRIMARY KEY,
\`time_completed\` integer NOT NULL
);
`)
yield* tx.run(`
CREATE TABLE \`account_state\` (
\`id\` integer PRIMARY KEY,
\`active_account_id\` text,
\`active_org_id\` text,
CONSTRAINT \`fk_account_state_active_account_id_account_id_fk\` FOREIGN KEY (\`active_account_id\`) REFERENCES \`account\`(\`id\`) ON DELETE SET NULL
);
`)
yield* tx.run(`
CREATE TABLE \`account\` (
\`id\` text PRIMARY KEY,
\`email\` text NOT NULL,
\`url\` text NOT NULL,
\`access_token\` text NOT NULL,
\`refresh_token\` text NOT NULL,
\`token_expiry\` integer,
\`time_created\` integer NOT NULL,
\`time_updated\` integer NOT NULL
);
`)
yield* tx.run(`
CREATE TABLE \`control_account\` (
\`email\` text NOT NULL,
\`url\` text NOT NULL,
\`access_token\` text NOT NULL,
\`refresh_token\` text NOT NULL,
\`token_expiry\` integer,
\`active\` integer NOT NULL,
\`time_created\` integer NOT NULL,
\`time_updated\` integer NOT NULL,
CONSTRAINT \`control_account_pk\` PRIMARY KEY(\`email\`, \`url\`)
);
`)
yield* tx.run(`
CREATE TABLE \`credential\` (
\`id\` text PRIMARY KEY,
\`integration_id\` text NOT NULL,
\`label\` text NOT NULL,
\`value\` text NOT NULL,
\`time_created\` integer NOT NULL,
\`time_updated\` integer NOT NULL
);
`)
yield* tx.run(`
CREATE TABLE \`event_sequence\` (
\`aggregate_id\` text PRIMARY KEY,
\`seq\` integer NOT NULL,
\`owner_id\` text
);
`)
yield* tx.run(`
CREATE TABLE \`event\` (
\`id\` text PRIMARY KEY,
\`aggregate_id\` text NOT NULL,
\`seq\` integer NOT NULL,
\`type\` text NOT NULL,
\`data\` text NOT NULL,
CONSTRAINT \`fk_event_aggregate_id_event_sequence_aggregate_id_fk\` FOREIGN KEY (\`aggregate_id\`) REFERENCES \`event_sequence\`(\`aggregate_id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`permission\` (
\`id\` text PRIMARY KEY,
\`project_id\` text NOT NULL,
\`action\` text NOT NULL,
\`resource\` text NOT NULL,
\`time_created\` integer NOT NULL,
\`time_updated\` integer NOT NULL,
CONSTRAINT \`fk_permission_project_id_project_id_fk\` FOREIGN KEY (\`project_id\`) REFERENCES \`project\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`project_directory\` (
\`project_id\` text NOT NULL,
\`directory\` text NOT NULL,
\`type\` text NOT NULL,
\`time_created\` integer NOT NULL,
CONSTRAINT \`project_directory_pk\` PRIMARY KEY(\`project_id\`, \`directory\`),
CONSTRAINT \`fk_project_directory_project_id_project_id_fk\` FOREIGN KEY (\`project_id\`) REFERENCES \`project\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`project\` (
\`id\` text PRIMARY KEY,
\`worktree\` text NOT NULL,
\`vcs\` text,
\`name\` text,
\`icon_url\` text,
\`icon_url_override\` text,
\`icon_color\` text,
\`time_created\` integer NOT NULL,
\`time_updated\` integer NOT NULL,
\`time_initialized\` integer,
\`sandboxes\` text NOT NULL,
\`commands\` text
);
`)
yield* tx.run(`
CREATE TABLE \`message\` (
\`id\` text PRIMARY KEY,
\`session_id\` text NOT NULL,
\`time_created\` integer NOT NULL,
\`time_updated\` integer NOT NULL,
\`data\` text NOT NULL,
CONSTRAINT \`fk_message_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`part\` (
\`id\` text PRIMARY KEY,
\`message_id\` text NOT NULL,
\`session_id\` text NOT NULL,
\`time_created\` integer NOT NULL,
\`time_updated\` integer NOT NULL,
\`data\` text NOT NULL,
CONSTRAINT \`fk_part_message_id_message_id_fk\` FOREIGN KEY (\`message_id\`) REFERENCES \`message\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`session_context_epoch\` (
\`session_id\` text PRIMARY KEY,
\`baseline\` text NOT NULL,
\`agent\` text DEFAULT 'build' NOT NULL,
\`snapshot\` text NOT NULL,
\`baseline_seq\` integer NOT NULL,
\`replacement_seq\` integer,
\`revision\` integer DEFAULT 0 NOT NULL,
CONSTRAINT \`fk_session_context_epoch_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`session_input\` (
\`id\` text PRIMARY KEY,
\`session_id\` text NOT NULL,
\`prompt\` text NOT NULL,
\`delivery\` text NOT NULL,
\`admitted_seq\` integer NOT NULL,
\`promoted_seq\` integer,
\`time_created\` integer NOT NULL,
CONSTRAINT \`fk_session_input_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`session_message\` (
\`id\` text PRIMARY KEY,
\`session_id\` text NOT NULL,
\`type\` text NOT NULL,
\`seq\` integer NOT NULL,
\`time_created\` integer NOT NULL,
\`time_updated\` integer NOT NULL,
\`data\` text NOT NULL,
CONSTRAINT \`fk_session_message_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`session\` (
\`id\` text PRIMARY KEY,
\`project_id\` text NOT NULL,
\`workspace_id\` text,
\`parent_id\` text,
\`slug\` text NOT NULL,
\`directory\` text NOT NULL,
\`path\` text,
\`title\` text NOT NULL,
\`version\` text NOT NULL,
\`share_url\` text,
\`summary_additions\` integer,
\`summary_deletions\` integer,
\`summary_files\` integer,
\`summary_diffs\` text,
\`metadata\` text,
\`cost\` real DEFAULT 0 NOT NULL,
\`tokens_input\` integer DEFAULT 0 NOT NULL,
\`tokens_output\` integer DEFAULT 0 NOT NULL,
\`tokens_reasoning\` integer DEFAULT 0 NOT NULL,
\`tokens_cache_read\` integer DEFAULT 0 NOT NULL,
\`tokens_cache_write\` integer DEFAULT 0 NOT NULL,
\`revert\` text,
\`permission\` text,
\`agent\` text,
\`model\` text,
\`time_created\` integer NOT NULL,
\`time_updated\` integer NOT NULL,
\`time_compacting\` integer,
\`time_archived\` integer,
CONSTRAINT \`fk_session_project_id_project_id_fk\` FOREIGN KEY (\`project_id\`) REFERENCES \`project\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`todo\` (
\`session_id\` text NOT NULL,
\`content\` text NOT NULL,
\`status\` text NOT NULL,
\`priority\` text NOT NULL,
\`position\` integer NOT NULL,
\`time_created\` integer NOT NULL,
\`time_updated\` integer NOT NULL,
CONSTRAINT \`todo_pk\` PRIMARY KEY(\`session_id\`, \`position\`),
CONSTRAINT \`fk_todo_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`
CREATE TABLE \`session_share\` (
\`session_id\` text PRIMARY KEY,
\`id\` text NOT NULL,
\`secret\` text NOT NULL,
\`url\` text NOT NULL,
\`time_created\` integer NOT NULL,
\`time_updated\` integer NOT NULL,
CONSTRAINT \`fk_session_share_session_id_session_id_fk\` FOREIGN KEY (\`session_id\`) REFERENCES \`session\`(\`id\`) ON DELETE CASCADE
);
`)
yield* tx.run(`CREATE UNIQUE INDEX \`event_aggregate_seq_idx\` ON \`event\` (\`aggregate_id\`,\`seq\`);`)
yield* tx.run(`CREATE INDEX \`event_aggregate_type_seq_idx\` ON \`event\` (\`aggregate_id\`,\`type\`,\`seq\`);`)
yield* tx.run(`CREATE UNIQUE INDEX \`permission_project_action_resource_idx\` ON \`permission\` (\`project_id\`,\`action\`,\`resource\`);`)
yield* tx.run(`CREATE INDEX \`message_session_time_created_id_idx\` ON \`message\` (\`session_id\`,\`time_created\`,\`id\`);`)
yield* tx.run(`CREATE INDEX \`part_message_id_id_idx\` ON \`part\` (\`message_id\`,\`id\`);`)
yield* tx.run(`CREATE INDEX \`part_session_idx\` ON \`part\` (\`session_id\`);`)
yield* tx.run(`CREATE INDEX \`session_input_session_pending_delivery_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`,\`delivery\`,\`admitted_seq\`);`)
yield* tx.run(`CREATE UNIQUE INDEX \`session_input_session_admitted_seq_idx\` ON \`session_input\` (\`session_id\`,\`admitted_seq\`);`)
yield* tx.run(`CREATE UNIQUE INDEX \`session_input_session_promoted_seq_idx\` ON \`session_input\` (\`session_id\`,\`promoted_seq\`);`)
yield* tx.run(`CREATE UNIQUE INDEX \`session_message_session_seq_idx\` ON \`session_message\` (\`session_id\`,\`seq\`);`)
yield* tx.run(`CREATE INDEX \`session_message_session_type_seq_idx\` ON \`session_message\` (\`session_id\`,\`type\`,\`seq\`);`)
yield* tx.run(`CREATE INDEX \`session_message_session_time_created_id_idx\` ON \`session_message\` (\`session_id\`,\`time_created\`,\`id\`);`)
yield* tx.run(`CREATE INDEX \`session_message_time_created_idx\` ON \`session_message\` (\`time_created\`);`)
yield* tx.run(`CREATE INDEX \`session_project_idx\` ON \`session\` (\`project_id\`);`)
yield* tx.run(`CREATE INDEX \`session_workspace_idx\` ON \`session\` (\`workspace_id\`);`)
yield* tx.run(`CREATE INDEX \`session_parent_idx\` ON \`session\` (\`parent_id\`);`)
yield* tx.run(`CREATE INDEX \`todo_session_idx\` ON \`todo\` (\`session_id\`);`)
})
},
} satisfies Omit<DatabaseMigration.Migration, "id">

524
packages/core/src/integration.ts Normal file
View file

@ -0,0 +1,524 @@
export * as Integration from "./integration"
import { Cause, Clock, Context, Duration, Effect, Exit, Layer, Schedule, Schema, Scope, SynchronizedRef } from "effect"
import { castDraft, enableMapSet, type Draft } from "immer"
import { Credential } from "./credential"
import { IntegrationSchema } from "./integration/schema"
import { withStatics } from "./schema"
import { State } from "./state"
import { Identifier } from "./util/identifier"
import { EventV2 } from "./event"
import { IntegrationConnection } from "./integration/connection"
export const ID = IntegrationSchema.ID
export type ID = IntegrationSchema.ID
export const MethodID = IntegrationSchema.MethodID
export type MethodID = IntegrationSchema.MethodID
export const AttemptID = Schema.String.pipe(
Schema.brand("Integration.AttemptID"),
withStatics((schema) => ({ create: () => schema.make("con_" + Identifier.ascending()) })),
)
export type AttemptID = typeof AttemptID.Type
export const When = Schema.Struct({
key: Schema.String,
op: Schema.Literals(["eq", "neq"]),
value: Schema.String,
}).annotate({ identifier: "Integration.When" })
export type When = typeof When.Type
export class TextPrompt extends Schema.Class<TextPrompt>("Integration.TextPrompt")({
type: Schema.Literal("text"),
key: Schema.String,
message: Schema.String,
placeholder: Schema.optional(Schema.String),
when: Schema.optional(When),
}) {}
export class SelectPrompt extends Schema.Class<SelectPrompt>("Integration.SelectPrompt")({
type: Schema.Literal("select"),
key: Schema.String,
message: Schema.String,
options: Schema.Array(
Schema.Struct({
label: Schema.String,
value: Schema.String,
hint: Schema.optional(Schema.String),
}),
),
when: Schema.optional(When),
}) {}
export const Prompt = Schema.Union([TextPrompt, SelectPrompt]).pipe(Schema.toTaggedUnion("type"))
export type Prompt = typeof Prompt.Type
export class OAuthMethod extends Schema.Class<OAuthMethod>("Integration.OAuthMethod")({
id: MethodID,
type: Schema.Literal("oauth"),
label: Schema.String,
prompts: Schema.optional(Schema.Array(Prompt)),
}) {}
export class KeyMethod extends Schema.Class<KeyMethod>("Integration.KeyMethod")({
type: Schema.Literal("key"),
label: Schema.optional(Schema.String),
}) {}
export class EnvMethod extends Schema.Class<EnvMethod>("Integration.EnvMethod")({
type: Schema.Literal("env"),
names: Schema.Array(Schema.String),
}) {}
export const Method = Schema.Union([OAuthMethod, KeyMethod, EnvMethod]).pipe(Schema.toTaggedUnion("type"))
export type Method = typeof Method.Type
export class Info extends Schema.Class<Info>("Integration.Info")({
id: ID,
name: Schema.String,
methods: Schema.Array(Method),
connections: Schema.Array(IntegrationConnection.Info),
}) {}
export type Inputs = Readonly<{ [key: string]: string }>
export type OAuthAuthorization = {
readonly url: string
readonly instructions: string
} & (
| {
readonly mode: "auto"
readonly callback: Effect.Effect<Credential.Info, unknown>
}
| {
readonly mode: "code"
readonly callback: (code: string) => Effect.Effect<Credential.Info, unknown>
}
)
export interface OAuthImplementation {
readonly integrationID: ID
readonly method: OAuthMethod
readonly authorize: (inputs: Inputs) => Effect.Effect<OAuthAuthorization, unknown, Scope.Scope>
readonly refresh?: (credential: Credential.OAuth) => Effect.Effect<Credential.OAuth, unknown>
}
export interface KeyImplementation {
readonly integrationID: ID
readonly method: KeyMethod
}
export interface EnvImplementation {
readonly integrationID: ID
readonly method: EnvMethod
}
export type Implementation = OAuthImplementation | KeyImplementation | EnvImplementation
function isOAuthImplementation(implementation: Implementation): implementation is OAuthImplementation {
return implementation.method.type === "oauth"
}
export class Attempt extends Schema.Class<Attempt>("Integration.Attempt")({
attemptID: AttemptID,
url: Schema.String,
instructions: Schema.String,
mode: Schema.Literals(["auto", "code"]),
time: Schema.Struct({
created: Schema.Number,
expires: Schema.Number,
}),
}) {}
const Time = Schema.Struct({
created: Schema.Number,
expires: Schema.Number,
})
export const AttemptStatus = Schema.Union([
Schema.Struct({ status: Schema.Literal("pending"), time: Time }),
Schema.Struct({ status: Schema.Literal("complete"), time: Time }),
Schema.Struct({ status: Schema.Literal("failed"), message: Schema.String, time: Time }),
Schema.Struct({ status: Schema.Literal("expired"), time: Time }),
]).pipe(Schema.toTaggedUnion("status"))
export type AttemptStatus = typeof AttemptStatus.Type
export class CodeRequiredError extends Schema.TaggedErrorClass<CodeRequiredError>()("Integration.CodeRequired", {
attemptID: AttemptID,
}) {}
export class AuthorizationError extends Schema.TaggedErrorClass<AuthorizationError>()("Integration.Authorization", {
cause: Schema.Defect,
}) {}
export type Error = CodeRequiredError | AuthorizationError
export const Event = {
Updated: EventV2.define({
type: "integration.updated",
schema: {},
}),
}
export type Ref = {
id: ID
name: string
}
type Entry = {
ref: Ref
methods: Method[]
implementations: Map<MethodID, OAuthImplementation>
}
type Data = {
integrations: Map<ID, Entry>
}
export type Editor = {
list: () => readonly Ref[]
get: (id: ID) => Ref | undefined
update: (id: ID, update: (integration: Draft<Ref>) => void) => void
remove: (id: ID) => void
method: {
list: (integrationID: ID) => readonly Method[]
update: (implementation: Implementation) => void
remove: (integrationID: ID, method: Method) => void
}
}
export interface Interface {
/** Registers a scoped transform over the integration registry. */
readonly transform: State.Interface<Data, Editor>["transform"]
/** Registers and immediately applies a scoped integration registry update. */
readonly update: State.Interface<Data, Editor>["update"]
/** Returns one integration with its methods and current connections. */
readonly get: (id: ID) => Effect.Effect<Info | undefined>
/** Returns all integrations with their methods and current connections. */
readonly list: () => Effect.Effect<Info[]>
readonly connect: {
/** Runs a key method and stores the resulting credential. */
readonly key: (input: {
/** Integration receiving the credential. */
readonly integrationID: ID
/** Secret entered by the user. */
readonly key: string
/** User-facing label for the stored credential. */
readonly label?: string
}) => Effect.Effect<void, AuthorizationError>
/** Starts a stateful OAuth attempt. */
readonly oauth: (input: {
/** Integration being authenticated. */
readonly integrationID: ID
/** OAuth method selected by the caller. */
readonly methodID: MethodID
/** Answers to the method's optional prompts. */
readonly inputs: Inputs
/** User-facing label for the credential created on completion. */
readonly label?: string
}) => Effect.Effect<Attempt, AuthorizationError>
}
readonly attempt: {
/** Returns the current state of an OAuth attempt. */
readonly status: (attemptID: AttemptID) => Effect.Effect<AttemptStatus>
/** Completes the attempt and stores its credential. */
readonly complete: (input: {
/** Opaque handle returned by `oauth`. */
readonly attemptID: AttemptID
/** Authorization code required by attempts in code mode. */
readonly code?: string
}) => Effect.Effect<void, CodeRequiredError | AuthorizationError>
/** Cancels an attempt and releases its resources. */
readonly cancel: (attemptID: AttemptID) => Effect.Effect<void>
}
}
export class Service extends Context.Service<Service, Interface>()("@opencode/v2/Integration") {}
enableMapSet()
const attemptLifetime = Duration.toMillis(Duration.minutes(10))
const terminalRetention = Duration.toMillis(Duration.minutes(1))
const scrubInterval = Duration.seconds(30)
type AttemptTime = { created: number; expires: number }
type PendingAttempt = {
status: "pending"
completing: boolean
authorization: OAuthAuthorization
integrationID: ID
methodID: MethodID
label?: string
scope: Scope.Closeable
time: AttemptTime
}
type TerminalAttempt = {
status: "complete" | "failed" | "expired"
message?: string
removeAt: number
time: AttemptTime
}
type AttemptEntry = PendingAttempt | TerminalAttempt
export const locationLayer = Layer.effect(
Service,
Effect.gen(function* () {
const credentials = yield* Credential.Service
const events = yield* EventV2.Service
const scope = yield* Scope.Scope
const attempts = SynchronizedRef.makeUnsafe(new Map<AttemptID, AttemptEntry>())
const state = State.create<Data, Editor>({
initial: () => ({ integrations: new Map<ID, Entry>() }),
editor: (draft) => ({
list: () => Array.from(draft.integrations.values(), (entry) => entry.ref) as Ref[],
get: (id) => draft.integrations.get(id)?.ref as Ref | undefined,
update: (id, update) => {
const current =
draft.integrations.get(id) ??
castDraft({ ref: { id, name: id } as Ref, methods: [], implementations: new Map() })
if (!draft.integrations.has(id)) draft.integrations.set(id, current)
update(current.ref)
current.ref.id = id
},
remove: (id) => draft.integrations.delete(id),
method: {
list: (integrationID) => (draft.integrations.get(integrationID)?.methods as Method[] | undefined) ?? [],
update: (implementation) => {
const current =
draft.integrations.get(implementation.integrationID) ??
castDraft({
ref: {
id: implementation.integrationID,
name: implementation.integrationID,
} as Ref,
methods: [],
implementations: new Map<MethodID, OAuthImplementation>(),
})
if (!draft.integrations.has(implementation.integrationID)) {
draft.integrations.set(implementation.integrationID, current)
}
const index = current.methods.findIndex((method) => {
if (method.type !== implementation.method.type) return false
if (method.type !== "oauth" || implementation.method.type !== "oauth") return true
return method.id === implementation.method.id
})
if (index === -1) current.methods.push(castDraft(implementation.method))
else current.methods[index] = castDraft(implementation.method)
if (isOAuthImplementation(implementation)) {
current.implementations.set(implementation.method.id, castDraft(implementation))
}
},
remove: (integrationID, method) => {
const current = draft.integrations.get(integrationID)
if (!current) return
const index = current.methods.findIndex((candidate) => {
if (candidate.type !== method.type) return false
if (candidate.type !== "oauth" || method.type !== "oauth") return true
return candidate.id === method.id
})
if (index !== -1) current.methods.splice(index, 1)
if (method.type === "oauth") current.implementations.delete(method.id)
},
},
}),
finalize: () => events.publish(Event.Updated, {}).pipe(Effect.asVoid),
})
const connections = (
entry: Entry,
saved: readonly Credential.Stored[],
): IntegrationConnection.Info[] => {
const connected = saved.map(
(credential) =>
new IntegrationConnection.CredentialInfo({ type: "credential", id: credential.id, label: credential.label }),
)
const detected = entry.methods
.filter((method) => method.type === "env")
.flatMap((method) => method.names.filter((name) => process.env[name]))
.map(
(name, index) =>
new IntegrationConnection.EnvInfo({
type: "env",
name,
}),
)
return [...connected, ...detected]
}
const project = (entry: Entry, saved: readonly Credential.Stored[]) =>
new Info({
id: entry.ref.id,
name: entry.ref.name,
methods: entry.methods,
connections: connections(entry, saved),
})
const authorize = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
effect.pipe(Effect.mapError((cause) => new AuthorizationError({ cause })))
const close = (attemptScope: Scope.Closeable) =>
Scope.close(attemptScope, Exit.void).pipe(Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid)
const message = (cause: Cause.Cause<unknown>) => {
const error = Cause.squash(cause)
return error instanceof Error ? error.message : String(error)
}
const settle = Effect.fnUntraced(function* (attemptID: AttemptID, exit: Exit.Exit<Credential.Info, unknown>) {
const now = yield* Clock.currentTimeMillis
const result = yield* SynchronizedRef.modify(attempts, (current) => {
const attempt = current.get(attemptID)
if (!attempt || attempt.status !== "pending") return [undefined, current]
const terminal: TerminalAttempt = Exit.isSuccess(exit)
? { status: "complete", time: attempt.time, removeAt: now + terminalRetention }
: { status: "failed", message: message(exit.cause), time: attempt.time, removeAt: now + terminalRetention }
return [attempt, new Map(current).set(attemptID, terminal)]
})
if (!result) return
if (Exit.isSuccess(exit)) {
yield* credentials.create({
integrationID: result.integrationID,
label: result.label,
value:
exit.value.type === "oauth"
? new Credential.OAuth({ ...exit.value, methodID: result.methodID })
: exit.value,
})
}
yield* close(result.scope)
})
const scrub = Effect.fnUntraced(function* () {
const now = yield* Clock.currentTimeMillis
const expired = yield* SynchronizedRef.modify(attempts, (current) => {
const next = new Map(current)
const scopes: Scope.Closeable[] = []
for (const [id, attempt] of current) {
if (attempt.status === "pending" && attempt.time.expires <= now) {
scopes.push(attempt.scope)
next.set(id, { status: "expired", time: attempt.time, removeAt: now + terminalRetention })
continue
}
if (attempt.status !== "pending" && attempt.removeAt <= now) next.delete(id)
}
return [scopes, next]
})
yield* Effect.forEach(expired, close, { discard: true })
})
yield* scrub().pipe(Effect.repeat(Schedule.spaced(scrubInterval)), Effect.forkIn(scope))
return Service.of({
transform: state.transform,
update: state.update,
get: Effect.fn("Integration.get")(function* (id) {
const entry = state.get().integrations.get(id)
if (!entry) return undefined
return project(entry, yield* credentials.list(id))
}),
list: Effect.fn("Integration.list")(function* () {
return (yield* Effect.forEach(state.get().integrations.values(), (entry) =>
Effect.gen(function* () {
return project(entry, yield* credentials.list(entry.ref.id))
}),
)).toSorted((a, b) => a.name.localeCompare(b.name))
}),
connect: {
key: Effect.fn("Integration.connect.key")(function* (input) {
const method = state
.get()
.integrations.get(input.integrationID)
?.methods.some((method) => method.type === "key")
if (!method) return yield* Effect.die(`Key method not found: ${input.integrationID}`)
yield* credentials.create({
integrationID: input.integrationID,
label: input.label,
value: new Credential.Key({ type: "key", key: input.key }),
})
}),
oauth: Effect.fn("Integration.connect.oauth")(function* (input) {
const method = state.get().integrations.get(input.integrationID)?.implementations.get(input.methodID)
if (!method) {
return yield* Effect.die(`OAuth method not found: ${input.integrationID}/${input.methodID}`)
}
const attemptScope = yield* Scope.fork(scope)
const authorization = yield* authorize(method.authorize(input.inputs)).pipe(
Scope.provide(attemptScope),
Effect.onExit((exit) => (Exit.isFailure(exit) ? Scope.close(attemptScope, exit) : Effect.void)),
)
const id = AttemptID.create()
const created = yield* Clock.currentTimeMillis
const time = { created, expires: created + attemptLifetime }
yield* SynchronizedRef.update(attempts, (current) =>
new Map(current).set(id, {
status: "pending",
completing: authorization.mode === "auto",
authorization,
integrationID: input.integrationID,
methodID: input.methodID,
label: input.label,
scope: attemptScope,
time,
}),
)
if (authorization.mode === "auto") {
yield* authorization.callback.pipe(
Effect.exit,
Effect.flatMap((exit) => settle(id, exit)),
Effect.forkIn(attemptScope, { startImmediately: true }),
)
}
return new Attempt({
attemptID: id,
url: authorization.url,
instructions: authorization.instructions,
mode: authorization.mode,
time,
})
}),
},
attempt: {
status: Effect.fn("Integration.attempt.status")(function* (attemptID) {
const attempt = (yield* SynchronizedRef.get(attempts)).get(attemptID)
if (!attempt) return yield* Effect.die(`OAuth attempt not found: ${attemptID}`)
if (attempt.status === "failed") {
return { status: attempt.status, message: attempt.message ?? "Authorization failed", time: attempt.time }
}
return { status: attempt.status, time: attempt.time }
}),
complete: Effect.fn("Integration.attempt.complete")(function* (input) {
const attempt = yield* SynchronizedRef.modify(attempts, (current) => {
const match = current.get(input.attemptID)
if (!match || match.status !== "pending" || match.completing) return [match, current]
if (match.authorization.mode === "code" && input.code === undefined) return [match, current]
return [match, new Map(current).set(input.attemptID, { ...match, completing: true })]
})
if (!attempt) return yield* Effect.die(`OAuth attempt not found: ${input.attemptID}`)
if (attempt.status !== "pending") return
if (attempt.authorization.mode === "code" && input.code === undefined) {
return yield* new CodeRequiredError({ attemptID: input.attemptID })
}
if (attempt.completing) return yield* Effect.die(`OAuth attempt already completing: ${input.attemptID}`)
const callback =
attempt.authorization.mode === "auto"
? attempt.authorization.callback
: attempt.authorization.callback(input.code as string)
const exit = yield* authorize(callback).pipe(Effect.exit)
yield* settle(input.attemptID, exit)
if (Exit.isFailure(exit)) return yield* exit
}),
cancel: Effect.fn("Integration.attempt.cancel")(function* (attemptID) {
const attempt = yield* SynchronizedRef.modify(attempts, (current) => {
const match = current.get(attemptID)
if (!match || match.status !== "pending") return [undefined, current]
const next = new Map(current)
next.delete(attemptID)
return [match, next]
})
if (attempt) yield* Scope.close(attempt.scope, Exit.void)
}),
},
})
}),
)

20
packages/core/src/integration/connection.ts Normal file
View file

@ -0,0 +1,20 @@
export * as IntegrationConnection from "./connection"
import { Schema } from "effect"
import { Credential } from "../credential"
export class CredentialInfo extends Schema.Class<CredentialInfo>("Connection.CredentialInfo")({
type: Schema.Literal("credential"),
id: Credential.ID,
label: Schema.String,
}) {}
export class EnvInfo extends Schema.Class<EnvInfo>("Connection.EnvInfo")({
type: Schema.Literal("env"),
name: Schema.String,
}) {}
export const Info = Schema.Union([CredentialInfo, EnvInfo])
.pipe(Schema.toTaggedUnion("type"))
.annotate({ identifier: "Connection.Info" })
export type Info = typeof Info.Type

9
packages/core/src/integration/schema.ts Normal file
View file

@ -0,0 +1,9 @@
export * as IntegrationSchema from "./schema"
import { Schema } from "effect"
export const ID = Schema.String.pipe(Schema.brand("Integration.ID"))
export type ID = typeof ID.Type
export const MethodID = Schema.String.pipe(Schema.brand("Integration.MethodID"))
export type MethodID = typeof MethodID.Type

4
packages/core/src/location-layer.ts
View file

@ -4,7 +4,7 @@ import { Policy } from "./policy"
import { Config } from "./config"
import { PluginV2 } from "./plugin"
import { Catalog } from "./catalog"
import { Connector } from "./connector"
import { Integration } from "./integration"
import { CommandV2 } from "./command"
import { AgentV2 } from "./agent"
import { PluginBoot } from "./plugin/boot"
@ -59,7 +59,7 @@ export class LocationServiceMap extends LayerMap.Service<LocationServiceMap>()("
Reference.locationLayer,
PluginV2.locationLayer,
Catalog.locationLayer,
Connector.locationLayer,
Integration.locationLayer,
CommandV2.locationLayer,
AgentV2.locationLayer,
PluginBoot.locationLayer,

10
packages/core/src/plugin/boot.ts
View file

@ -2,7 +2,7 @@ export * as PluginBoot from "./boot"
import { Context, Deferred, Effect, Layer } from "effect"
import { Credential } from "../credential"
import { Connector } from "../connector"
import { Integration } from "../integration"
import { AgentV2 } from "../agent"
import { Catalog } from "../catalog"
import { CommandV2 } from "../command"
@ -34,7 +34,7 @@ type Plugin = {
| Catalog.Service
| CommandV2.Service
| Credential.Service
| Connector.Service
| Integration.Service
| AgentV2.Service
| Npm.Service
| EventV2.Service
@ -62,7 +62,7 @@ export const layer = Layer.effect(
const commands = yield* CommandV2.Service
const plugin = yield* PluginV2.Service
const credentials = yield* Credential.Service
const connectors = yield* Connector.Service
const integrations = yield* Integration.Service
const agents = yield* AgentV2.Service
const config = yield* Config.Service
const location = yield* Location.Service
@ -82,7 +82,7 @@ export const layer = Layer.effect(
Effect.provideService(Catalog.Service, catalog),
Effect.provideService(CommandV2.Service, commands),
Effect.provideService(Credential.Service, credentials),
Effect.provideService(Connector.Service, connectors),
Effect.provideService(Integration.Service, integrations),
Effect.provideService(AgentV2.Service, agents),
Effect.provideService(Config.Service, config),
Effect.provideService(Location.Service, location),
@ -127,7 +127,7 @@ export const layer = Layer.effect(
)
export const locationLayer = layer.pipe(
Layer.provideMerge(Connector.locationLayer),
Layer.provideMerge(Integration.locationLayer),
Layer.provideMerge(Catalog.locationLayer),
Layer.provideMerge(CommandV2.locationLayer),
Layer.provideMerge(Config.locationLayer),

29
packages/core/src/plugin/models-dev.ts
View file

@ -1,7 +1,6 @@
import { DateTime, Effect, Scope, Stream } from "effect"
import { Catalog } from "../catalog"
import { Connector } from "../connector"
import { Credential } from "../credential"
import { Integration } from "../integration"
import { EventV2 } from "../event"
import { ModelV2 } from "../model"
import { ModelRequest } from "../model-request"
@ -56,27 +55,31 @@ export const ModelsDevPlugin = PluginV2.define({
id: PluginV2.ID.make("models-dev"),
effect: Effect.gen(function* () {
const catalog = yield* Catalog.Service
const connectors = yield* Connector.Service
const integrations = yield* Integration.Service
const modelsDev = yield* ModelsDev.Service
const events = yield* EventV2.Service
const scope = yield* Scope.Scope
const transform = yield* catalog.transform()
const connectorTransform = yield* connectors.transform()
const integrationTransform = yield* integrations.transform()
const refresh = Effect.fn("ModelsDevPlugin.refresh")(function* () {
const data = yield* modelsDev.get()
yield* connectorTransform((connectors) => {
yield* integrationTransform((integrations) => {
for (const item of Object.values(data)) {
if (item.env.length === 0) continue
const connectorID = Connector.ID.make(item.id)
connectors.update(connectorID, (connector) => (connector.name = item.name))
connectors.method.update({
connectorID,
method: new Connector.KeyMethod({
id: Connector.MethodID.make("api-key"),
const integrationID = Integration.ID.make(item.id)
integrations.update(integrationID, (integration) => (integration.name = item.name))
integrations.method.update({
integrationID,
method: new Integration.KeyMethod({
type: "key",
label: "API Key",
}),
authorize: (key: string) => Effect.succeed(new Credential.Key({ type: "key", key })),
})
integrations.method.update({
integrationID,
method: new Integration.EnvMethod({
type: "env",
names: [...item.env],
}),
})
}
})

29
packages/core/src/plugin/provider/openai-auth.ts
View file

@ -1,6 +1,6 @@
import { createServer } from "node:http"
import { Deferred, Effect } from "effect"
import { Connector } from "../../connector"
import { Integration } from "../../integration"
import { Credential } from "../../credential"
import { InstallationVersion } from "../../installation/version"
@ -27,10 +27,13 @@ type Claims = {
"https://api.openai.com/auth"?: { chatgpt_account_id?: string }
}
const browserMethodID = Integration.MethodID.make("chatgpt-browser")
const headlessMethodID = Integration.MethodID.make("chatgpt-headless")
export const browser = {
connectorID: Connector.ID.make("openai"),
method: new Connector.OAuthMethod({
id: Connector.MethodID.make("chatgpt-browser"),
integrationID: Integration.ID.make("openai"),
method: new Integration.OAuthMethod({
id: browserMethodID,
type: "oauth",
label: "ChatGPT Pro/Plus (browser)",
}),
@ -77,17 +80,17 @@ export const browser = {
instructions: "Complete authorization in your browser. This window will close automatically.",
callback: Deferred.await(code).pipe(
Effect.flatMap((value) => exchange(value, redirect, pkce)),
Effect.map(credential),
Effect.map((tokens) => credential(browserMethodID, tokens)),
),
}
}),
refresh: (value) => refresh(value),
} satisfies Connector.OAuthImplementation
} satisfies Integration.OAuthImplementation
export const headless = {
connectorID: Connector.ID.make("openai"),
method: new Connector.OAuthMethod({
id: Connector.MethodID.make("chatgpt-headless"),
integrationID: Integration.ID.make("openai"),
method: new Integration.OAuthMethod({
id: headlessMethodID,
type: "oauth",
label: "ChatGPT Pro/Plus (headless)",
}),
@ -124,6 +127,7 @@ export const headless = {
code_verifier: string
}
return credential(
headlessMethodID,
yield* exchange(data.authorization_code, `${issuer}/deviceauth/callback`, {
verifier: data.code_verifier,
challenge: "",
@ -139,7 +143,7 @@ export const headless = {
}
}),
refresh: (value) => refresh(value),
} satisfies Connector.OAuthImplementation
} satisfies Integration.OAuthImplementation
function headers(contentType: string) {
return { "Content-Type": contentType, "User-Agent": `opencode/${InstallationVersion}` }
@ -170,7 +174,7 @@ function refresh(value: Credential.OAuth) {
}).toString(),
}).pipe(
Effect.map((tokens) => {
const next = credential(tokens)
const next = credential(value.methodID, tokens)
return new Credential.OAuth({
...next,
metadata: next.metadata ?? value.metadata,
@ -190,10 +194,11 @@ function request<A>(url: string, init: RequestInit) {
})
}
function credential(tokens: TokenResponse) {
function credential(methodID: Integration.MethodID, tokens: TokenResponse) {
const accountID = extractAccountID(tokens)
return new Credential.OAuth({
type: "oauth",
methodID,
refresh: tokens.refresh_token,
access: tokens.access_token,
expires: Date.now() + (tokens.expires_in ?? 3600) * 1000,

6
packages/core/src/plugin/provider/openai.ts
View file

@ -2,14 +2,14 @@ import { Effect } from "effect"
import { ModelV2 } from "../../model"
import { PluginV2 } from "../../plugin"
import { ProviderV2 } from "../../provider"
import { Connector } from "../../connector"
import { Integration } from "../../integration"
import { browser, headless } from "./openai-auth"
export const OpenAIPlugin = PluginV2.define({
id: PluginV2.ID.make("openai"),
effect: Effect.gen(function* () {
const connectors = yield* Connector.Service
yield* connectors.update((editor) => {
const integrations = yield* Integration.Service
yield* integrations.update((editor) => {
editor.method.update(browser)
editor.method.update(headless)
})