Skip to content

refactor(sync): migrate SyncEvent to service layer #21988

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
kitlangton wants to merge 6 commits into
base: dev
Choose a base branch
from
Draft

refactor(sync): migrate SyncEvent to service layer #21988

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
af7ada9
refactor(sync): migrate SyncEvent to service layer
kitlangton Apr 11, 2026
f8580ab
Merge branch 'dev' into kit/effect-sync-event
kitlangton Apr 11, 2026
e2c41dc
Merge branch 'dev' into kit/effect-sync-event
kitlangton Apr 11, 2026
dbc136d
Merge branch 'dev' into kit/effect-sync-event
kitlangton Apr 11, 2026
397adb4
fix(sync): restore instance context for deferred publish
kitlangton Apr 11, 2026
c3eb241
fix(sync): preserve shared state and caller semantics
kitlangton Apr 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 30 additions & 30 deletions packages/opencode/src/session/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -366,11 +366,12 @@ export namespace Session {
const db = <T>(fn: (d: Parameters<typeof Database.use>[0] extends (trx: infer D) => any ? D : never) => T) =>
Effect.sync(() => Database.use(fn))

export const layer: Layer.Layer<Service, never, Bus.Service | Storage.Service> = Layer.effect(
export const layer: Layer.Layer<Service, never, Bus.Service | Storage.Service | SyncEvent.Service> = Layer.effect(
Service,
Effect.gen(function* () {
const bus = yield* Bus.Service
const storage = yield* Storage.Service
const sync = yield* SyncEvent.Service

const createNext = Effect.fn("Session.createNext")(function* (input: {
id?: SessionID
Expand Down Expand Up @@ -398,7 +399,7 @@ export namespace Session {
}
log.info("created", result)

yield* Effect.sync(() => SyncEvent.run(Event.Created, { sessionID: result.id, info: result }))
yield* sync.run(Event.Created, { sessionID: result.id, info: result })

if (!Flag.OPENCODE_EXPERIMENTAL_WORKSPACES) {
// This only exist for backwards compatibility. We should not be
Expand Down Expand Up @@ -446,30 +447,30 @@ export namespace Session {
Effect.catchCause(() => Effect.succeed(false)),
)

yield* Effect.sync(() => {
SyncEvent.run(Event.Deleted, { sessionID, info: session }, { publish: hasInstance })
SyncEvent.remove(sessionID)
})
if (hasInstance) {
yield* sync.run(Event.Deleted, { sessionID, info: session }, { publish: true })
} else {
yield* Effect.sync(() => SyncEvent.run(Event.Deleted, { sessionID, info: session }, { publish: false }))
}
yield* sync.remove(sessionID)
} catch (e) {
log.error(e)
}
})

const updateMessage = <T extends MessageV2.Info>(msg: T): Effect.Effect<T> =>
Effect.gen(function* () {
yield* Effect.sync(() => SyncEvent.run(MessageV2.Event.Updated, { sessionID: msg.sessionID, info: msg }))
yield* sync.run(MessageV2.Event.Updated, { sessionID: msg.sessionID, info: msg })
return msg
}).pipe(Effect.withSpan("Session.updateMessage"))

const updatePart = <T extends MessageV2.Part>(part: T): Effect.Effect<T> =>
Effect.gen(function* () {
yield* Effect.sync(() =>
SyncEvent.run(MessageV2.Event.PartUpdated, {
sessionID: part.sessionID,
part: structuredClone(part),
time: Date.now(),
}),
)
yield* sync.run(MessageV2.Event.PartUpdated, {
sessionID: part.sessionID,
part: structuredClone(part),
time: Date.now(),
})
return part
}).pipe(Effect.withSpan("Session.updatePart"))

Expand Down Expand Up @@ -549,8 +550,7 @@ export namespace Session {
return session
})

const patch = (sessionID: SessionID, info: Patch) =>
Effect.sync(() => SyncEvent.run(Event.Updated, { sessionID, info }))
const patch = (sessionID: SessionID, info: Patch) => sync.run(Event.Updated, { sessionID, info })

const touch = Effect.fn("Session.touch")(function* (sessionID: SessionID) {
yield* patch(sessionID, { time: { updated: Date.now() } })
Expand Down Expand Up @@ -607,12 +607,10 @@ export namespace Session {
sessionID: SessionID
messageID: MessageID
}) {
yield* Effect.sync(() =>
SyncEvent.run(MessageV2.Event.Removed, {
sessionID: input.sessionID,
messageID: input.messageID,
}),
)
yield* sync.run(MessageV2.Event.Removed, {
sessionID: input.sessionID,
messageID: input.messageID,
})
return input.messageID
})

Expand All @@ -621,13 +619,11 @@ export namespace Session {
messageID: MessageID
partID: PartID
}) {
yield* Effect.sync(() =>
SyncEvent.run(MessageV2.Event.PartRemoved, {
sessionID: input.sessionID,
messageID: input.messageID,
partID: input.partID,
}),
)
yield* sync.run(MessageV2.Event.PartRemoved, {
sessionID: input.sessionID,
messageID: input.messageID,
partID: input.partID,
})
return input.partID
})

Expand Down Expand Up @@ -678,7 +674,11 @@ export namespace Session {
}),
)

export const defaultLayer = layer.pipe(Layer.provide(Bus.layer), Layer.provide(Storage.defaultLayer))
export const defaultLayer = layer.pipe(
Layer.provide(Bus.layer),
Layer.provide(Storage.defaultLayer),
Layer.provide(SyncEvent.defaultLayer),
)

const { runPromise } = makeRuntime(Service, defaultLayer)

Expand Down
6 changes: 4 additions & 2 deletions packages/opencode/src/session/revert.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,6 +40,7 @@ export namespace SessionRevert {
const bus = yield* Bus.Service
const summary = yield* SessionSummary.Service
const state = yield* SessionRunState.Service
const sync = yield* SyncEvent.Service

const revert = Effect.fn("SessionRevert.revert")(function* (input: RevertInput) {
yield* state.assertNotBusy(input.sessionID)
Expand Down Expand Up @@ -123,7 +124,7 @@ export namespace SessionRevert {
remove.push(msg)
}
for (const msg of remove) {
SyncEvent.run(MessageV2.Event.Removed, {
yield* sync.run(MessageV2.Event.Removed, {
sessionID,
messageID: msg.info.id,
})
Expand All @@ -135,7 +136,7 @@ export namespace SessionRevert {
const removeParts = target.parts.slice(idx)
target.parts = target.parts.slice(0, idx)
for (const part of removeParts) {
SyncEvent.run(MessageV2.Event.PartRemoved, {
yield* sync.run(MessageV2.Event.PartRemoved, {
sessionID,
messageID: target.info.id,
partID: part.id,
Expand All @@ -158,6 +159,7 @@ export namespace SessionRevert {
Layer.provide(Storage.defaultLayer),
Layer.provide(Bus.layer),
Layer.provide(SessionSummary.defaultLayer),
Layer.provide(SyncEvent.defaultLayer),
),
)

Expand Down
8 changes: 4 additions & 4 deletions packages/opencode/src/share/session.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,20 +24,19 @@ export namespace SessionShare {
const session = yield* Session.Service
const shareNext = yield* ShareNext.Service
const scope = yield* Scope.Scope
const sync = yield* SyncEvent.Service

const share = Effect.fn("SessionShare.share")(function* (sessionID: SessionID) {
const conf = yield* cfg.get()
if (conf.share === "disabled") throw new Error("Sharing is disabled in configuration")
const result = yield* shareNext.create(sessionID)
yield* Effect.sync(() =>
SyncEvent.run(Session.Event.Updated, { sessionID, info: { share: { url: result.url } } }),
)
yield* sync.run(Session.Event.Updated, { sessionID, info: { share: { url: result.url } } })
return result
})

const unshare = Effect.fn("SessionShare.unshare")(function* (sessionID: SessionID) {
yield* shareNext.remove(sessionID)
yield* Effect.sync(() => SyncEvent.run(Session.Event.Updated, { sessionID, info: { share: { url: null } } }))
yield* sync.run(Session.Event.Updated, { sessionID, info: { share: { url: null } } })
})

const create = Effect.fn("SessionShare.create")(function* (input?: Parameters<typeof Session.create>[0]) {
Expand All @@ -57,6 +56,7 @@ export namespace SessionShare {
Layer.provide(ShareNext.defaultLayer),
Layer.provide(Session.defaultLayer),
Layer.provide(Config.defaultLayer),
Layer.provide(SyncEvent.defaultLayer),
)

const { runPromise } = makeRuntime(Service, defaultLayer)
Expand Down
Loading
Loading