diff --git a/packages/client/src/effect.ts b/packages/client/src/effect.ts index b580c7f48ac..ee8652c7009 100644 --- a/packages/client/src/effect.ts +++ b/packages/client/src/effect.ts @@ -4,6 +4,8 @@ export * from "./generated-effect/index" export { Agent } from "@opencode-ai/schema/agent" export { Command } from "@opencode-ai/schema/command" export { Credential } from "@opencode-ai/schema/credential" +export { Event } from "@opencode-ai/schema/event" +export { EventLog } from "@opencode-ai/schema/event-log" export { FileSystem } from "@opencode-ai/schema/filesystem" export { Integration } from "@opencode-ai/schema/integration" export { Location } from "@opencode-ai/schema/location" diff --git a/packages/client/src/generated-effect/client.ts b/packages/client/src/generated-effect/client.ts index 776862e7ebe..ff36429ec7f 100644 --- a/packages/client/src/generated-effect/client.ts +++ b/packages/client/src/generated-effect/client.ts @@ -80,10 +80,7 @@ const Endpoint4_1 = (raw: RawClient["server.session"]) => (input?: Endpoint4_1In ) const Endpoint4_2 = (raw: RawClient["server.session"]) => () => - raw["session.active"]({}).pipe( - Effect.mapError(mapClientError), - Effect.map((value) => value.data), - ) + raw["session.active"]({}).pipe(Effect.mapError(mapClientError)) type Endpoint4_3Request = Parameters[0] type Endpoint4_3Input = { readonly sessionID: Endpoint4_3Request["params"]["sessionID"] } @@ -282,47 +279,39 @@ const Endpoint4_20 = (raw: RawClient["server.session"]) => (input: Endpoint4_20I Effect.mapError(mapClientError), ) -type Endpoint4_21Request = Parameters[0] +type Endpoint4_21Request = Parameters[0] type Endpoint4_21Input = { readonly sessionID: Endpoint4_21Request["params"]["sessionID"] - readonly limit?: Endpoint4_21Request["query"]["limit"] readonly after?: Endpoint4_21Request["query"]["after"] + readonly follow?: Endpoint4_21Request["query"]["follow"] } const Endpoint4_21 = (raw: RawClient["server.session"]) => (input: Endpoint4_21Input) => - raw["session.history"]({ - params: { sessionID: input["sessionID"] }, - query: { limit: input["limit"], after: input["after"] }, - }).pipe(Effect.mapError(mapClientError)) - -type Endpoint4_22Request = Parameters[0] -type Endpoint4_22Input = { - readonly sessionID: Endpoint4_22Request["params"]["sessionID"] - readonly after?: Endpoint4_22Request["query"]["after"] -} -const Endpoint4_22 = (raw: RawClient["server.session"]) => (input: Endpoint4_22Input) => Stream.unwrap( - raw["session.events"]({ params: { sessionID: input["sessionID"] }, query: { after: input["after"] } }).pipe( + raw["session.log"]({ + params: { sessionID: input["sessionID"] }, + query: { after: input["after"], follow: input["follow"] }, + }).pipe( Effect.mapError(mapClientError), Effect.map((stream) => stream.pipe(Stream.mapError(mapClientError))), ), ) -type Endpoint4_23Request = Parameters[0] -type Endpoint4_23Input = { readonly sessionID: Endpoint4_23Request["params"]["sessionID"] } -const Endpoint4_23 = (raw: RawClient["server.session"]) => (input: Endpoint4_23Input) => +type Endpoint4_22Request = Parameters[0] +type Endpoint4_22Input = { readonly sessionID: Endpoint4_22Request["params"]["sessionID"] } +const Endpoint4_22 = (raw: RawClient["server.session"]) => (input: Endpoint4_22Input) => raw["session.interrupt"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)) -type Endpoint4_24Request = Parameters[0] -type Endpoint4_24Input = { readonly sessionID: Endpoint4_24Request["params"]["sessionID"] } -const Endpoint4_24 = (raw: RawClient["server.session"]) => (input: Endpoint4_24Input) => +type Endpoint4_23Request = Parameters[0] +type Endpoint4_23Input = { readonly sessionID: Endpoint4_23Request["params"]["sessionID"] } +const Endpoint4_23 = (raw: RawClient["server.session"]) => (input: Endpoint4_23Input) => raw["session.background"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)) -type Endpoint4_25Request = Parameters[0] -type Endpoint4_25Input = { - readonly sessionID: Endpoint4_25Request["params"]["sessionID"] - readonly messageID: Endpoint4_25Request["params"]["messageID"] +type Endpoint4_24Request = Parameters[0] +type Endpoint4_24Input = { + readonly sessionID: Endpoint4_24Request["params"]["sessionID"] + readonly messageID: Endpoint4_24Request["params"]["messageID"] } -const Endpoint4_25 = (raw: RawClient["server.session"]) => (input: Endpoint4_25Input) => +const Endpoint4_24 = (raw: RawClient["server.session"]) => (input: Endpoint4_24Input) => raw["session.message"]({ params: { sessionID: input["sessionID"], messageID: input["messageID"] } }).pipe( Effect.mapError(mapClientError), Effect.map((value) => value.data), @@ -350,11 +339,10 @@ const adaptGroup4 = (raw: RawClient["server.session"]) => ({ listContextEntries: Endpoint4_18(raw), putContextEntry: Endpoint4_19(raw), removeContextEntry: Endpoint4_20(raw), - history: Endpoint4_21(raw), - events: Endpoint4_22(raw), - interrupt: Endpoint4_23(raw), - background: Endpoint4_24(raw), - message: Endpoint4_25(raw), + log: Endpoint4_21(raw), + interrupt: Endpoint4_22(raw), + background: Endpoint4_23(raw), + message: Endpoint4_24(raw), }) type Endpoint5_0Request = Parameters[0] @@ -696,7 +684,15 @@ const Endpoint17_0 = (raw: RawClient["server.event"]) => () => ), ) -const adaptGroup17 = (raw: RawClient["server.event"]) => ({ subscribe: Endpoint17_0(raw) }) +const Endpoint17_1 = (raw: RawClient["server.event"]) => () => + Stream.unwrap( + raw["event.changes"]({}).pipe( + Effect.mapError(mapClientError), + Effect.map((stream) => stream.pipe(Stream.mapError(mapClientError))), + ), + ) + +const adaptGroup17 = (raw: RawClient["server.event"]) => ({ subscribe: Endpoint17_0(raw), changes: Endpoint17_1(raw) }) type Endpoint18_0Request = Parameters[0] type Endpoint18_0Input = { readonly location?: Endpoint18_0Request["query"]["location"] } diff --git a/packages/client/src/generated/client.ts b/packages/client/src/generated/client.ts index 485ee780391..cf841a43f10 100644 --- a/packages/client/src/generated/client.ts +++ b/packages/client/src/generated/client.ts @@ -47,10 +47,8 @@ import type { SessionPutContextEntryOutput, SessionRemoveContextEntryInput, SessionRemoveContextEntryOutput, - SessionHistoryInput, - SessionHistoryOutput, - SessionEventsInput, - SessionEventsOutput, + SessionLogInput, + SessionLogOutput, SessionInterruptInput, SessionInterruptOutput, SessionBackgroundInput, @@ -118,6 +116,7 @@ import type { SkillListInput, SkillListOutput, EventSubscribeOutput, + EventChangesOutput, PtyListInput, PtyListOutput, PtyCreateInput, @@ -380,7 +379,7 @@ export function make(options: ClientOptions) { requestOptions, ).then((value) => value.data), active: (requestOptions?: RequestOptions) => - request<{ readonly data: SessionActiveOutput }>( + request( { method: "GET", path: `/api/session/active`, @@ -389,7 +388,7 @@ export function make(options: ClientOptions) { empty: false, }, requestOptions, - ).then((value) => value.data), + ), get: (input: SessionGetInput, requestOptions?: RequestOptions) => request<{ readonly data: SessionGetOutput }>( { @@ -608,24 +607,12 @@ export function make(options: ClientOptions) { }, requestOptions, ), - history: (input: SessionHistoryInput, requestOptions?: RequestOptions) => - request( + log: (input: SessionLogInput, requestOptions?: RequestOptions): AsyncIterable => + sse( { method: "GET", - path: `/api/session/${encodeURIComponent(input.sessionID)}/history`, - query: { limit: input["limit"], after: input["after"] }, - successStatus: 200, - declaredStatuses: [404, 400, 401], - empty: false, - }, - requestOptions, - ), - events: (input: SessionEventsInput, requestOptions?: RequestOptions): AsyncIterable => - sse( - { - method: "GET", - path: `/api/session/${encodeURIComponent(input.sessionID)}/event`, - query: { after: input["after"] }, + path: `/api/session/${encodeURIComponent(input.sessionID)}/log`, + query: { after: input["after"], follow: input["follow"] }, successStatus: 200, declaredStatuses: [404, 400, 401], empty: false, @@ -1067,6 +1054,11 @@ 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/generated/types.ts b/packages/client/src/generated/types.ts index abfc8cd926d..9c6a388fd97 100644 --- a/packages/client/src/generated/types.ts +++ b/packages/client/src/generated/types.ts @@ -315,6 +315,7 @@ export type SessionListOutput = { }> } }> + readonly watermarks: { readonly [x: string]: number } readonly cursor: { readonly previous?: string | null; readonly next?: string | null } } @@ -379,7 +380,10 @@ export type SessionCreateOutput = { } }["data"] -export type SessionActiveOutput = { readonly data: { readonly [x: string]: { readonly type: "running" } } }["data"] +export type SessionActiveOutput = { + readonly data: { readonly [x: string]: { readonly type: "running" } } + readonly watermarks: { readonly [x: string]: number } +} export type SessionGetInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] } @@ -1046,994 +1050,501 @@ export type SessionRemoveContextEntryInput = { export type SessionRemoveContextEntryOutput = void -export type SessionHistoryInput = { +export type SessionLogInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] - readonly limit?: { readonly limit?: number | undefined; readonly after?: number | undefined }["limit"] - readonly after?: { readonly limit?: number | undefined; readonly after?: number | undefined }["after"] + readonly after?: { readonly after?: number | undefined; readonly follow?: boolean | undefined }["after"] + readonly follow?: { readonly after?: number | undefined; readonly follow?: boolean | undefined }["follow"] } -export type SessionHistoryOutput = { - readonly data: ReadonlyArray< - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.agent.switched" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly agent: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.model.switched" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly model: { readonly id: string; readonly providerID: string; readonly variant?: string } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.moved" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly location: { readonly directory: string; readonly workspaceID?: string } - readonly subdirectory?: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.renamed" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { readonly timestamp: number; readonly sessionID: string; readonly title: string } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.forked" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly parentID: string - readonly messageID?: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.prompted" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly prompt: { - readonly text: string - readonly files?: ReadonlyArray<{ - readonly uri: string - readonly mime: string - readonly name?: string - readonly description?: string - readonly source?: { readonly start: number; readonly end: number; readonly text: string } - }> - readonly agents?: ReadonlyArray<{ - readonly name: string - readonly source?: { readonly start: number; readonly end: number; readonly text: string } - }> - } - readonly delivery: "steer" | "queue" - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.prompt.admitted" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly prompt: { - readonly text: string - readonly files?: ReadonlyArray<{ - readonly uri: string - readonly mime: string - readonly name?: string - readonly description?: string - readonly source?: { readonly start: number; readonly end: number; readonly text: string } - }> - readonly agents?: ReadonlyArray<{ - readonly name: string - readonly source?: { readonly start: number; readonly end: number; readonly text: string } - }> - } - readonly delivery: "steer" | "queue" - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.context.updated" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly text: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.synthetic" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly text: string - readonly description?: string - readonly metadata?: { readonly [x: string]: JsonValue } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.skill.activated" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly name: string - readonly text: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.shell.started" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly callID: string - readonly command: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.shell.ended" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly callID: string - readonly output: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.step.started" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly agent: string - readonly model: { readonly id: string; readonly providerID: string; readonly variant?: string } - readonly snapshot?: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.step.ended" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly finish: string - readonly cost: number - readonly tokens: { - readonly input: number - readonly output: number - readonly reasoning: number - readonly cache: { readonly read: number; readonly write: number } - } - readonly snapshot?: string - readonly files?: ReadonlyArray - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.step.failed" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly error: { readonly type: "unknown"; readonly message: string } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.text.started" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly textID: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.text.ended" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly textID: string - readonly text: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.tool.input.started" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly callID: string - readonly name: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.tool.input.ended" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly callID: string - readonly text: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.tool.called" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly callID: string - readonly tool: string - readonly input: { readonly [x: string]: JsonValue } - readonly provider: { - readonly executed: boolean - readonly metadata?: { readonly [x: string]: { readonly [x: string]: JsonValue } } - } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.tool.progress" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly callID: string - readonly structured: { readonly [x: string]: JsonValue } - readonly content: ReadonlyArray< - | { readonly type: "text"; readonly text: string } - | { readonly type: "file"; readonly uri: string; readonly mime: string; readonly name?: string } - > - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.tool.success" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly callID: string - readonly structured: { readonly [x: string]: JsonValue } - readonly content: ReadonlyArray< - | { readonly type: "text"; readonly text: string } - | { readonly type: "file"; readonly uri: string; readonly mime: string; readonly name?: string } - > - readonly outputPaths?: ReadonlyArray - readonly result?: JsonValue - readonly provider: { - readonly executed: boolean - readonly metadata?: { readonly [x: string]: { readonly [x: string]: JsonValue } } - } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.tool.failed" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly callID: string - readonly error: { readonly type: "unknown"; readonly message: string } - readonly result?: JsonValue - readonly provider: { - readonly executed: boolean - readonly metadata?: { readonly [x: string]: { readonly [x: string]: JsonValue } } - } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.reasoning.started" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly reasoningID: string - readonly providerMetadata?: { readonly [x: string]: { readonly [x: string]: JsonValue } } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.reasoning.ended" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly reasoningID: string - readonly text: string - readonly providerMetadata?: { readonly [x: string]: { readonly [x: string]: JsonValue } } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.retried" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly attempt: number - readonly error: { - readonly message: string - readonly statusCode?: number - readonly isRetryable: boolean - readonly responseHeaders?: { readonly [x: string]: string } - readonly responseBody?: string - readonly metadata?: { readonly [x: string]: string } - } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.compaction.started" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly reason: "auto" | "manual" - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.compaction.ended" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly reason: "auto" | "manual" - readonly text: string - readonly recent: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.revert.staged" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly revert: { +export type SessionLogOutput = + | ( + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.agent.switched" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string readonly messageID: string - readonly partID?: string - readonly snapshot?: string - readonly diff?: string - readonly files?: ReadonlyArray<{ - readonly path: string - readonly status: "added" | "modified" | "deleted" - readonly additions: number - readonly deletions: number - readonly patch: string - }> + readonly agent: string } } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.revert.cleared" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { readonly timestamp: number; readonly sessionID: string } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: JsonValue } - readonly type: "session.next.revert.committed" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { readonly timestamp: number; readonly sessionID: string; readonly messageID: string } - } - > - readonly hasMore: boolean -} - -export type SessionEventsInput = { - readonly sessionID: { readonly sessionID: string }["sessionID"] - readonly after?: { readonly after?: number | undefined }["after"] -} - -export type SessionEventsOutput = - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.agent.switched" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly agent: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.model.switched" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly model: { readonly id: string; readonly providerID: string; readonly variant?: string } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.moved" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly location: { readonly directory: string; readonly workspaceID?: string } - readonly subdirectory?: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.renamed" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { readonly timestamp: number; readonly sessionID: string; readonly title: string } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.forked" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly parentID: string - readonly messageID?: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.prompted" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly prompt: { - readonly text: string - readonly files?: ReadonlyArray<{ - readonly uri: string - readonly mime: string - readonly name?: string + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.model.switched" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly messageID: string + readonly model: { readonly id: string; readonly providerID: string; readonly variant?: string } + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.moved" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly location: { readonly directory: string; readonly workspaceID?: string } + readonly subdirectory?: string + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.renamed" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { readonly timestamp: number; readonly sessionID: string; readonly title: string } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.forked" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly parentID: string + readonly messageID?: string + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.prompted" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly messageID: string + readonly prompt: { + readonly text: string + readonly files?: ReadonlyArray<{ + readonly uri: string + readonly mime: string + readonly name?: string + readonly description?: string + readonly source?: { readonly start: number; readonly end: number; readonly text: string } + }> + readonly agents?: ReadonlyArray<{ + readonly name: string + readonly source?: { readonly start: number; readonly end: number; readonly text: string } + }> + } + readonly delivery: "steer" | "queue" + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.prompt.admitted" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly messageID: string + readonly prompt: { + readonly text: string + readonly files?: ReadonlyArray<{ + readonly uri: string + readonly mime: string + readonly name?: string + readonly description?: string + readonly source?: { readonly start: number; readonly end: number; readonly text: string } + }> + readonly agents?: ReadonlyArray<{ + readonly name: string + readonly source?: { readonly start: number; readonly end: number; readonly text: string } + }> + } + readonly delivery: "steer" | "queue" + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.context.updated" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly messageID: string + readonly text: string + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.synthetic" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly messageID: string + readonly text: string readonly description?: string - readonly source?: { readonly start: number; readonly end: number; readonly text: string } - }> - readonly agents?: ReadonlyArray<{ + readonly metadata?: { readonly [x: string]: unknown } + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.skill.activated" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly messageID: string readonly name: string - readonly source?: { readonly start: number; readonly end: number; readonly text: string } - }> + readonly text: string + } } - readonly delivery: "steer" | "queue" - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.prompt.admitted" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly prompt: { - readonly text: string - readonly files?: ReadonlyArray<{ - readonly uri: string - readonly mime: string - readonly name?: string - readonly description?: string - readonly source?: { readonly start: number; readonly end: number; readonly text: string } - }> - readonly agents?: ReadonlyArray<{ + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.shell.started" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly messageID: string + readonly callID: string + readonly command: string + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.shell.ended" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly callID: string + readonly output: string + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.step.started" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly assistantMessageID: string + readonly agent: string + readonly model: { readonly id: string; readonly providerID: string; readonly variant?: string } + readonly snapshot?: string + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.step.ended" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly assistantMessageID: string + readonly finish: string + readonly cost: number + readonly tokens: { + readonly input: number + readonly output: number + readonly reasoning: number + readonly cache: { readonly read: number; readonly write: number } + } + readonly snapshot?: string + readonly files?: ReadonlyArray + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.step.failed" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly assistantMessageID: string + readonly error: { readonly type: "unknown"; readonly message: string } + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.text.started" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly assistantMessageID: string + readonly textID: string + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.text.ended" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly assistantMessageID: string + readonly textID: string + readonly text: string + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.tool.input.started" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly assistantMessageID: string + readonly callID: string readonly name: string - readonly source?: { readonly start: number; readonly end: number; readonly text: string } - }> + } } - readonly delivery: "steer" | "queue" - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.context.updated" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly text: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.synthetic" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly text: string - readonly description?: string - readonly metadata?: { readonly [x: string]: unknown } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.skill.activated" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly name: string - readonly text: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.shell.started" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly callID: string - readonly command: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.shell.ended" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly callID: string - readonly output: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.step.started" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly agent: string - readonly model: { readonly id: string; readonly providerID: string; readonly variant?: string } - readonly snapshot?: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.step.ended" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly finish: string - readonly cost: number - readonly tokens: { - readonly input: number - readonly output: number - readonly reasoning: number - readonly cache: { readonly read: number; readonly write: number } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.tool.input.ended" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly assistantMessageID: string + readonly callID: string + readonly text: string + } } - readonly snapshot?: string - readonly files?: ReadonlyArray - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.step.failed" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly error: { readonly type: "unknown"; readonly message: string } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.text.started" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly textID: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.text.ended" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly textID: string - readonly text: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.tool.input.started" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly callID: string - readonly name: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.tool.input.ended" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly callID: string - readonly text: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.tool.called" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly callID: string - readonly tool: string - readonly input: { readonly [x: string]: unknown } - readonly provider: { - readonly executed: boolean - readonly metadata?: { readonly [x: string]: { readonly [x: string]: unknown } } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.tool.called" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly assistantMessageID: string + readonly callID: string + readonly tool: string + readonly input: { readonly [x: string]: unknown } + readonly provider: { + readonly executed: boolean + readonly metadata?: { readonly [x: string]: { readonly [x: string]: unknown } } + } + } } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.tool.progress" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly callID: string - readonly structured: { readonly [x: string]: unknown } - readonly content: ReadonlyArray< - | { readonly type: "text"; readonly text: string } - | { readonly type: "file"; readonly uri: string; readonly mime: string; readonly name?: string } - > - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.tool.success" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly callID: string - readonly structured: { readonly [x: string]: unknown } - readonly content: ReadonlyArray< - | { readonly type: "text"; readonly text: string } - | { readonly type: "file"; readonly uri: string; readonly mime: string; readonly name?: string } - > - readonly outputPaths?: ReadonlyArray - readonly result?: unknown - readonly provider: { - readonly executed: boolean - readonly metadata?: { readonly [x: string]: { readonly [x: string]: unknown } } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.tool.progress" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly assistantMessageID: string + readonly callID: string + readonly structured: { readonly [x: string]: unknown } + readonly content: ReadonlyArray< + | { readonly type: "text"; readonly text: string } + | { readonly type: "file"; readonly uri: string; readonly mime: string; readonly name?: string } + > + } } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.tool.failed" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly callID: string - readonly error: { readonly type: "unknown"; readonly message: string } - readonly result?: unknown - readonly provider: { - readonly executed: boolean - readonly metadata?: { readonly [x: string]: { readonly [x: string]: unknown } } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.tool.success" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly assistantMessageID: string + readonly callID: string + readonly structured: { readonly [x: string]: unknown } + readonly content: ReadonlyArray< + | { readonly type: "text"; readonly text: string } + | { readonly type: "file"; readonly uri: string; readonly mime: string; readonly name?: string } + > + readonly outputPaths?: ReadonlyArray + readonly result?: unknown + readonly provider: { + readonly executed: boolean + readonly metadata?: { readonly [x: string]: { readonly [x: string]: unknown } } + } + } } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.reasoning.started" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly reasoningID: string - readonly providerMetadata?: { readonly [x: string]: { readonly [x: string]: unknown } } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.reasoning.ended" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly assistantMessageID: string - readonly reasoningID: string - readonly text: string - readonly providerMetadata?: { readonly [x: string]: { readonly [x: string]: unknown } } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.retried" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly attempt: number - readonly error: { - readonly message: string - readonly statusCode?: number - readonly isRetryable: boolean - readonly responseHeaders?: { readonly [x: string]: string } - readonly responseBody?: string - readonly metadata?: { readonly [x: string]: string } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.tool.failed" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly assistantMessageID: string + readonly callID: string + readonly error: { readonly type: "unknown"; readonly message: string } + readonly result?: unknown + readonly provider: { + readonly executed: boolean + readonly metadata?: { readonly [x: string]: { readonly [x: string]: unknown } } + } + } } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.compaction.started" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly reason: "auto" | "manual" - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.compaction.ended" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly messageID: string - readonly reason: "auto" | "manual" - readonly text: string - readonly recent: string - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.revert.staged" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { - readonly timestamp: number - readonly sessionID: string - readonly revert: { - readonly messageID: string - readonly partID?: string - readonly snapshot?: string - readonly diff?: string - readonly files?: ReadonlyArray<{ - readonly path: string - readonly status: "added" | "modified" | "deleted" - readonly additions: number - readonly deletions: number - readonly patch: string - }> + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.reasoning.started" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly assistantMessageID: string + readonly reasoningID: string + readonly providerMetadata?: { readonly [x: string]: { readonly [x: string]: unknown } } + } } - } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.revert.cleared" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { readonly timestamp: number; readonly sessionID: string } - } - | { - readonly id: string - readonly metadata?: { readonly [x: string]: unknown } - readonly type: "session.next.revert.committed" - readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } - readonly location?: { readonly directory: string; readonly workspaceID?: string } - readonly data: { readonly timestamp: number; readonly sessionID: string; readonly messageID: string } - } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.reasoning.ended" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly assistantMessageID: string + readonly reasoningID: string + readonly text: string + readonly providerMetadata?: { readonly [x: string]: { readonly [x: string]: unknown } } + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.retried" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly attempt: number + readonly error: { + readonly message: string + readonly statusCode?: number + readonly isRetryable: boolean + readonly responseHeaders?: { readonly [x: string]: string } + readonly responseBody?: string + readonly metadata?: { readonly [x: string]: string } + } + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.compaction.started" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly messageID: string + readonly reason: "auto" | "manual" + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.compaction.ended" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly messageID: string + readonly reason: "auto" | "manual" + readonly text: string + readonly recent: string + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.revert.staged" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { + readonly timestamp: number + readonly sessionID: string + readonly revert: { + readonly messageID: string + readonly partID?: string + readonly snapshot?: string + readonly diff?: string + readonly files?: ReadonlyArray<{ + readonly path: string + readonly status: "added" | "modified" | "deleted" + readonly additions: number + readonly deletions: number + readonly patch: string + }> + } + } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.revert.cleared" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { readonly timestamp: number; readonly sessionID: string } + } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.revert.committed" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { readonly timestamp: number; readonly sessionID: string; readonly messageID: string } + } + ) + | { readonly type: "log.caught_up"; readonly aggregateID: string; readonly seq?: number } export type SessionInterruptInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] } @@ -2390,6 +1901,7 @@ export type MessageListOutput = { readonly time: { readonly created: number } } > + readonly watermark?: number readonly cursor: { readonly previous?: string | null; readonly next?: string | null } } @@ -4508,6 +4020,10 @@ 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 37105dfa239..931a575d8b9 100644 --- a/packages/client/test/effect.test.ts +++ b/packages/client/test/effect.test.ts @@ -1,7 +1,9 @@ import { expect, test } from "bun:test" import { DateTime, Effect, Stream } from "effect" import { HttpClient, HttpClientResponse } from "effect/unstable/http" -import { AbsolutePath, Agent, Location, Model, OpenCode, Prompt, Session, SessionMessage } from "../src/effect" +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) } test("session.get returns the decoded Effect projection", async () => { const httpClient = HttpClient.make((request) => @@ -60,32 +62,20 @@ test("event.subscribe terminates on Effect protocol decode failures", async () = }) test("session methods retain decoded Effect inputs and outputs", async () => { - const historyQueries: Array> = [] - let historyPage = 0 + const logQueries: Array> = [] const httpClient = HttpClient.make((request) => { const url = request.url - if (url.includes("/event")) { + if (url.includes("/log")) { + logQueries.push(Object.fromEntries(request.urlParams.params)) return Effect.succeed( HttpClientResponse.fromWeb( request, - new Response(`data: ${JSON.stringify(modelSwitchedEvent)}\n\n`, { + new Response(`data: ${JSON.stringify(modelSwitchedEvent)}\n\ndata: ${JSON.stringify(caughtUp)}\n\n`, { headers: { "content-type": "text/event-stream" }, }), ), ) } - if (url.includes("/history")) { - historyPage++ - historyQueries.push(Object.fromEntries(request.urlParams.params)) - return Effect.succeed( - HttpClientResponse.fromWeb( - request, - Response.json( - historyPage === 1 ? { data: [modelSwitchedEvent], hasMore: true } : { data: [], hasMore: false }, - ), - ), - ) - } if (url.includes("/prompt")) { return Effect.succeed(HttpClientResponse.fromWeb(request, Response.json(admission))) } @@ -97,7 +87,10 @@ test("session methods retain decoded Effect inputs and outputs", async () => { } if (url.endsWith("/api/session/active")) { return Effect.succeed( - HttpClientResponse.fromWeb(request, Response.json({ data: { ses_test: { type: "running" } } })), + HttpClientResponse.fromWeb( + request, + Response.json({ data: { ses_test: { type: "running" } }, watermarks: { ses_test: 3 } }), + ), ) } if (request.method === "POST" && url.endsWith("/api/session")) { @@ -107,7 +100,10 @@ test("session methods retain decoded Effect inputs and outputs", async () => { return Effect.succeed(HttpClientResponse.fromWeb(request, new Response(null, { status: 204 }))) } return Effect.succeed( - HttpClientResponse.fromWeb(request, Response.json({ data: [session.data], cursor: { next: "next" } })), + HttpClientResponse.fromWeb( + request, + Response.json({ data: [session.data], watermarks: { ses_test: 3 }, cursor: { next: "next" } }), + ), ) }) const result = await Effect.gen(function* () { @@ -130,31 +126,20 @@ test("session methods retain decoded Effect inputs and outputs", async () => { yield* client.session.compact({ sessionID: Session.ID.make("ses_test") }) yield* client.session.wait({ sessionID: Session.ID.make("ses_test") }) const context = yield* client.session.context({ sessionID: Session.ID.make("ses_test") }) - const history = yield* client.session.history({ - sessionID: Session.ID.make("ses_test"), - after: 0, - limit: 1, - }) - const historyNext = history.hasMore - ? yield* client.session.history({ - sessionID: Session.ID.make("ses_test"), - after: history.data.at(-1)?.durable?.seq, - limit: 2, - }) - : undefined - const events = yield* client.session - .events({ sessionID: Session.ID.make("ses_test"), after: 0 }) + const log = yield* client.session + .log({ sessionID: Session.ID.make("ses_test"), after: Event.Seq.make(0) }) .pipe(Stream.runCollect) yield* client.session.interrupt({ sessionID: Session.ID.make("ses_test") }) const message = yield* client.session.message({ sessionID: Session.ID.make("ses_test"), messageID: SessionMessage.ID.make("msg_model"), }) - return { page, active, created, admitted, context, history, historyNext, events, message } + return { page, active, created, admitted, context, log, message } }).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({ ses_test: { type: "running" } }) + expect(result.active).toEqual({ data: { ses_test: { type: "running" } }, watermarks: { ses_test: 3 } }) + expect(result.page.watermarks).toEqual({ ses_test: 3 }) 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") @@ -162,16 +147,17 @@ test("session methods retain decoded Effect inputs and outputs", async () => { expect(Object.getPrototypeOf(result.admitted.prompt)).toBe(Object.prototype) expect(DateTime.toEpochMillis(result.admitted.timeCreated)).toBe(1_717_171_717_000) expect(result.context).toEqual([]) - expect(DateTime.toEpochMillis(result.history.data[0].data.timestamp)).toBe(1_717_171_717_000) - expect(result.history).toEqual(expect.objectContaining({ hasMore: true })) - expect(result.historyNext).toEqual({ data: [], hasMore: false }) - expect(historyQueries[0]).toEqual({ limit: "1", after: "0" }) - expect(historyQueries[1]).toEqual({ limit: "2", after: "1" }) - expect(DateTime.toEpochMillis(result.events[0].data.timestamp)).toBe(1_717_171_717_000) + 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[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(result.message).toEqual(expect.objectContaining({ id: "msg_model", type: "model-switched" })) }) -test("session.history retains the typed SessionNotFoundError", async () => { +test("session.log retains the typed SessionNotFoundError", async () => { const httpClient = HttpClient.make((request) => Effect.succeed( HttpClientResponse.fromWeb( @@ -185,11 +171,7 @@ test("session.history retains the typed SessionNotFoundError", async () => { ) const error = await Effect.gen(function* () { const client = yield* OpenCode.make({ baseUrl: "http://localhost:3000" }) - return yield* client.session - .history({ - sessionID: Session.ID.make("ses_missing"), - }) - .pipe(Effect.flip) + return yield* client.session.log({ sessionID: Session.ID.make("ses_missing") }).pipe(Stream.runCollect, Effect.flip) }).pipe(Effect.provideService(HttpClient.HttpClient, httpClient), Effect.runPromise) expect(error._tag).toBe("SessionNotFoundError") diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index bf7e441c09a..f0c648ff267 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -8,12 +8,14 @@ test("exposes every standard HTTP API group", () => { "health", "location", "agent", + "plugin", "session", "message", "model", "generate", "provider", "integration", + "server.mcp", "credential", "project", "permission", @@ -172,7 +174,6 @@ test("event.subscribe terminates on malformed Promise SSE data", async () => { test("session methods use the public HTTP contract", async () => { const requests: Array<{ url: string; init?: RequestInit }> = [] - let historyPage = 0 const client = OpenCode.make({ baseUrl: "http://localhost:3000", fetch: async (input, init) => { @@ -183,16 +184,16 @@ test("session methods use the public HTTP contract", async () => { headers: { "content-type": "text/event-stream" }, }) } - if (url.includes("/history")) { - historyPage++ - return Response.json( - historyPage === 1 ? { data: [modelSwitchedEvent], hasMore: true } : { data: [], hasMore: false }, - ) + if (url.includes("/log")) { + return new Response(`data: ${JSON.stringify(modelSwitchedEvent)}\n\ndata: ${JSON.stringify(caughtUp)}\n\n`, { + headers: { "content-type": "text/event-stream" }, + }) } if (url.includes("/prompt")) return Response.json(admission) 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" } } }) + if (url.endsWith("/api/session/active")) + return Response.json({ data: { ses_test: { type: "running" } }, watermarks: { ses_test: 3 } }) 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" } }) @@ -215,24 +216,17 @@ test("session methods use the public HTTP contract", async () => { await client.session.compact({ sessionID: "ses_test" }) await client.session.wait({ sessionID: "ses_test" }) const context = await client.session.context({ sessionID: "ses_test" }) - const history = await client.session.history({ sessionID: "ses_test", after: 0, limit: 1 }) - const historyAfter = history.data.at(-1)?.durable?.seq - const historyNext = history.hasMore - ? await client.session.history({ sessionID: "ses_test", after: historyAfter, limit: 2 }) - : undefined - const events = [] - for await (const event of client.session.events({ sessionID: "ses_test", after: 0 })) events.push(event) + const log = [] + for await (const item of client.session.log({ sessionID: "ses_test", after: 0 })) log.push(item) await client.session.interrupt({ sessionID: "ses_test" }) const message = await client.session.message({ sessionID: "ses_test", messageID: "msg_model" }) expect(page.cursor.next).toBe("next") - expect(active).toEqual({ ses_test: { type: "running" } }) + expect(active).toEqual({ data: { ses_test: { type: "running" } }, watermarks: { ses_test: 3 } }) expect(created.id).toBe("ses_test") expect(admitted.id).toBe("msg_test") expect(context).toEqual([]) - expect(history).toEqual({ data: [modelSwitchedEvent], hasMore: true }) - expect(historyNext).toEqual({ data: [], hasMore: false }) - expect(events).toEqual([modelSwitchedEvent]) + expect(log).toEqual([modelSwitchedEvent, caughtUp]) expect(message).toEqual(modelSwitchedMessage) expect(requests.map((request) => [request.init?.method, request.url])).toEqual([ ["GET", "http://localhost:3000/api/session?limit=10&order=desc"], @@ -244,9 +238,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/history?limit=1&after=0"], - ["GET", "http://localhost:3000/api/session/ses_test/history?limit=2&after=1"], - ["GET", "http://localhost:3000/api/session/ses_test/event?after=0"], + ["GET", "http://localhost:3000/api/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"], ]) @@ -273,7 +265,7 @@ test("middleware errors remain declared client errors", async () => { } }) -test("session.history decodes SessionNotFoundError", async () => { +test("session.log decodes SessionNotFoundError", async () => { const client = OpenCode.make({ baseUrl: "http://localhost:3000", fetch: async () => @@ -284,7 +276,7 @@ test("session.history decodes SessionNotFoundError", async () => { }) try { - await client.session.history({ sessionID: "ses_missing" }) + await client.session.log({ sessionID: "ses_missing" })[Symbol.asyncIterator]().next() throw new Error("Expected request to fail") } catch (error) { expect(isSessionNotFoundError(error)).toBe(true) @@ -329,6 +321,8 @@ const modelSwitchedMessage = { model: { id: "claude", providerID: "anthropic" }, } +const caughtUp = { type: "log.caught_up", aggregateID: "ses_test", seq: 1 } + const modelSwitchedEvent = { id: "evt_model", type: "session.next.model.switched", diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index 37233eb2965..0959bdc44bd 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -3,6 +3,7 @@ export * as EventV2 from "./event" import { Cause, Context, Effect, Layer, Option, PubSub, Queue, Schema, Stream } from "effect" 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 { Database } from "./database/database" import { EventSequenceTable, EventTable } from "./event/sql" @@ -13,6 +14,10 @@ import { Durable } from "@opencode-ai/schema/durable-event-manifest" export const ID = Event.ID export type ID = import("@opencode-ai/schema/event").ID +export const Seq = Event.Seq +export type Seq = import("@opencode-ai/schema/event").Seq +export const Version = Event.Version +export type Version = import("@opencode-ai/schema/event").Version export type { Data, Definition, Payload } from "@opencode-ai/schema/event" export type Subscriber = (event: Payload) => Effect.Effect @@ -63,6 +68,12 @@ export class InvalidDurableEventError extends Schema.TaggedErrorClass ({ + aggregateID, + seq: Seq.make(seq), + version: Version.make(version), +}) + const decodeSerializedEvent = (event: SerializedEvent): Payload => { const definition = Durable.get(event.type) if (!definition?.durable) { @@ -71,58 +82,11 @@ const decodeSerializedEvent = (event: SerializedEvent): Payload => { return { id: event.id, type: definition.type, - durable: { aggregateID: event.aggregateID, seq: event.seq, version: definition.durable.version }, + durable: envelope(event.aggregateID, event.seq, definition.durable.version), data: Schema.decodeUnknownSync(definition.data)(event.data), } } -export const readAggregate = Effect.fn("EventV2.readAggregate")(function* ( - db: Database.Interface["db"], - input: { - readonly aggregateID: string - readonly after?: number - readonly limit: number - readonly manifest: { - readonly definitions: ReadonlyMap - readonly schema: Schema.Decoder - } - }, -) { - const after = input.after ?? -1 - const rows = yield* db - .select() - .from(EventTable) - .where( - and( - eq(EventTable.aggregate_id, input.aggregateID), - gt(EventTable.seq, after), - inArray(EventTable.type, Array.from(input.manifest.definitions.keys())), - ), - ) - .orderBy(asc(EventTable.seq)) - .limit(input.limit + 1) - .all() - .pipe(Effect.orDie) - const page = rows.slice(0, input.limit) - const decode = Schema.decodeUnknownSync(input.manifest.schema) - const events = page.map((event) => - decode({ - id: event.id, - type: input.manifest.definitions.get(event.type)?.type ?? event.type, - durable: { - aggregateID: event.aggregate_id, - seq: event.seq, - version: input.manifest.definitions.get(event.type)?.durable?.version, - }, - data: event.data, - }), - ) - return { - events, - hasMore: rows.length > input.limit, - } -}) - export class SubscriberOverflowError extends Schema.TaggedErrorClass()( "EventV2.SubscriberOverflow", { capacity: Schema.Int }, @@ -139,6 +103,11 @@ 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 + +export const isCaughtUp = (item: LogItem): item is EventLog.CaughtUp => !("id" in item) + export interface Interface { readonly publish: ( definition: D, @@ -146,8 +115,31 @@ export interface Interface { options?: PublishOptions, ) => Effect.Effect> readonly subscribe: (definition: D) => Stream.Stream> - readonly all: () => Stream.Stream - readonly durable: (input: { readonly aggregateID: string; readonly after?: number }) => Stream.Stream + /** + * Volatile live channel: every event published from now on, nothing before, + * nothing across a disconnect. The only channel that carries non-durable + * events; consumers that need reliability combine `changes` with `log`. + */ + readonly live: () => Stream.Stream + /** + * 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. + */ + readonly log: (input: { + readonly aggregateID: string + 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. */ readonly listen: (listener: Subscriber) => Effect.Effect readonly project: (definition: D, projector: Subscriber) => Effect.Effect @@ -165,7 +157,7 @@ export interface Interface { export class Service extends Context.Service()("@opencode/Event") {} -export const allBounded = (events: Interface, capacity: number) => +export const liveBounded = (events: Interface, capacity: number) => Effect.gen(function* () { const queue = yield* Queue.dropping(capacity) const unsubscribe = yield* events.listen((event) => @@ -181,6 +173,11 @@ export const allBounded = (events: Interface, capacity: number) => 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 } export const layerWith = (options?: LayerOptions) => @@ -188,13 +185,19 @@ export const layerWith = (options?: LayerOptions) => Service, Effect.gen(function* () { const pubsub = { - all: yield* PubSub.unbounded(), + live: yield* PubSub.unbounded(), durable: new Map>>(), typed: new Map>(), } 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 getOrCreate = (definition: Definition) => @@ -208,13 +211,16 @@ export const layerWith = (options?: LayerOptions) => yield* Effect.addFinalizer(() => Effect.gen(function* () { - yield* PubSub.shutdown(pubsub.all) + yield* PubSub.shutdown(pubsub.live) yield* Effect.forEach( pubsub.durable.values(), (pubsubs) => Effect.forEach(pubsubs, PubSub.shutdown, { discard: true }), { discard: true }, ) yield* Effect.forEach(pubsub.typed.values(), PubSub.shutdown, { discard: true }) + yield* Effect.forEach(changesSubscribers, (subscriber) => PubSub.shutdown(subscriber.wake), { + discard: true, + }) }), ) @@ -373,6 +379,27 @@ 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 }), @@ -396,11 +423,7 @@ export const layerWith = (options?: LayerOptions) => if (committed) { event = { ...event, - durable: { - aggregateID: committed.aggregateID, - seq: committed.seq, - version: definition.durable.version, - }, + durable: envelope(committed.aggregateID, committed.seq, definition.durable.version), } yield* notify(event as Payload, true) return event @@ -428,7 +451,7 @@ export const layerWith = (options?: LayerOptions) => ) const typed = pubsub.typed.get(event.type) if (typed) yield* PubSub.publish(typed, event) - yield* PubSub.publish(pubsub.all, event) + yield* PubSub.publish(pubsub.live, event) }) } @@ -480,11 +503,7 @@ export const layerWith = (options?: LayerOptions) => yield* notify( { ...payload, - durable: { - aggregateID: committed.aggregateID, - seq: committed.seq, - version: definition.durable.version, - }, + durable: envelope(committed.aggregateID, committed.seq, definition.durable.version), }, true, ) @@ -552,7 +571,7 @@ export const layerWith = (options?: LayerOptions) => Stream.map((event) => event as Payload), ) - const streamAll = (): Stream.Stream => Stream.fromPubSub(pubsub.all) + const streamLive = (): Stream.Stream => Stream.fromPubSub(pubsub.live) const readAfter = (aggregateID: string, after: number) => (options?.beforeAggregateRead?.(aggregateID) ?? Effect.void).pipe( @@ -565,17 +584,24 @@ export const layerWith = (options?: LayerOptions) => .all(), ), Effect.orDie, - Effect.map((rows) => - rows.map((event) => - decodeSerializedEvent({ - id: event.id, - aggregateID: event.aggregate_id, - seq: event.seq, - type: event.type, - data: event.data, - }), - ), - ), + // Skip types missing from the durable manifest instead of failing the + // read: the aggregate may hold events this process cannot decode. The + // raw tail seq keeps cursors advancing across the resulting gaps. + Effect.map((rows) => ({ + seq: rows.at(-1)?.seq, + events: rows.flatMap((event) => { + if (!Durable.get(event.type)?.durable) return [] + return [ + decodeSerializedEvent({ + id: event.id, + aggregateID: event.aggregate_id, + seq: event.seq, + type: event.type, + data: event.data, + }), + ] + }), + })), ) const subscribeDurable = (aggregateID: string) => @@ -598,27 +624,95 @@ export const layerWith = (options?: LayerOptions) => return subscription }) - const durable = (input: { readonly aggregateID: string; readonly after?: number }): Stream.Stream => + const log = (input: { + readonly aggregateID: string + readonly after?: number + readonly follow?: boolean + }): Stream.Stream => Stream.unwrap( Effect.gen(function* () { - const wakes = yield* subscribeDurable(input.aggregateID) let sequence = input.after ?? -1 const read = Effect.suspend(() => readAfter(input.aggregateID, sequence)).pipe( - Effect.tap((events) => + Effect.tap((page) => Effect.sync(() => { - sequence = events.at(-1)?.durable?.seq ?? sequence + sequence = page.seq ?? sequence }), ), + Effect.map((page) => page.events), ) + // 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", + aggregateID: input.aggregateID, + ...(sequence >= 0 ? { seq: Seq.make(sequence) } : {}), + } + const replay = Stream.fromIterable(historical).pipe(Stream.concat(Stream.make(marker))) + if (!wakes) return replay const live = Stream.fromSubscription(wakes).pipe( Stream.mapEffect(() => read), Stream.flattenIterable, ) - return Stream.concat(Stream.fromIterable(historical), live) + return Stream.concat(replay, live) }), ) + 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 + .select({ aggregateID: EventSequenceTable.aggregate_id, seq: EventSequenceTable.seq }) + .from(EventSequenceTable) + .where(inArray(EventSequenceTable.aggregate_id, Array.from(aggregateIDs))) + .all() + .pipe( + Effect.orDie, + Effect.map((rows) => new Map(rows.map((row) => [row.aggregateID, Seq.make(row.seq)]))), + ) + } + const listen = (listener: Subscriber): Effect.Effect => Effect.sync(() => { listeners.push(listener) @@ -638,8 +732,10 @@ export const layerWith = (options?: LayerOptions) => return Service.of({ publish, subscribe, - all: streamAll, - durable, + live: streamLive, + log, + changes, + sequences, listen, project, replay, diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index 797546259f1..35a86b530ae 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -37,7 +37,7 @@ import { SessionCompaction } from "./session/compaction" import { SessionRevert } from "./session/revert" import { Revert } from "@opencode-ai/schema/revert" import { FSUtil } from "./fs-util" -import { SessionDurable } from "@opencode-ai/schema/durable-event-manifest" +import type { EventLog } from "@opencode-ai/schema/event-log" import { SkillV2 } from "./skill" import { Job } from "./job" import { CommandV2 } from "./command" @@ -136,7 +136,11 @@ export type Error = | MessageNotFoundError export interface Interface { - readonly list: (input?: ListInput) => Effect.Effect + readonly list: (input?: ListInput) => Effect.Effect<{ + readonly data: SessionSchema.Info[] + /** Per-session durable log watermark, read in the same transaction as the snapshot. Sessions without events are absent. */ + readonly watermarks: ReadonlyMap + }> readonly create: (input: CreateInput) => Effect.Effect readonly fork: (input: ForkInput) => Effect.Effect readonly get: (sessionID: SessionSchema.ID) => Effect.Effect @@ -156,15 +160,20 @@ export interface Interface { readonly context: ( sessionID: SessionSchema.ID, ) => Effect.Effect - readonly events: (input: { + /** + * 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. + * The marker's seq may exceed the last emitted event because non-public + * durable events share the aggregate's sequence space. + */ + readonly log: (input: { sessionID: SessionSchema.ID after?: number - }) => Stream.Stream - readonly history: (input: { - sessionID: SessionSchema.ID - after?: number - limit: number - }) => Effect.Effect<{ events: ReadonlyArray; hasMore: boolean }, NotFoundError> + follow?: boolean + }) => 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 readonly switchModel: (input: { sessionID: SessionSchema.ID @@ -189,7 +198,10 @@ export interface Interface { agents?: PromptInput.Prompt["agents"] delivery?: SessionInput.Delivery resume?: boolean - }) => Effect.Effect + }) => Effect.Effect< + SessionInput.Admitted, + NotFoundError | PromptConflictError | CommandV2.NotFoundError | CommandV2.EvaluationError + > readonly shell: (input: { id?: EventV2.ID sessionID: SessionSchema.ID @@ -373,10 +385,21 @@ const layer = Layer.effect( order === "asc" ? asc(sortColumn) : desc(sortColumn), order === "asc" ? asc(SessionTable.id) : desc(SessionTable.id), ) - const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe( - Effect.orDie, - ) - return (direction === "previous" ? rows.toReversed() : rows).map((row) => fromRow(row)) + // Watermarks must pair with the snapshot exactly, so both reads share a transaction: + // a higher watermark would let an attached tail skip events missing from the snapshot. + const snapshot = yield* db + .transaction(() => + Effect.gen(function* () { + const rows = yield* (input.limit === undefined ? query.all() : query.limit(input.limit).all()).pipe( + Effect.orDie, + ) + const watermarks = yield* events.sequences(rows.map((row) => row.id)) + return { rows, watermarks } + }), + ) + .pipe(Effect.orDie) + const rows = direction === "previous" ? snapshot.rows.toReversed() : snapshot.rows + return { data: rows.map((row) => fromRow(row)), watermarks: snapshot.watermarks } }), messages: Effect.fn("V2Session.messages")(function* (input) { yield* result.get(input.sessionID) @@ -420,19 +443,19 @@ const layer = Layer.effect( yield* result.get(sessionID) return yield* store.context(sessionID) }), - events: (input) => + log: (input) => Stream.unwrap( result .get(input.sessionID) - .pipe(Effect.as(events.durable({ aggregateID: input.sessionID, after: input.after }))), - ).pipe(Stream.filter((event): event is SessionEvent.DurableEvent => isDurableSessionEvent(event))), - history: Effect.fn("V2Session.history")(function* (input) { - yield* result.get(input.sessionID) - return yield* EventV2.readAggregate(db, { - ...input, - aggregateID: input.sessionID, - manifest: SessionDurable, - }) + .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), + ), + ), + watermarks: Effect.fn("V2Session.watermarks")(function* (sessionIDs) { + return yield* events.sequences(sessionIDs) }), prompt: Effect.fn("V2Session.prompt")((input) => Effect.uninterruptible( diff --git a/packages/core/test/event.test.ts b/packages/core/test/event.test.ts index e4329a2dde9..cd2f8cc98d1 100644 --- a/packages/core/test/event.test.ts +++ b/packages/core/test/event.test.ts @@ -78,6 +78,12 @@ const durableData = (sessionID: Session.ID, text: string) => ({ messageID: SessionV1.MessageID.ascending(`msg_${text}`), }) +/** 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))) + const it = testEffect( AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, Location.node]), [[Location.node, locationLayer]]), ) @@ -119,7 +125,7 @@ describe("EventV2", () => { const event = yield* events.publish(VersionedMessage, { id: "one", text: "hello" }) expect(event.type).toBe("test.versioned") - expect(event.durable?.version).toBe(2) + expect(event.durable?.version).toBe(EventV2.Version.make(2)) }), ) @@ -145,7 +151,7 @@ describe("EventV2", () => { Effect.gen(function* () { const events = yield* EventV2.Service const typed = yield* events.subscribe(Message).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) - const wildcard = yield* events.all().pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + const wildcard = yield* events.live().pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) yield* Effect.yieldNow const event = yield* events.publish(Message, { text: "hello" }) @@ -226,7 +232,7 @@ describe("EventV2", () => { Effect.gen(function* () { const events = yield* EventV2.Service const received = new Array() - const fiber = yield* events.all().pipe( + const fiber = yield* events.live().pipe( Stream.take(1), Stream.runForEach(() => Effect.sync(() => received.push("stream"))), Effect.forkScoped, @@ -325,8 +331,8 @@ describe("EventV2", () => { const events = yield* EventV2.Service const consuming = yield* Deferred.make() const release = yield* Deferred.make() - const slowStream = yield* EventV2.allBounded(events, 1) - const fastStream = yield* EventV2.allBounded(events, 8) + const slowStream = yield* EventV2.liveBounded(events, 1) + const fastStream = yield* EventV2.liveBounded(events, 8) const slow = yield* slowStream.pipe( Stream.runForEach(() => Deferred.succeed(consuming, undefined).pipe(Effect.andThen(Deferred.await(release)))), Effect.forkScoped, @@ -425,9 +431,11 @@ describe("EventV2", () => { const aggregateID = Session.ID.create() yield* events.publish(DurableMessage, durableData(aggregateID, "zero")) yield* events.publish(DurableMessage, durableData(aggregateID, "one")) - const fiber = yield* events - .durable({ aggregateID, after: 0 }) - .pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped) + const fiber = yield* tail(events, { aggregateID, after: 0 }).pipe( + Stream.take(2), + Stream.runCollect, + Effect.forkScoped, + ) yield* Effect.yieldNow yield* events.publish(DurableMessage, durableData(aggregateID, "two")) @@ -444,7 +452,7 @@ describe("EventV2", () => { const events = yield* EventV2.Service const aggregateID = Session.ID.create() yield* events.publish(DurableMessage, durableData(aggregateID, "zero")) - const fiber = yield* events.durable({ aggregateID }).pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped) + const fiber = yield* tail(events, { aggregateID }).pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped) yield* events.publish(DurableMessage, durableData(aggregateID, "one")) @@ -470,7 +478,7 @@ describe("EventV2", () => { yield* Effect.gen(function* () { const events = yield* EventV2.Service const aggregateID = Session.ID.create() - const fiber = yield* events.durable({ aggregateID }).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + const fiber = yield* tail(events, { aggregateID }).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) yield* Deferred.await(readStarted) pause = false @@ -489,9 +497,7 @@ describe("EventV2", () => { const events = yield* EventV2.Service const aggregateID = Session.ID.create() const count = 64 - const fiber = yield* events - .durable({ aggregateID }) - .pipe(Stream.take(count), Stream.runCollect, Effect.forkScoped) + const fiber = yield* tail(events, { aggregateID }).pipe(Stream.take(count), Stream.runCollect, Effect.forkScoped) yield* Effect.yieldNow for (let index = 0; index < count; index++) { @@ -508,7 +514,7 @@ describe("EventV2", () => { Effect.gen(function* () { const events = yield* EventV2.Service const aggregateID = Session.ID.create() - const fiber = yield* events.durable({ aggregateID }).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) + const fiber = yield* tail(events, { aggregateID }).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped) yield* Effect.yieldNow yield* events.publish(Message, { text: "live only" }) @@ -1121,4 +1127,125 @@ describe("EventV2", () => { expect(received[0]?.data).toEqual(durableData(aggregateID, "replayed")) }), ) + + it.effect("log without follow replays events and completes with a caught-up marker", () => + 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")) + + const items = Array.from(yield* Stream.runCollect(events.log({ aggregateID }))) + + expect(items.map((item) => (EventV2.isCaughtUp(item) ? item.type : item.durable?.seq))).toEqual([ + EventV2.Seq.make(0), + EventV2.Seq.make(1), + "log.caught_up", + ]) + expect(items.at(-1)).toEqual({ type: "log.caught_up", aggregateID, seq: EventV2.Seq.make(1) }) + }), + ) + + it.effect("log caught-up 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() + + const empty = Array.from(yield* Stream.runCollect(events.log({ aggregateID }))) + 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[0]).not.toHaveProperty("seq") + expect(drained).toEqual([{ type: "log.caught_up", aggregateID, seq: EventV2.Seq.make(0) }]) + }), + ) + + it.effect("log with follow emits the caught-up marker at the replay-to-live boundary", () => + 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* Effect.yieldNow + + 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([ + EventV2.Seq.make(0), + { type: "log.caught_up", aggregateID, seq: EventV2.Seq.make(0) }, + EventV2.Seq.make(1), + ]) + }), + ) + + 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 + const first = Session.ID.create() + const second = Session.ID.create() + yield* events.publish(DurableMessage, durableData(first, "zero")) + yield* events.publish(DurableMessage, durableData(first, "one")) + yield* events.publish(DurableMessage, durableData(second, "zero")) + + const sequences = yield* events.sequences([first, second, Session.ID.create()]) + + expect(sequences).toEqual( + new Map([ + [first, EventV2.Seq.make(1)], + [second, EventV2.Seq.make(0)], + ]), + ) + expect(yield* events.sequences([])).toEqual(new Map()) + }), + ) }) diff --git a/packages/core/test/session-create.test.ts b/packages/core/test/session-create.test.ts index 5554f9782a2..27d8c6fe04c 100644 --- a/packages/core/test/session-create.test.ts +++ b/packages/core/test/session-create.test.ts @@ -49,6 +49,12 @@ 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. */ +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))) + const assertCreateInputTypes = (session: SessionV2.Interface) => { // @ts-expect-error location or parentID is required. session.create({}) @@ -66,7 +72,7 @@ describe("SessionV2.create", () => { const second = yield* session.create({ location }) expect(second.id).not.toBe(first.id) - expect(yield* session.list()).toHaveLength(2) + expect((yield* session.list()).data).toHaveLength(2) }), ) @@ -79,7 +85,7 @@ describe("SessionV2.create", () => { const retried = yield* session.create(input) expect(retried).toEqual(first) - expect(yield* session.list()).toEqual([first]) + expect((yield* session.list()).data).toEqual([first]) }), ) @@ -146,7 +152,7 @@ describe("SessionV2.create", () => { const forked = yield* session.fork({ sessionID: parent.id }) const parentContext = yield* session.context(parent.id) const forkContext = yield* session.context(forked.id) - const history = yield* session.history({ sessionID: forked.id, limit: 10 }) + const history = Array.from(yield* Stream.runCollect(logEvents(session, forked.id))) expect(forked).toMatchObject({ parentID: parent.id, title: "Parent (fork #1)" }) expect(forkContext).toMatchObject([ @@ -154,8 +160,8 @@ describe("SessionV2.create", () => { { type: "synthetic", text: "parent note", sessionID: forked.id }, ]) expect(forkContext.map((message) => message.id)).not.toEqual(parentContext.map((message) => message.id)) - expect(history.events).toHaveLength(1) - expect(history.events[0]).toMatchObject({ + expect(history).toHaveLength(1) + expect(history[0]).toMatchObject({ type: "session.next.forked", durable: { seq: 0 }, data: { sessionID: forked.id, parentID: parent.id }, @@ -175,7 +181,9 @@ describe("SessionV2.create", () => { expect((yield* session.context(forked.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"]) expect((yield* session.context(forked.id)).at(-1)).toMatchObject({ text: "Child continues" }) expect( - (yield* session.history({ sessionID: forked.id, limit: 10 })).events.map((event) => event.durable?.seq), + Array.from(yield* Stream.runCollect(logEvents(session, forked.id))).map( + (event): number | undefined => event.durable?.seq, + ), ).toEqual([0, 4, 5]) expect(yield* SessionInput.find(db, admitted.id)).toMatchObject({ sessionID: parent.id }) }), @@ -203,10 +211,10 @@ describe("SessionV2.create", () => { const forked = yield* session.fork({ sessionID: parent.id, messageID: second.id }) const context = yield* session.context(forked.id) - const history = yield* session.history({ sessionID: forked.id, limit: 10 }) + const history = Array.from(yield* Stream.runCollect(logEvents(session, forked.id))) expect(context).toMatchObject([{ text: "First" }]) expect(context[0]?.id).not.toBe(first.id) - expect(history.events[0]).toMatchObject({ data: { messageID: second.id } }) + expect(history[0]).toMatchObject({ data: { messageID: second.id } }) }), ) @@ -227,7 +235,7 @@ describe("SessionV2.create", () => { for (const input of changed) { expect(yield* session.create(input)).toEqual(created) } - expect(yield* session.list()).toHaveLength(1) + expect((yield* session.list()).data).toHaveLength(1) }), ) @@ -239,7 +247,7 @@ describe("SessionV2.create", () => { const created = yield* Effect.all([session.create(input), session.create(input)], { concurrency: "unbounded" }) expect(created[1]).toEqual(created[0]) - expect(yield* session.list()).toEqual([created[0]]) + expect((yield* session.list()).data).toEqual([created[0]]) }), ) @@ -317,7 +325,7 @@ describe("SessionV2.create", () => { yield* SessionInput.promoteSteers(db, events, created.id, Number.MAX_SAFE_INTEGER) expect( - Array.from(yield* session.events({ sessionID: created.id }).pipe(Stream.take(2), Stream.runCollect)), + Array.from(yield* logEvents(session, created.id, true).pipe(Stream.take(2), Stream.runCollect)), ).toMatchObject([ { durable: { seq: 1 }, type: "session.next.prompt.admitted", data: { prompt: { text: "Hello" } } }, { durable: { seq: 2 }, type: "session.next.prompted" }, @@ -447,7 +455,7 @@ describe("SessionV2.create", () => { expect(yield* session.get(created.id)).toMatchObject({ agent: "plan" }) expect( - Array.from(yield* session.events({ sessionID: created.id }).pipe(Stream.take(1), Stream.runCollect)), + Array.from(yield* logEvents(session, created.id, true).pipe(Stream.take(1), Stream.runCollect)), ).toMatchObject([{ type: "session.next.agent.switched", data: { agent: "plan" } }]) }), ) @@ -480,7 +488,7 @@ describe("SessionV2.create", () => { expect(yield* session.get(created.id)).toMatchObject({ model }) expect( - Array.from(yield* session.events({ sessionID: created.id }).pipe(Stream.take(1), Stream.runCollect)), + Array.from(yield* logEvents(session, created.id, true).pipe(Stream.take(1), Stream.runCollect)), ).toMatchObject([{ type: "session.next.model.switched", data: { model } }]) }), ) diff --git a/packages/core/test/session-history.test.ts b/packages/core/test/session-history.test.ts deleted file mode 100644 index 560982c8756..00000000000 --- a/packages/core/test/session-history.test.ts +++ /dev/null @@ -1,166 +0,0 @@ -import { describe, expect } from "bun:test" -import { Effect, Layer, Schema } from "effect" -import { Database } from "@opencode-ai/core/database/database" -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 { Job } from "@opencode-ai/core/job" -import { Location } from "@opencode-ai/core/location" -import { ProjectV2 } from "@opencode-ai/core/project" -import { ProjectTable } from "@opencode-ai/core/project/sql" -import { AbsolutePath } from "@opencode-ai/core/schema" -import { SessionV2 } from "@opencode-ai/core/session" -import { SessionExecution } from "@opencode-ai/core/session/execution" -import { SessionProjector } from "@opencode-ai/core/session/projector" -import { SessionStore } from "@opencode-ai/core/session/store" -import { SessionTable } from "@opencode-ai/core/session/sql" -import { testEffect } from "./lib/effect" - -const projects = Layer.succeed( - ProjectV2.Service, - ProjectV2.Service.of({ - resolve: (directory) => Effect.succeed({ id: ProjectV2.ID.global, directory }), - directories: () => Effect.succeed([]), - commit: () => Effect.void, - }), -) -const it = testEffect( - AppNodeBuilder.build( - LayerNode.group([Database.node, EventV2.node, SessionProjector.node, SessionStore.node, SessionV2.node]), - [ - [ProjectV2.node, projects], - [SessionExecution.node, SessionExecution.noopLayer], - ], - ), -) -const location = Location.Ref.make({ directory: AbsolutePath.make("/project") }) - -const GapEvent = EventV2.define({ - type: "test.session.history.gap", - durable: { aggregate: "sessionID", version: 1 }, - schema: { sessionID: SessionV2.ID, value: Schema.String }, -}) - -describe("SessionV2.history", () => { - it.effect("returns an exhausted page for a migrated Session with no event sequence", () => - Effect.gen(function* () { - const db = (yield* Database.Service).db - const session = yield* SessionV2.Service - const sessionID = SessionV2.ID.make("ses_empty_history") - yield* db - .insert(ProjectTable) - .values({ id: ProjectV2.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) - .onConflictDoNothing() - .run() - yield* db - .insert(SessionTable) - .values({ - id: sessionID, - project_id: ProjectV2.ID.global, - slug: "empty-history", - directory: "/project", - title: "Empty history", - version: "test", - }) - .run() - - const first = yield* session.history({ sessionID, limit: 10 }) - - expect(first).toEqual({ events: [], hasMore: false }) - }), - ) - - it.effect("treats after as an exclusive aggregate sequence", () => - Effect.gen(function* () { - const session = yield* SessionV2.Service - const created = yield* session.create({ location }) - yield* session.switchAgent({ sessionID: created.id, agent: "one" }) - yield* session.switchAgent({ sessionID: created.id, agent: "two" }) - - const page = yield* session.history({ sessionID: created.id, after: 1, limit: 10 }) - - expect(page.events.map((event) => event.durable?.seq)).toEqual([2]) - expect(page.hasMore).toBe(false) - }), - ) - - it.effect("paginates public events in aggregate order across filtered gaps without duplicates", () => - Effect.gen(function* () { - const session = yield* SessionV2.Service - const events = yield* EventV2.Service - const created = yield* session.create({ location }) - yield* session.switchAgent({ sessionID: created.id, agent: "one" }) - yield* events.publish(GapEvent, { sessionID: created.id, value: "filtered" }) - yield* session.switchAgent({ sessionID: created.id, agent: "two" }) - yield* session.switchAgent({ sessionID: created.id, agent: "three" }) - - const first = yield* session.history({ sessionID: created.id, limit: 2 }) - const after = first.events.at(-1)?.durable?.seq - const second = yield* session.history({ - sessionID: created.id, - after, - limit: 2, - }) - const sequence = [...first.events, ...second.events].map((event) => event.durable?.seq) - - expect(first.hasMore).toBe(true) - expect(second.hasMore).toBe(false) - expect(sequence).toEqual([1, 3, 4]) - expect(new Set(sequence).size).toBe(sequence.length) - }), - ) - - it.effect("includes events committed between pages", () => - Effect.gen(function* () { - const session = yield* SessionV2.Service - const created = yield* session.create({ location }) - yield* session.switchAgent({ sessionID: created.id, agent: "one" }) - yield* session.switchAgent({ sessionID: created.id, agent: "two" }) - - const first = yield* session.history({ sessionID: created.id, limit: 1 }) - yield* session.switchAgent({ sessionID: created.id, agent: "later" }) - const second = yield* session.history({ - sessionID: created.id, - after: first.events.at(-1)?.durable?.seq, - limit: 10, - }) - - expect(first.hasMore).toBe(true) - expect([...first.events, ...second.events].map((event) => event.durable?.seq)).toEqual([1, 2, 3]) - expect(second.hasMore).toBe(false) - }), - ) - - it.effect("reports exhaustion for exact-limit and limit-plus-one pages", () => - Effect.gen(function* () { - const session = yield* SessionV2.Service - const created = yield* session.create({ location }) - yield* session.switchAgent({ sessionID: created.id, agent: "one" }) - yield* session.switchAgent({ sessionID: created.id, agent: "two" }) - - const exact = yield* session.history({ sessionID: created.id, limit: 2 }) - const oneMore = yield* session.history({ sessionID: created.id, limit: 1 }) - const exhausted = yield* session.history({ - sessionID: created.id, - after: oneMore.events.at(-1)?.durable?.seq, - limit: 1, - }) - - expect(exact.events).toHaveLength(2) - expect(exact.hasMore).toBe(false) - expect(oneMore.events).toHaveLength(1) - expect(oneMore.hasMore).toBe(true) - expect(exhausted.events).toHaveLength(1) - expect(exhausted.hasMore).toBe(false) - }), - ) - - it.effect("fails with NotFoundError for a missing Session", () => - Effect.gen(function* () { - const session = yield* SessionV2.Service - const error = yield* session.history({ sessionID: SessionV2.ID.make("ses_missing"), limit: 10 }).pipe(Effect.flip) - - expect(error._tag).toBe("Session.NotFoundError") - }), - ) -}) diff --git a/packages/core/test/session-log.test.ts b/packages/core/test/session-log.test.ts new file mode 100644 index 00000000000..16f1e97ff1e --- /dev/null +++ b/packages/core/test/session-log.test.ts @@ -0,0 +1,161 @@ +import { describe, expect } from "bun:test" +import { Effect, Fiber, Layer, Schema, Stream } from "effect" +import { Database } from "@opencode-ai/core/database/database" +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 { Location } from "@opencode-ai/core/location" +import { ProjectV2 } from "@opencode-ai/core/project" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { SessionV2 } from "@opencode-ai/core/session" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionExecution } from "@opencode-ai/core/session/execution" +import { SessionStore } from "@opencode-ai/core/session/store" +import { SessionTable } from "@opencode-ai/core/session/sql" +import { testEffect } from "./lib/effect" + +const projects = Layer.succeed( + ProjectV2.Service, + ProjectV2.Service.of({ + resolve: (directory) => Effect.succeed({ id: ProjectV2.ID.global, directory }), + directories: () => Effect.succeed([]), + commit: () => Effect.void, + }), +) +const it = testEffect( + AppNodeBuilder.build( + LayerNode.group([Database.node, EventV2.node, SessionProjector.node, SessionStore.node, SessionV2.node]), + [ + [ProjectV2.node, projects], + [SessionExecution.node, SessionExecution.noopLayer], + ], + ), +) +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", () => + Effect.gen(function* () { + const session = yield* SessionV2.Service + const events = yield* EventV2.Service + const created = yield* session.create({ location }) + yield* session.rename({ sessionID: created.id, title: "renamed" }) + + const items = Array.from(yield* Stream.runCollect(session.log({ sessionID: created.id }))) + const watermark = (yield* events.sequences([created.id])).get(created.id) + + // 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 }) + }), + ) + + it.effect("continues with live public events when following", () => + Effect.gen(function* () { + const session = yield* SessionV2.Service + const created = yield* session.create({ location }) + const fiber = yield* session + .log({ sessionID: created.id, follow: true }) + .pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped) + yield* Effect.yieldNow + + 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"]) + }), + ) + + it.effect("fails with NotFound for an unknown session", () => + Effect.gen(function* () { + const session = yield* SessionV2.Service + const error = yield* Effect.flip(Stream.runCollect(session.log({ sessionID: SessionV2.ID.create() }))) + expect(error._tag).toBe("Session.NotFoundError") + }), + ) + + it.effect("reads across undecodable gaps in aggregate order and marks the true log position", () => + Effect.gen(function* () { + const GapEvent = EventV2.define({ + type: "test.session.log.gap", + durable: { aggregate: "sessionID", version: 1 }, + schema: { sessionID: SessionV2.ID, value: Schema.String }, + }) + const session = yield* SessionV2.Service + const events = yield* EventV2.Service + const created = yield* session.create({ location }) + yield* session.switchAgent({ sessionID: created.id, agent: "one" }) + // Not in the durable manifest, so reads must skip it without failing. + yield* events.publish(GapEvent, { sessionID: created.id, value: "filtered" }) + yield* session.switchAgent({ sessionID: created.id, agent: "two" }) + yield* session.switchAgent({ sessionID: created.id, agent: "three" }) + + 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) }) + }), + ) + + it.effect("completes with a bare caught-up marker for a migrated Session with no event sequence", () => + Effect.gen(function* () { + const db = (yield* Database.Service).db + const session = yield* SessionV2.Service + const sessionID = SessionV2.ID.make("ses_empty_log") + yield* db + .insert(ProjectTable) + .values({ id: ProjectV2.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .onConflictDoNothing() + .run() + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: ProjectV2.ID.global, + slug: "empty-log", + directory: "/project", + title: "Empty log", + version: "test", + }) + .run() + + const items = Array.from(yield* Stream.runCollect(session.log({ sessionID }))) + + expect(items).toEqual([{ type: "log.caught_up", aggregateID: sessionID }]) + }), + ) +}) + +describe("SessionV2 watermarks", () => { + it.effect("list pairs each session snapshot with its durable log watermark", () => + Effect.gen(function* () { + const session = yield* SessionV2.Service + const events = yield* EventV2.Service + const first = yield* session.create({ location }) + const second = yield* session.create({ location }) + yield* session.rename({ sessionID: first.id, title: "renamed" }) + + const page = yield* session.list() + const sequences = yield* events.sequences([first.id, second.id]) + + expect(page.data.map((info) => info.id).toSorted()).toEqual([first.id, second.id].toSorted()) + expect(page.watermarks).toEqual(sequences) + expect(page.watermarks.get(first.id)).toBeGreaterThan(page.watermarks.get(second.id)!) + }), + ) + + it.effect("watermarks omits sessions without durable events", () => + Effect.gen(function* () { + const session = yield* SessionV2.Service + const created = yield* session.create({ location }) + + const watermarks = yield* session.watermarks([created.id, SessionV2.ID.create()]) + + expect(Array.from(watermarks.keys())).toEqual([created.id]) + }), + ) +}) diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index 8eb8fb4a279..9445424f0b4 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/packages/core/test/session-prompt.test.ts @@ -245,7 +245,11 @@ describe("SessionV2.prompt", () => { const session = yield* SessionV2.Service const events = yield* EventV2.Service const { db } = yield* Database.Service - const fiber = yield* session.events({ sessionID }).pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped) + const publicEvents = (input: { sessionID: SessionV2.ID; after?: number }) => + session + .log({ ...input, follow: true }) + .pipe(Stream.filter((item): item is SessionEvent.DurableEvent => !EventV2.isCaughtUp(item))) + const fiber = yield* publicEvents({ sessionID }).pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped) yield* Effect.yieldNow yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false }) @@ -253,7 +257,7 @@ describe("SessionV2.prompt", () => { yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER) const streamed = Array.from(yield* Fiber.join(fiber)) - expect(streamed.map((event) => [event.durable?.seq, event.type])).toEqual([ + expect(streamed.map((event): [number | undefined, string] => [event.durable?.seq, event.type])).toEqual([ [0, "session.next.prompt.admitted"], [1, "session.next.prompt.admitted"], [2, "session.next.prompted"], @@ -261,10 +265,8 @@ describe("SessionV2.prompt", () => { ]) expect( Array.from( - yield* session - .events({ sessionID, after: streamed[0]!.durable?.seq }) - .pipe(Stream.take(1), Stream.runCollect), - ).map((event) => [event.durable?.seq, event.type]), + 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/core/test/session-runner-tool-events.test.ts b/packages/core/test/session-runner-tool-events.test.ts index f96ea4dea2b..048766e4396 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -27,8 +27,10 @@ const capture = () => { return event }), subscribe: () => Stream.empty, - all: () => Stream.empty, - durable: () => Stream.empty, + live: () => Stream.empty, + log: () => Stream.empty, + changes: () => Stream.empty, + sequences: () => Effect.succeed(new Map()), listen: () => Effect.succeed(Effect.void), project: () => Effect.void, replay: () => Effect.void, diff --git a/packages/plugin/src/v2/effect/generated/api.ts b/packages/plugin/src/v2/effect/generated/api.ts index 13c5cb80f53..fe24821a1a3 100644 --- a/packages/plugin/src/v2/effect/generated/api.ts +++ b/packages/plugin/src/v2/effect/generated/api.ts @@ -65,7 +65,7 @@ export type Endpoint4_1Input = { export type Endpoint4_1Output = EffectValue>["data"] export type SessionCreateOperation = (input?: Endpoint4_1Input) => Effect.Effect -export type Endpoint4_2Output = EffectValue>["data"] +export type Endpoint4_2Output = EffectValue> export type SessionActiveOperation = () => Effect.Effect type Endpoint4_3Request = Parameters[0] @@ -216,40 +216,32 @@ export type SessionRemoveContextEntryOperation = ( input: Endpoint4_20Input, ) => Effect.Effect -type Endpoint4_21Request = Parameters[0] +type Endpoint4_21Request = Parameters[0] export type Endpoint4_21Input = { readonly sessionID: Endpoint4_21Request["params"]["sessionID"] - readonly limit?: Endpoint4_21Request["query"]["limit"] readonly after?: Endpoint4_21Request["query"]["after"] + readonly follow?: Endpoint4_21Request["query"]["follow"] } -export type Endpoint4_21Output = EffectValue> -export type SessionHistoryOperation = (input: Endpoint4_21Input) => Effect.Effect +export type Endpoint4_21Output = StreamValue>> +export type SessionLogOperation = (input: Endpoint4_21Input) => Stream.Stream -type Endpoint4_22Request = Parameters[0] -export type Endpoint4_22Input = { - readonly sessionID: Endpoint4_22Request["params"]["sessionID"] - readonly after?: Endpoint4_22Request["query"]["after"] -} -export type Endpoint4_22Output = StreamValue>> -export type SessionEventsOperation = (input: Endpoint4_22Input) => Stream.Stream +type Endpoint4_22Request = Parameters[0] +export type Endpoint4_22Input = { readonly sessionID: Endpoint4_22Request["params"]["sessionID"] } +export type Endpoint4_22Output = EffectValue> +export type SessionInterruptOperation = (input: Endpoint4_22Input) => Effect.Effect -type Endpoint4_23Request = Parameters[0] +type Endpoint4_23Request = Parameters[0] export type Endpoint4_23Input = { readonly sessionID: Endpoint4_23Request["params"]["sessionID"] } -export type Endpoint4_23Output = EffectValue> -export type SessionInterruptOperation = (input: Endpoint4_23Input) => Effect.Effect +export type Endpoint4_23Output = EffectValue> +export type SessionBackgroundOperation = (input: Endpoint4_23Input) => Effect.Effect -type Endpoint4_24Request = Parameters[0] -export type Endpoint4_24Input = { readonly sessionID: Endpoint4_24Request["params"]["sessionID"] } -export type Endpoint4_24Output = EffectValue> -export type SessionBackgroundOperation = (input: Endpoint4_24Input) => Effect.Effect - -type Endpoint4_25Request = Parameters[0] -export type Endpoint4_25Input = { - readonly sessionID: Endpoint4_25Request["params"]["sessionID"] - readonly messageID: Endpoint4_25Request["params"]["messageID"] +type Endpoint4_24Request = Parameters[0] +export type Endpoint4_24Input = { + readonly sessionID: Endpoint4_24Request["params"]["sessionID"] + readonly messageID: Endpoint4_24Request["params"]["messageID"] } -export type Endpoint4_25Output = EffectValue>["data"] -export type SessionMessageOperation = (input: Endpoint4_25Input) => Effect.Effect +export type Endpoint4_24Output = EffectValue>["data"] +export type SessionMessageOperation = (input: Endpoint4_24Input) => Effect.Effect export interface SessionApi { readonly list: SessionListOperation @@ -273,8 +265,7 @@ export interface SessionApi { readonly listContextEntries: SessionListContextEntriesOperation readonly putContextEntry: SessionPutContextEntryOperation readonly removeContextEntry: SessionRemoveContextEntryOperation - readonly history: SessionHistoryOperation - readonly events: SessionEventsOperation + readonly log: SessionLogOperation readonly interrupt: SessionInterruptOperation readonly background: SessionBackgroundOperation readonly message: SessionMessageOperation @@ -586,8 +577,12 @@ export interface SkillApi { export type Endpoint17_0Output = StreamValue>> export type EventSubscribeOperation = () => Stream.Stream +export type Endpoint17_1Output = StreamValue>> +export type EventChangesOperation = () => Stream.Stream + export interface EventApi { readonly subscribe: EventSubscribeOperation + readonly changes: EventChangesOperation } type Endpoint18_0Request = Parameters[0] diff --git a/packages/protocol/src/groups/event.ts b/packages/protocol/src/groups/event.ts index a6fc5692eae..f5a0c0c7bbe 100644 --- a/packages/protocol/src/groups/event.ts +++ b/packages/protocol/src/groups/event.ts @@ -1,4 +1,5 @@ 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" @@ -8,7 +9,7 @@ import { HttpApiEndpoint, HttpApiGroup, HttpApiSchema, OpenApi } from "effect/un const fields = { id: Event.ID, metadata: Schema.optional(Schema.Record(Schema.String, Schema.Unknown)), - durable: Schema.optional(Schema.Struct({ aggregateID: Schema.String, seq: Schema.Int, version: Schema.Int })), + durable: Schema.optional(Schema.Struct({ aggregateID: Schema.String, seq: Event.Seq, version: Event.Version })), location: Schema.optional(Location.Ref), } @@ -38,11 +39,24 @@ const make = >(definitions: OpenApi.annotations({ identifier: "v2.event.subscribe", summary: "Subscribe to events", - description: "Subscribe to native event payloads for the server.", + 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.", }), ), ) - .annotateMerge(OpenApi.annotations({ title: "events", description: "Experimental event stream route." })), + .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.", + }), + ), + ) + .annotateMerge(OpenApi.annotations({ title: "events", description: "Experimental event stream routes." })), } } diff --git a/packages/protocol/src/groups/message.ts b/packages/protocol/src/groups/message.ts index 31853810718..e6d9a483778 100644 --- a/packages/protocol/src/groups/message.ts +++ b/packages/protocol/src/groups/message.ts @@ -1,5 +1,7 @@ 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" @@ -28,6 +30,10 @@ 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 d7285a69c9d..4a8e5f2351a 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -4,9 +4,10 @@ import { PromptInput } from "@opencode-ai/schema/prompt-input" import { Session } from "@opencode-ai/schema/session" import { SessionContextEntry } from "@opencode-ai/schema/session-context-entry" import { Project } from "@opencode-ai/schema/project" -import { AbsolutePath, NonNegativeInt, PositiveInt, RelativePath, statics } from "@opencode-ai/schema/schema" +import { AbsolutePath, PositiveInt, RelativePath, statics } from "@opencode-ai/schema/schema" +import { Event } from "@opencode-ai/schema/event" import { Workspace } from "@opencode-ai/schema/workspace" -import { Context, Effect, Encoding, Result, Schema, Struct } from "effect" +import { Context, Effect, Encoding, Result, Schema, SchemaGetter, Struct } from "effect" import { HttpApiEndpoint, HttpApiGroup, HttpApiMiddleware, HttpApiSchema, OpenApi } from "effect/unstable/httpapi" import { ConflictError, @@ -26,6 +27,7 @@ import { Model } from "@opencode-ai/schema/model" import { Location } from "@opencode-ai/schema/location" import { Revert } from "@opencode-ai/schema/revert" import { SessionEvent } from "@opencode-ai/schema/session-event" +import { EventLog } from "@opencode-ai/schema/event-log" const SessionsQueryFields = { workspace: Workspace.ID.pipe(Schema.optional), @@ -89,13 +91,19 @@ const SessionActive = Schema.Struct({ type: Schema.Literal("running"), }).annotate({ identifier: "SessionActive" }) -const SessionHistoryLimit = PositiveInt.check(Schema.isLessThanOrEqualTo(100)) - -export const SessionHistoryQuery = Schema.Struct({ - limit: Schema.NumberFromString.pipe(Schema.decodeTo(SessionHistoryLimit), Schema.optional), - after: Schema.NumberFromString.pipe(Schema.decodeTo(NonNegativeInt), Schema.optional), +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"), + encode: SchemaGetter.transform((value): "true" | "false" => (value ? "true" : "false")), + }), +) + const SessionsQueryCursor = SessionsCursor.annotate({ description: "Opaque pagination cursor returned as cursor.previous or cursor.next in the previous response.", }) @@ -115,6 +123,7 @@ 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), @@ -149,13 +158,13 @@ export const makeSessionGroup = (sessionLo ) .add( HttpApiEndpoint.get("session.active", "/api/session/active", { - success: Schema.Struct({ data: Schema.Record(Session.ID, SessionActive) }), + success: Schema.Struct({ data: Schema.Record(Session.ID, SessionActive), watermarks: SessionWatermarks }), }).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.", + "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.", }), ), ) @@ -282,7 +291,8 @@ export const makeSessionGroup = (sessionLo OpenApi.annotations({ identifier: "v2.session.command", summary: "Run command", - description: "Resolve a slash command into prompt input, admit it durably, and schedule execution unless resume is false.", + description: + "Resolve a slash command into prompt input, admit it durably, and schedule execution unless resume is false.", }), ), ) @@ -455,40 +465,24 @@ export const makeSessionGroup = (sessionLo ), ) .add( - HttpApiEndpoint.get("session.history", "/api/session/:sessionID/history", { - params: { sessionID: Session.ID }, - query: SessionHistoryQuery, - success: Schema.Struct({ - data: Schema.Array(SessionEvent.Durable), - hasMore: Schema.Boolean, - }).annotate({ identifier: "SessionHistory" }), - error: SessionNotFoundError, - }) - .middleware(sessionLocationMiddleware) - .annotateMerge( - OpenApi.annotations({ - identifier: "v2.session.history", - summary: "Get session history", - description: - "Read one finite page of public durable Session events after an exclusive aggregate sequence. Newly committed events may appear on later pages.", - }), - ), - ) - .add( - HttpApiEndpoint.get("session.events", "/api/session/:sessionID/event", { + HttpApiEndpoint.get("session.log", "/api/session/:sessionID/log", { params: { sessionID: Session.ID }, query: { - after: Schema.NumberFromString.pipe(Schema.decodeTo(NonNegativeInt), Schema.optional), + after: Schema.NumberFromString.pipe(Schema.decodeTo(Event.Seq), Schema.optional), + follow: BooleanFromString.pipe(Schema.optional), }, - success: HttpApiSchema.StreamSse({ data: SessionEvent.Durable }), + success: HttpApiSchema.StreamSse({ + data: Schema.Union([SessionEvent.Durable, EventLog.CaughtUp]).annotate({ identifier: "SessionLogItem" }), + }), error: SessionNotFoundError, }) .middleware(sessionLocationMiddleware) .annotateMerge( OpenApi.annotations({ - identifier: "v2.session.events", - summary: "Subscribe to session events", - description: "Replay durable events after an aggregate sequence, then continue with new durable events.", + 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.", }), ), ) diff --git a/packages/protocol/test/session-cursor.test.ts b/packages/protocol/test/session-cursor.test.ts index 2d44a11f5cf..86c78d9503a 100644 --- a/packages/protocol/test/session-cursor.test.ts +++ b/packages/protocol/test/session-cursor.test.ts @@ -1,6 +1,6 @@ import { describe, expect, test } from "bun:test" -import { Effect, Schema } from "effect" -import { SessionHistoryQuery, SessionsCursor } from "../src/groups/session.js" +import { Effect } from "effect" +import { SessionsCursor } from "../src/groups/session.js" import { Session } from "@opencode-ai/schema/session" describe("SessionsCursor", () => { @@ -16,11 +16,3 @@ describe("SessionsCursor", () => { expect(await Effect.runPromise(SessionsCursor.parse(cursor))).toEqual(input) }) }) - -describe("SessionHistoryQuery", () => { - test("decodes numeric paging inputs", async () => { - const query = await Effect.runPromise(Schema.decodeUnknownEffect(SessionHistoryQuery)({ after: "3", limit: "10" })) - - expect(query).toEqual({ after: 3, limit: 10 }) - }) -}) diff --git a/packages/schema/src/event-log.ts b/packages/schema/src/event-log.ts new file mode 100644 index 00000000000..dc4f363ccc6 --- /dev/null +++ b/packages/schema/src/event-log.ts @@ -0,0 +1,56 @@ +export * as EventLog from "./event-log.js" + +import { Schema } from "effect" +import { Event } from "./event.js" +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. + */ +export const CaughtUp = Schema.Struct({ + type: Schema.Literal("log.caught_up"), + aggregateID: Schema.String, + seq: optional(Event.Seq), +}).annotate({ + identifier: "EventLog.CaughtUp", + 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.", +}) +export interface CaughtUp 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 9b1aca354fb..cd2fd6532d0 100644 --- a/packages/schema/src/event.ts +++ b/packages/schema/src/event.ts @@ -12,6 +12,18 @@ export const ID = Schema.String.check(Schema.isStartsWith("evt_")).pipe( ) 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; + * `after` cursors accept only values that came from those sources. + */ +export const Seq = Schema.Int.check(Schema.isGreaterThanOrEqualTo(0)).pipe(Schema.brand("Event.Seq")) +export type Seq = typeof Seq.Type + +/** Durable schema version of one event type, from the event definition that committed it. */ +export const Version = Schema.Int.check(Schema.isGreaterThanOrEqualTo(1)).pipe(Schema.brand("Event.Version")) +export type Version = typeof Version.Type + export type Definition< Type extends string = string, DataSchema extends Schema.Codec = Schema.Codec, @@ -32,8 +44,8 @@ export type Payload = { readonly data: Data readonly durable?: { readonly aggregateID: string - readonly seq: number - readonly version: number + readonly seq: Seq + readonly version: Version } readonly location?: Location.Ref readonly metadata?: Record @@ -55,7 +67,7 @@ export function define< id: ID, metadata: optional(Schema.Record(Schema.String, Schema.Unknown)), type: Schema.Literal(input.type), - durable: optional(Schema.Struct({ aggregateID: Schema.String, seq: Schema.Int, version: Schema.Int })), + durable: optional(Schema.Struct({ aggregateID: Schema.String, seq: Seq, version: Version })), location: optional(Location.Ref), data, }) diff --git a/packages/sdk-next/test/embedded.test.ts b/packages/sdk-next/test/embedded.test.ts index ae8b54d648c..a34bd628a6b 100644 --- a/packages/sdk-next/test/embedded.test.ts +++ b/packages/sdk-next/test/embedded.test.ts @@ -77,16 +77,19 @@ it.live( sessionID: id, prompt: fixture.sdk.Prompt.make({ text: "Promote this input" }), }) - const prompted = yield* opencode.sessions.events({ sessionID: id }).pipe( + const prompted = yield* opencode.sessions.log({ sessionID: id, follow: true }).pipe( Stream.filter((event) => event.type === "session.next.prompted" && event.data.messageID === wake.id), Stream.runHead, Effect.timeout("10 seconds"), Effect.map(Option.getOrThrow), ) const wakeContext = yield* opencode.sessions.context({ sessionID: id }) - const event = yield* opencode.sessions - .events({ sessionID: id }) - .pipe(Stream.take(1), Stream.runHead, Effect.map(Option.getOrUndefined)) + const event = yield* opencode.sessions.log({ sessionID: id }).pipe( + Stream.filter((item) => item.type !== "log.caught_up"), + Stream.take(1), + Stream.runHead, + Effect.map(Option.getOrUndefined), + ) const modelMessage = Option.fromNullishOr(context.find((message) => message.type === "model-switched")).pipe( Option.getOrThrow, ) @@ -96,7 +99,7 @@ it.live( const missingSessionID = fixture.sdk.Session.ID.create() const missing = yield* Effect.all( [ - opencode.sessions.events({ sessionID: missingSessionID }).pipe(Stream.runHead, Effect.flip), + opencode.sessions.log({ sessionID: missingSessionID }).pipe(Stream.runHead, Effect.flip), opencode.sessions.interrupt({ sessionID: missingSessionID }).pipe(Effect.flip), opencode.sessions.message({ sessionID: missingSessionID, messageID: modelMessage.id }).pipe(Effect.flip), opencode.sessions.listContextEntries({ sessionID: missingSessionID }).pipe(Effect.flip), @@ -114,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({}) + expect(active).toEqual({ data: {}, watermarks: {} }) expect(admitted.sessionID).toBe(id) expect(prompted.type).toBe("session.next.prompted") expect(wakeContext).toContainEqual(expect.objectContaining({ id: wake.id, type: "user" })) diff --git a/packages/server/src/handlers/event.ts b/packages/server/src/handlers/event.ts index 47e94731ee0..f412137abaa 100644 --- a/packages/server/src/handlers/event.ts +++ b/packages/server/src/handlers/event.ts @@ -20,33 +20,38 @@ 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.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.allBounded(events, subscriberCapacity) - 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 + .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, subscriberCapacity) + 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 93734c628d8..85de308fca4 100644 --- a/packages/server/src/handlers/message.ts +++ b/packages/server/src/handlers/message.ts @@ -38,6 +38,10 @@ 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, @@ -70,6 +74,7 @@ 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 d53574758ed..9f44b9dce71 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -19,7 +19,6 @@ import { import { AbsolutePath } from "@opencode-ai/core/schema" const DefaultSessionsLimit = 50 -const DefaultSessionHistoryLimit = 50 export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handlers) => Effect.gen(function* () { @@ -35,15 +34,17 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl Effect.mapError(() => new InvalidCursorError({ message: "Invalid cursor" })), ) : ctx.query - const sessions = yield* session.list({ + const page = yield* session.list({ ...query, workspaceID: query.workspace, limit: ctx.query.limit ?? DefaultSessionsLimit, }) + const sessions = page.data const first = sessions[0] const last = sessions.at(-1) return { data: sessions, + watermarks: Object.fromEntries(page.watermarks), cursor: { previous: first ? SessionsCursor.make({ @@ -87,10 +88,11 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl .handle( "session.active", Effect.fn(function* () { + const active = yield* session.active + const watermarks = yield* session.watermarks(Array.from(active)) return { - data: Object.fromEntries( - Array.from(yield* session.active, (sessionID) => [sessionID, { type: "running" as const }]), - ), + data: Object.fromEntries(Array.from(active, (sessionID) => [sessionID, { type: "running" as const }])), + watermarks: Object.fromEntries(watermarks), } }), ) @@ -547,35 +549,12 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl }), ) .handle( - "session.history", - Effect.fn(function* (ctx) { - return yield* session - .history({ - sessionID: ctx.params.sessionID, - after: ctx.query.after, - limit: ctx.query.limit ?? DefaultSessionHistoryLimit, - }) - .pipe( - Effect.map((page) => ({ - data: page.events, - hasMore: page.hasMore, - })), - Effect.catchTag( - "Session.NotFoundError", - (error) => - new SessionNotFoundError({ - sessionID: error.sessionID, - message: `Session not found: ${error.sessionID}`, - }), - ), - ) - }), - ) - .handle( - "session.events", + "session.log", Effect.fn((ctx) => Effect.succeed( - session.events({ sessionID: ctx.params.sessionID, after: ctx.query.after }).pipe(Stream.orDie), + session + .log({ sessionID: ctx.params.sessionID, after: ctx.query.after, follow: ctx.query.follow }) + .pipe(Stream.orDie), ), ), ) diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index 869c18d1fd3..df0ca9ee23b 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -869,7 +869,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ setStore( "session", "status", - Object.fromEntries(Object.keys(active).map((sessionID) => [sessionID, "running" as const])), + Object.fromEntries(Object.keys(active.data).map((sessionID) => [sessionID, "running" as const])), ), ), result.location.refresh(), diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index bdf928cb48c..c6d1756f7be 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -232,7 +232,8 @@ test("connectedOnce is false until first connect and persists across disconnect" 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" } } }) + if (url.pathname === "/api/session/active") + return json({ data: { "session-active": { type: "running" } }, watermarks: {} }) }, events) let data!: ReturnType diff --git a/packages/tui/test/fixture/tui-sdk.ts b/packages/tui/test/fixture/tui-sdk.ts index 52d50bf3acc..472ba4f78f1 100644 --- a/packages/tui/test/fixture/tui-sdk.ts +++ b/packages/tui/test/fixture/tui-sdk.ts @@ -98,7 +98,7 @@ export function createFetch(override?: FetchHandler, events?: ReturnType