diff --git a/packages/cli/src/mini/stream-v2.transport.ts b/packages/cli/src/mini/stream-v2.transport.ts index 7cec6f9e87..d6998f200b 100644 --- a/packages/cli/src/mini/stream-v2.transport.ts +++ b/packages/cli/src/mini/stream-v2.transport.ts @@ -544,8 +544,8 @@ export async function createSessionTransport(input: StreamInput): Promise>["data"] export type SessionCreateOperation = (input?: Endpoint4_1Input) => Effect.Effect -export type Endpoint4_2Output = EffectValue> +export type Endpoint4_2Output = EffectValue>["data"] export type SessionActiveOperation = () => Effect.Effect type Endpoint4_3Request = Parameters[0] @@ -658,12 +658,8 @@ export interface SkillApi { export type Endpoint18_0Output = StreamValue>> export type EventSubscribeOperation = () => Stream.Stream -export type Endpoint18_1Output = StreamValue>> -export type EventChangesOperation = () => Stream.Stream - export interface EventApi { readonly subscribe: EventSubscribeOperation - readonly changes: EventChangesOperation } type Endpoint19_0Request = Parameters[0] diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index 1001c1e047..b7dc996aba 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -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[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[0] type Endpoint19_0Input = { readonly location?: Endpoint19_0Request["query"]["location"] } diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index d3e18279e0..ad94f07fd3 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -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( + 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( { 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 => - sse( - { method: "GET", path: `/api/event/changes`, successStatus: 200, declaredStatuses: [401, 400], empty: false }, - requestOptions, - ), }, pty: { list: (input?: PtyListInput, requestOptions?: RequestOptions) => diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index ac0efadb5c..dc28955f49 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -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 diff --git a/packages/client/test/effect.test.ts b/packages/client/test/effect.test.ts index 2077c27ca5..2073af42ac 100644 --- a/packages/client/test/effect.test.ts +++ b/packages/client/test/effect.test.ts @@ -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") diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index 2dbd58a858..04737612e2 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -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"], ]) diff --git a/packages/codemode/test/openapi.test.ts b/packages/codemode/test/openapi.test.ts index a50460e8da..7ad3d48624 100644 --- a/packages/codemode/test/openapi.test.ts +++ b/packages/codemode/test/openapi.test.ts @@ -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() }) diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index a5b052b42f..6b6d7813e5 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -135,12 +135,6 @@ export interface Interface { readonly after?: number readonly follow?: boolean }) => Stream.Stream - /** - * 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 /** Latest committed seq per aggregate. Aggregates without events are absent. */ readonly sequences: (aggregateIDs: ReadonlyArray) => Effect.Effect> /** @deprecated Use `all()` and consume the returned stream. */ @@ -183,11 +177,6 @@ export const liveBounded = ( export interface LayerOptions { readonly beforeAggregateRead?: (aggregateID: string) => Effect.Effect - /** - * 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() // TODO: Bind durable projectors to exact type+version before supporting incompatible historical payloads. const listeners = new Array() - const changesKeyCapacity = options?.changesKeyCapacity ?? 4096 - const changesSubscribers = new Set<{ - readonly hints: Map - sweepRequired: boolean - readonly wake: PubSub.PubSub - }>() 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 => - Stream.unwrap( - Effect.gen(function* () { - const wake = yield* PubSub.sliding(1) - const subscription = yield* PubSub.subscribe(wake) - const subscriber = { hints: new Map(), 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 => { - 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): Effect.Effect> => { 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, diff --git a/packages/core/test/event.test.ts b/packages/core/test/event.test.ts index 70a8083d16..96e767084f 100644 --- a/packages/core/test/event.test.ts +++ b/packages/core/test/event.test.ts @@ -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 diff --git a/packages/core/test/session-runner-tool-events.test.ts b/packages/core/test/session-runner-tool-events.test.ts index d016b2694b..983dca8c2c 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -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, diff --git a/packages/protocol/src/groups/event.ts b/packages/protocol/src/groups/event.ts index e555abe3f9..242aa745ef 100644 --- a/packages/protocol/src/groups/event.ts +++ b/packages/protocol/src/groups/event.ts @@ -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 = >(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.", }), ), ) diff --git a/packages/protocol/src/groups/message.ts b/packages/protocol/src/groups/message.ts index b2a360cd10..b8c387576f 100644 --- a/packages/protocol/src/groups/message.ts +++ b/packages/protocol/src/groups/message.ts @@ -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), diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index d89918f515..3d82cd3297 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -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 = (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 = (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 = (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 = (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.", }), ), ) diff --git a/packages/schema/src/event-log.ts b/packages/schema/src/event-log.ts index 1b0bbcb924..4091087056 100644 --- a/packages/schema/src/event-log.ts +++ b/packages/schema/src/event-log.ts @@ -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 {} - -/** - * 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 {} - -/** - * 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 {} - -export const Change = Schema.Union([Hint, SweepRequired]).annotate({ identifier: "EventLog.Change" }) -export type Change = typeof Change.Type diff --git a/packages/schema/src/event.ts b/packages/schema/src/event.ts index 9f7c347b03..34e4495807 100644 --- a/packages/schema/src/event.ts +++ b/packages/schema/src/event.ts @@ -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")) diff --git a/packages/sdk-next/test/embedded.test.ts b/packages/sdk-next/test/embedded.test.ts index 4ac31b70ab..a6184da837 100644 --- a/packages/sdk-next/test/embedded.test.ts +++ b/packages/sdk-next/test/embedded.test.ts @@ -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" })) diff --git a/packages/sdk/js/src/v2/gen/sdk.gen.ts b/packages/sdk/js/src/v2/gen/sdk.gen.ts index 59484dddfb..2243f8e89d 100644 --- a/packages/sdk/js/src/v2/gen/sdk.gen.ts +++ b/packages/sdk/js/src/v2/gen/sdk.gen.ts @@ -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(options?: Options) { return (options?.client ?? this.client).get({ @@ -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( parameters: { @@ -6364,7 +6364,7 @@ export class Session3 extends HeyApiClient { ], ) return (options?.client ?? this.client).sse.get({ - 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(options?: Options) { + return (options?.client ?? this.client).get({ + 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(options?: Options) { return (options?.client ?? this.client).sse.get({ @@ -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(options?: Options) { - return (options?.client ?? this.client).sse.get({ - url: "/api/event/changes", - ...options, - }) - } } export class Pty2 extends HeyApiClient { diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 4d2cf1922c..de17b34bb6 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -2796,13 +2796,8 @@ export type UnauthorizedError = { message: string } -export type SessionWatermarks = { - [key: string]: unknown | number -} - export type SessionsResponse = { data: Array - watermarks: SessionWatermarks cursor: { previous?: string next?: string @@ -2930,7 +2925,6 @@ export type SessionLogItemStream = string export type SessionMessagesResponse = { data: Array - 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 - watermarks: SessionWatermarksV2 cursor: { previous?: string | null next?: string | null @@ -9059,7 +9031,6 @@ export type SessionLogItemStreamV2 = string export type SessionMessagesResponseV2 = { data: Array - 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 +} + 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 +} + +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 diff --git a/packages/server/src/handlers/event.ts b/packages/server/src/handlers/event.ts index 1c8b94d124..e17c815cb8 100644 --- a/packages/server/src/handlers/event.ts +++ b/packages/server/src/handlers/event.ts @@ -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", }, - ) - }), - ) + }, + ) + }), + ) }), ) diff --git a/packages/server/src/handlers/message.ts b/packages/server/src/handlers/message.ts index 85de308fca..93734c628d 100644 --- a/packages/server/src/handlers/message.ts +++ b/packages/server/src/handlers/message.ts @@ -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, diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index 8a3cf1b130..72d238e684 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -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), } }), ) diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index d9b29c1fdd..aa27572ac4 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -911,7 +911,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ .then((active) => { if (generation !== connectionGeneration) return const status: Record = 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)) diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index 6c33f8496f..1820e58f29 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -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((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 diff --git a/packages/tui/test/fixture/tui-sdk.ts b/packages/tui/test/fixture/tui-sdk.ts index 5ebe0a63f4..8c6e476db0 100644 --- a/packages/tui/test/fixture/tui-sdk.ts +++ b/packages/tui/test/fixture/tui-sdk.ts @@ -100,7 +100,7 @@ export function createFetch(override?: FetchHandler, events?: ReturnType