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

fix(sdk): preserve session stream ordering

Browse source
This commit is contained in:
Kit Langton 2026-06-25 23:37:44 -04:00
commit 5146f01e0a
8 changed files with 252 additions and 95 deletions
Download patch file Download diff file Expand all files Collapse all files

103
packages/core/src/event.ts
View file

@ -72,7 +72,11 @@ export interface Interface {
readonly after?: number
readonly live: (event: Payload) => boolean
}) => Effect.Effect<
{ readonly replay: ReadonlyArray<Payload>; readonly updates: Stream.Stream<Payload> },
{
readonly replay: ReadonlyArray<Payload>
readonly updates: Stream.Stream<Payload>
readonly offer: (event: Payload, position?: "after" | "before") => boolean
},
never,
Scope.Scope
>
@ -103,12 +107,12 @@ export const layerWith = (options?: LayerOptions) =>
Effect.gen(function* () {
const pubsub = {
all: yield* PubSub.unbounded<Payload>(),
durable: new Map<string, Set<PubSub.PubSub<void>>>(),
durable: new Map<string, Set<PubSub.PubSub<number>>>(),
typed: new Map<string, PubSub.PubSub<Payload>>(),
}
const projectors = new Map<string, Subscriber[]>()
// TODO: Bind durable projectors to exact type+version before supporting incompatible historical payloads.
const listeners = new Array<Subscriber>()
const listeners = new Set<Subscriber>()
const { db } = yield* Database.Service
const getOrCreate = (definition: Definition) =>
@ -284,7 +288,7 @@ export const layerWith = (options?: LayerOptions) =>
if (committed) {
yield* Effect.forEach(
pubsub.durable.get(committed.aggregateID) ?? [],
(wake) => PubSub.publish(wake, undefined),
(wake) => PubSub.publish(wake, committed.seq),
{ discard: true },
)
}
@ -513,7 +517,7 @@ export const layerWith = (options?: LayerOptions) =>
const subscribeDurable = (aggregateID: string) =>
Effect.gen(function* () {
const wake = yield* PubSub.sliding<void>(1)
const wake = yield* PubSub.sliding<number>(1)
const subscription = yield* PubSub.subscribe(wake)
yield* Effect.acquireRelease(
Effect.sync(() => {
@ -531,70 +535,83 @@ export const layerWith = (options?: LayerOptions) =>
return subscription
})
const aggregateDrain = (
aggregateID: string,
after: number,
): ((through?: number) => Effect.Effect<ReadonlyArray<Payload>>) => {
let sequence = after
return (through?: number) =>
through !== undefined && through <= sequence
? Effect.succeed<ReadonlyArray<Payload>>([])
: Effect.suspend(() => readAfter(aggregateID, sequence, through)).pipe(
Effect.tap((events) =>
Effect.sync(() => {
sequence = events.at(-1)?.durable?.seq ?? sequence
}),
),
)
}
const durable = (input: { readonly aggregateID: string; readonly after?: number }): Stream.Stream<Payload> =>
Stream.unwrap(
Effect.gen(function* () {
const wakes = yield* subscribeDurable(input.aggregateID)
let sequence = input.after ?? -1
const read = Effect.suspend(() => readAfter(input.aggregateID, sequence)).pipe(
Effect.tap((events) =>
Effect.sync(() => {
sequence = events.at(-1)?.durable?.seq ?? sequence
}),
),
)
const historical = yield* read
const live = Stream.fromSubscription(wakes).pipe(
Stream.mapEffect(() => read),
Stream.flattenIterable,
)
const drain = aggregateDrain(input.aggregateID, input.after ?? -1)
const historical = yield* drain()
const live = Stream.fromSubscription(wakes).pipe(Stream.mapEffect(drain), Stream.flattenIterable)
return Stream.concat(Stream.fromIterable(historical), live)
}),
)
const observeAggregate: Interface["observeAggregate"] = (input) =>
Effect.gen(function* () {
type Signal = { readonly _tag: "durable" } | { readonly _tag: "live"; readonly event: Payload }
type Signal =
| { readonly _tag: "durable" }
| { readonly _tag: "transient"; readonly event: Payload; readonly position: "after" | "before" }
const signals = yield* Queue.dropping<Signal, Cause.Done>(256)
const wakes = yield* subscribeDurable(input.aggregateID)
let durableQueued = false
let durableThrough = -1
const offer = (event: Payload, position: "after" | "before" = "after") => {
const offered = Queue.offerUnsafe(signals, { _tag: "transient", event, position })
if (!offered) Queue.endUnsafe(signals)
return offered
}
const unsubscribe = yield* listen((event) =>
Effect.sync(() => {
if (event.durable?.aggregateID === input.aggregateID) {
// Durable payloads are only wakeups; a full queue already contains work that will drain the database.
Queue.offerUnsafe(signals, { _tag: "durable" })
return
}
if (!input.live(event)) return
if (!Queue.offerUnsafe(signals, { _tag: "live", event })) Queue.endUnsafe(signals)
if (input.live(event)) offer(event)
}),
)
yield* Effect.addFinalizer(() => unsubscribe.pipe(Effect.andThen(Queue.shutdown(signals))))
yield* Stream.runForEach(Stream.fromSubscription(wakes), (sequence) =>
Effect.sync(() => {
durableThrough = Math.max(durableThrough, sequence)
if (durableQueued) return
durableQueued = Queue.offerUnsafe(signals, { _tag: "durable" })
}),
).pipe(Effect.forkScoped)
const cutoff = yield* latestSequence(db, input.aggregateID)
const replay = yield* readAfter(input.aggregateID, input.after ?? -1, cutoff)
let sequence = cutoff
const drain = Effect.suspend(() => readAfter(input.aggregateID, sequence)).pipe(
Effect.tap((events) =>
Effect.sync(() => {
sequence = events.at(-1)?.durable?.seq ?? sequence
}),
),
)
const drain = aggregateDrain(input.aggregateID, cutoff)
const updates = Stream.fromQueue(signals).pipe(
Stream.mapEffect((signal) =>
drain.pipe(Effect.map((events) => (signal._tag === "live" ? [...events, signal.event] : events))),
),
Stream.mapEffect((signal) => {
if (signal._tag === "durable") {
durableQueued = false
return drain(durableThrough)
}
if (signal.position === "before") return Effect.succeed([signal.event])
return drain().pipe(Effect.map((events) => [...events, signal.event]))
}),
Stream.flattenIterable,
)
return { replay, updates }
return { replay, updates, offer }
})
const listen = (listener: Subscriber): Effect.Effect<Unsubscribe> =>
Effect.sync(() => {
listeners.push(listener)
return Effect.sync(() => {
const index = listeners.indexOf(listener)
if (index >= 0) listeners.splice(index, 1)
})
listeners.add(listener)
return Effect.sync(() => listeners.delete(listener)).pipe(Effect.asVoid)
})
const project = <D extends Definition>(definition: D, projector: Subscriber<D>): Effect.Effect<void> =>

45
packages/core/src/session.ts
View file

@ -185,8 +185,11 @@ export const layer = Layer.unwrap(
const store = yield* SessionStore.Service
const locations = yield* LocationServiceMap
const decodeMessage = Schema.decodeUnknownEffect(SessionMessage.Message)
const isDurableSessionEvent = Schema.is(SessionEvent.Durable)
const sessionEventTypes = new Set<string>(SessionEvent.Definitions.map((definition) => definition.type))
const isSessionEvent = (event: EventV2.Payload): event is SessionEvent.Event =>
sessionEventTypes.has(event.type)
const isDurableSessionEvent = (event: EventV2.Payload): event is SessionEvent.DurableEvent =>
event.durable !== undefined && isSessionEvent(event)
const decode = (row: typeof SessionMessageTable.$inferSelect) =>
decodeMessage({ ...row.data, id: row.id, type: row.type }).pipe(
Effect.mapError(
@ -350,36 +353,24 @@ export const layer = Layer.unwrap(
aggregateID: input.sessionID,
after: input.after,
live: (event) =>
event.durable === undefined &&
sessionEventTypes.has(event.type) &&
typeof event.data === "object" &&
event.data !== null &&
"sessionID" in event.data &&
event.data.sessionID === input.sessionID,
event.durable === undefined && isSessionEvent(event) && event.data.sessionID === input.sessionID,
})
const initialActivity: SessionEvent.Activity = {
id: EventV2.ID.create(),
type: SessionEvent.Activity.type,
data: { sessionID: input.sessionID, active: yield* activity.snapshot },
}
const replay = observed.replay.filter((event): event is SessionEvent.DurableEvent =>
isDurableSessionEvent(event),
)
const updates = observed.updates.pipe(
Stream.filter((event): event is SessionEvent.Event => Schema.is(SessionEvent.All)(event)),
)
const activityChanges = activity.changes.pipe(
Stream.map(
(active): SessionEvent.Activity => ({
id: EventV2.ID.create(),
type: SessionEvent.Activity.type,
data: { sessionID: input.sessionID, active },
}),
const initialActivity = SessionEvent.makeActivity(
input.sessionID,
yield* activity.attach((active) =>
observed.offer(SessionEvent.makeActivity(input.sessionID, active), active ? "before" : "after"),
),
)
return Stream.fromIterable(replay).pipe(
return Stream.fromIterable(observed.replay.filter(isDurableSessionEvent)).pipe(
Stream.concat(Stream.make(initialActivity)),
Stream.concat(Stream.merge(updates, activityChanges)),
Stream.concat(
observed.updates.pipe(
Stream.filter(
(event): event is SessionEvent.StreamEvent =>
event.type === SessionEvent.Activity.type || isSessionEvent(event),
),
),
),
)
}),
),

13
packages/core/src/session/execution.ts
View file

@ -1,20 +1,15 @@
export * as SessionExecution from "./execution"
import { Context, Effect, Layer, Scope, Stream } from "effect"
import { Context, Effect, Layer, Scope } from "effect"
import { SessionRunner } from "./runner/index"
import { SessionRunCoordinator } from "./run-coordinator"
import { SessionSchema } from "./schema"
export interface Interface {
/** Snapshots active execution owned by this process. */
readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
/** Observes foreground ownership with an authoritative initial snapshot. */
readonly activity: (
sessionID: SessionSchema.ID,
) => Effect.Effect<
{ readonly snapshot: Effect.Effect<boolean>; readonly changes: Stream.Stream<boolean> },
never,
Scope.Scope
>
readonly activity: (sessionID: SessionSchema.ID) => Effect.Effect<SessionRunCoordinator.Activity, never, Scope.Scope>
/** Starts execution while idle or joins the active execution. */
readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
/** Registers newly recorded work. Repeated wakeups may coalesce. */
@ -31,7 +26,7 @@ export const noopLayer = Layer.succeed(
Service,
Service.of({
active: Effect.succeed(new Set()),
activity: () => Effect.succeed({ snapshot: Effect.succeed(false), changes: Stream.never }),
activity: () => Effect.succeed({ attach: () => Effect.succeed(false) }),
resume: () => Effect.void,
wake: () => Effect.void,
interrupt: () => Effect.void,

20
packages/core/src/session/run-coordinator.ts
View file

@ -1,11 +1,10 @@
export * as SessionRunCoordinator from "./run-coordinator"
import { Cause, Deferred, Effect, Exit, Fiber, FiberSet, Queue, Scope, Stream } from "effect"
import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect"
export interface Activity {
/** Discards earlier transitions and snapshots current ownership. */
readonly snapshot: Effect.Effect<boolean>
readonly changes: Stream.Stream<boolean>
/** Installs a synchronous observer and snapshots current ownership atomically. */
readonly attach: (observer: (active: boolean) => void) => Effect.Effect<boolean>
}
/** Serializes execution for each key while allowing different keys to run concurrently. */
@ -119,9 +118,9 @@ export const make = <Key, E>(options: {
const activity = (key: Key) =>
Effect.gen(function* () {
const queue = yield* Queue.dropping<boolean, Cause.Done>(256)
let attached: ((active: boolean) => void) | undefined
const observer = (value: boolean) => {
if (!Queue.offerUnsafe(queue, value)) Queue.endUnsafe(queue)
attached?.(value)
}
yield* Effect.acquireRelease(
Effect.sync(() => {
@ -134,11 +133,14 @@ export const make = <Key, E>(options: {
const observers = activityObservers.get(key)
observers?.delete(observer)
if (observers?.size === 0) activityObservers.delete(key)
}).pipe(Effect.andThen(Queue.shutdown(queue))),
}),
)
return {
snapshot: Queue.clear(queue).pipe(Effect.andThen(Effect.sync(() => active.has(key)))),
changes: Stream.fromQueue(queue),
attach: (next: (active: boolean) => void) =>
Effect.sync(() => {
attached = next
return active.has(key)
}),
}
})