refactor(session): remove message content mutation API (#48043)

This commit is contained in:
Aiden Cline
2026-09-08 20:27:22 -05:00
committed by GitHub
parent 594635b5ae
commit 6ee2ed7510
19 changed files with 25 additions and 1214 deletions
+13 -26
View File
@@ -1029,19 +1029,6 @@ 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<SessionMessage.AssistantContentEncoded>
}
}
| {
readonly id: Event.ID
readonly created: number
@@ -1061,6 +1048,19 @@ export type SessionLogOutput =
}
}
}
| {
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<SessionMessage.AssistantContentEncoded>
}
}
)
| EventLog.Synced
export type SessionLogOperation<E = never> = (input: SessionLogInput) => Stream.Stream<SessionLogOutput, E>
@@ -1081,18 +1081,6 @@ export type SessionMessageInput = { readonly sessionID: Session.ID; readonly mes
export type SessionMessageOutput = SessionMessage.Info
export type SessionMessageOperation<E = never> = (input: SessionMessageInput) => Effect.Effect<SessionMessageOutput, E>
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<E = never> = (
input: SessionMessageUpdateInput,
) => Effect.Effect<SessionMessageUpdateOutput, E>
export type SessionEnvironmentInput = {
readonly sessionID: Session.ID
readonly variables: { readonly [x: string]: string }
@@ -1151,7 +1139,6 @@ export interface SessionApi<E = never> {
readonly interrupt: SessionInterruptOperation<E>
readonly background: SessionBackgroundOperation<E>
readonly message: SessionMessageOperation<E>
readonly messageUpdate: SessionMessageUpdateOperation<E>
readonly environment: SessionEnvironmentOperation<E>
readonly view: SessionViewOperation<E>
}
@@ -92,8 +92,6 @@ import type {
SessionBackgroundOutput,
SessionMessageInput,
SessionMessageOutput,
SessionMessageUpdateInput,
SessionMessageUpdateOutput,
SessionEnvironmentInput,
SessionEnvironmentOutput,
SessionViewInput,
@@ -691,17 +689,6 @@ const EndpointSessionMessage = (raw: RawClient["server.session"]) => (input: Ses
),
)
const EndpointSessionMessageUpdate = (raw: RawClient["server.session"]) => (input: SessionMessageUpdateInput) =>
preserveEffect<SessionMessageUpdateOutput>()(
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<SessionEnvironmentOutput>()(
raw["session.environment"]({
@@ -762,7 +749,6 @@ const adaptGroupSession = (raw: RawClient["server.session"]) => ({
interrupt: EndpointSessionInterrupt(raw),
background: EndpointSessionBackground(raw),
message: EndpointSessionMessage(raw),
messageUpdate: EndpointSessionMessageUpdate(raw),
environment: EndpointSessionEnvironment(raw),
view: EndpointSessionView(raw),
})
@@ -86,8 +86,6 @@ import type {
SessionBackgroundOutput,
SessionMessageInput,
SessionMessageOutput,
SessionMessageUpdateInput,
SessionMessageUpdateOutput,
SessionEnvironmentInput,
SessionEnvironmentOutput,
SessionViewInput,
@@ -986,18 +984,6 @@ 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: [400, 401, 404, 409],
empty: false,
},
requestOptions,
).then((value) => value.data),
environment: (input: SessionEnvironmentInput, requestOptions?: RequestOptions) =>
request<SessionEnvironmentOutput>(
{
+1 -87
View File
@@ -2252,8 +2252,8 @@ export type SessionEventDurable =
| SessionRevertStaged
| SessionRevertCleared
| SessionRevertCommitted
| SessionMessageContentUpdated
| SessionUsageRecorded
| SessionMessageContentUpdated
export type IntegrationInfo = {
id: string
@@ -2317,7 +2317,6 @@ export type V2Event =
| SessionRevertStaged
| SessionRevertCleared
| SessionRevertCommitted
| SessionMessageContentUpdated
| FilesystemChanged
| ReferenceUpdated
| PermissionAsked
@@ -4270,91 +4269,6 @@ 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"]
-10
View File
@@ -827,16 +827,6 @@ export function createData(config: CreateDataInput) {
match.time.completed = event.created
})
return
case "session.message.content.updated": {
if (store.session.message[event.data.sessionID])
message.editAssistant(event.data.sessionID, event.data.messageID, (assistant) => {
assistant.content = [...event.data.content]
})
if (!sync.pending(`session.message:${event.data.sessionID}`)) return
result.session.message.invalidate(event.data.sessionID)
refresh(() => 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)
-71
View File
@@ -744,77 +744,6 @@ test.each(["success", "failure", "cancel", "cancel-retry", "cancel-page", "join-
},
)
test("preserves assistant content replacement events across an active message read", async () => {
const listeners = new Set<Parameters<CreateDataInput["event"]["listen"]>[0]>()
const release = Promise.withResolvers<void>()
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()
}
})
test.each([
"session.execution.succeeded",
"session.execution.failed",
-10
View File
@@ -31,10 +31,7 @@ import {
ForkEmptyError,
InboxConflictError,
MessageDecodeError,
MessageIncompleteError,
MessageNotAssistantError,
MessageNotFoundError,
MessageToolIncompleteError,
NotFoundError,
PromptConflictError,
SkillNotFoundError,
@@ -101,10 +98,7 @@ export {
CompactionConflictError,
InboxConflictError,
MessageDecodeError,
MessageIncompleteError,
MessageNotAssistantError,
MessageNotFoundError,
MessageToolIncompleteError,
NotFoundError,
PromptConflictError,
SkillNotFoundError,
@@ -136,9 +130,6 @@ export interface Interface {
sessionID: SessionSchema.ID
messageID: SessionMessage.ID
}) => Effect.Effect<SessionMessage.Info | undefined>
readonly updateMessage: (
input: Parameters<Session.Handle["updateMessage"]>[0] & { readonly sessionID: SessionSchema.ID },
) => ReturnType<Session.Handle["updateMessage"]>
readonly context: (
sessionID: SessionSchema.ID,
) => Effect.Effect<SessionMessage.Info[], NotFoundError | MessageDecodeError>
@@ -357,7 +348,6 @@ const layer = Layer.effect(
return yield* store.messages(input)
}),
message: (input) => sessions.forSession(input.sessionID).message(input.messageID),
updateMessage: (input) => sessions.forSession(input.sessionID).updateMessage(input),
context: Effect.fn("Session.context")(function* (sessionID) {
yield* result.get(sessionID)
return yield* store.context(sessionID)
-24
View File
@@ -16,30 +16,6 @@ export class MessageNotFoundError extends Schema.TaggedError<MessageNotFoundErro
messageID: SessionMessage.ID,
}) {}
export class MessageNotAssistantError extends Schema.TaggedError<MessageNotAssistantError>()(
"Session.MessageNotAssistantError",
{
sessionID: SessionSchema.ID,
messageID: SessionMessage.ID,
},
) {}
export class MessageIncompleteError extends Schema.TaggedError<MessageIncompleteError>()(
"Session.MessageIncompleteError",
{
sessionID: SessionSchema.ID,
messageID: SessionMessage.ID,
},
) {}
export class MessageToolIncompleteError extends Schema.TaggedError<MessageToolIncompleteError>()(
"Session.MessageToolIncompleteError",
{
sessionID: SessionSchema.ID,
messageID: SessionMessage.ID,
},
) {}
export class ForkEmptyError extends Schema.TaggedError<ForkEmptyError>()("Session.ForkEmptyError", {
sessionID: SessionSchema.ID,
}) {
+1 -31
View File
@@ -1,6 +1,6 @@
export * as Session from "./session.js"
import { DateTime, Effect, Fiber, Schema, Scope } from "effect"
import { DateTime, Effect, Fiber, Scope } from "effect"
import type { Agent } from "@opencode/schema/agent"
import type { Model } from "@opencode/schema/model"
import { Event } from "@opencode/schema/event"
@@ -14,10 +14,7 @@ import {
BusyError,
CompactionConflictError,
InboxConflictError,
MessageIncompleteError,
MessageNotAssistantError,
MessageNotFoundError,
MessageToolIncompleteError,
NotFoundError,
PromptConflictError,
SyntheticConflictError,
@@ -61,26 +58,6 @@ export const make = Effect.fn("Session.make")(function* () {
const stored = yield* store.message(messageID)
return stored?.sessionID === sessionID ? stored.message : undefined
})
const updateMessage = Effect.fn("Session.updateMessage")(function* (
sessionID: SessionSchema.ID,
input: { readonly messageID: SessionMessage.ID; readonly content: readonly SessionMessage.AssistantContent[] },
) {
const ref = { sessionID, messageID: input.messageID }
yield* get(sessionID)
if (yield* execution.isActive(sessionID)) return yield* new BusyError({ sessionID })
const current = yield* message(sessionID, input.messageID)
if (!current) return yield* new MessageNotFoundError(ref)
if (current.type !== "assistant") return yield* new MessageNotAssistantError(ref)
if (!current.time.completed) return yield* new MessageIncompleteError(ref)
if (input.content.some(isUnfinishedTool)) return yield* new MessageToolIncompleteError(ref)
yield* bus.publish(SessionEvent.MessageContentUpdated, {
...ref,
content: Schema.encodeSync(Schema.Array(SessionMessage.AssistantContent))(input.content),
})
const updated = yield* message(sessionID, input.messageID)
if (updated?.type !== "assistant") return yield* new MessageNotFoundError(ref)
return updated
})
const view = Effect.fn("Session.view")(function* (sessionID: SessionSchema.ID, input: { idle: number }) {
const session = yield* get(sessionID)
if (
@@ -355,7 +332,6 @@ export const make = Effect.fn("Session.make")(function* () {
const operations = {
get,
message,
updateMessage,
view,
rename,
switchAgent,
@@ -378,7 +354,6 @@ export const make = Effect.fn("Session.make")(function* () {
const forSession = (sessionID: SessionSchema.ID) => {
const get = operations.get.bind(undefined, sessionID)
const message = operations.message.bind(undefined, sessionID)
const updateMessage = operations.updateMessage.bind(undefined, sessionID)
const view = operations.view.bind(undefined, sessionID)
const rename = operations.rename.bind(undefined, sessionID)
const switchAgent = operations.switchAgent.bind(undefined, sessionID)
@@ -404,7 +379,6 @@ export const make = Effect.fn("Session.make")(function* () {
id: sessionID,
get,
message,
updateMessage,
view,
rename,
switchAgent,
@@ -429,8 +403,4 @@ export const make = Effect.fn("Session.make")(function* () {
export type Handle = ReturnType<Effect.Success<ReturnType<typeof make>>["forSession"]>
function isUnfinishedTool(content: SessionMessage.AssistantContent) {
return content.type === "tool" && (content.state.status === "streaming" || content.state.status === "running")
}
// Mirrors the shell tool's in-memory preview safety limit.
@@ -1,280 +0,0 @@
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/core/agent"
import { Bus } from "@opencode/core/bus"
import { Database } from "@opencode/core/database/database"
import { AppNodeBuilder } from "@opencode/core/effect/app-node-builder"
import { EventTable } from "@opencode/core/event/sql"
import { Location } from "@opencode/core/location"
import { Model } from "@opencode/core/model"
import { Project } from "@opencode/core/project"
import { ProjectTable } from "@opencode/core/project/sql"
import { Provider } from "@opencode/core/provider"
import { AbsolutePath } from "@opencode/core/schema"
import { Session } from "@opencode/core/session"
import { SessionEvent } from "@opencode/core/session/event"
import { SessionExecution } from "@opencode/core/session/execution"
import { SessionMessage } from "@opencode/core/session/message"
import { SessionProjector } from "@opencode/core/session/projector"
import { SessionStore } from "@opencode/core/session/store"
import { Money } from "@opencode/schema/money"
import { LayerNode } from "@opencode/util/effect/layer-node"
import { tmpdirScoped } from "./fixture/tmpdir"
import { testEffect } from "./lib/effect"
import { globalProjectNode } from "./lib/project"
const active = new Set<Session.ID>()
const it = testEffect(
AppNodeBuilder.build(
LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node, Session.node]),
[
Bus.node.replace(Bus.configured({ persist: true })),
Project.node.replace(globalProjectNode),
SessionExecution.node.replace(
Layer.succeed(
SessionExecution.Service,
SessionExecution.Service.of({
active: Effect.sync(() => active),
isActive: (sessionID) => Effect.sync(() => active.has(sessionID)),
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* tmpdirScoped()
const target = AppNodeBuilder.build(
LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node]),
[
Database.node.replace(Database.configured({ path: path.join(tmp.path, "target.sqlite") })),
Bus.node.replace(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)
yield* Effect.forEach(
[
SessionMessage.ToolStateStreaming.make({ status: "streaming", input: "" }),
SessionMessage.ToolStateRunning.make({ status: "running", input: {}, metadata: {} }),
],
Effect.fnUntraced(function* (state) {
const unfinished = SessionMessage.AssistantTool.make({
type: "tool",
id: "call_unfinished",
name: "read",
state,
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("accepts completed and failed tool content", () =>
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)
const content = [
SessionMessage.ToolStateCompleted.make({
status: "completed",
input: {},
content: [{ type: "text", text: "result" }],
}),
SessionMessage.ToolStateError.make({ status: "error", input: {}, error: { type: "tool", message: "failed" } }),
].map((state) =>
SessionMessage.AssistantTool.make({
type: "tool",
id: `call_${state.status}`,
name: "read",
state,
time: { created: created.time.created },
}),
)
expect((yield* session.updateMessage({ sessionID: created.id, messageID, content })).content).toEqual(content)
}),
)
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 }))
}),
)
})
+3 -5
View File
@@ -166,7 +166,7 @@ const setup = Effect.fnUntraced(function* (options?: {
})
describe("Session-owned handles", () => {
it.live("owns state changes and message editing without caller services or Location acquisition", () =>
it.live("owns state changes and message reads without caller services or Location acquisition", () =>
Effect.gen(function* () {
const fixture = yield* setup()
const handle = fixture.sessions.forSession(sessionID)
@@ -191,7 +191,7 @@ describe("Session-owned handles", () => {
.where(eq(SessionTable.id, sessionID))
.run()
.pipe(Effect.orDie)
const { rename, switchAgent, switchModel, view, message, updateMessage } = handle
const { rename, switchAgent, switchModel, view, message } = handle
yield* Effect.gen(function* () {
yield* rename({ title: "Renamed" })
@@ -200,9 +200,7 @@ describe("Session-owned handles", () => {
yield* switchModel({ model })
yield* view({ idle: 0 })
yield* view({ idle: 0 })
const content = [SessionMessage.AssistantText.make({ type: "text", text: "Edited" })]
expect((yield* updateMessage({ messageID, content })).content).toEqual(content)
expect(yield* message(messageID)).toMatchObject({ type: "assistant", content })
expect(yield* message(messageID)).toMatchObject({ type: "assistant", content: [] })
}).pipe(Effect.satisfiesServicesType<never>(), Effect.setContext(Context.empty()))
const session = yield* handle.get()
+1 -145
View File
@@ -1621,14 +1621,7 @@
"content": {
"application/json": {
"schema": {
"anyOf": [
{
"$ref": "#/components/schemas/SessionNotFoundErrorEncoded"
},
{
"$ref": "#/components/schemas/SessionNotFoundErrorEncoded"
}
]
"$ref": "#/components/schemas/SessionNotFoundErrorEncoded"
}
}
}
@@ -4285,143 +4278,6 @@
},
"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": {
-14
View File
@@ -710,20 +710,6 @@ export const makeSessionGroup = <I extends HttpApiMiddleware.AnyId, S>(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 },
+3 -2
View File
@@ -120,6 +120,7 @@ export const Viewed = Event.durable({
})
export type Viewed = typeof Viewed.Type
// Replay-only: older releases allowed replacing completed assistant content.
export const MessageContentUpdated = Event.durable({
type: "session.message.content.updated",
...options,
@@ -669,13 +670,13 @@ 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.
// Internal and replay-only events are excluded from the public manifest.
export const DurableDefinitions = Event.inventory(
...Definitions.filter((definition) => definition.durability === "durable"),
UsageRecorded,
MessageContentUpdated,
)
export const EphemeralDefinitions = Event.inventory(
...Definitions.filter((definition) => definition.durability === "ephemeral"),
@@ -158,6 +158,7 @@ describe("public event manifest", () => {
expect(SessionEvent.DurableDefinitions).toEqual([
...SessionEvent.Definitions.filter((definition) => definition.durability === "durable"),
SessionEvent.UsageRecorded,
SessionEvent.MessageContentUpdated,
])
expect(SessionEvent.UsageRecorded.durability).toBe("durable")
expect(EventManifest.Durable.get("session.usage.recorded.1")).toBe(SessionEvent.UsageRecorded)
-31
View File
@@ -637,36 +637,5 @@ 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 }
}),
)
}),
)
@@ -1,160 +0,0 @@
import { expect, setDefaultTimeout } from "bun:test"
import { Agent } from "@opencode/core/agent"
import { Bus } from "@opencode/core/bus"
import { Model } from "@opencode/core/model"
import { Provider } from "@opencode/core/provider"
import { Session } from "@opencode/core/session"
import { SessionEvent } from "@opencode/core/session/event"
import { SessionExecution } from "@opencode/core/session/execution"
import { SessionMessage } from "@opencode/core/session/message"
import { Money } from "@opencode/schema/money"
import { makeGlobalNode } from "@opencode/util/effect/app-node"
import { Effect, Layer } from "effect"
import { it } from "../../core/test/lib/effect"
import { ServerFetch } from "../src/fetch"
setDefaultTimeout(30_000)
it.live("updates completed assistant message content through the session HTTP API", () =>
Effect.gen(function* () {
const state = {
active: new Set<Session.ID>(),
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),
isActive: (sessionID) => Effect.sync(() => state.active.has(sessionID)),
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 },
models: { fetch: false },
},
{
overrides: [
SessionExecution.node.replace(
makeGlobalNode({ service: SessionExecution.Service, layer: execution, deps: [Bus.node] }),
),
],
},
)
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,
})
}),
)
+1 -145
View File
@@ -1621,14 +1621,7 @@
"content": {
"application/json": {
"schema": {
"anyOf": [
{
"$ref": "#/components/schemas/SessionNotFoundErrorEncoded"
},
{
"$ref": "#/components/schemas/SessionNotFoundErrorEncoded"
}
]
"$ref": "#/components/schemas/SessionNotFoundErrorEncoded"
}
}
}
@@ -4285,143 +4278,6 @@
},
"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": {
+1 -145
View File
@@ -1621,14 +1621,7 @@
"content": {
"application/json": {
"schema": {
"anyOf": [
{
"$ref": "#/components/schemas/SessionNotFoundErrorEncoded"
},
{
"$ref": "#/components/schemas/SessionNotFoundErrorEncoded"
}
]
"$ref": "#/components/schemas/SessionNotFoundErrorEncoded"
}
}
}
@@ -4285,143 +4278,6 @@
},
"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": {