vrr/opencode
2
0
Fork 0
mirror of https://github.com/anomalyco/opencode.git synced 2026-07-10 04:48:32 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

refactor(api): remove watermark sync surface

Browse source
This commit is contained in:
Dax Raad 2026-07-06 12:59:26 -04:00
parent 4cee5bd824
commit ff499c3603
25 changed files with 141 additions and 392 deletions
Download patch file Download diff file Expand all files Collapse all files

4
packages/cli/src/mini/stream-v2.transport.ts
View file

@ -544,8 +544,8 @@ export async function createSessionTransport(input: StreamInput): Promise<Sessio
state.permissions = permissions.map(permission)
state.questions = questions.map(question)
syncBlockers()
await subagents.hydrate({ messages: [...projected], active: active.data })
const running = input.sessionID in active.data
await subagents.hydrate({ messages: [...projected], active })
const running = input.sessionID in active
write([], { phase: running ? "running" : "idle", status: running ? "assistant responding" : "" })
if (!running && state.wait && (state.wait.promoted || state.wait.interrupted)) {
const current = state.wait

6
packages/client/src/effect/api/api.ts
View file

@ -66,7 +66,7 @@ export type Endpoint4_1Input = {
export type Endpoint4_1Output = EffectValue<ReturnType<RawClient["server.session"]["session.create"]>>["data"]
export type SessionCreateOperation<E = never> = (input?: Endpoint4_1Input) => Effect.Effect<Endpoint4_1Output, E>
export type Endpoint4_2Output = EffectValue<ReturnType<RawClient["server.session"]["session.active"]>>
export type Endpoint4_2Output = EffectValue<ReturnType<RawClient["server.session"]["session.active"]>>["data"]
export type SessionActiveOperation<E = never> = () => Effect.Effect<Endpoint4_2Output, E>
type Endpoint4_3Request = Parameters<RawClient["server.session"]["session.get"]>[0]
@ -658,12 +658,8 @@ export interface SkillApi<E = never> {
export type Endpoint18_0Output = StreamValue<EffectValue<ReturnType<RawClient["server.event"]["event.subscribe"]>>>
export type EventSubscribeOperation<E = never> = () => Stream.Stream<Endpoint18_0Output, E>
export type Endpoint18_1Output = StreamValue<EffectValue<ReturnType<RawClient["server.event"]["event.changes"]>>>
export type EventChangesOperation<E = never> = () => Stream.Stream<Endpoint18_1Output, E>
export interface EventApi<E = never> {
readonly subscribe: EventSubscribeOperation<E>
readonly changes: EventChangesOperation<E>
}
type Endpoint19_0Request = Parameters<RawClient["server.pty"]["pty.list"]>[0]

15
packages/client/src/effect/generated/client.ts
View file

@ -82,7 +82,10 @@ const Endpoint4_1 = (raw: RawClient["server.session"]) => (input?: Endpoint4_1In
)
const Endpoint4_2 = (raw: RawClient["server.session"]) => () =>
raw["session.active"]({}).pipe(Effect.mapError(mapClientError))
raw["session.active"]({}).pipe(
Effect.mapError(mapClientError),
Effect.map((value) => value.data),
)
type Endpoint4_3Request = Parameters<RawClient["server.session"]["session.get"]>[0]
type Endpoint4_3Input = { readonly sessionID: Endpoint4_3Request["params"]["sessionID"] }
@ -796,15 +799,7 @@ const Endpoint18_0 = (raw: RawClient["server.event"]) => () =>
),
)
const Endpoint18_1 = (raw: RawClient["server.event"]) => () =>
Stream.unwrap(
raw["event.changes"]({}).pipe(
Effect.mapError(mapClientError),
Effect.map((stream) => stream.pipe(Stream.mapError(mapClientError))),
),
)
const adaptGroup18 = (raw: RawClient["server.event"]) => ({ subscribe: Endpoint18_0(raw), changes: Endpoint18_1(raw) })
const adaptGroup18 = (raw: RawClient["server.event"]) => ({ subscribe: Endpoint18_0(raw) })
type Endpoint19_0Request = Parameters<RawClient["server.pty"]["pty.list"]>[0]
type Endpoint19_0Input = { readonly location?: Endpoint19_0Request["query"]["location"] }

12
packages/client/src/promise/generated/client.ts
View file

@ -133,7 +133,6 @@ import type {
SkillListInput,
SkillListOutput,
EventSubscribeOutput,
EventChangesOutput,
PtyListInput,
PtyListOutput,
PtyCreateInput,
@ -402,7 +401,7 @@ export function make(options: ClientOptions) {
requestOptions,
).then((value) => value.data),
active: (requestOptions?: RequestOptions) =>
request<SessionActiveOutput>(
request<{ readonly data: SessionActiveOutput }>(
{
method: "GET",
path: `/api/session/active`,
@ -411,7 +410,7 @@ export function make(options: ClientOptions) {
empty: false,
},
requestOptions,
),
).then((value) => value.data),
get: (input: SessionGetInput, requestOptions?: RequestOptions) =>
request<{ readonly data: SessionGetOutput }>(
{
@ -646,7 +645,7 @@ export function make(options: ClientOptions) {
sse<SessionLogOutput>(
{
method: "GET",
path: `/api/session/${encodeURIComponent(input.sessionID)}/log`,
path: `/api/experimental/session/${encodeURIComponent(input.sessionID)}/log`,
query: { after: input["after"], follow: input["follow"] },
successStatus: 200,
declaredStatuses: [404, 400, 401],
@ -1183,11 +1182,6 @@ export function make(options: ClientOptions) {
{ method: "GET", path: `/api/event`, successStatus: 200, declaredStatuses: [401, 400], empty: false },
requestOptions,
),
changes: (requestOptions?: RequestOptions): AsyncIterable<EventChangesOutput> =>
sse<EventChangesOutput>(
{ method: "GET", path: `/api/event/changes`, successStatus: 200, declaredStatuses: [401, 400], empty: false },
requestOptions,
),
},
pty: {
list: (input?: PtyListInput, requestOptions?: RequestOptions) =>

11
packages/client/src/promise/generated/types.ts
View file

@ -354,7 +354,6 @@ export type SessionListOutput = {
}>
}
}>
readonly watermarks: { readonly [x: string]: number }
readonly cursor: { readonly previous?: string | null; readonly next?: string | null }
}
@ -419,10 +418,7 @@ export type SessionCreateOutput = {
}
}["data"]
export type SessionActiveOutput = {
readonly data: { readonly [x: string]: { readonly type: "running" } }
readonly watermarks: { readonly [x: string]: number }
}
export type SessionActiveOutput = { readonly data: { readonly [x: string]: { readonly type: "running" } } }["data"]
export type SessionGetInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] }
@ -1987,7 +1983,6 @@ export type MessageListOutput = {
readonly time: { readonly created: number }
}
>
readonly watermark?: number
readonly cursor: { readonly previous?: string | null; readonly next?: string | null }
}
@ -5546,10 +5541,6 @@ export type EventSubscribeOutput =
readonly data: {}
}
export type EventChangesOutput =
| { readonly type: "log.hint"; readonly aggregateID: string; readonly seq: number }
| { readonly type: "log.sweep_required" }
export type PtyListInput = {
readonly location?: {
readonly location?: { readonly directory?: string | undefined; readonly workspace?: string | undefined } | undefined

7
packages/client/test/effect.test.ts
View file

@ -99,7 +99,7 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
return Effect.succeed(
HttpClientResponse.fromWeb(
request,
Response.json({ data: { ses_test: { type: "running" } }, watermarks: { ses_test: 3 } }),
Response.json({ data: { ses_test: { type: "running" } } }),
),
)
}
@ -112,7 +112,7 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
return Effect.succeed(
HttpClientResponse.fromWeb(
request,
Response.json({ data: [session.data], watermarks: { ses_test: 3 }, cursor: { next: "next" } }),
Response.json({ data: [session.data], cursor: { next: "next" } }),
),
)
})
@ -148,8 +148,7 @@ test("session methods retain decoded Effect inputs and outputs", async () => {
}).pipe(Effect.provideService(HttpClient.HttpClient, httpClient), Effect.runPromise)
expect(DateTime.toEpochMillis(result.page.data[0].time.created)).toBe(1_717_171_717_000)
expect(result.active).toEqual({ data: { ses_test: { type: "running" } }, watermarks: { ses_test: 3 } })
expect(result.page.watermarks).toEqual({ ses_test: 3 })
expect(result.active).toEqual({ ses_test: { type: "running" } })
expect(Object.getPrototypeOf(result.page.data[0])).toBe(Object.prototype)
expect(Object.getPrototypeOf(result.created)).toBe(Object.prototype)
expect(result.created.id).toBe("ses_test")

