From e82aa92e6422a72c89c2ba76efa5db9a1d633d7f Mon Sep 17 00:00:00 2001 From: Shoubhit Dash Date: Wed, 26 Aug 2026 18:31:59 +0530 Subject: [PATCH] fix(session): support assistant message content updates (#45015) --- packages/client/src/effect/api/api.ts | 26 ++ .../client/src/effect/generated/client.ts | 14 + .../client/src/promise/generated/client.ts | 14 + .../client/src/promise/generated/types.ts | 219 +++++++++++++--- packages/client/src/solid/data.ts | 15 ++ packages/client/test/solid-data.test.ts | 71 +++++ packages/core/src/session.ts | 57 ++++ packages/core/src/session/message-updater.ts | 8 +- packages/core/src/session/projector.ts | 1 + .../core/test/session-message-update.test.ts | 246 ++++++++++++++++++ packages/protocol/openapi.json | 137 ++++++++++ packages/protocol/src/groups/session.ts | 14 + packages/schema/src/session-event.ts | 13 + packages/schema/src/session-message.ts | 5 + packages/schema/test/event-manifest.test.ts | 1 + packages/server/src/handlers/session.ts | 31 +++ .../test/session-message-update.test.ts | 145 +++++++++++ 17 files changed, 980 insertions(+), 37 deletions(-) create mode 100644 packages/core/test/session-message-update.test.ts create mode 100644 packages/server/test/session-message-update.test.ts diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index 0d737b6b698..5450a173e0b 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -974,6 +974,19 @@ export type SessionLogOutput = readonly location?: Location.Ref | undefined readonly data: { readonly sessionID: Session.ID; readonly to: SessionMessage.ID } } + | { + readonly id: Event.ID + readonly created: number + readonly metadata?: { readonly [x: string]: unknown } | undefined + readonly type: "session.message.content.updated" + readonly durable: { readonly aggregateID: string; readonly seq: Event.Seq; readonly version: Event.Version } + readonly location?: Location.Ref | undefined + readonly data: { + readonly sessionID: Session.ID + readonly messageID: SessionMessage.ID + readonly content: ReadonlyArray + } + } | { readonly id: Event.ID readonly created: number @@ -1013,6 +1026,18 @@ export type SessionMessageInput = { readonly sessionID: Session.ID; readonly mes export type SessionMessageOutput = SessionMessage.Info export type SessionMessageOperation = (input: SessionMessageInput) => Effect.Effect +export type SessionMessageUpdateInput = { + readonly sessionID: Session.ID + readonly messageID: SessionMessage.ID + readonly content: ReadonlyArray< + SessionMessage.AssistantText | SessionMessage.AssistantReasoning | SessionMessage.AssistantTool + > +} +export type SessionMessageUpdateOutput = SessionMessage.Assistant +export type SessionMessageUpdateOperation = ( + input: SessionMessageUpdateInput, +) => Effect.Effect + export type SessionEnvironmentInput = { readonly sessionID: Session.ID readonly variables: { readonly [x: string]: string } @@ -1071,6 +1096,7 @@ export interface SessionApi { readonly interrupt: SessionInterruptOperation readonly background: SessionBackgroundOperation readonly message: SessionMessageOperation + readonly messageUpdate: SessionMessageUpdateOperation readonly environment: SessionEnvironmentOperation readonly view: SessionViewOperation } diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index d00a3430a89..697d771e6f4 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -86,6 +86,8 @@ import type { SessionBackgroundOutput, SessionMessageInput, SessionMessageOutput, + SessionMessageUpdateInput, + SessionMessageUpdateOutput, SessionEnvironmentInput, SessionEnvironmentOutput, SessionViewInput, @@ -651,6 +653,17 @@ const EndpointSessionMessage = (raw: RawClient["server.session"]) => (input: Ses ), ) +const EndpointSessionMessageUpdate = (raw: RawClient["server.session"]) => (input: SessionMessageUpdateInput) => + preserveEffect()( + raw["session.messageUpdate"]({ + params: { sessionID: input["sessionID"], messageID: input["messageID"] }, + payload: { content: input["content"] }, + }).pipe( + Effect.mapError(mapClientError), + Effect.map((value) => value.data), + ), + ) + const EndpointSessionEnvironment = (raw: RawClient["server.session"]) => (input: SessionEnvironmentInput) => preserveEffect()( raw["session.environment"]({ @@ -711,6 +724,7 @@ const adaptGroupSession = (raw: RawClient["server.session"]) => ({ interrupt: EndpointSessionInterrupt(raw), background: EndpointSessionBackground(raw), message: EndpointSessionMessage(raw), + messageUpdate: EndpointSessionMessageUpdate(raw), environment: EndpointSessionEnvironment(raw), view: EndpointSessionView(raw), }) diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index a1911eb959d..95a0a90d077 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -80,6 +80,8 @@ import type { SessionBackgroundOutput, SessionMessageInput, SessionMessageOutput, + SessionMessageUpdateInput, + SessionMessageUpdateOutput, SessionEnvironmentInput, SessionEnvironmentOutput, SessionViewInput, @@ -929,6 +931,18 @@ export function make(options: ClientOptions) { }, requestOptions, ).then((value) => value.data), + messageUpdate: (input: SessionMessageUpdateInput, requestOptions?: RequestOptions) => + request<{ readonly data: SessionMessageUpdateOutput }>( + { + method: "PATCH", + path: `/api/session/${encodeURIComponent(input.sessionID)}/message/${encodeURIComponent(input.messageID)}`, + body: { content: input["content"] }, + successStatus: 200, + declaredStatuses: [404, 400, 409, 401], + empty: false, + }, + requestOptions, + ).then((value) => value.data), environment: (input: SessionEnvironmentInput, requestOptions?: RequestOptions) => request( { diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index f24623020e0..1ab9c646de7 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -174,6 +174,12 @@ export type SessionMessageProviderState1 = { [x: string]: any } export type ToolFileContent1 = { type: "file"; uri: string; mime: string; name?: string | undefined } +export type SessionMessageToolStateRunning1 = { + status: "running" + input: { [x: string]: any } + metadata: { [x: string]: JsonValue } +} + export type EventLogSynced = { type: "log.synced"; aggregateID: string; seq?: number } export type SessionInterruptResponse = { interrupted: boolean } @@ -1315,6 +1321,15 @@ export type SessionToolCalled = { } } +export type SessionMessageAssistantText1 = { type: "text"; text: string; state?: SessionMessageProviderState1 } + +export type SessionMessageAssistantReasoning1 = { + type: "reasoning" + text: string + state?: SessionMessageProviderState1 + time?: { created: number; completed?: number } +} + export type ToolContent1 = ToolTextContent | ToolFileContent1 export type ModelCompatibility = { @@ -1743,6 +1758,21 @@ export type SessionToolFailed = { } } +export type SessionMessageToolStateCompleted1 = { + status: "completed" + input: { [x: string]: any } + content: [ToolContent1, ...Array] + metadata?: { [x: string]: JsonValue } +} + +export type SessionMessageToolStateError1 = { + status: "error" + input: { [x: string]: any } + error: SessionStructuredError + content?: [ToolContent1, ...Array] + metadata?: { [x: string]: JsonValue } +} + export type ModelInfo = { id: string modelID: string @@ -2009,6 +2039,21 @@ export type SessionMessageAssistantTool = { time: { created: number; ran?: number; completed?: number } } +export type SessionMessageAssistantTool1 = { + type: "tool" + id: string + name: string + executed?: boolean + providerState?: SessionMessageProviderState1 + providerResultState?: SessionMessageProviderState1 + state: + | SessionMessageToolStateStreaming + | SessionMessageToolStateRunning1 + | SessionMessageToolStateCompleted1 + | SessionMessageToolStateError1 + time: { created: number; ran?: number; completed?: number } +} + export type FormFields = [FormField, ...Array] export type FormFields2 = [FormField1, ...Array] @@ -2043,6 +2088,11 @@ export type SessionMessageAssistant = { retry?: SessionMessageAssistantRetry } +export type SessionMessageAssistantContentEncoded = + | SessionMessageAssistantText1 + | SessionMessageAssistantReasoning1 + | SessionMessageAssistantTool1 + export type IntegrationOAuthMethod = { id: string; type: "oauth"; label: string; form?: FormFields } export type IntegrationKeyMethod = { type: "key"; label?: string; form?: FormFields } @@ -2051,6 +2101,50 @@ export type FormInfo = { id: string; sessionID: string; title: string; metadata? export type FormInfo1 = { id: string; sessionID: string; title: string; metadata?: FormMetadata1; fields: FormFields2 } +export type SessionMessageInfo = + | SessionMessageAgentSelected + | SessionMessageModelSelected + | SessionMessageLocationSwitched + | SessionMessageUser + | SessionMessageSynthetic + | SessionMessageSystem + | SessionMessageSkill + | SessionMessageShell + | SessionMessageAssistant + | SessionMessageCompaction + +export type SessionMessageContentUpdated = { + id: string + created: number + metadata?: { [x: string]: any } + type: "session.message.content.updated" + durable: { aggregateID: string; seq: number; version: 1 } + location?: LocationRef + data: { sessionID: string; messageID: string; content: Array } +} + +export type IntegrationMethod = + | IntegrationOAuthMethod + | IntegrationCommandMethod + | IntegrationKeyMethod + | IntegrationEnvMethod + +export type FormCreated = { + id: string + created: number + metadata?: { [x: string]: any } + type: "form.created" + location?: LocationRef + data: { form: FormInfo1 } +} + +export type SessionTransferData = { info: SessionInfo; messages: Array } + +export type SessionMessagesResponse = { + data: Array + cursor: { previous?: string | null; next?: string | null } +} + export type SessionEventDurable = | SessionCreated | SessionAgentSelected @@ -2092,44 +2186,9 @@ export type SessionEventDurable = | SessionRevertStaged | SessionRevertCleared | SessionRevertCommitted + | SessionMessageContentUpdated | SessionUsageRecorded -export type SessionMessageInfo = - | SessionMessageAgentSelected - | SessionMessageModelSelected - | SessionMessageLocationSwitched - | SessionMessageUser - | SessionMessageSynthetic - | SessionMessageSystem - | SessionMessageSkill - | SessionMessageShell - | SessionMessageAssistant - | SessionMessageCompaction - -export type IntegrationMethod = - | IntegrationOAuthMethod - | IntegrationCommandMethod - | IntegrationKeyMethod - | IntegrationEnvMethod - -export type FormCreated = { - id: string - created: number - metadata?: { [x: string]: any } - type: "form.created" - location?: LocationRef - data: { form: FormInfo1 } -} - -export type SessionLogItem = SessionEventDurable | EventLogSynced - -export type SessionTransferData = { info: SessionInfo; messages: Array } - -export type SessionMessagesResponse = { - data: Array - cursor: { previous?: string | null; next?: string | null } -} - export type IntegrationInfo = { id: string name: string @@ -2191,6 +2250,7 @@ export type V2Event = | SessionRevertStaged | SessionRevertCleared | SessionRevertCommitted + | SessionMessageContentUpdated | FilesystemChanged | ReferenceUpdated | PermissionAsked @@ -2229,6 +2289,8 @@ export type V2Event = | McpResourcesChanged | V2EventServerConnected +export type SessionLogItem = SessionEventDurable | EventLogSynced + export type UnauthorizedError = { readonly _tag: "UnauthorizedError"; readonly message: string } export const isUnauthorizedError = (value: unknown): value is UnauthorizedError => typeof value === "object" && value !== null && "_tag" in value && value["_tag"] === "UnauthorizedError" @@ -4036,6 +4098,91 @@ export type SessionMessageInput = { export type SessionMessageOutput = { data: SessionMessageInfo }["data"] +export type SessionMessageUpdateInput = { + readonly sessionID: { readonly sessionID: string; readonly messageID: string }["sessionID"] + readonly messageID: { readonly sessionID: string; readonly messageID: string }["messageID"] + readonly content: { + readonly content: ReadonlyArray< + | { readonly type: "text"; readonly text: string; readonly state?: { readonly [x: string]: JsonValue } } + | { + readonly type: "reasoning" + readonly text: string + readonly state?: { readonly [x: string]: JsonValue } + readonly time?: { readonly created: number; readonly completed?: number } + } + | { + readonly type: "tool" + readonly id: string + readonly name: string + readonly executed?: boolean + readonly providerState?: { readonly [x: string]: JsonValue } + readonly providerResultState?: { readonly [x: string]: JsonValue } + readonly state: + | { readonly status: "streaming"; readonly input: string } + | { + readonly status: "running" + readonly input: { readonly [x: string]: JsonValue } + readonly metadata: { readonly [x: string]: JsonValue } + } + | { + readonly status: "completed" + readonly input: { readonly [x: string]: JsonValue } + readonly content: readonly [ + ( + | { readonly type: "text"; readonly text: string } + | { + readonly type: "file" + readonly uri: string + readonly mime: string + readonly name?: string | null + } + ), + ...Array< + | { readonly type: "text"; readonly text: string } + | { + readonly type: "file" + readonly uri: string + readonly mime: string + readonly name?: string | null + } + >, + ] + readonly metadata?: { readonly [x: string]: JsonValue } + } + | { + readonly status: "error" + readonly input: { readonly [x: string]: JsonValue } + readonly error: { readonly type: string; readonly message: string; readonly status?: number } + readonly content?: readonly [ + ( + | { readonly type: "text"; readonly text: string } + | { + readonly type: "file" + readonly uri: string + readonly mime: string + readonly name?: string | null + } + ), + ...Array< + | { readonly type: "text"; readonly text: string } + | { + readonly type: "file" + readonly uri: string + readonly mime: string + readonly name?: string | null + } + >, + ] + readonly metadata?: { readonly [x: string]: JsonValue } + } + readonly time: { readonly created: number; readonly ran?: number; readonly completed?: number } + } + > + }["content"] +} + +export type SessionMessageUpdateOutput = { data: SessionMessageAssistant }["data"] + export type SessionEnvironmentInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] readonly variables: { readonly variables: { readonly [x: string]: string } }["variables"] diff --git a/packages/client/src/solid/data.ts b/packages/client/src/solid/data.ts index 53196acc3e5..a55570183c4 100644 --- a/packages/client/src/solid/data.ts +++ b/packages/client/src/solid/data.ts @@ -167,6 +167,10 @@ function createSync() { has(key: string) { return state.has(key) }, + pending(key: string) { + const active = state.get(key) + return active !== undefined && active !== true + }, invalidate(key?: string) { if (key) { const active = state.get(key) @@ -723,6 +727,17 @@ export function createData(config: CreateDataInput) { match.time.completed = event.created }) return + case "session.message.content.updated": { + if (store.session.message[event.data.sessionID]) + message.update(event.data.sessionID, (draft, index) => { + const assistant = message.assistant(draft, index, event.data.messageID) + if (assistant) assistant.content = [...event.data.content] + }) + if (!sync.pending(`session.message:${event.data.sessionID}`)) return + result.session.message.invalidate(event.data.sessionID) + void result.session.message.sync(event.data.sessionID) + return + } case "session.step.started": message.update(event.data.sessionID, (draft, index) => { const position = index.get(event.data.assistantMessageID) diff --git a/packages/client/test/solid-data.test.ts b/packages/client/test/solid-data.test.ts index 36d6b14e8b2..9869c8ad790 100644 --- a/packages/client/test/solid-data.test.ts +++ b/packages/client/test/solid-data.test.ts @@ -414,6 +414,77 @@ test("loads bounded message pages", async () => { } }) +test("preserves assistant content replacement events across an active message read", async () => { + const listeners = new Set[0]>() + const release = Promise.withResolvers() + let requests = 0 + const content = [ + { type: "text" as const, text: "replacement" }, + { type: "reasoning" as const, text: "reasoning", time: { created: 3 } }, + ] + const api = OpenCode.make({ + baseUrl: "http://opencode.local", + fetch: async () => { + const current = ++requests + if (current === 2) await release.promise + return Response.json({ + data: [ + { + id: "msg_assistant", + type: "assistant", + agent: "build", + model: { id: "model", providerID: "provider" }, + content: current === 3 ? content : [{ type: "text", text: "original" }], + time: { created: 1, completed: 2 }, + }, + ], + cursor: {}, + }) + }, + }) + const setup = createRoot((dispose) => ({ + data: createData({ + api: () => api, + directory: "/project", + event: { + on: () => () => {}, + listen(handler) { + listeners.add(handler) + return () => listeners.delete(handler) + }, + }, + }), + dispose, + })) + + try { + await setup.data.session.message.sync("ses_refresh") + setup.data.session.message.invalidate("ses_refresh") + const stale = setup.data.session.message.sync("ses_refresh") + await wait(() => requests === 2) + const updated: OpenCodeEvent = { + id: "evt_message_updated", + created: 3, + type: "session.message.content.updated", + durable: { aggregateID: "ses_refresh", seq: 3, version: 1 }, + data: { + sessionID: "ses_refresh", + messageID: "msg_assistant", + content, + }, + } + listeners.forEach((listener) => listener({ name: updated.type, details: updated })) + + expect(setup.data.session.message.list("ses_refresh")[0]).toMatchObject({ content }) + release.resolve() + await stale + await wait(() => requests === 3) + expect(setup.data.session.message.list("ses_refresh")[0]).toMatchObject({ content }) + } finally { + setup.dispose() + } +}) + async function wait(check: () => boolean) { const started = Date.now() while (!check()) { diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index d16d9875fb2..80a113ed5db 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -139,6 +139,27 @@ export class CompactionConflictError extends Schema.TaggedError()("Session.BusyError", { sessionID: SessionSchema.ID, }) {} +export class MessageNotAssistantError extends Schema.TaggedError()( + "Session.MessageNotAssistantError", + { + sessionID: SessionSchema.ID, + messageID: SessionMessage.ID, + }, +) {} +export class MessageIncompleteError extends Schema.TaggedError()( + "Session.MessageIncompleteError", + { + sessionID: SessionSchema.ID, + messageID: SessionMessage.ID, + }, +) {} +export class MessageToolIncompleteError extends Schema.TaggedError()( + "Session.MessageToolIncompleteError", + { + sessionID: SessionSchema.ID, + messageID: SessionMessage.ID, + }, +) {} export class InboxConflictError extends Schema.TaggedError()("Session.InboxConflictError", { sessionID: SessionSchema.ID, inboxID: SessionMessage.ID, @@ -193,6 +214,19 @@ export interface Interface { sessionID: SessionSchema.ID messageID: SessionMessage.ID }) => Effect.Effect + readonly updateMessage: (input: { + readonly sessionID: SessionSchema.ID + readonly messageID: SessionMessage.ID + readonly content: readonly SessionMessage.AssistantContent[] + }) => Effect.Effect< + SessionMessage.Assistant, + | NotFoundError + | MessageNotFoundError + | BusyError + | MessageNotAssistantError + | MessageIncompleteError + | MessageToolIncompleteError + > readonly context: ( sessionID: SessionSchema.ID, ) => Effect.Effect @@ -560,6 +594,29 @@ const layer = Layer.effect( const stored = yield* store.message(input.messageID) return stored?.sessionID === input.sessionID ? stored.message : undefined }), + updateMessage: Effect.fn("Session.updateMessage")(function* (input) { + const ref = { sessionID: input.sessionID, messageID: input.messageID } + yield* result.get(ref.sessionID) + if ((yield* execution.active).has(ref.sessionID)) return yield* new BusyError({ sessionID: ref.sessionID }) + const message = yield* result.message(ref) + if (!message) return yield* new MessageNotFoundError(ref) + if (message.type !== "assistant") return yield* new MessageNotAssistantError(ref) + if (!message.time.completed) return yield* new MessageIncompleteError(ref) + if ( + input.content.some( + (content) => + content.type === "tool" && (content.state.status === "streaming" || content.state.status === "running"), + ) + ) + return yield* new MessageToolIncompleteError(ref) + yield* bus.publish(SessionEvent.MessageContentUpdated, { + ...ref, + content: Schema.encodeSync(Schema.Array(SessionMessage.AssistantContent))(input.content), + }) + const updated = yield* result.message(ref) + if (updated?.type !== "assistant") return yield* new MessageNotFoundError(ref) + return updated + }), context: Effect.fn("Session.context")(function* (sessionID) { yield* result.get(sessionID) return yield* store.context(sessionID) diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index f3df3a8151a..179aeb86769 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -1,5 +1,5 @@ import { castDraft, produce, type WritableDraft } from "immer" -import { DateTime, Effect, Match, pipe } from "effect" +import { DateTime, Effect, Match, pipe, Schema } from "effect" import { SessionEvent } from "./event.js" import { SessionMessage } from "./message.js" @@ -71,6 +71,12 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { Match.discriminatorsExhaustive("type")({ "session.created": () => Effect.void, "session.viewed": () => Effect.void, + "session.message.content.updated": (event) => + updateOwnedAssistant(event.data.messageID, (draft) => { + draft.content = castDraft( + Schema.decodeUnknownSync(Schema.Array(SessionMessage.AssistantContent))(event.data.content), + ) + }), "session.usage.recorded": () => Effect.void, "session.agent.selected": (event) => Effect.gen(function* () { diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index f4be4a8f9c3..ce5b816db32 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -581,6 +581,7 @@ const layer = Layer.effectDiscard( .run() .pipe(Effect.orDie) }) + yield* bus.project(SessionEvent.MessageContentUpdated, (event) => run(db, event)) yield* bus.project(SessionEvent.UsageRecorded, (event) => applyUsage(db, event.data.sessionID, event.data)) yield* bus.project(SessionEvent.Forked, (event) => projectFork(db, event)) yield* bus.project(SessionEvent.InboxDelivered, (event) => diff --git a/packages/core/test/session-message-update.test.ts b/packages/core/test/session-message-update.test.ts new file mode 100644 index 00000000000..23810107e1d --- /dev/null +++ b/packages/core/test/session-message-update.test.ts @@ -0,0 +1,246 @@ +import { describe, expect } from "bun:test" +import path from "path" +import { Effect, Layer, Stream } from "effect" +import { asc, eq } from "drizzle-orm" +import { Agent } from "@opencode-ai/core/agent" +import { Bus } from "@opencode-ai/core/bus" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { EventTable } from "@opencode-ai/core/event/sql" +import { Location } from "@opencode-ai/core/location" +import { Model } from "@opencode-ai/core/model" +import { Project } from "@opencode-ai/core/project" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { Provider } from "@opencode-ai/core/provider" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { Session } from "@opencode-ai/core/session" +import { SessionEvent } from "@opencode-ai/core/session/event" +import { SessionExecution } from "@opencode-ai/core/session/execution" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionStore } from "@opencode-ai/core/session/store" +import { Money } from "@opencode-ai/schema/money" +import { LayerNode } from "@opencode-ai/util/effect/layer-node" +import { tmpdir } from "./fixture/tmpdir" +import { testEffect } from "./lib/effect" +import { globalProjectLayer } from "./lib/project" + +const active = new Set() +const it = testEffect( + AppNodeBuilder.build( + LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node, Session.node]), + [ + [Bus.node, Bus.configured({ persist: true })], + [Project.node, globalProjectLayer], + [ + SessionExecution.node, + Layer.succeed( + SessionExecution.Service, + SessionExecution.Service.of({ + active: Effect.sync(() => active), + resume: () => Effect.void, + wake: () => Effect.void, + interrupt: () => Effect.succeed(false), + awaitIdle: () => Effect.void, + }), + ), + ], + ], + ), +) +const location = Location.Ref.make({ directory: AbsolutePath.make("/project") }) +const model = { id: Model.ID.make("model"), providerID: Provider.ID.make("provider") } + +const start = (bus: Bus.Interface, sessionID: Session.ID, messageID: SessionMessage.ID) => + bus.publish(SessionEvent.Step.Started, { + sessionID, + assistantMessageID: messageID, + agent: Agent.defaultID, + model, + }) + +const complete = (bus: Bus.Interface, sessionID: Session.ID, messageID: SessionMessage.ID) => + bus.publish(SessionEvent.Step.Ended, { + sessionID, + assistantMessageID: messageID, + finish: "stop", + cost: Money.USD.make(0), + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + }) + +describe("Session.updateMessage", () => { + it.effect("replaces assistant content through a durable projected event", () => + Effect.gen(function* () { + const session = yield* Session.Service + const bus = yield* Bus.Service + const db = (yield* Database.Service).db + const created = yield* session.create({ location }) + const messageID = SessionMessage.ID.create() + yield* start(bus, created.id, messageID) + yield* complete(bus, created.id, messageID) + + const content = [ + SessionMessage.AssistantText.make({ type: "text", text: "replacement" }), + SessionMessage.AssistantReasoning.make({ + type: "reasoning", + text: "updated reasoning", + time: { created: created.time.created }, + }), + ] + const updated = yield* session.updateMessage({ sessionID: created.id, messageID, content }) + + expect(updated.content).toEqual(content) + expect(yield* session.message({ sessionID: created.id, messageID })).toMatchObject({ content }) + expect((yield* session.messages({ sessionID: created.id }))[0]).toMatchObject({ id: messageID, content }) + + const events = Array.from(yield* Stream.runCollect(session.log({ sessionID: created.id }))) + expect(events.at(-2)).toMatchObject({ + type: "session.message.content.updated", + data: { + sessionID: created.id, + messageID, + content: [ + { type: "text", text: "replacement" }, + { type: "reasoning", text: "updated reasoning", time: { created: expect.any(Number) } }, + ], + }, + }) + expect( + yield* db + .select() + .from(EventTable) + .where(eq(EventTable.type, Bus.versionedType(SessionEvent.MessageContentUpdated.type, 1))) + .get(), + ).toMatchObject({ aggregate_id: created.id, data: { messageID } }) + + expect((yield* session.updateMessage({ sessionID: created.id, messageID, content: [] })).content).toEqual([]) + }), + ) + + it.effect("replays updated assistant content into a fresh projection", () => + Effect.gen(function* () { + const session = yield* Session.Service + const bus = yield* Bus.Service + const db = (yield* Database.Service).db + const created = yield* session.create({ location }) + const messageID = SessionMessage.ID.create() + yield* start(bus, created.id, messageID) + yield* complete(bus, created.id, messageID) + const content = [ + SessionMessage.AssistantReasoning.make({ + type: "reasoning", + text: "replayed reasoning", + time: { created: created.time.created }, + }), + ] + yield* session.updateMessage({ sessionID: created.id, messageID, content }) + + const serialized = (yield* db + .select() + .from(EventTable) + .where(eq(EventTable.aggregate_id, created.id)) + .orderBy(asc(EventTable.seq)) + .all() + .pipe(Effect.orDie)).map((event) => ({ + id: event.id, + created: event.created, + aggregateID: event.aggregate_id, + seq: event.seq, + type: event.type, + data: event.data, + })) + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const target = AppNodeBuilder.build( + LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node]), + [ + [Database.node, Database.configured({ path: path.join(tmp.path, "target.sqlite") })], + [Bus.node, Bus.configured({ persist: true })], + ], + ) + + yield* Effect.gen(function* () { + const database = (yield* Database.Service).db + const replay = yield* Bus.Service + const store = yield* SessionStore.Service + yield* database + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: location.directory, sandboxes: [] }) + .run() + .pipe(Effect.orDie) + yield* Effect.forEach(serialized, (event) => replay.replay(event), { discard: true }) + expect((yield* store.message(messageID))?.message).toMatchObject({ content }) + }).pipe(Effect.provide(Layer.fresh(target))) + }), + ) + + it.effect("rejects missing and cross-session messages", () => + Effect.gen(function* () { + const session = yield* Session.Service + const bus = yield* Bus.Service + const created = yield* session.create({ location }) + const other = yield* session.create({ location }) + const messageID = SessionMessage.ID.create() + yield* start(bus, created.id, messageID) + yield* complete(bus, created.id, messageID) + + expect(yield* Effect.flip(session.updateMessage({ sessionID: other.id, messageID, content: [] }))).toEqual( + new Session.MessageNotFoundError({ sessionID: other.id, messageID }), + ) + const missing = Session.ID.create() + expect(yield* Effect.flip(session.updateMessage({ sessionID: missing, messageID, content: [] }))).toEqual( + new Session.NotFoundError({ sessionID: missing }), + ) + }), + ) + + it.effect("rejects non-assistant messages, incomplete assistants, and unfinished tools", () => + Effect.gen(function* () { + const session = yield* Session.Service + const bus = yield* Bus.Service + const created = yield* session.create({ location }) + const synthetic = yield* bus.publish(SessionEvent.Synthetic, { sessionID: created.id, text: "synthetic" }) + const syntheticID = SessionMessage.ID.fromEvent(synthetic.id) + + expect( + yield* Effect.flip(session.updateMessage({ sessionID: created.id, messageID: syntheticID, content: [] })), + ).toEqual(new Session.MessageNotAssistantError({ sessionID: created.id, messageID: syntheticID })) + + const messageID = SessionMessage.ID.create() + yield* start(bus, created.id, messageID) + expect(yield* Effect.flip(session.updateMessage({ sessionID: created.id, messageID, content: [] }))).toEqual( + new Session.MessageIncompleteError({ sessionID: created.id, messageID }), + ) + + yield* complete(bus, created.id, messageID) + const unfinished = SessionMessage.AssistantTool.make({ + type: "tool", + id: "call_unfinished", + name: "read", + state: { status: "streaming", input: "" }, + time: { created: created.time.created }, + }) + expect( + yield* Effect.flip(session.updateMessage({ sessionID: created.id, messageID, content: [unfinished] })), + ).toEqual(new Session.MessageToolIncompleteError({ sessionID: created.id, messageID })) + }), + ) + + it.effect("rejects a completed assistant while its session is active", () => + Effect.gen(function* () { + const session = yield* Session.Service + const bus = yield* Bus.Service + const created = yield* session.create({ location }) + const messageID = SessionMessage.ID.create() + yield* start(bus, created.id, messageID) + yield* complete(bus, created.id, messageID) + active.add(created.id) + const failure = yield* Effect.flip(session.updateMessage({ sessionID: created.id, messageID, content: [] })) + active.delete(created.id) + + expect(failure).toEqual(new Session.BusyError({ sessionID: created.id })) + }), + ) +}) diff --git a/packages/protocol/openapi.json b/packages/protocol/openapi.json index 238ca367b22..3539338410c 100644 --- a/packages/protocol/openapi.json +++ b/packages/protocol/openapi.json @@ -3950,6 +3950,143 @@ }, "description": "Retrieve one projected message owned by the Session.", "summary": "Get session message" + }, + "patch": { + "tags": ["session"], + "operationId": "v2.session.messageUpdate", + "parameters": [ + { + "name": "sessionID", + "in": "path", + "schema": { + "type": "string", + "pattern": "^ses" + }, + "required": true + }, + { + "name": "messageID", + "in": "path", + "schema": { + "type": "string", + "pattern": "^msg_" + }, + "required": true + } + ], + "security": [], + "responses": { + "200": { + "description": "Success", + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "data": { + "$ref": "#/components/schemas/Session.Message.Assistant" + } + }, + "required": ["data"], + "additionalProperties": false + } + } + } + }, + "400": { + "description": "InvalidRequestError", + "content": { + "application/json": { + "schema": { + "anyOf": [ + { + "$ref": "#/components/schemas/InvalidRequestErrorEncoded" + }, + { + "$ref": "#/components/schemas/InvalidRequestErrorEncoded" + } + ] + } + } + } + }, + "401": { + "description": "UnauthorizedError", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/UnauthorizedErrorEncoded" + } + } + } + }, + "404": { + "description": "SessionNotFoundError | MessageNotFoundError", + "content": { + "application/json": { + "schema": { + "anyOf": [ + { + "$ref": "#/components/schemas/SessionNotFoundErrorEncoded" + }, + { + "$ref": "#/components/schemas/MessageNotFoundErrorEncoded" + } + ] + } + } + } + }, + "409": { + "description": "SessionBusyError | ConflictError", + "content": { + "application/json": { + "schema": { + "anyOf": [ + { + "$ref": "#/components/schemas/SessionBusyErrorEncoded" + }, + { + "$ref": "#/components/schemas/ConflictErrorEncoded" + } + ] + } + } + } + } + }, + "description": "Replace the content of a completed assistant message in an idle session.", + "summary": "Update assistant message content", + "requestBody": { + "content": { + "application/json": { + "schema": { + "type": "object", + "properties": { + "content": { + "type": "array", + "items": { + "anyOf": [ + { + "$ref": "#/components/schemas/Session.Message.Assistant.Text" + }, + { + "$ref": "#/components/schemas/Session.Message.Assistant.Reasoning" + }, + { + "$ref": "#/components/schemas/Session.Message.Assistant.Tool" + } + ] + } + } + }, + "required": ["content"], + "additionalProperties": false + } + } + }, + "required": true + } } }, "/api/session/{sessionID}/environment": { diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index f0f2b46e8e5..4613b05faa6 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -710,6 +710,20 @@ export const makeSessionGroup = (sessionLo }), ), ) + .add( + HttpApiEndpoint.patch("session.messageUpdate", "/api/session/:sessionID/message/:messageID", { + params: { sessionID: Session.ID, messageID: SessionMessage.ID }, + payload: Schema.Struct({ content: Schema.Array(SessionMessage.AssistantContent) }), + success: Schema.Struct({ data: SessionMessage.Assistant }), + error: [SessionNotFoundError, MessageNotFoundError, InvalidRequestError, SessionBusyError, ConflictError], + }).annotateMerge( + OpenApi.annotations({ + identifier: "v2.session.messageUpdate", + summary: "Update assistant message content", + description: "Replace the content of a completed assistant message in an idle session.", + }), + ), + ) .add( HttpApiEndpoint.put("session.environment", "/api/session/:sessionID/environment", { params: { sessionID: Session.ID }, diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index 6db772c0757..72fcb4cae98 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -117,6 +117,18 @@ export const Viewed = Event.durable({ }) export type Viewed = typeof Viewed.Type +export const MessageContentUpdated = Event.durable({ + type: "session.message.content.updated", + ...options, + schema: { + ...Base, + messageID: SessionMessage.ID, + // Public events are framed directly, so timestamps must already be encoded. + content: Schema.Array(SessionMessage.AssistantContentEncoded), + }, +}) +export type MessageContentUpdated = typeof MessageContentUpdated.Type + export const UsageRecorded = Event.durable({ type: "session.usage.recorded", ...options, @@ -639,6 +651,7 @@ export const Definitions = Event.inventory( RevertEvent.Staged, RevertEvent.Cleared, RevertEvent.Committed, + MessageContentUpdated, ) // UsageRecorded is durable but internal: excluded from Definitions so it never reaches the public manifest. diff --git a/packages/schema/src/session-message.ts b/packages/schema/src/session-message.ts index 9323ca58d65..d4ab3bfca7e 100644 --- a/packages/schema/src/session-message.ts +++ b/packages/schema/src/session-message.ts @@ -195,6 +195,11 @@ export const AssistantContent = Schema.Union([AssistantText, AssistantReasoning, ) export type AssistantContent = AssistantText | AssistantReasoning | AssistantTool +export const AssistantContentEncoded = Schema.toEncoded(AssistantContent).annotate({ + identifier: "Session.Message.AssistantContent.Encoded", +}) +export type AssistantContentEncoded = typeof AssistantContentEncoded.Type + export interface AssistantRetry extends Schema.Schema.Type {} export const AssistantRetry = Schema.Struct({ attempt: PositiveInt, diff --git a/packages/schema/test/event-manifest.test.ts b/packages/schema/test/event-manifest.test.ts index 92535fd4e2c..ed67448c5c3 100644 --- a/packages/schema/test/event-manifest.test.ts +++ b/packages/schema/test/event-manifest.test.ts @@ -118,6 +118,7 @@ describe("public event manifest", () => { "session.moved.1", "session.renamed.1", "session.viewed.1", + "session.message.content.updated.1", "session.usage.recorded.1", "session.forked.2", "session.inbox.delivered.1", diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index 353f95f0904..282eeeb4264 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -635,5 +635,36 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl }) }), ) + .handle( + "session.messageUpdate", + Effect.fn(function* (ctx) { + const message = yield* session.updateMessage({ ...ctx.params, content: ctx.payload.content }).pipe( + Effect.catchTag("Session.NotFoundError", missingSession), + Effect.catchTag( + "Session.MessageNotFoundError", + (error) => + new MessageNotFoundError({ + sessionID: error.sessionID, + messageID: error.messageID, + message: `Message not found: ${error.messageID}`, + }), + ), + Effect.catchTag("Session.BusyError", busySession), + Effect.catchTag( + "Session.MessageNotAssistantError", + () => new InvalidRequestError({ message: "Only assistant messages can be updated", field: "messageID" }), + ), + Effect.catchTag( + "Session.MessageIncompleteError", + (error) => new ConflictError({ message: "Assistant message is incomplete", resource: error.messageID }), + ), + Effect.catchTag( + "Session.MessageToolIncompleteError", + () => new InvalidRequestError({ message: "Tool content must be completed", field: "content" }), + ), + ) + return { data: message } + }), + ) }), ) diff --git a/packages/server/test/session-message-update.test.ts b/packages/server/test/session-message-update.test.ts new file mode 100644 index 00000000000..a7f5c55e508 --- /dev/null +++ b/packages/server/test/session-message-update.test.ts @@ -0,0 +1,145 @@ +import { expect } from "bun:test" +import { Agent } from "@opencode-ai/core/agent" +import { Bus } from "@opencode-ai/core/bus" +import { Model } from "@opencode-ai/core/model" +import { Provider } from "@opencode-ai/core/provider" +import { Session } from "@opencode-ai/core/session" +import { SessionEvent } from "@opencode-ai/core/session/event" +import { SessionExecution } from "@opencode-ai/core/session/execution" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { Money } from "@opencode-ai/schema/money" +import { Effect, Layer } from "effect" +import { it } from "../../core/test/lib/effect" +import { ServerFetch } from "../src/fetch" + +it.live("updates completed assistant message content through the session HTTP API", () => + Effect.gen(function* () { + const state = { + active: new Set(), + user: SessionMessage.ID.create(), + assistant: SessionMessage.ID.create(), + complete: true, + } + const execution = Layer.effect( + SessionExecution.Service, + Effect.gen(function* () { + const bus = yield* Bus.Service + return SessionExecution.Service.of({ + active: Effect.sync(() => state.active), + resume: () => Effect.void, + wake: (sessionID) => + Effect.gen(function* () { + yield* bus.publish(SessionEvent.InboxDelivered, { sessionID, inboxID: state.user }) + yield* bus.publish(SessionEvent.Step.Started, { + sessionID, + assistantMessageID: state.assistant, + agent: Agent.defaultID, + model: { id: Model.ID.make("model"), providerID: Provider.ID.make("provider") }, + }) + if (!state.complete) return + yield* bus.publish(SessionEvent.Step.Ended, { + sessionID, + assistantMessageID: state.assistant, + finish: "stop", + cost: Money.USD.make(0), + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + }) + }), + interrupt: () => Effect.succeed(false), + awaitIdle: () => Effect.void, + }) + }), + ) + const handler = yield* ServerFetch.make( + { app: { version: "test-version" }, database: { path: ":memory:" }, fs: { filewatcher: false } }, + { overrides: [[SessionExecution.node, execution]] }, + ) + const created = yield* Effect.promise(() => + handler( + new Request("http://opencode.local/api/session", { + method: "POST", + headers: { "content-type": "application/json" }, + body: "{}", + }), + ).then((response) => response.json()), + ) + const sessionID = Session.ID.make(created.data.id) + const prompt = () => + Effect.promise(() => + handler( + new Request(`http://opencode.local/api/session/${sessionID}/prompt`, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ id: state.user, text: "prompt" }), + }), + ), + ) + const update = (messageID: SessionMessage.ID, body: unknown, id = sessionID) => + Effect.promise(() => + handler( + new Request(`http://opencode.local/api/session/${id}/message/${messageID}`, { + method: "PATCH", + headers: { "content-type": "application/json" }, + body: JSON.stringify(body), + }), + ), + ) + + expect((yield* prompt()).status).toBe(200) + const content = [ + { type: "text", text: "edited assistant response" }, + { type: "reasoning", text: "edited reasoning", time: { created: 123 } }, + ] + const updated = yield* update(state.assistant, { content }) + expect(updated.status).toBe(200) + expect(yield* Effect.promise(() => updated.json())).toMatchObject({ + data: { id: state.assistant, type: "assistant", content }, + }) + + const projected = yield* Effect.promise(() => + handler(new Request(`http://opencode.local/api/session/${sessionID}/message/${state.assistant}`)).then( + (response) => response.json(), + ), + ) + expect(projected.data.content).toEqual(content) + expect((yield* update(state.assistant, { text: "not a content array" })).status).toBe(400) + const unfinished = yield* update(state.assistant, { + content: [ + { + type: "tool", + id: "call_unfinished", + name: "read", + state: { status: "streaming", input: "" }, + time: { created: 123 }, + }, + ], + }) + expect(unfinished.status).toBe(400) + expect(yield* Effect.promise(() => unfinished.json())).toMatchObject({ + _tag: "InvalidRequestError", + field: "content", + }) + const nonAssistant = yield* update(state.user, { content: [] }) + expect(nonAssistant.status).toBe(400) + expect(yield* Effect.promise(() => nonAssistant.json())).toMatchObject({ _tag: "InvalidRequestError" }) + expect((yield* update(SessionMessage.ID.create(), { content: [] })).status).toBe(404) + expect((yield* update(state.assistant, { content: [] }, Session.ID.create())).status).toBe(404) + + state.active.add(sessionID) + const busy = yield* update(state.assistant, { content: [] }) + state.active.delete(sessionID) + expect(busy.status).toBe(409) + expect(yield* Effect.promise(() => busy.json())).toMatchObject({ _tag: "SessionBusyError", sessionID }) + + state.user = SessionMessage.ID.create() + state.assistant = SessionMessage.ID.create() + state.complete = false + expect((yield* prompt()).status).toBe(200) + const incomplete = yield* update(state.assistant, { content: [] }) + expect(incomplete.status).toBe(409) + expect(yield* Effect.promise(() => incomplete.json())).toMatchObject({ + _tag: "ConflictError", + resource: state.assistant, + }) + }).pipe(Effect.scoped), +)