diff --git a/packages/client/src/generated/types.ts b/packages/client/src/generated/types.ts index 01a359f3e89..42371572207 100644 --- a/packages/client/src/generated/types.ts +++ b/packages/client/src/generated/types.ts @@ -1575,7 +1575,7 @@ export type SessionLogOutput = readonly data: { readonly timestamp: number; readonly sessionID: string; readonly messageID: string } } ) - | { readonly type: "log.caught_up"; readonly aggregateID: string; readonly seq?: number } + | { readonly type: "log.synced"; readonly aggregateID: string; readonly seq?: number } export type SessionInterruptInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] } diff --git a/packages/client/test/effect.test.ts b/packages/client/test/effect.test.ts index 931a575d8b9..b6ee443aa65 100644 --- a/packages/client/test/effect.test.ts +++ b/packages/client/test/effect.test.ts @@ -3,7 +3,7 @@ import { DateTime, Effect, Stream } from "effect" import { HttpClient, HttpClientResponse } from "effect/unstable/http" import { AbsolutePath, Agent, Event, Location, Model, OpenCode, Prompt, Session, SessionMessage } from "../src/effect" -const caughtUp = { type: "log.caught_up" as const, aggregateID: "ses_test", seq: Event.Seq.make(1) } +const synced = { type: "log.synced" as const, aggregateID: "ses_test", seq: Event.Seq.make(1) } test("session.get returns the decoded Effect projection", async () => { const httpClient = HttpClient.make((request) => @@ -70,7 +70,7 @@ test("session methods retain decoded Effect inputs and outputs", async () => { return Effect.succeed( HttpClientResponse.fromWeb( request, - new Response(`data: ${JSON.stringify(modelSwitchedEvent)}\n\ndata: ${JSON.stringify(caughtUp)}\n\n`, { + new Response(`data: ${JSON.stringify(modelSwitchedEvent)}\n\ndata: ${JSON.stringify(synced)}\n\n`, { headers: { "content-type": "text/event-stream" }, }), ), @@ -149,11 +149,11 @@ test("session methods retain decoded Effect inputs and outputs", async () => { expect(result.context).toEqual([]) expect(logQueries[0]).toEqual({ after: "0" }) const logged = Array.from(result.log) - expect(logged.map((item) => item.type)).toEqual(["session.next.model.switched", "log.caught_up"]) + expect(logged.map((item) => item.type)).toEqual(["session.next.model.switched", "log.synced"]) expect(logged[0]?.type === "session.next.model.switched" && DateTime.toEpochMillis(logged[0].data.timestamp)).toBe( 1_717_171_717_000, ) - expect(logged.at(-1)).toEqual(caughtUp) + expect(logged.at(-1)).toEqual(synced) expect(result.message).toEqual(expect.objectContaining({ id: "msg_model", type: "model-switched" })) }) diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index bf1b373c3b6..b868f21ddf8 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -232,7 +232,7 @@ test("session methods use the public HTTP contract", async () => { }) } if (url.includes("/log")) { - return new Response(`data: ${JSON.stringify(modelSwitchedEvent)}\n\ndata: ${JSON.stringify(caughtUp)}\n\n`, { + return new Response(`data: ${JSON.stringify(modelSwitchedEvent)}\n\ndata: ${JSON.stringify(synced)}\n\n`, { headers: { "content-type": "text/event-stream" }, }) } @@ -273,7 +273,7 @@ test("session methods use the public HTTP contract", async () => { expect(created.id).toBe("ses_test") expect(admitted.id).toBe("msg_test") expect(context).toEqual([]) - expect(log).toEqual([modelSwitchedEvent, caughtUp]) + expect(log).toEqual([modelSwitchedEvent, synced]) expect(message).toEqual(modelSwitchedMessage) expect(requests.map((request) => [request.init?.method, request.url])).toEqual([ ["GET", "http://localhost:3000/api/session?limit=10&order=desc&parentID=null"], @@ -368,7 +368,7 @@ const modelSwitchedMessage = { model: { id: "claude", providerID: "anthropic" }, } -const caughtUp = { type: "log.caught_up", aggregateID: "ses_test", seq: 1 } +const synced = { type: "log.synced", aggregateID: "ses_test", seq: 1 } const modelSwitchedEvent = { id: "evt_model", diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index 0959bdc44bd..66ee91c6c68 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -4,7 +4,7 @@ import { Cause, Context, Effect, Layer, Option, PubSub, Queue, Schema, Stream } import { Event } from "@opencode-ai/schema/event" import type { Data, Definition, Payload } from "@opencode-ai/schema/event" import type { EventLog } from "@opencode-ai/schema/event-log" -import { and, asc, eq, gt, inArray, sql } from "drizzle-orm" +import { and, asc, eq, gt, inArray, lte, sql } from "drizzle-orm" import { Database } from "./database/database" import { EventSequenceTable, EventTable } from "./event/sql" import { Location } from "./location" @@ -103,10 +103,10 @@ export interface PublishOptions { readonly commit?: (seq: number) => Effect.Effect } -/** Marker/event union emitted by `log`. Markers carry no event `id`. */ -export type LogItem = Payload | EventLog.CaughtUp +/** Marker/event union emitted by `log`. */ +export type LogItem = Payload | EventLog.Synced -export const isCaughtUp = (item: LogItem): item is EventLog.CaughtUp => !("id" in item) +export const isSynced = (item: LogItem): item is EventLog.Synced => item.type === "log.synced" export interface Interface { readonly publish: ( @@ -124,8 +124,8 @@ export interface Interface { /** * Durable, ordered, gap-free per-aggregate log read. `follow: false` * completes at the end of the log; `follow: true` replays then transitions - * to live. Both modes emit a `CaughtUp` marker at the replay boundary; the - * marker may be re-emitted after internal re-attaches. + * to live. Both modes emit one `Synced` marker at the captured replay + * watermark. */ readonly log: (input: { readonly aggregateID: string @@ -178,6 +178,8 @@ export interface LayerOptions { * 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 } export const layerWith = (options?: LayerOptions) => @@ -199,6 +201,7 @@ export const layerWith = (options?: LayerOptions) => readonly wake: PubSub.PubSub }>() const { db } = yield* Database.Service + const logReadPageSize = options?.logReadPageSize ?? 512 const getOrCreate = (definition: Definition) => Effect.gen(function* () { @@ -573,15 +576,27 @@ export const layerWith = (options?: LayerOptions) => const streamLive = (): Stream.Stream => Stream.fromPubSub(pubsub.live) - const readAfter = (aggregateID: string, after: number) => + const readAfter = ( + aggregateID: string, + after: number, + input: { readonly through: number; readonly limit: number }, + ) => (options?.beforeAggregateRead?.(aggregateID) ?? Effect.void).pipe( Effect.andThen( - db - .select() - .from(EventTable) - .where(and(eq(EventTable.aggregate_id, aggregateID), gt(EventTable.seq, after))) - .orderBy(asc(EventTable.seq)) - .all(), + Effect.suspend(() => { + const query = db + .select() + .from(EventTable) + .where( + and( + eq(EventTable.aggregate_id, aggregateID), + gt(EventTable.seq, after), + lte(EventTable.seq, input.through), + ), + ) + .orderBy(asc(EventTable.seq)) + return query.limit(input.limit).all() + }), ), Effect.orDie, // Skip types missing from the durable manifest instead of failing the @@ -632,28 +647,42 @@ export const layerWith = (options?: LayerOptions) => Stream.unwrap( Effect.gen(function* () { let sequence = input.after ?? -1 - const read = Effect.suspend(() => readAfter(input.aggregateID, sequence)).pipe( - Effect.tap((page) => - Effect.sync(() => { - sequence = page.seq ?? sequence - }), - ), - Effect.map((page) => page.events), - ) + const readThrough = (through: number): Stream.Stream => + Stream.paginate(sequence, (cursor) => + readAfter(input.aggregateID, cursor, { through, limit: logReadPageSize }).pipe( + Effect.tap((page) => + Effect.sync(() => { + sequence = page.seq ?? sequence + }), + ), + Effect.map( + (page) => + [ + page.events, + page.seq !== undefined && page.seq < through ? Option.some(page.seq) : Option.none(), + ] as const, + ), + ), + ) // Subscribing before the historical read means events committed during // replay either appear in the read or arrive through a post-marker wake. const wakes = input.follow ? yield* subscribeDurable(input.aggregateID) : undefined - const historical = yield* read - const marker: EventLog.CaughtUp = { - type: "log.caught_up", + const target = yield* latestSequence(db, input.aggregateID) + const marker: EventLog.Synced = { + type: "log.synced", aggregateID: input.aggregateID, - ...(sequence >= 0 ? { seq: Seq.make(sequence) } : {}), + ...(target >= 0 ? { seq: Seq.make(target) } : {}), } - const replay = Stream.fromIterable(historical).pipe(Stream.concat(Stream.make(marker))) + const replay: Stream.Stream = readThrough(target).pipe( + Stream.map((event): LogItem => event), + Stream.concat(Stream.make(marker)), + ) if (!wakes) return replay - const live = Stream.fromSubscription(wakes).pipe( - Stream.mapEffect(() => read), - Stream.flattenIterable, + const live: Stream.Stream = Stream.fromSubscription(wakes).pipe( + Stream.mapEffect(() => latestSequence(db, input.aggregateID)), + Stream.filter((target) => target > sequence), + Stream.flatMap((target) => readThrough(target)), + Stream.map((event): LogItem => event), ) return Stream.concat(replay, live) }), diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 4c7e43066b9..4f1be4151c8 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -163,8 +163,9 @@ export interface Interface { ) => Effect.Effect /** * Durable, ordered, gap-free session log read. Replays public durable - * session events after the exclusive `after` cursor, emits a `CaughtUp` - * marker at the replay boundary, then continues live when `follow` is set. + * session events after the exclusive `after` cursor, emits a `Synced` + * marker at the captured replay watermark, then continues live when `follow` + * is set. * The marker's seq may exceed the last emitted event because non-public * durable events share the aggregate's sequence space. */ @@ -172,7 +173,7 @@ export interface Interface { sessionID: SessionSchema.ID after?: number follow?: boolean - }) => Stream.Stream + }) => Stream.Stream /** Latest durable log seq per session. Sessions without events are absent. */ readonly watermarks: (sessionIDs: ReadonlyArray) => Effect.Effect> readonly switchAgent: (input: { sessionID: SessionSchema.ID; agent: string }) => Effect.Effect @@ -455,8 +456,8 @@ const layer = Layer.effect( .pipe(Effect.as(events.log({ aggregateID: input.sessionID, after: input.after, follow: input.follow }))), ).pipe( Stream.filter( - (item): item is SessionEvent.DurableEvent | EventLog.CaughtUp => - EventV2.isCaughtUp(item) || isDurableSessionEvent(item), + (item): item is SessionEvent.DurableEvent | EventLog.Synced => + EventV2.isSynced(item) || isDurableSessionEvent(item), ), ), watermarks: Effect.fn("V2Session.watermarks")(function* (sessionIDs) { diff --git a/packages/core/test/event.test.ts b/packages/core/test/event.test.ts index cd2f8cc98d1..b40f088c5d2 100644 --- a/packages/core/test/event.test.ts +++ b/packages/core/test/event.test.ts @@ -1,5 +1,5 @@ import { describe, expect } from "bun:test" -import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Option, Schema, Stream } from "effect" +import { Cause, DateTime, Deferred, Effect, Exit, Fiber, Layer, Option, Ref, Schema, Stream } from "effect" import { EventV2 } from "@opencode-ai/core/event" import { Event } from "@opencode-ai/schema/event" import { Session } from "@opencode-ai/schema/session" @@ -80,9 +80,7 @@ const durableData = (sessionID: Session.ID, text: string) => ({ /** Followed log read without markers: the old `durable` stream shape. */ const tail = (events: EventV2.Interface, input: { aggregateID: string; after?: number }) => - events - .log({ ...input, follow: true }) - .pipe(Stream.filter((item): item is EventV2.Payload => !EventV2.isCaughtUp(item))) + events.log({ ...input, follow: true }).pipe(Stream.filter((item): item is EventV2.Payload => !EventV2.isSynced(item))) const it = testEffect( AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, Location.node]), [[Location.node, locationLayer]]), @@ -1128,7 +1126,7 @@ describe("EventV2", () => { }), ) - it.effect("log without follow replays events and completes with a caught-up marker", () => + it.effect("log without follow replays events and completes with a synced marker", () => Effect.gen(function* () { const events = yield* EventV2.Service const aggregateID = Session.ID.create() @@ -1137,16 +1135,16 @@ describe("EventV2", () => { const items = Array.from(yield* Stream.runCollect(events.log({ aggregateID }))) - expect(items.map((item) => (EventV2.isCaughtUp(item) ? item.type : item.durable?.seq))).toEqual([ + expect(items.map((item) => (EventV2.isSynced(item) ? item.type : item.durable?.seq))).toEqual([ EventV2.Seq.make(0), EventV2.Seq.make(1), - "log.caught_up", + "log.synced", ]) - expect(items.at(-1)).toEqual({ type: "log.caught_up", aggregateID, seq: EventV2.Seq.make(1) }) + expect(items.at(-1)).toEqual({ type: "log.synced", aggregateID, seq: EventV2.Seq.make(1) }) }), ) - it.effect("log caught-up marker omits seq for an empty log and keeps the cursor otherwise", () => + it.effect("log synced marker omits seq for an empty log and keeps the cursor otherwise", () => Effect.gen(function* () { const events = yield* EventV2.Service const aggregateID = Session.ID.create() @@ -1155,13 +1153,13 @@ describe("EventV2", () => { yield* events.publish(DurableMessage, durableData(aggregateID, "zero")) const drained = Array.from(yield* Stream.runCollect(events.log({ aggregateID, after: 0 }))) - expect(empty).toEqual([{ type: "log.caught_up", aggregateID }]) + expect(empty).toEqual([{ type: "log.synced", aggregateID }]) expect(empty[0]).not.toHaveProperty("seq") - expect(drained).toEqual([{ type: "log.caught_up", aggregateID, seq: EventV2.Seq.make(0) }]) + expect(drained).toEqual([{ type: "log.synced", aggregateID, seq: EventV2.Seq.make(0) }]) }), ) - it.effect("log with follow emits the caught-up marker at the replay-to-live boundary", () => + it.effect("log with follow emits the synced marker at the replay-to-live boundary", () => Effect.gen(function* () { const events = yield* EventV2.Service const aggregateID = Session.ID.create() @@ -1174,14 +1172,79 @@ describe("EventV2", () => { yield* events.publish(DurableMessage, durableData(aggregateID, "one")) const items = Array.from(yield* Fiber.join(fiber)) - expect(items.map((item) => (EventV2.isCaughtUp(item) ? item : item.durable?.seq))).toEqual([ + expect(items.map((item) => (EventV2.isSynced(item) ? item : item.durable?.seq))).toEqual([ EventV2.Seq.make(0), - { type: "log.caught_up", aggregateID, seq: EventV2.Seq.make(0) }, + { type: "log.synced", aggregateID, seq: EventV2.Seq.make(0) }, EventV2.Seq.make(1), ]) }), ) + it.effect("log replays across configured read pages", () => + Effect.gen(function* () { + const eventLayer = EventV2.layerWith({ logReadPageSize: 2 }).pipe(Layer.provide(LayerNode.compile(Database.node))) + + yield* Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = Session.ID.create() + yield* events.publish(DurableMessage, durableData(aggregateID, "zero")) + yield* events.publish(DurableMessage, durableData(aggregateID, "one")) + yield* events.publish(DurableMessage, durableData(aggregateID, "two")) + yield* events.publish(DurableMessage, durableData(aggregateID, "three")) + yield* events.publish(DurableMessage, durableData(aggregateID, "four")) + + const items = Array.from(yield* Stream.runCollect(events.log({ aggregateID }))) + + expect(items.map((item) => (EventV2.isSynced(item) ? item.type : item.durable?.seq))).toEqual([ + EventV2.Seq.make(0), + EventV2.Seq.make(1), + EventV2.Seq.make(2), + EventV2.Seq.make(3), + EventV2.Seq.make(4), + "log.synced", + ]) + expect(items.at(-1)).toEqual({ type: "log.synced", aggregateID, seq: EventV2.Seq.make(4) }) + }).pipe(Effect.provide(Layer.merge(LayerNode.compile(Database.node), eventLayer))) + }), + ) + + it.effect("log with follow emits events committed during replay after the synced marker", () => + Effect.gen(function* () { + const readStarted = yield* Deferred.make() + const releaseRead = yield* Deferred.make() + const firstRead = yield* Ref.make(true) + const eventLayer = EventV2.layerWith({ + beforeAggregateRead: () => + Ref.getAndSet(firstRead, false).pipe( + Effect.flatMap((shouldBlock) => { + if (!shouldBlock) return Effect.void + return Deferred.succeed(readStarted, undefined).pipe(Effect.andThen(Deferred.await(releaseRead))) + }), + ), + }).pipe(Layer.provide(LayerNode.compile(Database.node))) + + yield* Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = Session.ID.create() + yield* events.publish(DurableMessage, durableData(aggregateID, "zero")) + const fiber = yield* events + .log({ aggregateID, follow: true }) + .pipe(Stream.take(3), Stream.runCollect, Effect.forkScoped) + + yield* Deferred.await(readStarted) + yield* events.publish(DurableMessage, durableData(aggregateID, "one")) + yield* Deferred.succeed(releaseRead, undefined) + + const items = Array.from(yield* Fiber.join(fiber)) + expect(items.map((item) => (EventV2.isSynced(item) ? item : item.durable?.seq))).toEqual([ + EventV2.Seq.make(0), + { type: "log.synced", aggregateID, seq: EventV2.Seq.make(0) }, + EventV2.Seq.make(1), + ]) + }).pipe(Effect.provide(Layer.merge(LayerNode.compile(Database.node), eventLayer))) + }), + ) + it.effect("changes emits sweep-required on subscribe then coalesced hints per aggregate", () => Effect.gen(function* () { const events = yield* EventV2.Service diff --git a/packages/core/test/session-create.test.ts b/packages/core/test/session-create.test.ts index 5afd2cd9c4f..90d460ecb79 100644 --- a/packages/core/test/session-create.test.ts +++ b/packages/core/test/session-create.test.ts @@ -8,7 +8,6 @@ import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { EventV2 } from "@opencode-ai/core/event" import { EventTable } from "@opencode-ai/core/event/sql" -import { Job } from "@opencode-ai/core/job" import { Location } from "@opencode-ai/core/location" import { ModelV2 } from "@opencode-ai/core/model" import { ProjectV2 } from "@opencode-ai/core/project" @@ -49,11 +48,11 @@ const it = testEffect( const location = Location.Ref.make({ directory: AbsolutePath.make("/project") }) const id = SessionV2.ID.create() -/** Public session events from a `log` read, without caught-up markers. */ +/** Public session events from a `log` read, without synced markers. */ const logEvents = (session: SessionV2.Interface, sessionID: SessionV2.ID, follow?: boolean) => session .log({ sessionID, follow }) - .pipe(Stream.filter((item): item is SessionEvent.DurableEvent => !EventV2.isCaughtUp(item))) + .pipe(Stream.filter((item): item is SessionEvent.DurableEvent => !EventV2.isSynced(item))) const assertCreateInputTypes = (session: SessionV2.Interface) => { // @ts-expect-error location or parentID is required. @@ -213,7 +212,7 @@ describe("SessionV2.create", () => { durable: { seq: 0 }, data: { sessionID: forked.id, parentID: parent.id }, }) - expect(yield* SessionInput.find(db, forkContext[0]!.id)).toMatchObject({ + expect(yield* SessionInput.find(db, forkContext[0].id)).toMatchObject({ sessionID: forked.id, prompt: { text: "First" }, promotedSeq: 2, diff --git a/packages/core/test/session-log.test.ts b/packages/core/test/session-log.test.ts index 16f1e97ff1e..b80dd2c2060 100644 --- a/packages/core/test/session-log.test.ts +++ b/packages/core/test/session-log.test.ts @@ -35,7 +35,7 @@ const it = testEffect( const location = Location.Ref.make({ directory: AbsolutePath.make("/project") }) describe("SessionV2.log", () => { - it.effect("replays public session events and marks caught-up at the aggregate watermark", () => + it.effect("replays public session events and marks synced at the aggregate watermark", () => Effect.gen(function* () { const session = yield* SessionV2.Service const events = yield* EventV2.Service @@ -47,8 +47,8 @@ describe("SessionV2.log", () => { // Session creation commits a non-public durable event, so the marker's // seq covers more of the aggregate than the public events emitted. - expect(items.map((item) => item.type)).toEqual(["session.next.renamed", "log.caught_up"]) - expect(items.at(-1)).toEqual({ type: "log.caught_up", aggregateID: created.id, seq: watermark }) + expect(items.map((item) => item.type)).toEqual(["session.next.renamed", "log.synced"]) + expect(items.at(-1)).toEqual({ type: "log.synced", aggregateID: created.id, seq: watermark }) }), ) @@ -64,7 +64,7 @@ describe("SessionV2.log", () => { yield* session.rename({ sessionID: created.id, title: "renamed live" }) const items = Array.from(yield* Fiber.join(fiber)) - expect(items.map((item) => item.type)).toEqual(["log.caught_up", "session.next.renamed"]) + expect(items.map((item) => item.type)).toEqual(["log.synced", "session.next.renamed"]) }), ) @@ -95,13 +95,13 @@ describe("SessionV2.log", () => { const items = Array.from(yield* Stream.runCollect(session.log({ sessionID: created.id, after: 1 }))) expect( - items.map((item): number | string | undefined => (EventV2.isCaughtUp(item) ? item.type : item.durable?.seq)), - ).toEqual([3, 4, "log.caught_up"]) - expect(items.at(-1)).toEqual({ type: "log.caught_up", aggregateID: created.id, seq: EventV2.Seq.make(4) }) + items.map((item): number | string | undefined => (EventV2.isSynced(item) ? item.type : item.durable?.seq)), + ).toEqual([3, 4, "log.synced"]) + expect(items.at(-1)).toEqual({ type: "log.synced", aggregateID: created.id, seq: EventV2.Seq.make(4) }) }), ) - it.effect("completes with a bare caught-up marker for a migrated Session with no event sequence", () => + it.effect("completes with a bare synced marker for a migrated Session with no event sequence", () => Effect.gen(function* () { const db = (yield* Database.Service).db const session = yield* SessionV2.Service @@ -125,7 +125,7 @@ describe("SessionV2.log", () => { const items = Array.from(yield* Stream.runCollect(session.log({ sessionID }))) - expect(items).toEqual([{ type: "log.caught_up", aggregateID: sessionID }]) + expect(items).toEqual([{ type: "log.synced", aggregateID: sessionID }]) }), ) }) diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index 9445424f0b4..baf3c303b2d 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/packages/core/test/session-prompt.test.ts @@ -6,7 +6,6 @@ import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { EventV2 } from "@opencode-ai/core/event" import { EventTable } from "@opencode-ai/core/event/sql" -import { Job } from "@opencode-ai/core/job" import { SessionEvent } from "@opencode-ai/core/session/event" import { ModelV2 } from "@opencode-ai/core/model" import { ProviderV2 } from "@opencode-ai/core/provider" @@ -248,7 +247,7 @@ describe("SessionV2.prompt", () => { const publicEvents = (input: { sessionID: SessionV2.ID; after?: number }) => session .log({ ...input, follow: true }) - .pipe(Stream.filter((item): item is SessionEvent.DurableEvent => !EventV2.isCaughtUp(item))) + .pipe(Stream.filter((item): item is SessionEvent.DurableEvent => !EventV2.isSynced(item))) const fiber = yield* publicEvents({ sessionID }).pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped) yield* Effect.yieldNow @@ -265,7 +264,7 @@ describe("SessionV2.prompt", () => { ]) expect( Array.from( - yield* publicEvents({ sessionID, after: streamed[0]!.durable?.seq }).pipe(Stream.take(1), Stream.runCollect), + yield* publicEvents({ sessionID, after: streamed[0].durable?.seq }).pipe(Stream.take(1), Stream.runCollect), ).map((event): [number | undefined, string] => [event.durable?.seq, event.type]), ).toEqual([[1, "session.next.prompt.admitted"]]) }), diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index aca615220f9..d2114ddee99 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -485,7 +485,7 @@ export const makeSessionGroup = (sessionLo follow: BooleanFromString.pipe(Schema.optional), }, success: HttpApiSchema.StreamSse({ - data: Schema.Union([SessionEvent.Durable, EventLog.CaughtUp]).annotate({ identifier: "SessionLogItem" }), + data: Schema.Union([SessionEvent.Durable, EventLog.Synced]).annotate({ identifier: "SessionLogItem" }), }), error: SessionNotFoundError, }) @@ -495,7 +495,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 caught-up marker once the replay reaches the end of the log, 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.", + "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.", }), ), ) diff --git a/packages/schema/src/event-log.ts b/packages/schema/src/event-log.ts index dc4f363ccc6..1b0bbcb924b 100644 --- a/packages/schema/src/event-log.ts +++ b/packages/schema/src/event-log.ts @@ -6,20 +6,19 @@ import { optional } from "./schema.js" /** * Replay-to-live boundary marker for a durable log read. The reader now holds - * every event committed at or below `seq`; `seq` is absent when the log holds - * no events at or before the read position. May be re-emitted after the - * reader internally re-attaches. + * every event committed at or below this watermark; `seq` is absent when the + * captured watermark is empty. Emitted once for the captured watermark. */ -export const CaughtUp = Schema.Struct({ - type: Schema.Literal("log.caught_up"), +export const Synced = Schema.Struct({ + type: Schema.Literal("log.synced"), aggregateID: Schema.String, seq: optional(Event.Seq), }).annotate({ - identifier: "EventLog.CaughtUp", + identifier: "EventLog.Synced", description: - "Marker emitted when a log read transitions from replay to live (or completes a finite read). The reader holds every event committed at or below seq.", + "Marker emitted once when a log read reaches its captured watermark. The reader holds every event committed at or below seq.", }) -export interface CaughtUp extends Schema.Schema.Type {} +export interface Synced extends Schema.Schema.Type {} /** * A payload-free doorbell: the aggregate's log advanced to at least `seq`. diff --git a/packages/schema/src/event.ts b/packages/schema/src/event.ts index cd2fd6532d0..1157a2bc44f 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, caught-up markers, change hints, and snapshot watermarks; + * event envelope, synced markers, change hints, and snapshot watermarks; * `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/schema/test/event.test.ts b/packages/schema/test/event.test.ts index b643b0e601c..a05dca0c6e3 100644 --- a/packages/schema/test/event.test.ts +++ b/packages/schema/test/event.test.ts @@ -1,6 +1,7 @@ import { describe, expect, test } from "bun:test" import { Schema } from "effect" import { Event } from "../src/event.js" +import { EventLog } from "../src/event-log.js" describe("public event schemas", () => { test("definition is pure", () => { @@ -34,4 +35,18 @@ describe("public event schemas", () => { expect(Event.durable([definition]).get("test.durable.1")).toBe(definition) }) + + test("synced marker encodes the captured watermark", () => { + expect( + Schema.encodeSync(EventLog.Synced)({ + type: "log.synced", + aggregateID: "ses_test", + seq: Event.Seq.make(1), + }), + ).toEqual({ type: "log.synced", aggregateID: "ses_test", seq: 1 }) + expect(Schema.encodeSync(EventLog.Synced)({ type: "log.synced", aggregateID: "ses_test" })).toEqual({ + type: "log.synced", + aggregateID: "ses_test", + }) + }) }) diff --git a/packages/sdk-next/test/embedded.test.ts b/packages/sdk-next/test/embedded.test.ts index a34bd628a6b..572cdd96654 100644 --- a/packages/sdk-next/test/embedded.test.ts +++ b/packages/sdk-next/test/embedded.test.ts @@ -85,7 +85,7 @@ it.live( ) const wakeContext = yield* opencode.sessions.context({ sessionID: id }) const event = yield* opencode.sessions.log({ sessionID: id }).pipe( - Stream.filter((item) => item.type !== "log.caught_up"), + Stream.filter((item) => item.type !== "log.synced"), Stream.take(1), Stream.runHead, Effect.map(Option.getOrUndefined), diff --git a/specs/v2/schema-changelog.md b/specs/v2/schema-changelog.md index c0bf822532a..794db93c959 100644 --- a/specs/v2/schema-changelog.md +++ b/specs/v2/schema-changelog.md @@ -1,5 +1,16 @@ # V2 Schema Changelog +## 2026-07-02: Rename Session Log Replay Marker + +- Rename the replay boundary marker from `log.caught_up` / `EventLog.CaughtUp` to `log.synced` / `EventLog.Synced`. +- Define the marker as a single notification that replay reached the captured aggregate watermark; omit `seq` only when the captured watermark is empty. +- Page durable Session log reads toward the captured watermark before switching to live tailing. + +Compatibility: + +- The V2 event stream is still experimental. No durable event rows change because replay markers are stream-only and are not stored. +- Generated clients are intentionally regenerated by the audit integrator lane, not by this parallel branch. + ## 2026-07-02: Add Default Model Endpoint - Add `GET /api/model/default` (`v2.model.default`) returning the Location's resolved default model, or `undefined` when no model is available. diff --git a/specs/v2/session.md b/specs/v2/session.md index b0931fdb2fe..15bc970b471 100644 --- a/specs/v2/session.md +++ b/specs/v2/session.md @@ -172,13 +172,15 @@ Inbox promotion coalesces pending steers in durable admission order. Once contin Eager local-tool execution is intentionally unbounded in the current local slice. This minimizes tool latency but does not increase SQLite settlement throughput: Session-event publication remains serialized per provider turn. Before broadening exposure, revisit per-turn call limits, output truncation, and operational backpressure using observed workloads. The `session.next.*` event schemas remain experimental and unshipped; databases created by earlier experimental builds are disposable rather than compatibility targets. -The synchronized `session.next.*` event family and projected Session-message model predate this branch. This slice refines their replay contract: projected Session messages retain their source aggregate sequence so canonical context ordering and `sessions.messages(...)` pagination follow durable event order even when caller-supplied IDs or timestamps do not. Consumers can use `sessions.events({ sessionID, after? })` to replay durable `session.next.*` events after an aggregate sequence cursor, then tail durable events without a race. Live-only text, reasoning, and tool-input fragments remain available through EventV2 subscriptions for connected renderers; they are intentionally absent from the replayable Session stream. +The synchronized `session.next.*` event family and projected Session-message model predate this branch. This slice refines their replay contract: projected Session messages retain their source aggregate sequence so canonical context ordering and `sessions.messages(...)` pagination follow durable event order even when caller-supplied IDs or timestamps do not. Consumers can use `sessions.log({ sessionID, after? })` to replay durable `session.next.*` events after an aggregate sequence cursor, then tail durable events without a race. Live-only text, reasoning, and tool-input fragments remain available through EventV2 subscriptions for connected renderers; they are intentionally absent from the replayable Session stream. -The first `sessions.events(...)` contract is durable-only during both replay and live tailing. This keeps one cursor equal to one persisted aggregate sequence and is sufficient for reconnect-safe consumers. A later UI-facing API may optionally interleave live-only deltas while connected, but those fragments must remain explicitly ephemeral: they cannot advance the durable cursor, replay after reconnect, or be mistaken for publication boundaries. +The first `sessions.log(...)` contract is durable-only during both replay and live tailing. This keeps one cursor equal to one persisted aggregate sequence and is sufficient for reconnect-safe consumers. A later UI-facing API may optionally interleave live-only deltas while connected, but those fragments must remain explicitly ephemeral: they cannot advance the durable cursor, replay after reconnect, or be mistaken for publication boundaries. + +`sessions.log(...)` captures the aggregate's durable watermark before replay and emits one `log.synced` marker when replay reaches that watermark. The marker carries `seq` when the captured watermark is non-empty and omits it when the log has no sequence. With `follow=true`, the tail subscribes before the replay watermark is captured, so events committed during replay are delivered after `log.synced` with a sequence greater than the marker watermark. Durable reads are paged internally; consumers observe only the ordered event stream plus the single marker. `sessions.history({ sessionID, after?, limit? })` is the finite counterpart for request/response consumers. `after` is an exclusive aggregate sequence, and omission starts before sequence zero. The response is `{ data, hasMore }`; callers derive the next `after` from the final event's durable sequence when `hasMore` is true. Public durable Session events are selected before pagination, which permits gaps from private or historical aggregate events while preserving strictly increasing unique sequences. The log has a moving head, so events committed between pages may appear on the next page. -The finite endpoint is `GET /api/session/:sessionID/history`, uses the normal Session Location and authorization middleware, defaults to 50 events, and accepts at most 100. It returns only events in the public durable Session schema. The existing `sessions.events()` replay-and-tail stream is unchanged. +The finite endpoint is `GET /api/session/:sessionID/history`, uses the normal Session Location and authorization middleware, defaults to 50 events, and accepts at most 100. It returns only events in the public durable Session schema. The existing `sessions.log()` replay-and-tail stream is unchanged except for its explicit `log.synced` replay marker. Durable event tail wakeups are advisory and edge-triggered. Each active tail owns one sliding-capacity-1 dirty signal for its aggregate and re-queries SQLite after a wake. Repeated commits coalesce while the tail is busy because durable rows, not in-memory notifications, preserve every event and sequence. Subscribe and register the dirty signal before historical replay, then remove it when the tail closes, so replay handoff cannot miss a commit and inactive aggregates retain no wake state.