8
packages/client/test/promise.test.ts
View file

@ -47,7 +47,7 @@ test("exposes every standard HTTP API group", () => {
expect(Object.keys(client.vcs)).toEqual(["status", "diff"])
expect(Object.keys(client.pty)).toEqual(["list", "create", "get", "update", "remove"])
expect(Object.keys(client.shell)).toEqual(["list", "create", "get", "output", "remove"])
expect(Object.keys(client.project)).toEqual(["current", "directories"])
expect(Object.keys(client.project)).toEqual(["list", "current", "directories"])
})
test("file.read returns binary content from the public HTTP contract", async () => {
@ -198,7 +198,7 @@ test("session methods use the public HTTP contract", async () => {
if (url.includes("/context")) return Response.json({ data: [] })
if (url.includes("/message/")) return Response.json({ data: modelSwitchedMessage })
if (url.endsWith("/api/session/active"))
return Response.json({ data: { ses_test: { type: "running" } }, watermarks: { ses_test: 3 } })
return Response.json({ data: { ses_test: { type: "running" } } })
if (init?.method === "POST" && url.endsWith("/api/session")) return Response.json(session)
if (init?.method === "POST") return new Response(null, { status: 204 })
return Response.json({ data: [session.data], cursor: { next: "next" } })
@ -227,7 +227,7 @@ test("session methods use the public HTTP contract", async () => {
const message = await client.session.message({ sessionID: "ses_test", messageID: "msg_model" })
expect(page.cursor.next).toBe("next")
expect(active).toEqual({ data: { ses_test: { type: "running" } }, watermarks: { ses_test: 3 } })
expect(active).toEqual({ ses_test: { type: "running" } })
expect(created.id).toBe("ses_test")
expect(admitted.id).toBe("msg_test")
expect(context).toEqual([])
@ -243,7 +243,7 @@ test("session methods use the public HTTP contract", async () => {
["POST", "http://localhost:3000/api/session/ses_test/compact"],
["POST", "http://localhost:3000/api/session/ses_test/wait"],
["GET", "http://localhost:3000/api/session/ses_test/context"],
["GET", "http://localhost:3000/api/session/ses_test/log?after=0"],
["GET", "http://localhost:3000/api/experimental/session/ses_test/log?after=0"],
["POST", "http://localhost:3000/api/session/ses_test/interrupt"],
["GET", "http://localhost:3000/api/session/ses_test/message/msg_model"],
])

1
packages/codemode/test/openapi.test.ts
View file

@ -204,7 +204,6 @@ describe("OpenAPI.fromSpec", () => {
expect(toolAt(result.tools, "v2.pty.connect")).toBeUndefined()
expect(toolAt(result.tools, "v2.session.log")).toBeUndefined()
expect(toolAt(result.tools, "v2.event.subscribe")).toBeUndefined()
expect(toolAt(result.tools, "v2.event.changes")).toBeUndefined()
expect(toolAt(result.tools, "v2.fs.read")).toBeUndefined()
expect(toolAt(result.tools, "v2.pty.connectToken")).not.toBeUndefined()
})

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

@ -135,12 +135,6 @@ export interface Interface {
readonly after?: number
readonly follow?: boolean
}) => Stream.Stream<LogItem>
/**
* Coalescing hint channel: latest committed seq per aggregate, never a
* delivery guarantee. Emits `SweepRequired` first on every subscribe and
* whenever per-key retention is exceeded. Never fails under backpressure.
*/
readonly changes: () => Stream.Stream<EventLog.Change>
/** Latest committed seq per aggregate. Aggregates without events are absent. */
readonly sequences: (aggregateIDs: ReadonlyArray<string>) => Effect.Effect<ReadonlyMap<string, Seq>>
/** @deprecated Use `all()` and consume the returned stream. */
@ -183,11 +177,6 @@ export const liveBounded = (
export interface LayerOptions {
readonly beforeAggregateRead?: (aggregateID: string) => Effect.Effect<void>
/**
* Maximum distinct aggregates buffered per changes subscriber before the
* buffer is abandoned and the subscriber is told to sweep.
*/
readonly changesKeyCapacity?: number
/** Maximum durable rows read per page while replaying or tailing an aggregate log. */
readonly logReadPageSize?: number
}
@ -204,12 +193,6 @@ export const layerWith = (options?: LayerOptions) =>
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 changesKeyCapacity = options?.changesKeyCapacity ?? 4096
const changesSubscribers = new Set<{
readonly hints: Map<string, number>
sweepRequired: boolean
readonly wake: PubSub.PubSub<void>
}>()
const { db } = yield* Database.Service
const logReadPageSize = options?.logReadPageSize ?? 512
@ -231,9 +214,6 @@ export const layerWith = (options?: LayerOptions) =>
{ discard: true },
)
yield* Effect.forEach(pubsub.typed.values(), PubSub.shutdown, { discard: true })
yield* Effect.forEach(changesSubscribers, (subscriber) => PubSub.shutdown(subscriber.wake), {
discard: true,
})
}),
)
@ -394,27 +374,6 @@ export const layerWith = (options?: LayerOptions) =>
(wake) => PubSub.publish(wake, undefined),
{ discard: true },
)
yield* Effect.forEach(
changesSubscribers,
(subscriber) =>
Effect.sync(() => {
// Coalesce to the latest seq per aggregate. Overflowing key
// cardinality abandons the buffer instead of dropping hints silently.
if (
subscriber.hints.size >= changesKeyCapacity &&
!subscriber.hints.has(committed.aggregateID)
) {
subscriber.hints.clear()
subscriber.sweepRequired = true
} else if (!subscriber.sweepRequired) {
subscriber.hints.set(
committed.aggregateID,
Math.max(subscriber.hints.get(committed.aggregateID) ?? -1, committed.seq),
)
}
}).pipe(Effect.andThen(PubSub.publish(subscriber.wake, undefined)), Effect.asVoid),
{ discard: true },
)
}
return committed
}),
@ -723,47 +682,6 @@ export const layerWith = (options?: LayerOptions) =>
}),
)
const changes = (): Stream.Stream<EventLog.Change> =>
Stream.unwrap(
Effect.gen(function* () {
const wake = yield* PubSub.sliding<void>(1)
const subscription = yield* PubSub.subscribe(wake)
const subscriber = { hints: new Map<string, number>(), sweepRequired: false, wake }
yield* Effect.acquireRelease(
Effect.sync(() => changesSubscribers.add(subscriber)),
() =>
Effect.sync(() => changesSubscribers.delete(subscriber)).pipe(
Effect.andThen(PubSub.shutdown(wake)),
Effect.asVoid,
),
)
const drain = Effect.sync((): ReadonlyArray<EventLog.Change> => {
if (subscriber.sweepRequired) {
subscriber.sweepRequired = false
subscriber.hints.clear()
return [{ type: "log.sweep_required" }]
}
const hints = Array.from(
subscriber.hints,
([aggregateID, seq]): EventLog.Change => ({ type: "log.hint", aggregateID, seq: Seq.make(seq) }),
)
subscriber.hints.clear()
return hints
})
// Hints missed while unsubscribed were never buffered, so every
// (re)subscribe starts from the sweep contract.
const initial: EventLog.Change = { type: "log.sweep_required" }
return Stream.make(initial).pipe(
Stream.concat(
Stream.fromSubscription(subscription).pipe(
Stream.mapEffect(() => drain),
Stream.flattenIterable,
),
),
)
}),
)
const sequences = (aggregateIDs: ReadonlyArray<string>): Effect.Effect<ReadonlyMap<string, Seq>> => {
if (aggregateIDs.length === 0) return Effect.succeed(new Map())
return db
@ -798,7 +716,6 @@ export const layerWith = (options?: LayerOptions) =>
subscribe,
live: streamLive,
log,
changes,
sequences,
listen,
project,

46
packages/core/test/event.test.ts
View file

@ -1289,52 +1289,6 @@ describe("EventV2", () => {
}),
)
it.effect("changes emits sweep-required on subscribe then coalesced hints per aggregate", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service
const first = Session.ID.create()
const second = Session.ID.create()
const pull = yield* Stream.toPull(events.changes())
expect(Array.from(yield* pull)).toEqual([{ type: "log.sweep_required" }])
yield* events.publish(DurableMessage, durableData(first, "zero"))
yield* events.publish(DurableMessage, durableData(first, "one"))
yield* events.publish(DurableMessage, durableData(first, "two"))
yield* events.publish(DurableMessage, durableData(second, "zero"))
expect(Array.from(yield* pull)).toEqual([
{ type: "log.hint", aggregateID: first, seq: EventV2.Seq.make(2) },
{ type: "log.hint", aggregateID: second, seq: EventV2.Seq.make(0) },
])
}),
)
it.effect("changes abandons the hint buffer for a sweep when key retention is exceeded", () =>
Effect.gen(function* () {
const eventLayer = EventV2.layerWith({ changesKeyCapacity: 2 }).pipe(
Layer.provide(LayerNode.compile(Database.node)),
)
yield* Effect.gen(function* () {
const events = yield* EventV2.Service
const pull = yield* Stream.toPull(events.changes())
expect(Array.from(yield* pull)).toEqual([{ type: "log.sweep_required" }])
yield* events.publish(DurableMessage, durableData(Session.ID.create(), "a"))
yield* events.publish(DurableMessage, durableData(Session.ID.create(), "b"))
yield* events.publish(DurableMessage, durableData(Session.ID.create(), "c"))
expect(Array.from(yield* pull)).toEqual([{ type: "log.sweep_required" }])
const late = Session.ID.create()
yield* events.publish(DurableMessage, durableData(late, "d"))
expect(Array.from(yield* pull)).toEqual([{ type: "log.hint", aggregateID: late, seq: EventV2.Seq.make(0) }])
}).pipe(Effect.provide(Layer.merge(LayerNode.compile(Database.node), eventLayer)))
}),
)
it.effect("sequences returns the latest committed seq per aggregate and omits unknown aggregates", () =>
Effect.gen(function* () {
const events = yield* EventV2.Service

1
packages/core/test/session-runner-tool-events.test.ts
View file

@ -29,7 +29,6 @@ const capture = () => {
subscribe: () => Stream.empty,
live: () => Stream.empty,
log: () => Stream.empty,
changes: () => Stream.empty,
sequences: () => Effect.succeed(new Map()),
listen: () => Effect.succeed(Effect.void),
project: () => Effect.void,

15
packages/protocol/src/groups/event.ts
View file

@ -1,5 +1,4 @@
import { Event } from "@opencode-ai/schema/event"
import { EventLog } from "@opencode-ai/schema/event-log"
import { EventManifest } from "@opencode-ai/schema/event-manifest"
import { Location } from "@opencode-ai/schema/location"
import type { Definition } from "@opencode-ai/schema/event"
@ -39,19 +38,7 @@ const make = <const Definitions extends ReadonlyArray<Definition>>(definitions:
identifier: "v2.event.subscribe",
summary: "Subscribe to events",
description:
"Subscribe to native event payloads for the server. Volatile by contract: a slow consumer overflows and fails the stream, and events during disconnection are missed. Consumers that need reliability should combine the changes feed with durable session log reads.",
}),
),
)
.add(
HttpApiEndpoint.get("event.changes", "/api/event/changes", {
success: HttpApiSchema.StreamSse({ data: EventLog.Change }),
}).annotateMerge(
OpenApi.annotations({
identifier: "v2.event.changes",
summary: "Subscribe to change hints",
description:
"Payload-free hint channel: after an event commits, a subscriber eventually receives a hint for that aggregate with seq at or beyond the event, or a sweep-required marker. Hints coalesce to the latest seq per aggregate under backpressure and the stream never fails from overflow. No consumer may derive correctness from receiving a hint; correctness always comes from durable log reads plus the consumer's own checkpoint. A sweep-required marker is emitted first on every (re)subscribe and whenever hint retention is exceeded: treat every aggregate as potentially dirty and recover via bounded sweep plus log reads.",
"Subscribe to native event payloads for the server. Volatile by contract: a slow consumer overflows and fails the stream, and events during disconnection are missed.",
}),
),
)

6
packages/protocol/src/groups/message.ts
View file

@ -1,7 +1,5 @@
import { Session } from "@opencode-ai/schema/session"
import { SessionMessage } from "@opencode-ai/schema/session-message"
import { optional } from "@opencode-ai/schema/schema"
import { Event } from "@opencode-ai/schema/event"
import { Schema } from "effect"
import { HttpApiEndpoint, HttpApiGroup, OpenApi } from "effect/unstable/httpapi"
import { InvalidCursorError, SessionNotFoundError, UnknownError } from "../errors.js"
@ -30,10 +28,6 @@ export const MessageGroup = HttpApiGroup.make("server.message")
query: SessionMessagesQuery,
success: Schema.Struct({
data: Schema.Array(SessionMessage.Message),
watermark: optional(Event.Seq).annotate({
description:
"Durable log seq this snapshot was computed at, read before the snapshot. Attach a live log read after the watermark to compose fetch and stream gap-free; events between the watermark and the snapshot read may be redelivered by the tail and are safe to re-apply. Absent when the session has no durable events.",
}),
cursor: Schema.Struct({
previous: Schema.String.pipe(Schema.optional),
next: Schema.String.pipe(Schema.optional),

15
packages/protocol/src/groups/session.ts
View file

@ -104,12 +104,6 @@ const SessionActive = Schema.Struct({
type: Schema.Literal("running"),
}).annotate({ identifier: "SessionActive" })
const SessionWatermarks = Schema.Record(Session.ID, Event.Seq).annotate({
identifier: "SessionWatermarks",
description:
"Durable log seq each session's snapshot was computed at. Attach a live log read after the watermark to compose fetch and stream gap-free; apply a snapshot only where its watermark is at or beyond already-applied events. Sessions without durable events are absent.",
})
const BooleanFromString = Schema.Literals(["true", "false"]).pipe(
Schema.decodeTo(Schema.Boolean, {
decode: SchemaGetter.transform((value) => value === "true"),
@ -136,7 +130,6 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
query: SessionsQuery,
success: Schema.Struct({
data: Schema.Array(Session.Info),
watermarks: SessionWatermarks,
cursor: Schema.Struct({
previous: SessionsCursor.pipe(Schema.optional),
next: SessionsCursor.pipe(Schema.optional),
@ -171,13 +164,13 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
)
.add(
HttpApiEndpoint.get("session.active", "/api/session/active", {
success: Schema.Struct({ data: Schema.Record(Session.ID, SessionActive), watermarks: SessionWatermarks }),
success: Schema.Struct({ data: Schema.Record(Session.ID, SessionActive) }),
}).annotateMerge(
OpenApi.annotations({
identifier: "v2.session.active",
summary: "List active sessions",
description:
"Retrieve foreground Session drains currently owned by this OpenCode process. Sessions absent from the result are inactive. Watermarks are the durable log positions read alongside the activity snapshot; activity itself is process state, so the pairing is advisory rather than transactional.",
"Retrieve foreground Session drains currently owned by this OpenCode process. Sessions absent from the result are inactive.",
}),
),
)
@ -498,7 +491,7 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
),
)
.add(
HttpApiEndpoint.get("session.log", "/api/session/:sessionID/log", {
HttpApiEndpoint.get("session.log", "/api/experimental/session/:sessionID/log", {
params: { sessionID: Session.ID },
query: {
after: Schema.NumberFromString.pipe(Schema.decodeTo(Event.Seq), Schema.optional),
@ -515,7 +508,7 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(sessionLo
identifier: "v2.session.log",
summary: "Read the session log",
description:
"Durable, ordered, gap-free read of public session events after an exclusive aggregate sequence. Emits a synced marker once replay reaches the captured watermark, then completes; with follow=true it continues with live events instead. The only event API that promises reliability: attach after a snapshot watermark to compose fetch and stream without a race window.",
"Experimental durable session event log. Reads events after an exclusive aggregate sequence and continues with live events when follow=true.",
}),
),
)

34
packages/schema/src/event-log.ts
View file

@ -19,37 +19,3 @@ export const Synced = Schema.Struct({
"Marker emitted once when a log read reaches its captured watermark. The reader holds every event committed at or below seq.",
})
export interface Synced extends Schema.Schema.Type<typeof Synced> {}
/**
* A payload-free doorbell: the aggregate's log advanced to at least `seq`.
* Hints are a latency optimization only; no consumer may derive correctness
* from receiving one. Correctness always comes from a durable log read plus
* the consumer's own checkpoint.
*/
export const Hint = Schema.Struct({
type: Schema.Literal("log.hint"),
aggregateID: Schema.String,
seq: Event.Seq,
}).annotate({
identifier: "EventLog.Hint",
description:
"Payload-free change hint: the aggregate's durable log advanced to at least seq. Hints coalesce under backpressure (latest per aggregate) and are never a delivery guarantee.",
})
export interface Hint extends Schema.Schema.Type<typeof Hint> {}
/**
* Hints may have been lost. Treat every aggregate as potentially dirty and
* recover via bounded sweep plus durable log reads. Also emitted first on
* every (re)subscribe, since hints during disconnection were never buffered.
*/
export const SweepRequired = Schema.Struct({
type: Schema.Literal("log.sweep_required"),
}).annotate({
identifier: "EventLog.SweepRequired",
description:
"Hints may have been lost; treat every aggregate as potentially dirty and recover via bounded sweep plus durable log reads. Emitted first on every (re)subscribe.",
})
export interface SweepRequired extends Schema.Schema.Type<typeof SweepRequired> {}
export const Change = Schema.Union([Hint, SweepRequired]).annotate({ identifier: "EventLog.Change" })
export type Change = typeof Change.Type

2
packages/schema/src/event.ts
View file

@ -14,7 +14,7 @@ export type ID = typeof ID.Type
/**
* Position in one aggregate's durable log. Values originate from the durable
* event envelope, synced markers, change hints, and snapshot watermarks;
* event envelope and synced markers;
* `after` cursors accept only values that came from those sources.
*/
export const Seq = Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)).pipe(Schema.brand("Event.Seq"))

2
packages/sdk-next/test/embedded.test.ts
View file

@ -117,7 +117,7 @@ it.live(
expect(selected.model?.id).toBe(model.id)
expect(selected.model?.providerID).toBe(model.providerID)
expect(page.data.some((session) => session.id === id)).toBe(true)
expect(active).toEqual({ data: {}, watermarks: {} })
expect(active).toEqual({})
expect(admitted.sessionID).toBe(id)
expect(prompted.type).toBe("session.prompt.promoted")
expect(wakeContext).toContainEqual(expect.objectContaining({ id: wake.id, type: "user" }))

36
packages/sdk/js/src/v2/gen/sdk.gen.ts
View file

@ -278,8 +278,6 @@ import type {
V2CredentialUpdateResponses,
V2DebugLocationErrors,
V2DebugLocationResponses,
V2EventChangesErrors,
V2EventChangesResponses,
V2EventSubscribeErrors,
V2EventSubscribeResponses,
V2FormRequestListErrors,
@ -334,6 +332,8 @@ import type {
V2ProjectCurrentResponses,
V2ProjectDirectoriesErrors,
V2ProjectDirectoriesResponses,
V2ProjectListErrors,
V2ProjectListResponses,
V2ProviderGetErrors,
V2ProviderGetResponses,
V2ProviderListErrors,
@ -5896,7 +5896,7 @@ export class Session3 extends HeyApiClient {
/**
* List active sessions
*
* Retrieve foreground Session drains currently owned by this OpenCode process. Sessions absent from the result are inactive. Watermarks are the durable log positions read alongside the activity snapshot; activity itself is process state, so the pairing is advisory rather than transactional.
* Retrieve foreground Session drains currently owned by this OpenCode process. Sessions absent from the result are inactive.
*/
public active<ThrowOnError extends boolean = false>(options?: Options<never, ThrowOnError>) {
return (options?.client ?? this.client).get<V2SessionActiveResponses, V2SessionActiveErrors, ThrowOnError>({
@ -6341,7 +6341,7 @@ export class Session3 extends HeyApiClient {
/**
* Read the session log
*
* Durable, ordered, gap-free read of public session events after an exclusive aggregate sequence. Emits a synced marker once replay reaches the captured watermark, then completes; with follow=true it continues with live events instead. The only event API that promises reliability: attach after a snapshot watermark to compose fetch and stream without a race window.
* Experimental durable session event log. Reads events after an exclusive aggregate sequence and continues with live events when follow=true.
*/
public log<ThrowOnError extends boolean = false>(
parameters: {
@ -6364,7 +6364,7 @@ export class Session3 extends HeyApiClient {
],
)
return (options?.client ?? this.client).sse.get<V2SessionLogResponses, V2SessionLogErrors, ThrowOnError>({
url: "/api/session/{sessionID}/log",
url: "/api/experimental/session/{sessionID}/log",
...options,
...params,
})
@ -7032,6 +7032,18 @@ export class Credential extends HeyApiClient {
}
export class Project2 extends HeyApiClient {
/**
* List projects
*
* List known projects.
*/
public list<ThrowOnError extends boolean = false>(options?: Options<never, ThrowOnError>) {
return (options?.client ?? this.client).get<V2ProjectListResponses, V2ProjectListErrors, ThrowOnError>({
url: "/api/project",
...options,
})
}
/**
* Get current project
*
@ -7357,7 +7369,7 @@ export class Event2 extends HeyApiClient {
/**
* Subscribe to events
*
* Subscribe to native event payloads for the server. Volatile by contract: a slow consumer overflows and fails the stream, and events during disconnection are missed. Consumers that need reliability should combine the changes feed with durable session log reads.
* Subscribe to native event payloads for the server. Volatile by contract: a slow consumer overflows and fails the stream, and events during disconnection are missed.
*/
public subscribe<ThrowOnError extends boolean = false>(options?: Options<never, ThrowOnError>) {
return (options?.client ?? this.client).sse.get<V2EventSubscribeResponses, V2EventSubscribeErrors, ThrowOnError>({
@ -7365,18 +7377,6 @@ export class Event2 extends HeyApiClient {
...options,
})
}
/**
* Subscribe to change hints
*
* Payload-free hint channel: after an event commits, a subscriber eventually receives a hint for that aggregate with seq at or beyond the event, or a sweep-required marker. Hints coalesce to the latest seq per aggregate under backpressure and the stream never fails from overflow. No consumer may derive correctness from receiving a hint; correctness always comes from durable log reads plus the consumer's own checkpoint. A sweep-required marker is emitted first on every (re)subscribe and whenever hint retention is exceeded: treat every aggregate as potentially dirty and recover via bounded sweep plus log reads.
*/
public changes<ThrowOnError extends boolean = false>(options?: Options<never, ThrowOnError>) {
return (options?.client ?? this.client).sse.get<V2EventChangesResponses, V2EventChangesErrors, ThrowOnError>({
url: "/api/event/changes",
...options,
})
}
}
export class Pty2 extends HeyApiClient {

146
packages/sdk/js/src/v2/gen/types.gen.ts
View file

@ -2796,13 +2796,8 @@ export type UnauthorizedError = {
message: string
}
export type SessionWatermarks = {
[key: string]: unknown | number
}
export type SessionsResponse = {
data: Array<SessionV2Info>
watermarks: SessionWatermarks
cursor: {
previous?: string
next?: string
@ -2930,7 +2925,6 @@ export type SessionLogItemStream = string
export type SessionMessagesResponse = {
data: Array<SessionMessage>
watermark?: number
cursor: {
previous?: string
next?: string
@ -6615,20 +6609,6 @@ export type GlobalDisposed = {
}
}
export type EventLogHint = {
type: "log.hint"
aggregateID: string
seq: number
}
export type EventLogSweepRequired = {
type: "log.sweep_required"
}
export type EventLogChange = EventLogHint | EventLogSweepRequired
export type EventLogChangeStream = string
export type QuestionV2Request = {
id: string
sessionID: string
@ -7859,16 +7839,8 @@ export type SessionV2Info2 = {
revert?: RevertState2
}
/**
* Durable log seq each session's snapshot was computed at. Attach a live log read after the watermark to compose fetch and stream gap-free; apply a snapshot only where its watermark is at or beyond already-applied events. Sessions without durable events are absent.
*/
export type SessionWatermarksV2 = {
[key: string]: unknown | number
}
export type SessionsResponseV2 = {
data: Array<SessionV2Info2>
watermarks: SessionWatermarksV2
cursor: {
previous?: string | null
next?: string | null
@ -9059,7 +9031,6 @@ export type SessionLogItemStreamV2 = string
export type SessionMessagesResponseV2 = {
data: Array<SessionMessage2>
watermark?: number
cursor: {
previous?: string | null
next?: string | null
@ -9329,6 +9300,38 @@ export type McpServer2 = {
integrationID?: string
}
export type ProjectVcs2 = "git" | "hg"
export type ProjectIcon2 = {
url?: string
override?: string
color?: string
}
export type ProjectCommands2 = {
/**
* Startup script to run when creating a new workspace (worktree)
*/
start?: string
}
export type ProjectTime2 = {
created: number
updated: number
initialized?: number
}
export type ProjectV2 = {
id: string
worktree: string
vcs?: ProjectVcs2
name?: string
icon?: ProjectIcon2
commands?: ProjectCommands2
time: ProjectTime2
sandboxes: Array<string>
}
export type ProjectCurrent2 = {
id: string
directory: string
@ -11276,26 +11279,6 @@ export type V2EventV2 =
export type V2EventStreamV2 = string
/**
* Payload-free change hint: the aggregate's durable log advanced to at least seq. Hints coalesce under backpressure (latest per aggregate) and are never a delivery guarantee.
*/
export type EventLogHint2 = {
type: "log.hint"
aggregateID: string
seq: number
}
/**
* Hints may have been lost; treat every aggregate as potentially dirty and recover via bounded sweep plus durable log reads. Emitted first on every (re)subscribe.
*/
export type EventLogSweepRequired2 = {
type: "log.sweep_required"
}
export type EventLogChange2 = EventLogHint2 | EventLogSweepRequired2
export type EventLogChangeStream2 = string
export type PtyNotFoundErrorV2 = {
_tag: "PtyNotFoundError"
ptyID: string
@ -15770,7 +15753,6 @@ export type V2SessionActiveResponses = {
data: {
[key: string]: unknown | SessionActiveV2
}
watermarks: SessionWatermarksV2
}
}
@ -16563,7 +16545,7 @@ export type V2SessionLogData = {
after?: number | null
follow?: "true" | "false" | null
}
url: "/api/session/{sessionID}/log"
url: "/api/experimental/session/{sessionID}/log"
}
export type V2SessionLogErrors = {
@ -17353,6 +17335,35 @@ export type V2CredentialUpdateResponses = {
export type V2CredentialUpdateResponse = V2CredentialUpdateResponses[keyof V2CredentialUpdateResponses]
export type V2ProjectListData = {
body?: never
path?: never
query?: never
url: "/api/project"
}
export type V2ProjectListErrors = {
/**
* InvalidRequestError
*/
400: InvalidRequestErrorV2
/**
* UnauthorizedError
*/
401: UnauthorizedErrorV2
}
export type V2ProjectListError = V2ProjectListErrors[keyof V2ProjectListErrors]
export type V2ProjectListResponses = {
/**
* Success
*/
200: Array<ProjectV2>
}
export type V2ProjectListResponse = V2ProjectListResponses[keyof V2ProjectListResponses]
export type V2ProjectCurrentData = {
body?: never
path?: never
@ -18176,39 +18187,6 @@ export type V2EventSubscribeResponses = {
export type V2EventSubscribeResponse = V2EventSubscribeResponses[keyof V2EventSubscribeResponses]
export type V2EventChangesData = {
body?: never
path?: never
query?: never
url: "/api/event/changes"
}
export type V2EventChangesErrors = {
/**
* InvalidRequestError
*/
400: InvalidRequestErrorV2
/**
* UnauthorizedError
*/
401: UnauthorizedErrorV2
}
export type V2EventChangesError = V2EventChangesErrors[keyof V2EventChangesErrors]
export type V2EventChangesResponses = {
/**
* Success
*/
200: {
id: string | null
event: string
data: EventLogChangeStream2
}
}
export type V2EventChangesResponse = V2EventChangesResponses[keyof V2EventChangesResponses]
export type V2PtyListData = {
body?: never
path?: never

65
packages/server/src/handlers/event.ts
View file

@ -20,41 +20,36 @@ function eventData(data: unknown): Sse.Event {
export const EventHandler = HttpApiBuilder.group(Api, "server.event", (handlers) =>
Effect.gen(function* () {
const events = yield* EventV2.Service
return handlers
.handle(
"event.changes",
Effect.fn(() => Effect.succeed(events.changes())),
)
.handleRaw("event.subscribe", () =>
Effect.gen(function* () {
const connected = {
id: EventV2.ID.create(),
type: "server.connected",
data: {},
}
const output = Stream.unwrap(
Effect.gen(function* () {
// Acquiring the bounded stream installs its listener before readiness is observable.
const live = yield* EventV2.liveBounded(events, {
capacity: subscriberCapacity,
accept: isOpenCodeEvent,
})
return Stream.make(connected).pipe(Stream.concat(live))
}),
).pipe(Stream.map(eventData), Stream.pipeThroughChannel(Sse.encode()))
const heartbeat = Stream.tick("15 seconds").pipe(Stream.map(() => ": heartbeat\n\n"))
return HttpServerResponse.stream(
output.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }), Stream.encodeText),
{
contentType: "text/event-stream",
headers: {
"Cache-Control": "no-cache, no-transform",
"X-Accel-Buffering": "no",
"X-Content-Type-Options": "nosniff",
},
return handlers.handleRaw("event.subscribe", () =>
Effect.gen(function* () {
const connected = {
id: EventV2.ID.create(),
type: "server.connected",
data: {},
}
const output = Stream.unwrap(
Effect.gen(function* () {
// Acquiring the bounded stream installs its listener before readiness is observable.
const live = yield* EventV2.liveBounded(events, {
capacity: subscriberCapacity,
accept: isOpenCodeEvent,
})
return Stream.make(connected).pipe(Stream.concat(live))
}),
).pipe(Stream.map(eventData), Stream.pipeThroughChannel(Sse.encode()))
const heartbeat = Stream.tick("15 seconds").pipe(Stream.map(() => ": heartbeat\n\n"))
return HttpServerResponse.stream(
output.pipe(Stream.merge(heartbeat, { haltStrategy: "left" }), Stream.encodeText),
{
contentType: "text/event-stream",
headers: {
"Cache-Control": "no-cache, no-transform",
"X-Accel-Buffering": "no",
"X-Content-Type-Options": "nosniff",
},
)
}),
)
},
)
}),
)
}),
)

5
packages/server/src/handlers/message.ts
View file

@ -38,10 +38,6 @@ export const MessageHandler = HttpApiBuilder.group(Api, "server.message", (handl
catch: () => new InvalidCursorError({ message: "Invalid cursor" }),
})
const order = decoded?.order ?? ctx.query.order ?? "desc"
// Read the watermark before the snapshot: an understated watermark only
// redelivers already-reflected events, while an overstated one would let
// an attached tail skip events missing from the snapshot.
const watermark = (yield* session.watermarks([ctx.params.sessionID])).get(ctx.params.sessionID)
const messages = yield* session
.messages({
sessionID: ctx.params.sessionID,
@ -74,7 +70,6 @@ export const MessageHandler = HttpApiBuilder.group(Api, "server.message", (handl
const last = messages.at(-1)
return {
data: messages,
watermark,
cursor: {
previous: first ? cursor.encode(first, order, "previous") : undefined,
next: last ? cursor.encode(last, order, "next") : undefined,

3
packages/server/src/handlers/session.ts
View file

@ -44,7 +44,6 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl
const last = sessions.at(-1)
return {
data: sessions,
watermarks: Object.fromEntries(page.watermarks),
cursor: {
previous: first
? SessionsCursor.make({
@ -89,10 +88,8 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl
"session.active",
Effect.fn(function* () {
const active = yield* session.active
const watermarks = yield* session.watermarks(Array.from(active))
return {
data: Object.fromEntries(Array.from(active, (sessionID) => [sessionID, { type: "running" as const }])),
watermarks: Object.fromEntries(watermarks),
}
}),
)

2
packages/tui/src/context/data.tsx
View file

@ -911,7 +911,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({
.then((active) => {
if (generation !== connectionGeneration) return
const status: Record<string, DataSessionStatus> = Object.fromEntries(
Object.keys(active.data).map((sessionID) => [sessionID, "running" as const]),
Object.keys(active).map((sessionID) => [sessionID, "running" as const]),
)
for (const sessionID of changed) status[sessionID] = store.session.status[sessionID]
setStore("session", "status", reconcile(status))

6
packages/tui/test/cli/tui/data.test.tsx
View file

@ -119,7 +119,7 @@ test("reconnects the event stream and bootstraps fresh data", async () => {
}
if (url.pathname === "/api/session/active") {
requests.active++
if (requests.active === 1) return json({ data: { "session-stale": { type: "running" } }, watermarks: {} })
if (requests.active === 1) return json({ data: { "session-stale": { type: "running" } } })
return new Promise<Response>((resolve) => {
resolveActive = resolve
})
@ -189,7 +189,7 @@ test("reconnects the event stream and bootstraps fresh data", async () => {
},
})
await wait(() => data.session.status("session-new") === "running")
resolveActive(json({ data: {}, watermarks: {} }))
resolveActive(json({ data: {} }))
await wait(() => data.location.model.list()?.[0]?.id === "model-2", 4000)
await wait(() => data.session.status("session-stale") === "idle")
@ -350,7 +350,7 @@ test("tracks session status from active sessions and execution events", async ()
const events = createEventStream()
const calls = createFetch((url) => {
if (url.pathname === "/api/session/active")
return json({ data: { "session-active": { type: "running" } }, watermarks: {} })
return json({ data: { "session-active": { type: "running" } } })
}, events)
let data!: ReturnType<typeof useData>

2
packages/tui/test/fixture/tui-sdk.ts
View file

@ -100,7 +100,7 @@ export function createFetch(override?: FetchHandler, events?: ReturnType<typeof
if (url.pathname === "/api/mcp")
return json({ location: { directory, project: { id: "proj_test", directory: worktree } }, data: [] })
if (url.pathname === "/api/session") return json({ data: [], cursor: {} })
if (url.pathname === "/api/session/active") return json({ data: {}, watermarks: {} })
if (url.pathname === "/api/session/active") return json({ data: {} })
if (/^\/api\/session\/[^/]+\/form$/.test(url.pathname)) return json({ data: [] })
if (
["/api/agent", "/api/model", "/api/provider", "/api/integration", "/api/command", "/api/skill"].includes(