fix(opencode): restore global sync event compatibility
This commit is contained in:
parent
1ed76ccace
commit
5e3026b5ce
9 changed files with 766 additions and 303 deletions
|
|
@ -30,6 +30,7 @@ export type Payload<D extends Definition = Definition> = {
|
||||||
readonly type: D["type"]
|
readonly type: D["type"]
|
||||||
readonly data: Data<D>
|
readonly data: Data<D>
|
||||||
readonly version?: number
|
readonly version?: number
|
||||||
|
readonly sync?: SyncMetadata
|
||||||
readonly location?: Location.Info
|
readonly location?: Location.Info
|
||||||
readonly metadata?: Record<string, unknown>
|
readonly metadata?: Record<string, unknown>
|
||||||
}
|
}
|
||||||
|
|
@ -48,6 +49,11 @@ export type SerializedEvent = {
|
||||||
readonly data: Record<string, unknown>
|
readonly data: Record<string, unknown>
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export type SyncMetadata = {
|
||||||
|
readonly seq: number
|
||||||
|
readonly aggregateID: string
|
||||||
|
}
|
||||||
|
|
||||||
export class InvalidSyncEventError extends Schema.TaggedErrorClass<InvalidSyncEventError>()(
|
export class InvalidSyncEventError extends Schema.TaggedErrorClass<InvalidSyncEventError>()(
|
||||||
"EventV2.InvalidSyncEvent",
|
"EventV2.InvalidSyncEvent",
|
||||||
{
|
{
|
||||||
|
|
@ -77,6 +83,12 @@ export function define<const Type extends string, Fields extends Schema.Struct.F
|
||||||
metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
|
metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)),
|
||||||
type: Schema.Literal(input.type),
|
type: Schema.Literal(input.type),
|
||||||
version: Schema.optional(Schema.Number),
|
version: Schema.optional(Schema.Number),
|
||||||
|
sync: Schema.optional(
|
||||||
|
Schema.Struct({
|
||||||
|
seq: Schema.Finite,
|
||||||
|
aggregateID: Schema.String,
|
||||||
|
}),
|
||||||
|
),
|
||||||
location: Schema.optional(Location.Info),
|
location: Schema.optional(Location.Info),
|
||||||
data: Data,
|
data: Data,
|
||||||
}).annotate({ identifier: input.type })
|
}).annotate({ identifier: input.type })
|
||||||
|
|
@ -186,7 +198,7 @@ export const layer = Layer.effect(
|
||||||
)
|
)
|
||||||
} else {
|
} else {
|
||||||
const list = projectors.get(event.type) ?? []
|
const list = projectors.get(event.type) ?? []
|
||||||
yield* db
|
return yield* db
|
||||||
.transaction(
|
.transaction(
|
||||||
() =>
|
() =>
|
||||||
Effect.gen(function* () {
|
Effect.gen(function* () {
|
||||||
|
|
@ -208,8 +220,12 @@ export const layer = Layer.effect(
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
const serialized = {
|
||||||
|
seq,
|
||||||
|
aggregateID,
|
||||||
|
}
|
||||||
for (const projector of list) {
|
for (const projector of list) {
|
||||||
yield* projector(event as Payload)
|
yield* projector({ ...event, sync: serialized })
|
||||||
}
|
}
|
||||||
yield* db
|
yield* db
|
||||||
.insert(EventSequenceTable)
|
.insert(EventSequenceTable)
|
||||||
|
|
@ -233,6 +249,7 @@ export const layer = Layer.effect(
|
||||||
])
|
])
|
||||||
.run()
|
.run()
|
||||||
.pipe(Effect.orDie)
|
.pipe(Effect.orDie)
|
||||||
|
return serialized
|
||||||
}),
|
}),
|
||||||
{ behavior: "immediate" },
|
{ behavior: "immediate" },
|
||||||
)
|
)
|
||||||
|
|
@ -247,14 +264,15 @@ export const layer = Layer.effect(
|
||||||
for (const sync of syncHandlers) {
|
for (const sync of syncHandlers) {
|
||||||
yield* sync(event as Payload)
|
yield* sync(event as Payload)
|
||||||
}
|
}
|
||||||
yield* commitSyncEvent(event as Payload)
|
const sync = yield* commitSyncEvent(event as Payload)
|
||||||
|
const payload = sync ? { ...event, sync } : event
|
||||||
for (const listener of listeners) {
|
for (const listener of listeners) {
|
||||||
yield* listener(event as Payload)
|
yield* listener(payload as Payload)
|
||||||
}
|
}
|
||||||
const pubsub = typed.get(event.type)
|
const pubsub = typed.get(event.type)
|
||||||
if (pubsub) yield* PubSub.publish(pubsub, event as Payload)
|
if (pubsub) yield* PubSub.publish(pubsub, payload as Payload)
|
||||||
yield* PubSub.publish(all, event as Payload)
|
yield* PubSub.publish(all, payload as Payload)
|
||||||
return event
|
return payload
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -109,6 +109,19 @@ describe("EventV2", () => {
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
it.effect("publishes sync metadata", () =>
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const events = yield* EventV2.Service
|
||||||
|
const aggregateID = EventV2.ID.create()
|
||||||
|
const event = yield* events.publish(SyncMessage, { id: aggregateID, text: "hello" })
|
||||||
|
|
||||||
|
expect(event.sync).toEqual({
|
||||||
|
seq: 0,
|
||||||
|
aggregateID,
|
||||||
|
})
|
||||||
|
}),
|
||||||
|
)
|
||||||
|
|
||||||
it.effect("stores definitions in the exported registry", () =>
|
it.effect("stores definitions in the exported registry", () =>
|
||||||
Effect.sync(() => {
|
Effect.sync(() => {
|
||||||
expect(EventV2.registry.get(Message.type)).toBe(Message)
|
expect(EventV2.registry.get(Message.type)).toBe(Message)
|
||||||
|
|
|
||||||
|
|
@ -291,7 +291,7 @@ export const { use: useSyncV2, provider: SyncProviderV2 } = createSimpleContext(
|
||||||
message: {
|
message: {
|
||||||
async sync(sessionID: string) {
|
async sync(sessionID: string) {
|
||||||
const response = await sdk.client.v2.session.messages({ sessionID })
|
const response = await sdk.client.v2.session.messages({ sessionID })
|
||||||
setStore("messages", sessionID, reconcile(response.data?.items ?? []))
|
setStore("messages", sessionID, reconcile(response.data?.data.items ?? []))
|
||||||
},
|
},
|
||||||
fromSession(sessionID: string) {
|
fromSession(sessionID: string) {
|
||||||
const messages = store.messages[sessionID]
|
const messages = store.messages[sessionID]
|
||||||
|
|
|
||||||
|
|
@ -44,6 +44,21 @@ export const layer = Layer.effect(
|
||||||
workspace: workspaceID,
|
workspace: workspaceID,
|
||||||
payload: { id: event.id, type: event.type, properties: event.data },
|
payload: { id: event.id, type: event.type, properties: event.data },
|
||||||
})
|
})
|
||||||
|
if (!event.sync || event.version === undefined) return
|
||||||
|
GlobalBus.emit("event", {
|
||||||
|
directory: event.location?.directory ?? ctx?.directory,
|
||||||
|
project: ctx?.project.id,
|
||||||
|
workspace: workspaceID,
|
||||||
|
payload: {
|
||||||
|
type: "sync",
|
||||||
|
syncEvent: {
|
||||||
|
id: event.id,
|
||||||
|
type: EventV2.versionedType(event.type, event.version),
|
||||||
|
...event.sync,
|
||||||
|
data: event.data,
|
||||||
|
},
|
||||||
|
},
|
||||||
|
})
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
yield* Effect.addFinalizer(() => unsubscribe)
|
yield* Effect.addFinalizer(() => unsubscribe)
|
||||||
|
|
|
||||||
|
|
@ -20,11 +20,14 @@ const SyncEventSchemas = EventV2.registry
|
||||||
return [
|
return [
|
||||||
Schema.Struct({
|
Schema.Struct({
|
||||||
type: Schema.Literal("sync"),
|
type: Schema.Literal("sync"),
|
||||||
name: Schema.Literal(EventV2.versionedType(definition.type, definition.sync.version)),
|
|
||||||
id: Schema.String,
|
id: Schema.String,
|
||||||
seq: Schema.Finite,
|
syncEvent: Schema.Struct({
|
||||||
aggregateID: Schema.Literal(definition.sync.aggregate),
|
type: Schema.Literal(EventV2.versionedType(definition.type, definition.sync.version)),
|
||||||
data: definition.data,
|
id: Schema.String,
|
||||||
|
seq: Schema.Finite,
|
||||||
|
aggregateID: Schema.String,
|
||||||
|
data: definition.data,
|
||||||
|
}),
|
||||||
}).annotate({ identifier: `SyncEvent.${definition.type}` }),
|
}).annotate({ identifier: `SyncEvent.${definition.type}` }),
|
||||||
]
|
]
|
||||||
})
|
})
|
||||||
|
|
|
||||||
|
|
@ -3,7 +3,13 @@ import { OpenApi } from "effect/unstable/httpapi"
|
||||||
import { PublicApi } from "../../src/server/routes/instance/httpapi/public"
|
import { PublicApi } from "../../src/server/routes/instance/httpapi/public"
|
||||||
|
|
||||||
type Method = "get" | "post" | "put" | "delete" | "patch"
|
type Method = "get" | "post" | "put" | "delete" | "patch"
|
||||||
type OpenApiSchema = { readonly $ref?: string }
|
type OpenApiSchema = {
|
||||||
|
readonly $ref?: string
|
||||||
|
readonly type?: string
|
||||||
|
readonly enum?: readonly unknown[]
|
||||||
|
readonly properties?: Record<string, OpenApiSchema>
|
||||||
|
readonly required?: readonly string[]
|
||||||
|
}
|
||||||
type OpenApiResponse = {
|
type OpenApiResponse = {
|
||||||
readonly description?: string
|
readonly description?: string
|
||||||
readonly content?: Record<string, { readonly schema?: OpenApiSchema }>
|
readonly content?: Record<string, { readonly schema?: OpenApiSchema }>
|
||||||
|
|
@ -19,7 +25,10 @@ type OpenApiOperation = {
|
||||||
readonly security?: unknown
|
readonly security?: unknown
|
||||||
}
|
}
|
||||||
type OpenApiPathItem = Partial<Record<Method, OpenApiOperation>>
|
type OpenApiPathItem = Partial<Record<Method, OpenApiOperation>>
|
||||||
type OpenApiSpec = { readonly paths: Record<string, OpenApiPathItem> }
|
type OpenApiSpec = {
|
||||||
|
readonly paths: Record<string, OpenApiPathItem>
|
||||||
|
readonly components: { readonly schemas: Record<string, OpenApiSchema> }
|
||||||
|
}
|
||||||
|
|
||||||
const methods = ["get", "post", "put", "delete", "patch"] as const
|
const methods = ["get", "post", "put", "delete", "patch"] as const
|
||||||
|
|
||||||
|
|
@ -49,6 +58,23 @@ function isBuiltInEndpointError(name: string) {
|
||||||
}
|
}
|
||||||
|
|
||||||
describe("PublicApi OpenAPI v2 errors", () => {
|
describe("PublicApi OpenAPI v2 errors", () => {
|
||||||
|
test("documents nested legacy global sync events", () => {
|
||||||
|
const spec = OpenApi.fromApi(PublicApi) as OpenApiSpec
|
||||||
|
const schema = spec.components.schemas.SyncEventSessionCreated
|
||||||
|
|
||||||
|
expect(schema?.required).toEqual(["type", "id", "syncEvent"])
|
||||||
|
expect(schema?.properties?.type?.enum).toEqual(["sync"])
|
||||||
|
expect(schema?.properties?.syncEvent).toMatchObject({
|
||||||
|
required: ["type", "id", "seq", "aggregateID", "data"],
|
||||||
|
properties: {
|
||||||
|
type: { enum: ["session.created.1"] },
|
||||||
|
id: { type: "string" },
|
||||||
|
seq: { type: "number" },
|
||||||
|
aggregateID: { type: "string" },
|
||||||
|
},
|
||||||
|
})
|
||||||
|
})
|
||||||
|
|
||||||
test("preserves /api auth responses", () => {
|
test("preserves /api auth responses", () => {
|
||||||
const spec = OpenApi.fromApi(PublicApi) as OpenApiSpec
|
const spec = OpenApi.fromApi(PublicApi) as OpenApiSpec
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,6 +1,7 @@
|
||||||
import { describe, expect } from "bun:test"
|
import { describe, expect } from "bun:test"
|
||||||
import { SessionV1 } from "@opencode-ai/core/v1/session"
|
import { SessionV1 } from "@opencode-ai/core/v1/session"
|
||||||
import { Database } from "@opencode-ai/core/database/database"
|
import { Database } from "@opencode-ai/core/database/database"
|
||||||
|
import { EventV2 } from "@opencode-ai/core/event"
|
||||||
import { SessionProjector } from "@opencode-ai/core/session/projector"
|
import { SessionProjector } from "@opencode-ai/core/session/projector"
|
||||||
import { Deferred, Effect, Exit, Layer } from "effect"
|
import { Deferred, Effect, Exit, Layer } from "effect"
|
||||||
import { Session as SessionNs } from "@/session/session"
|
import { Session as SessionNs } from "@/session/session"
|
||||||
|
|
@ -14,6 +15,7 @@ import { Storage } from "@/storage/storage"
|
||||||
import { RuntimeFlags } from "@/effect/runtime-flags"
|
import { RuntimeFlags } from "@/effect/runtime-flags"
|
||||||
import { BackgroundJob } from "@/background/job"
|
import { BackgroundJob } from "@/background/job"
|
||||||
import { EventV2Bridge } from "@/event-v2-bridge"
|
import { EventV2Bridge } from "@/event-v2-bridge"
|
||||||
|
import { GlobalBus } from "@/bus/global"
|
||||||
|
|
||||||
void Log.init({ print: false })
|
void Log.init({ print: false })
|
||||||
|
|
||||||
|
|
@ -101,6 +103,31 @@ describe("session.created event", () => {
|
||||||
yield* session.remove(info.id)
|
yield* session.remove(info.id)
|
||||||
}),
|
}),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
it.instance("emits legacy global sync payload", () =>
|
||||||
|
Effect.gen(function* () {
|
||||||
|
const session = yield* SessionNs.Service
|
||||||
|
const received = yield* Deferred.make<{ syncEvent: EventV2.SerializedEvent }>()
|
||||||
|
const listener = (event: { payload: { type?: string; syncEvent?: EventV2.SerializedEvent } }) => {
|
||||||
|
if (event.payload.type === "sync" && event.payload.syncEvent)
|
||||||
|
Deferred.doneUnsafe(received, Effect.succeed({ syncEvent: event.payload.syncEvent }))
|
||||||
|
}
|
||||||
|
GlobalBus.on("event", listener)
|
||||||
|
yield* Effect.addFinalizer(() => Effect.sync(() => GlobalBus.off("event", listener)))
|
||||||
|
|
||||||
|
const info = yield* session.create({})
|
||||||
|
const event = yield* awaitDeferred(received, "timed out waiting for legacy global sync event")
|
||||||
|
|
||||||
|
expect(event.syncEvent).toMatchObject({
|
||||||
|
type: EventV2.versionedType(SessionNs.Event.Created.type, 1),
|
||||||
|
seq: 0,
|
||||||
|
aggregateID: info.id,
|
||||||
|
data: { sessionID: info.id },
|
||||||
|
})
|
||||||
|
|
||||||
|
yield* session.remove(info.id)
|
||||||
|
}),
|
||||||
|
)
|
||||||
})
|
})
|
||||||
|
|
||||||
describe("step-finish token propagation via event", () => {
|
describe("step-finish token propagation via event", () => {
|
||||||
|
|
|
||||||
|
|
@ -255,6 +255,10 @@ import type {
|
||||||
TuiShowToastResponses,
|
TuiShowToastResponses,
|
||||||
TuiSubmitPromptErrors,
|
TuiSubmitPromptErrors,
|
||||||
TuiSubmitPromptResponses,
|
TuiSubmitPromptResponses,
|
||||||
|
V2CommandListErrors,
|
||||||
|
V2CommandListResponses,
|
||||||
|
V2EventSubscribeErrors,
|
||||||
|
V2EventSubscribeResponses,
|
||||||
V2FsListErrors,
|
V2FsListErrors,
|
||||||
V2FsListResponses,
|
V2FsListResponses,
|
||||||
V2FsReadErrors,
|
V2FsReadErrors,
|
||||||
|
|
@ -287,6 +291,8 @@ import type {
|
||||||
V2SessionPromptResponses,
|
V2SessionPromptResponses,
|
||||||
V2SessionWaitErrors,
|
V2SessionWaitErrors,
|
||||||
V2SessionWaitResponses,
|
V2SessionWaitResponses,
|
||||||
|
V2SkillListErrors,
|
||||||
|
V2SkillListResponses,
|
||||||
VcsApplyErrors,
|
VcsApplyErrors,
|
||||||
VcsApplyResponses,
|
VcsApplyResponses,
|
||||||
VcsDiffErrors,
|
VcsDiffErrors,
|
||||||
|
|
@ -4990,6 +4996,78 @@ export class Fs extends HeyApiClient {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export class Command2 extends HeyApiClient {
|
||||||
|
/**
|
||||||
|
* List v2 commands
|
||||||
|
*
|
||||||
|
* Retrieve currently registered v2 commands.
|
||||||
|
*/
|
||||||
|
public list<ThrowOnError extends boolean = false>(
|
||||||
|
parameters?: {
|
||||||
|
location?: {
|
||||||
|
directory?: string
|
||||||
|
workspace?: string
|
||||||
|
}
|
||||||
|
},
|
||||||
|
options?: Options<never, ThrowOnError>,
|
||||||
|
) {
|
||||||
|
const params = buildClientParams([parameters], [{ args: [{ in: "query", key: "location" }] }])
|
||||||
|
return (options?.client ?? this.client).get<V2CommandListResponses, V2CommandListErrors, ThrowOnError>({
|
||||||
|
url: "/api/command",
|
||||||
|
...options,
|
||||||
|
...params,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export class Skill extends HeyApiClient {
|
||||||
|
/**
|
||||||
|
* List v2 skills
|
||||||
|
*
|
||||||
|
* Retrieve currently registered v2 skills.
|
||||||
|
*/
|
||||||
|
public list<ThrowOnError extends boolean = false>(
|
||||||
|
parameters?: {
|
||||||
|
location?: {
|
||||||
|
directory?: string
|
||||||
|
workspace?: string
|
||||||
|
}
|
||||||
|
},
|
||||||
|
options?: Options<never, ThrowOnError>,
|
||||||
|
) {
|
||||||
|
const params = buildClientParams([parameters], [{ args: [{ in: "query", key: "location" }] }])
|
||||||
|
return (options?.client ?? this.client).get<V2SkillListResponses, V2SkillListErrors, ThrowOnError>({
|
||||||
|
url: "/api/skill",
|
||||||
|
...options,
|
||||||
|
...params,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
export class Event2 extends HeyApiClient {
|
||||||
|
/**
|
||||||
|
* Subscribe to v2 events
|
||||||
|
*
|
||||||
|
* Subscribe to native EventV2 payloads for a location.
|
||||||
|
*/
|
||||||
|
public subscribe<ThrowOnError extends boolean = false>(
|
||||||
|
parameters?: {
|
||||||
|
location?: {
|
||||||
|
directory?: string
|
||||||
|
workspace?: string
|
||||||
|
}
|
||||||
|
},
|
||||||
|
options?: Options<never, ThrowOnError>,
|
||||||
|
) {
|
||||||
|
const params = buildClientParams([parameters], [{ args: [{ in: "query", key: "location" }] }])
|
||||||
|
return (options?.client ?? this.client).sse.get<V2EventSubscribeResponses, V2EventSubscribeErrors, ThrowOnError>({
|
||||||
|
url: "/api/event",
|
||||||
|
...options,
|
||||||
|
...params,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
export class V2 extends HeyApiClient {
|
export class V2 extends HeyApiClient {
|
||||||
private _session?: Session3
|
private _session?: Session3
|
||||||
get session(): Session3 {
|
get session(): Session3 {
|
||||||
|
|
@ -5015,6 +5093,21 @@ export class V2 extends HeyApiClient {
|
||||||
get fs(): Fs {
|
get fs(): Fs {
|
||||||
return (this._fs ??= new Fs({ client: this.client }))
|
return (this._fs ??= new Fs({ client: this.client }))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private _command?: Command2
|
||||||
|
get command(): Command2 {
|
||||||
|
return (this._command ??= new Command2({ client: this.client }))
|
||||||
|
}
|
||||||
|
|
||||||
|
private _skill?: Skill
|
||||||
|
get skill(): Skill {
|
||||||
|
return (this._skill ??= new Skill({ client: this.client }))
|
||||||
|
}
|
||||||
|
|
||||||
|
private _event?: Event2
|
||||||
|
get event(): Event2 {
|
||||||
|
return (this._event ??= new Event2({ client: this.client }))
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
export class Control extends HeyApiClient {
|
export class Control extends HeyApiClient {
|
||||||
|
|
|
||||||
File diff suppressed because it is too large
Load diff
Loading…
Add table
Add a link
Reference in a new issue