refactor(core): isolate Session admission and controls (#46019)

Separate ID-bound Session policy from host routing. Bind Inbox and Location preparation dependencies at construction, preserve admission and execution semantics, and cover the extracted ownership contracts directly.
This commit is contained in:
Kit Langton
2026-08-28 23:22:25 -04:00
committed by GitHub
parent 0116a98371
commit 6e954f75ee
38 changed files with 2341 additions and 914 deletions
+4
View File
@@ -45,6 +45,8 @@ import { InstructionBuiltIns } from "./instructions/builtins.js"
import { InstructionEntry } from "./session/instruction-entry.js"
import { SessionInstructions } from "./session/instructions.js"
import { SessionGenerateNode } from "./session/generate-node.js"
import { SessionPromptNode } from "./session/prompt-node.js"
import { SessionRevertNode } from "./session/revert-node.js"
import { McpTool } from "./tool/mcp.js"
import { ReadToolFileSystem } from "./tool/read-filesystem.js"
import { Tool } from "./tool.js"
@@ -92,6 +94,8 @@ const nodes = [
Form.node,
Generate.node,
SessionGenerateNode.node,
SessionPromptNode.node,
SessionRevertNode.node,
ReadToolFileSystem.node,
McpTool.node,
SessionInstructions.node,
+1
View File
@@ -62,6 +62,7 @@ export const root = Effect.fn("Project.root")(function* (
export interface Interface {
readonly list: () => Effect.Effect<ReadonlyArray<Info>>
readonly update: (input: UpdateInput) => Effect.Effect<Info, NotFoundError>
/** Resolves and persists the owning Project. */
readonly resolve: (input: AbsolutePath, options?: { readonly discovery?: boolean }) => Effect.Effect<Resolved>
}
+79 -615
View File
@@ -1,7 +1,7 @@
export * as Session from "./session.js"
export * from "./session/schema.js"
import { Cause, Effect, Fiber, Layer, Schema, Context, RcMap, Stream, Scope } from "effect"
import { Cause, Effect, Layer, Schema, Context, RcMap, Stream, Scope } from "effect"
import { ListAnchor } from "@opencode-ai/schema/session"
import { and, asc, desc, eq, gt, isNull, like, lt, or, type SQL } from "drizzle-orm"
import { Project } from "./project.js"
@@ -9,7 +9,6 @@ import { Workspace } from "@opencode-ai/schema/workspace"
import { Model } from "@opencode-ai/schema/model"
import { Location } from "./location.js"
import { SessionMessage } from "./session/message.js"
import { Base64, FileAttachment, Prompt } from "@opencode-ai/schema/prompt"
import { PromptInput } from "@opencode-ai/schema/prompt-input"
import { Bus } from "./bus.js"
import { Database } from "./database/database.js"
@@ -18,17 +17,30 @@ import { SessionMessageTable, SessionTable } from "./session/sql.js"
import { SessionSchema } from "./session/schema.js"
import { AbsolutePath, PositiveInt, RelativePath } from "./schema.js"
import { Agent } from "@opencode-ai/schema/agent"
import { Money } from "@opencode-ai/schema/money"
import { App } from "./app.js"
import { Slug } from "./util/slug.js"
import { upsertProject } from "./project/sql.js"
import path from "path"
import { fromRow } from "./session/info.js"
import { SessionRunner } from "./session/runner/index.js"
import { SessionStore } from "./session/store.js"
import { SessionExecution } from "./session/execution.js"
import { SessionModelTransport } from "./session/model-transport.js"
import { ForkEmptyError, MessageDecodeError, NotFoundError } from "./session/error.js"
import {
AttachmentError,
BusyError,
CompactionConflictError,
ForkEmptyError,
InboxConflictError,
MessageDecodeError,
MessageIncompleteError,
MessageNotAssistantError,
MessageNotFoundError,
MessageToolIncompleteError,
NotFoundError,
PromptConflictError,
SkillNotFoundError,
SyntheticConflictError,
} from "./session/error.js"
import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
import { LocationServiceMap } from "./location-service-map.js"
import { SessionEvent } from "./session/event.js"
@@ -36,22 +48,15 @@ import { SessionInbox } from "./session/inbox.js"
import { InstructionState } from "./session/instruction-state.js"
import { SessionGenerate } from "./session/generate.js"
import { Snapshot } from "./snapshot.js"
import { SessionRevert } from "./session/revert.js"
import { Session } from "@opencode-ai/schema/session"
import { Session } from "./session/session.js"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Image } from "./image.js"
import { PluginSupervisor } from "./plugin/supervisor-service.js"
import { PluginHooks } from "./plugin/hooks.js"
import { Mime } from "./mime.js"
import type { EventLog } from "@opencode-ai/schema/event-log"
import { Event } from "@opencode-ai/schema/event"
import { Skill } from "./skill.js"
import { Job } from "./job.js"
import { Command } from "./command.js"
import { Shell } from "./shell.js"
import { Global } from "@opencode-ai/util/global"
import { ShellResult } from "./shell/result.js"
import { fileURLToPath } from "url"
import { SessionEnvironment } from "./session/environment.js"
import { SessionHistory } from "./session/history.js"
import { InstructionEntry } from "./session/instruction-entry.js"
@@ -102,73 +107,29 @@ type CreateBaseInput = {
type CreateInput = CreateBaseInput &
({ location: Location.Ref; parentID?: never } | { parentID: SessionSchema.ID; location?: never })
type CompactInput = {
id?: SessionMessage.ID
sessionID: SessionSchema.ID
delivery?: SessionInbox.Delivery
}
type CompactInput = Parameters<Session.Handle["compact"]>[0] & { sessionID: SessionSchema.ID }
type ForkInput = {
sessionID: SessionSchema.ID
boundary: Session.ForkRequestBoundary
boundary: SessionSchema.ForkRequestBoundary
}
export { MessageDecodeError, NotFoundError }
export class PromptConflictError extends Schema.TaggedError<PromptConflictError>()("Session.PromptConflictError", {
sessionID: SessionSchema.ID,
messageID: SessionMessage.ID,
}) {}
export class SyntheticConflictError extends Schema.TaggedError<SyntheticConflictError>()(
"Session.SyntheticConflictError",
{
sessionID: SessionSchema.ID,
inputID: SessionMessage.ID,
},
) {}
export class AttachmentError extends Schema.TaggedError<AttachmentError>()("Session.AttachmentError", {
uri: Schema.String,
message: Schema.String,
}) {}
export class CompactionConflictError extends Schema.TaggedError<CompactionConflictError>()(
"Session.CompactionConflictError",
{
sessionID: SessionSchema.ID,
inputID: SessionMessage.ID,
},
) {}
export class BusyError extends Schema.TaggedError<BusyError>()("Session.BusyError", {
sessionID: SessionSchema.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 InboxConflictError extends Schema.TaggedError<InboxConflictError>()("Session.InboxConflictError", {
sessionID: SessionSchema.ID,
inboxID: SessionMessage.ID,
}) {}
export {
AttachmentError,
BusyError,
CompactionConflictError,
InboxConflictError,
MessageDecodeError,
MessageIncompleteError,
MessageNotAssistantError,
MessageNotFoundError,
MessageToolIncompleteError,
NotFoundError,
PromptConflictError,
SkillNotFoundError,
SyntheticConflictError,
}
type InboxItemRef = { readonly sessionID: SessionSchema.ID; readonly inboxID: SessionMessage.ID }
export class SkillNotFoundError extends Schema.TaggedError<SkillNotFoundError>()("Session.SkillNotFoundError", {
skill: Skill.ID,
}) {}
export class DestinationNotFoundError extends Schema.TaggedError<DestinationNotFoundError>()(
"Session.DestinationNotFoundError",
@@ -184,8 +145,6 @@ export class DestinationUnavailableError extends Schema.TaggedError<DestinationU
"Session.DestinationUnavailableError",
{ directory: AbsolutePath },
) {}
export const MessageNotFoundError = SessionRevert.MessageNotFoundError
export type MessageNotFoundError = SessionRevert.MessageNotFoundError
export interface Interface {
readonly list: (input?: ListInput) => Effect.Effect<{
@@ -215,19 +174,9 @@ export interface Interface {
sessionID: SessionSchema.ID
messageID: SessionMessage.ID
}) => Effect.Effect<SessionMessage.Info | undefined>
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 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>
@@ -264,17 +213,9 @@ export interface Interface {
void,
NotFoundError | DestinationNotFoundError | DestinationNotDirectoryError | DestinationUnavailableError
>
readonly prompt: (input: {
id?: SessionMessage.ID
sessionID: SessionSchema.ID
text: string
files?: PromptInput.Prompt["files"]
agents?: PromptInput.Prompt["agents"]
skills?: PromptInput.Prompt["skills"]
metadata?: Record<string, unknown>
delivery?: SessionInbox.Delivery
resume?: boolean
}) => Effect.Effect<SessionInbox.User, NotFoundError | PromptConflictError | AttachmentError | SkillNotFoundError>
readonly prompt: (
input: Parameters<Session.Handle["prompt"]>[0] & { sessionID: SessionSchema.ID },
) => ReturnType<Session.Handle["prompt"]>
/** Generates text from current Session context without admitting input or mutating history. */
readonly generate: (input: {
sessionID: SessionSchema.ID
@@ -289,11 +230,9 @@ export interface Interface {
skills?: PromptInput.Prompt["skills"]
delivery?: SessionInbox.Delivery
}) => Effect.Effect<void, NotFoundError | Command.NotFoundError | Command.ExecutionError>
readonly shell: (input: {
id?: Event.ID
sessionID: SessionSchema.ID
command: string
}) => Effect.Effect<void, NotFoundError>
readonly shell: (
input: Parameters<Session.Handle["shell"]>[0] & { sessionID: SessionSchema.ID },
) => ReturnType<Session.Handle["shell"]>
readonly skill: (input: {
id?: SessionMessage.ID
sessionID: SessionSchema.ID
@@ -308,21 +247,15 @@ export interface Interface {
readonly background: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError>
readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError | SessionRunner.RunError>
readonly interrupt: (sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) => Effect.Effect<boolean>
readonly synthetic: (input: {
id?: SessionMessage.ID
sessionID: SessionSchema.ID
text: string
description?: string
metadata?: Record<string, unknown>
delivery?: SessionInbox.Delivery
resume?: boolean
}) => Effect.Effect<SessionInbox.Synthetic, NotFoundError | SyntheticConflictError>
readonly synthetic: (
input: Parameters<Session.Handle["synthetic"]>[0] & { sessionID: SessionSchema.ID },
) => ReturnType<Session.Handle["synthetic"]>
readonly revert: {
readonly stage: (input: {
sessionID: SessionSchema.ID
messageID: SessionMessage.ID
files?: boolean
}) => Effect.Effect<Session.Revert, NotFoundError | MessageNotFoundError | BusyError | Snapshot.Error>
}) => Effect.Effect<SessionSchema.Revert, NotFoundError | MessageNotFoundError | BusyError | Snapshot.Error>
readonly clear: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError | BusyError | Snapshot.Error>
readonly commit: (sessionID: SessionSchema.ID) => Effect.Effect<void, NotFoundError | BusyError>
}
@@ -346,6 +279,8 @@ const layer = Layer.effect(
const jobs = yield* Job.Service
const environments = yield* SessionEnvironment.Service
const scope = yield* Scope.Scope
const sessions = yield* Session.make((ref) => locations.get(ref))
const admission = yield* SessionInbox.Service
const closeTransport = Effect.fn("Session.closeTransport")(function* (session: SessionSchema.Info) {
const location = Location.Ref.make({
directory: session.location.directory,
@@ -357,30 +292,6 @@ const layer = Layer.effect(
)
})
const isDurableSessionEvent = Schema.is(SessionEvent.Durable)
const persistProject = (project: Project.Resolved) => upsertProject(db, project).pipe(Effect.orDie)
const pendingConflict = Effect.fn("Session.pendingConflict")(function* (input: InboxItemRef) {
yield* result.get(input.sessionID)
return yield* new InboxConflictError(input)
})
const mutatePending = (
input: InboxItemRef,
mutation: (
bus: Bus.Interface,
input: { readonly id: SessionMessage.ID; readonly sessionID: SessionSchema.ID },
) => Effect.Effect<unknown>,
wake = false,
) =>
Effect.uninterruptible(
Effect.gen(function* () {
yield* mutation(bus, { sessionID: input.sessionID, id: input.inboxID }).pipe(
Effect.catchDefect((defect) =>
defect instanceof SessionInbox.LifecycleConflict ? pendingConflict(input) : Effect.die(defect),
),
)
if (wake) yield* execution.wake(input.sessionID)
}),
)
const result = Service.of({
create: Effect.fn("Session.create")(function* (input) {
@@ -393,7 +304,6 @@ const layer = Layer.effect(
if (location === undefined)
return yield* Effect.die(new Error("Session.create requires either location or an existing parentID"))
const project = yield* projects.resolve(location.directory)
yield* persistProject(project)
const projected = yield* bus
.publish(
SessionEvent.Created,
@@ -481,28 +391,13 @@ const layer = Layer.effect(
})
return yield* result.get(sessionID).pipe(Effect.orDie)
}),
get: Effect.fn("Session.get")(function* (sessionID) {
const session = yield* store.get(sessionID)
if (!session) return yield* new NotFoundError({ sessionID })
return session
}),
get: (sessionID) => sessions.forSession(sessionID).get(),
environment: Effect.fn("Session.environment")(function* (input) {
yield* result.get(input.sessionID)
if (input.variables !== undefined) yield* environments.set(input.sessionID, input.variables)
return yield* environments.get(input.sessionID)
}),
view: Effect.fn("Session.view")(function* (input) {
const row = yield* db
.select({ idle: SessionTable.time_idle, viewed: SessionTable.time_viewed })
.from(SessionTable)
.where(eq(SessionTable.id, input.sessionID))
.get()
.pipe(Effect.orDie)
if (!row) return yield* new NotFoundError({ sessionID: input.sessionID })
if (row.idle === null || input.idle > row.idle || (row.viewed !== null && row.viewed >= input.idle))
return yield* Effect.void
yield* bus.publish(SessionEvent.Viewed, { sessionID: input.sessionID, idle: input.idle })
}),
view: (input) => sessions.forSession(input.sessionID).view(input),
remove: Effect.fn("Session.remove")(function* (sessionID) {
const session = yield* result.get(sessionID)
yield* execution.interrupt(sessionID)
@@ -592,44 +487,16 @@ const layer = Layer.effect(
SessionHistory.decodeMessageRow,
)
}),
message: Effect.fn("Session.message")(function* (input) {
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
}),
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)
}),
inbox: Effect.fn("Session.inbox")(function* (sessionID) {
yield* result.get(sessionID)
return yield* SessionInbox.list(db, sessionID)
}),
cancelInbox: Effect.fn("Session.cancelInbox")((input) => mutatePending(input, SessionInbox.cancel)),
steerInbox: Effect.fn("Session.steerInbox")((input) => mutatePending(input, SessionInbox.steer, true)),
queueInbox: Effect.fn("Session.queueInbox")((input) => mutatePending(input, SessionInbox.queue)),
inbox: (sessionID) => sessions.forSession(sessionID).inbox(),
cancelInbox: (input) => sessions.forSession(input.sessionID).cancelInbox(input.inboxID),
steerInbox: (input) => sessions.forSession(input.sessionID).steerInbox(input.inboxID),
queueInbox: (input) => sessions.forSession(input.sessionID).queueInbox(input.inboxID),
log: (input) =>
Stream.unwrap(
result
@@ -641,47 +508,7 @@ const layer = Layer.effect(
Bus.isSynced(item) || isDurableSessionEvent(item),
),
),
prompt: Effect.fn("Session.prompt")((input) =>
Effect.uninterruptibleMask((restore) =>
Effect.gen(function* () {
const session = yield* result.get(input.sessionID)
const messageID = input.id ?? SessionMessage.ID.create()
const admitted = yield* Effect.gen(function* () {
const existing = yield* SessionInbox.reconcile(db, {
id: messageID,
sessionID: session.id,
delivery: input.delivery ?? "steer",
})
if (existing) return existing
const item = yield* restore(
preparePrompt(input, messageID).pipe(
Effect.provide(locations.get(session.location)),
Effect.provideService(FSUtil.Service, fs),
),
)
// Commit a staged revert only after preparation succeeds, before admitting new work.
if (session.revert) yield* SessionRevert.commit(session).pipe(Effect.provideService(Bus.Service, bus))
return yield* SessionInbox.admit(db, bus, {
id: messageID,
sessionID: session.id,
item,
})
}).pipe(
Effect.catchDefect((defect) =>
defect instanceof SessionInbox.LifecycleConflict
? new PromptConflictError({ sessionID: input.sessionID, messageID })
: Effect.die(defect),
),
)
// First admission wins: same-session reuse is idempotent and ignores the
// retried payload, metadata, and delivery mode.
if (admitted.type !== "user" || admitted.sessionID !== input.sessionID)
return yield* new PromptConflictError({ sessionID: input.sessionID, messageID })
if (input.resume !== false) yield* execution.wake(admitted.sessionID)
return admitted
}),
),
),
prompt: (input) => sessions.forSession(input.sessionID).prompt(input),
generate: Effect.fn("Session.generate")(function* (input) {
const session = yield* result.get(input.sessionID)
const generate = yield* SessionGenerate.Service.pipe(Effect.provide(locations.get(session.location)))
@@ -709,65 +536,7 @@ const layer = Layer.effect(
},
})
}),
shell: Effect.fn("Session.shell")(function* (input) {
const session = yield* result.get(input.sessionID)
// The server owns completion recording even if the submitting client disconnects.
const running = yield* Effect.gen(function* () {
// Resolve shell services here without pinning Session events to this Location after a move.
const shell = yield* Effect.gen(function* () {
const plugins = yield* PluginSupervisor.Service
yield* plugins.flush
return yield* Shell.Service
}).pipe(Effect.provide(locations.get(session.location)))
const started = yield* shell
.create({
command: input.command,
cwd: session.location.directory,
timeout: 0,
metadata: { sessionID: input.sessionID, background: true },
})
.pipe(
Effect.tapError((error) =>
result.synthetic({
sessionID: input.sessionID,
text: `User shell command failed to start:\n${input.command}\n\n${error.message}`,
description: input.command,
metadata: { source: "shell", state: "error" },
resume: false,
}),
),
Effect.orDie,
)
yield* bus.publish(
SessionEvent.Shell.Started,
{
sessionID: input.sessionID,
shell: started,
},
{ id: input.id },
)
const terminal = yield* shell.result(started)
const preview = yield* shell
.output(started.id, { limit: SHELL_MAX_CAPTURE_BYTES })
.pipe(Effect.catchTag("Shell.NotFoundError", () => Effect.succeed(ShellResult.unavailable)))
yield* bus.publish(SessionEvent.Shell.Ended, {
sessionID: input.sessionID,
shell: terminal.info,
output: preview,
})
yield* result
.synthetic({
...ShellResult.userNotification(terminal),
sessionID: input.sessionID,
resume: false,
})
.pipe(
Effect.catchTag("Session.NotFoundError", () => Effect.void),
Effect.orDie,
)
}).pipe(Effect.forkIn(scope, { startImmediately: true }))
yield* Fiber.join(running)
}),
shell: (input) => sessions.forSession(input.sessionID).shell(input),
skill: Effect.fn("Session.skill")(function* (input) {
const session = yield* result.get(input.sessionID)
const skills = yield* Skill.Service.pipe(Effect.provide(locations.get(session.location)))
@@ -788,35 +557,9 @@ const layer = Layer.effect(
.resume(input.sessionID)
.pipe(Effect.ignore, Effect.forkIn(scope, { startImmediately: true }), Effect.asVoid)
}),
switchAgent: Effect.fn("Session.switchAgent")(function* (input) {
const session = yield* result.get(input.sessionID)
yield* bus.publish(SessionEvent.AgentSelected, {
sessionID: input.sessionID,
agent: input.agent,
previous: session.agent,
})
}),
switchModel: Effect.fn("Session.switchModel")(function* (input) {
const session = yield* result.get(input.sessionID)
if (
session.model?.providerID === input.model.providerID &&
session.model.id === input.model.id &&
(session.model.variant ?? "default") === (input.model.variant ?? "default")
)
return
yield* bus.publish(SessionEvent.ModelSelected, {
sessionID: input.sessionID,
model: input.model,
previous: session.model,
})
}),
rename: Effect.fn("Session.rename")(function* (input) {
yield* result.get(input.sessionID)
yield* bus.publish(SessionEvent.Renamed, {
sessionID: input.sessionID,
title: input.title,
})
}),
switchAgent: (input) => sessions.forSession(input.sessionID).switchAgent(input),
switchModel: (input) => sessions.forSession(input.sessionID).switchModel(input),
rename: (input) => sessions.forSession(input.sessionID).rename(input),
move: Effect.fn("Session.move")(function* (input) {
const current = yield* result.get(input.sessionID)
const value = input.directory.trim()
@@ -842,7 +585,6 @@ const layer = Layer.effect(
)
}),
)
yield* persistProject(project)
const item = SessionInbox.Item.make({
type: "move",
payload,
@@ -862,36 +604,19 @@ const layer = Layer.effect(
if (!first) return yield* bus.publish(...moved).pipe(Effect.asVoid)
return yield* bus.publishAll([first, ...cancellations.slice(1), moved])
}
yield* SessionInbox.admit(db, bus, {
id: SessionMessage.ID.create(),
sessionID: input.sessionID,
item,
})
yield* admission
.admit({
id: SessionMessage.ID.create(),
sessionID: input.sessionID,
item,
})
.pipe(Effect.orDie)
}),
)
yield* execution.wake(input.sessionID)
}),
compact: Effect.fn("Session.compact")(function* (input) {
yield* result.get(input.sessionID)
const inputID = input.id ?? SessionMessage.ID.create()
const admitted = yield* SessionInbox.admitCompaction(db, bus, {
id: inputID,
sessionID: input.sessionID,
delivery: input.delivery ?? "steer",
}).pipe(
Effect.catchDefect((defect) =>
defect instanceof SessionInbox.LifecycleConflict
? new CompactionConflictError({ sessionID: input.sessionID, inputID })
: Effect.die(defect),
),
)
yield* execution.wake(input.sessionID)
return admitted
}),
wait: Effect.fn("Session.wait")(function* (sessionID) {
yield* result.get(sessionID)
yield* execution.awaitIdle(sessionID)
}),
compact: (input) => sessions.forSession(input.sessionID).compact(input),
wait: (sessionID) => sessions.forSession(sessionID).wait(),
active: execution.active,
background: Effect.fn("Session.background")(function* (sessionID) {
yield* result.get(sessionID)
@@ -911,78 +636,13 @@ const layer = Layer.effect(
})
.pipe(Effect.catchTag("Session.SyntheticConflictError", Effect.die))
}),
resume: Effect.fn("Session.resume")(function* (sessionID) {
yield* result.get(sessionID)
yield* execution.resume(sessionID)
}),
synthetic: Effect.fn("Session.synthetic")((input) =>
Effect.uninterruptible(
Effect.gen(function* () {
yield* result.get(input.sessionID)
const inputID = input.id ?? SessionMessage.ID.create()
const admittedInput = SessionInbox.Item.make({
type: "synthetic",
payload: {
text: input.text,
description: input.description,
metadata: input.metadata,
},
delivery: input.delivery ?? "steer",
})
const admitted = yield* SessionInbox.admit(db, bus, {
id: inputID,
sessionID: input.sessionID,
item: admittedInput,
}).pipe(
Effect.catchDefect((defect) =>
defect instanceof SessionInbox.LifecycleConflict
? new SyntheticConflictError({ sessionID: input.sessionID, inputID })
: Effect.die(defect),
),
)
// First admission wins: same-session reuse is idempotent and ignores the
// retried payload, metadata, and delivery mode.
if (admitted.type !== "synthetic" || admitted.sessionID !== input.sessionID)
return yield* new SyntheticConflictError({ sessionID: input.sessionID, inputID })
if (input.resume !== false && !(yield* result.get(input.sessionID)).revert)
yield* execution.wake(input.sessionID)
return admitted
}),
),
),
interrupt: Effect.fn("Session.interrupt")((sessionID, options) =>
Effect.uninterruptible(execution.interrupt(sessionID, options)),
),
resume: (sessionID) => sessions.forSession(sessionID).resume(),
synthetic: (input) => sessions.forSession(input.sessionID).synthetic(input),
interrupt: (sessionID, options) => sessions.forSession(sessionID).interrupt(options),
revert: {
stage: Effect.fn("Session.revert.stage")(function* (input) {
const session = yield* result.get(input.sessionID)
if ((yield* execution.active).has(input.sessionID))
return yield* new BusyError({ sessionID: input.sessionID })
return yield* Effect.gen(function* () {
const plugins = yield* PluginSupervisor.Service
yield* plugins.flush
return yield* SessionRevert.stage({ session, messageID: input.messageID, files: input.files }).pipe(
Effect.provideService(Database.Service, database),
Effect.provideService(Bus.Service, bus),
)
}).pipe(Effect.provide(locations.get(session.location)))
}),
clear: Effect.fn("Session.revert.clear")(function* (sessionID) {
const session = yield* result.get(sessionID)
if ((yield* execution.active).has(sessionID)) return yield* new BusyError({ sessionID })
const revert = yield* Effect.gen(function* () {
const plugins = yield* PluginSupervisor.Service
yield* plugins.flush
return yield* SessionRevert.clear(session).pipe(Effect.provideService(Bus.Service, bus))
}).pipe(Effect.provide(locations.get(session.location)))
yield* execution.wake(sessionID)
return revert
}),
commit: Effect.fn("Session.revert.commit")(function* (sessionID) {
const session = yield* result.get(sessionID)
if ((yield* execution.active).has(sessionID)) return yield* new BusyError({ sessionID })
return yield* SessionRevert.commit(session).pipe(Effect.provideService(Bus.Service, bus))
}),
stage: (input) => sessions.forSession(input.sessionID).revert.stage(input),
clear: (sessionID) => sessions.forSession(sessionID).revert.clear(),
commit: (sessionID) => sessions.forSession(sessionID).revert.commit(),
},
})
@@ -990,203 +650,6 @@ const layer = Layer.effect(
}),
)
const preparePrompt = Effect.fn("Session.preparePrompt")(function* (
request: Parameters<Interface["prompt"]>[0],
messageID: SessionMessage.ID,
) {
const plugins = yield* PluginSupervisor.Service
yield* plugins.flush
const hooks = yield* PluginHooks.Service
const event = yield* hooks.trigger("session", "prompt", {
sessionID: request.sessionID,
messageID,
prompt: structuredClone({
text: request.text,
files: request.files?.slice(),
agents: request.agents?.slice(),
skills: request.skills?.slice(),
}),
metadata: structuredClone(request.metadata),
delivery: request.delivery ?? "steer",
})
const input = event.prompt
const fs = yield* FSUtil.Service
const files = input.files
? yield* Effect.forEach(input.files, (file) => materializeAttachment(fs, file), { concurrency: 8 })
: undefined
const requested = input.skills
const selected = yield* Effect.gen(function* () {
if (!requested?.length) return undefined
const skillService = yield* Skill.Service
const prepared = new Map<Skill.ID, Skill.Name>()
return yield* Effect.forEach(requested, (attachment) =>
Effect.gen(function* () {
const name = prepared.get(attachment.id)
if (name !== undefined) return { id: attachment.id, name, mention: attachment.mention }
const skill = yield* skillService.get(attachment.id)
if (!skill) return yield* new SkillNotFoundError({ skill: attachment.id })
prepared.set(skill.id, skill.name)
return {
id: skill.id,
name: skill.name,
text: (yield* Skill.prepare(fs, skill).pipe(Effect.orDie)).output,
mention: attachment.mention,
}
}),
)
})
return SessionInbox.Item.make({
type: "user",
payload: {
...Prompt.make({
text: input.text,
agents: input.agents,
files,
skills: selected?.length ? selected : undefined,
}),
metadata: event.metadata,
},
delivery: event.delivery,
})
})
const MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024
const materializeAttachment = Effect.fn("Session.materializeAttachment")(function* (
fs: FSUtil.Interface,
input: PromptInput.FileAttachment,
) {
const resolved = input.uri.startsWith("data:")
? {
bytes: yield* decodeDataURL(input.uri),
source: { type: "inline" as const },
start: undefined,
end: undefined,
name: undefined,
mime: undefined,
}
: yield* readFileAttachment(fs, input.uri)
if (resolved.bytes.byteLength > MAX_ATTACHMENT_BYTES)
return yield* new AttachmentError({
uri: input.uri,
message: `Attachment exceeds the ${MAX_ATTACHMENT_BYTES} byte limit: ${input.uri}`,
})
const mime = resolved.mime ?? Mime.detect(resolved.bytes)
const content =
mime === "text/plain" && resolved.start !== undefined
? Buffer.from(
Buffer.from(resolved.bytes)
.toString("utf8")
.split("\n")
.slice(resolved.start - 1, resolved.end)
.join("\n"),
)
: resolved.bytes
const normalized = yield* normalizeImageAttachment(input, Buffer.from(content).toString("base64"), mime)
return FileAttachment.create({
data: normalized.data,
mime: normalized.mime,
source: resolved.source,
name: input.name ?? resolved.name,
description: input.description,
mention: input.mention,
})
})
const normalizeImageAttachment = Effect.fn("Session.normalizeImageAttachment")(function* (
input: PromptInput.FileAttachment,
data: string,
mime: string,
) {
if (!mime.startsWith("image/")) return { data: Base64.make(data), mime }
const service = yield* Image.Service
const label = input.name ?? (input.uri.startsWith("data:") ? "inline attachment" : input.uri)
const content = { uri: label, content: data, encoding: "base64" as const, mime }
const normalized = yield* service.normalize(label, content).pipe(
Effect.catchTag("Image.ResizerUnavailableError", () => Effect.succeed(content)),
Effect.mapError((error) => new AttachmentError({ uri: label, message: error.message })),
)
return { data: Base64.make(normalized.content), mime: normalized.mime }
})
const readFileAttachment = Effect.fn("Session.readFileAttachment")(function* (fs: FSUtil.Interface, uri: string) {
const url = yield* Effect.try({
try: () => new URL(uri),
catch: () => new AttachmentError({ uri, message: `Invalid attachment URI: ${uri}` }),
})
if (url.protocol !== "file:")
return yield* new AttachmentError({ uri, message: `Unsupported attachment URI: ${uri}` })
const start = positiveInt(url.searchParams.get("start"))
const end = positiveInt(url.searchParams.get("end"))
const target = yield* Effect.try({
try: () => {
url.search = ""
url.hash = ""
return fileURLToPath(url)
},
catch: () => new AttachmentError({ uri, message: `Invalid file URI: ${uri}` }),
})
const info = yield* fs
.stat(target)
.pipe(Effect.mapError(() => new AttachmentError({ uri, message: `Unable to read attachment: ${uri}` })))
if (info.type === "Directory") {
const entries = yield* fs
.readDirectoryEntries(target)
.pipe(Effect.mapError(() => new AttachmentError({ uri, message: `Unable to read attachment: ${uri}` })))
return {
bytes: Buffer.from(
entries
.filter((entry) => entry.type === "file" || entry.type === "directory")
.sort((a, b) => (a.type === b.type ? a.name.localeCompare(b.name) : a.type === "directory" ? -1 : 1))
.map((entry) => entry.name + (entry.type === "directory" ? path.sep : ""))
.join("\n"),
),
source: { type: "uri" as const, uri },
start: undefined,
end: undefined,
name: path.basename(target),
mime: "application/x-directory",
}
}
if (info.type !== "File") return yield* new AttachmentError({ uri, message: `Attachment is not a file: ${uri}` })
if (Number(info.size) > MAX_ATTACHMENT_BYTES)
return yield* new AttachmentError({
uri,
message: `Attachment exceeds the ${MAX_ATTACHMENT_BYTES} byte limit: ${uri}`,
})
const bytes = yield* fs
.readFile(target)
.pipe(Effect.mapError(() => new AttachmentError({ uri, message: `Unable to read attachment: ${uri}` })))
return { bytes, source: { type: "uri" as const, uri }, start, end, name: path.basename(target), mime: undefined }
})
function decodeDataURL(uri: string) {
return Effect.try({
try: () => {
const comma = uri.indexOf(",")
if (comma === -1) throw new Error("Invalid data URL")
const metadata = uri.slice(5, comma)
const payload = uri.slice(comma + 1)
if (!metadata.split(";").some((part) => part.toLowerCase() === "base64"))
return Buffer.from(decodeURIComponent(payload))
const bytes = Buffer.from(payload, "base64")
if (bytes.toString("base64") !== payload) throw new Error("Non-canonical base64")
return bytes
},
catch: () => new AttachmentError({ uri, message: "Invalid attachment data URL" }),
})
}
function positiveInt(value: string | null) {
if (value === null) return
const parsed = Number(value)
return Number.isInteger(parsed) && parsed > 0 ? parsed : undefined
}
// Mirrors the shell tool's in-memory preview safety limit.
const SHELL_MAX_CAPTURE_BYTES = 1024 * 1024
export const node = makeGlobalNode({
service: Service,
layer,
@@ -1198,6 +661,7 @@ export const node = makeGlobalNode({
Project.node,
SessionExecution.node,
SessionStore.node,
SessionInbox.node,
LocationServiceMap.node,
SessionProjector.node,
FSUtil.node,
+69
View File
@@ -2,6 +2,7 @@ export * as SessionErrors from "./error.js"
import { Schema } from "effect"
import { Agent } from "@opencode-ai/schema/agent"
import { Skill } from "@opencode-ai/schema/skill"
import { SessionMessage } from "./message.js"
import { SessionSchema } from "./schema.js"
import { SessionError } from "@opencode-ai/schema/session-error"
@@ -10,6 +11,35 @@ export class NotFoundError extends Schema.TaggedError<NotFoundError>()("Session.
sessionID: SessionSchema.ID,
}) {}
export class MessageNotFoundError extends Schema.TaggedError<MessageNotFoundError>()("Session.MessageNotFoundError", {
sessionID: SessionSchema.ID,
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,
}) {
@@ -52,3 +82,42 @@ export class UserInterruptedError extends Schema.TaggedError<UserInterruptedErro
return "Session interrupted by user"
}
}
export class PromptConflictError extends Schema.TaggedError<PromptConflictError>()("Session.PromptConflictError", {
sessionID: SessionSchema.ID,
messageID: SessionMessage.ID,
}) {}
export class SyntheticConflictError extends Schema.TaggedError<SyntheticConflictError>()(
"Session.SyntheticConflictError",
{
sessionID: SessionSchema.ID,
inputID: SessionMessage.ID,
},
) {}
export class AttachmentError extends Schema.TaggedError<AttachmentError>()("Session.AttachmentError", {
uri: Schema.String,
message: Schema.String,
}) {}
export class CompactionConflictError extends Schema.TaggedError<CompactionConflictError>()(
"Session.CompactionConflictError",
{
sessionID: SessionSchema.ID,
inputID: SessionMessage.ID,
},
) {}
export class BusyError extends Schema.TaggedError<BusyError>()("Session.BusyError", {
sessionID: SessionSchema.ID,
}) {}
export class InboxConflictError extends Schema.TaggedError<InboxConflictError>()("Session.InboxConflictError", {
sessionID: SessionSchema.ID,
inboxID: SessionMessage.ID,
}) {}
export class SkillNotFoundError extends Schema.TaggedError<SkillNotFoundError>()("Session.SkillNotFoundError", {
skill: Skill.ID,
}) {}
+4
View File
@@ -18,6 +18,8 @@ import { SessionInbox } from "./inbox.js"
export interface Interface {
/** Snapshots active execution owned by this process. */
readonly active: Effect.Effect<ReadonlySet<SessionSchema.ID>>
/** Checks process-local ownership, including interruption cleanup and terminal settlement. */
readonly isActive: (sessionID: SessionSchema.ID) => Effect.Effect<boolean>
/** Starts execution while idle or joins the active execution. */
readonly resume: (sessionID: SessionSchema.ID) => Effect.Effect<void, SessionRunner.RunError>
/** Registers newly recorded work. Repeated wakeups may coalesce. */
@@ -142,6 +144,7 @@ export const layer = Layer.effect(
return Service.of({
active: coordinator.active,
isActive: coordinator.isActive,
interrupt: (sessionID, options) =>
Effect.gen(function* () {
const interrupted = yield* coordinator.interrupt(sessionID, "user")
@@ -178,6 +181,7 @@ export const noopLayer = Layer.succeed(
Service,
Service.of({
active: Effect.succeed(new Set()),
isActive: () => Effect.succeed(false),
resume: () => Effect.void,
wake: () => Effect.void,
interrupt: () => Effect.succeed(false),
@@ -157,7 +157,7 @@ export const layer = (options?: Options) =>
yield* notify(background)
return
}
if ((yield* execution.active).has(recovery.childSessionID)) return
if (yield* execution.isActive(recovery.childSessionID)) return
if (!(yield* prepareResume(recovery.childSessionID))) {
yield* notify({ status: "error", error: RESUME_EXHAUSTED.message })
return
+143 -111
View File
@@ -1,7 +1,8 @@
export * as SessionInbox from "./inbox.js"
import { and, asc, eq, or } from "drizzle-orm"
import { DateTime, Effect, Schema } from "effect"
import { Context, DateTime, Effect, Layer, Schema } from "effect"
import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
import {
Compaction,
CompactionPayload,
@@ -15,7 +16,7 @@ import {
User,
UserPayload,
} from "@opencode-ai/schema/session-inbox"
import type { Database } from "../database/database.js"
import { Database } from "../database/database.js"
import { Bus } from "../bus.js"
import { KeyedMutex } from "../effect/keyed-mutex.js"
import { SessionEvent } from "./event.js"
@@ -65,6 +66,13 @@ export class LifecycleConflict extends Schema.TaggedError<LifecycleConflict>()("
id: SessionMessage.ID,
}) {}
function matches<Type extends Item["type"]>(
stored: Info,
request: PendingRef & { readonly type: Type },
): stored is Extract<Info, { readonly type: Type }> {
return stored.sessionID === request.sessionID && stored.type === request.type
}
const fromRow = (row: typeof SessionInboxTable.$inferSelect): Info => {
const base = {
id: SessionMessage.ID.make(row.id),
@@ -116,7 +124,7 @@ const promotedFromMessage = Effect.fn("SessionInbox.promotedFromMessage")(functi
.pipe(Effect.orDie)
if (row === undefined) return undefined
if (row.session_id !== sessionID || (row.type !== "user" && row.type !== "synthetic"))
return yield* Effect.die(new LifecycleConflict({ id }))
return yield* new LifecycleConflict({ id })
const message = decodeMessage({ ...row.data, id: row.id, type: row.type })
const base = { id, sessionID, timeCreated: message.time.created, delivery }
if (message.type === "user")
@@ -131,89 +139,140 @@ const promotedFromMessage = Effect.fn("SessionInbox.promotedFromMessage")(functi
type: "synthetic",
payload: decodeSynthetic(message),
})
return yield* Effect.die(new LifecycleConflict({ id }))
return yield* new LifecycleConflict({ id })
})
/** Reconciles pending or delivered work without preparing a new admission payload. */
export const reconcile = Effect.fn("SessionInbox.reconcile")(function* (
db: DatabaseService,
request: {
export type Interface = Effect.Success<ReturnType<typeof make>>
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionInbox") {}
export const make = Effect.fn("SessionInbox.make")(function* () {
const database = yield* Database.Service
const db = database.db
const bus = yield* Bus.Service
/** First admission wins for matching Session and type, without preparing a new payload. */
const reconcile = Effect.fn("SessionInbox.reconcile")(function* <Type extends Item["type"]>(request: {
readonly id: SessionMessage.ID
readonly sessionID: SessionSchema.ID
readonly type: Type
readonly delivery: Delivery
}) {
const existing =
(yield* find(db, request.id)) ?? (yield* promotedFromMessage(db, request.sessionID, request.id, request.delivery))
if (existing === undefined) return undefined
if (existing.type === "compaction" || !matches(existing, request))
return yield* new LifecycleConflict({ id: request.id })
return existing
})
const admit = Effect.fn("SessionInbox.admit")(function* <Type extends Item["type"]>(request: {
readonly id: SessionMessage.ID
readonly sessionID: SessionSchema.ID
readonly item: Item & { readonly type: Type }
}) {
const existing = yield* reconcile({ ...request, type: request.item.type, delivery: request.item.delivery })
if (existing !== undefined) return existing
const admitted = yield* bus
.publish(SessionEvent.InboxEnqueued, {
inboxID: request.id,
sessionID: request.sessionID,
item: request.item,
})
.pipe(
Effect.map((event) =>
Info.make({
id: request.id,
sessionID: request.sessionID,
timeCreated: DateTime.makeUnsafe(event.created),
...request.item,
}),
),
Effect.catchDefect((defect) =>
defect instanceof LifecycleConflict
? find(db, request.id).pipe(
Effect.flatMap((stored) => (stored === undefined ? Effect.fail(defect) : Effect.succeed(stored))),
)
: Effect.die(defect),
),
)
if (!matches(admitted, { ...request, type: request.item.type }))
return yield* new LifecycleConflict({ id: request.id })
return admitted
})
const admitCompaction = Effect.fn("SessionInbox.admitCompaction")(function* (input: {
readonly id: SessionMessage.ID
readonly sessionID: SessionSchema.ID
readonly delivery: Delivery
},
) {
const existing = yield* find(db, request.id)
if (existing !== undefined) {
if (existing.type === "compaction") return yield* Effect.die(new LifecycleConflict({ id: request.id }))
return existing
}
return yield* promotedFromMessage(db, request.sessionID, request.id, request.delivery)
})
export const admit = Effect.fn("SessionInbox.admit")(function* (
db: DatabaseService,
bus: Bus.Interface,
request: {
readonly id: SessionMessage.ID
readonly sessionID: SessionSchema.ID
readonly item: Item
},
) {
const existing = yield* reconcile(db, { ...request, delivery: request.item.delivery })
if (existing !== undefined) return existing
return yield* bus
.publish(SessionEvent.InboxEnqueued, {
inboxID: request.id,
sessionID: request.sessionID,
item: request.item,
})
.pipe(
Effect.map((event) =>
Info.make({
id: request.id,
sessionID: request.sessionID,
timeCreated: DateTime.makeUnsafe(event.created),
...request.item,
}),
),
Effect.catchDefect((defect) =>
find(db, request.id).pipe(
Effect.flatMap((stored) =>
stored?.type === request.item.type ? Effect.succeed(stored) : Effect.die(defect),
),
),
),
}) {
return yield* serialized(
input.sessionID,
Effect.gen(function* () {
const exact = yield* find(db, input.id)
if (exact) {
if (exact.type === "compaction" && exact.sessionID === input.sessionID) return exact
return yield* new LifecycleConflict({ id: input.id })
}
if (yield* promotedFromMessage(db, input.sessionID, input.id, input.delivery))
return yield* new LifecycleConflict({ id: input.id })
const pending = (yield* list(db, input.sessionID)).find((item) => item.type === "compaction")
if (pending) return pending
return yield* admit({
id: input.id,
sessionID: input.sessionID,
item: { type: "compaction", payload: {}, delivery: Delivery.make(input.delivery) },
})
}),
)
})
const cancel = Effect.fn("SessionInbox.cancel")((input: PendingRef) =>
publishMutation(
input,
bus.publish(SessionEvent.InboxCancelled, {
sessionID: input.sessionID,
inboxID: input.id,
}),
),
)
const steer = Effect.fn("SessionInbox.steer")((input: PendingRef) =>
publishMutation(
input,
bus.publish(SessionEvent.InboxDeliveryChanged, {
sessionID: input.sessionID,
inboxID: input.id,
delivery: "steer",
}),
),
)
const queue = Effect.fn("SessionInbox.queue")((input: PendingRef) =>
publishMutation(
input,
bus.publish(SessionEvent.InboxDeliveryChanged, {
sessionID: input.sessionID,
inboxID: input.id,
delivery: "queue",
}),
),
)
return {
list: (sessionID: SessionSchema.ID) => list(db, sessionID),
reconcile,
admit,
admitCompaction,
cancel,
steer,
queue,
}
})
export const admitCompaction = Effect.fn("SessionInbox.admitCompaction")(function* (
db: DatabaseService,
bus: Bus.Interface,
input: { readonly id: SessionMessage.ID; readonly sessionID: SessionSchema.ID; readonly delivery: Delivery },
) {
return yield* serialized(
input.sessionID,
Effect.gen(function* () {
const exact = yield* find(db, input.id)
if (exact) {
if (exact.type === "compaction" && exact.sessionID === input.sessionID) return exact
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
}
if (yield* promotedFromMessage(db, input.sessionID, input.id, input.delivery))
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
const pending = (yield* list(db, input.sessionID)).find((item) => item.type === "compaction")
if (pending) return pending
const admitted = yield* admit(db, bus, {
id: input.id,
sessionID: input.sessionID,
item: Item.make({ type: "compaction", payload: {}, delivery: input.delivery }),
})
if (admitted.type === "compaction") return admitted
return yield* Effect.die(new LifecycleConflict({ id: input.id }))
}),
)
})
export const layer = Layer.effect(Service, make())
export const node = makeGlobalNode({ service: Service, layer, deps: [Database.node, Bus.node] })
export const projectAdmitted = Effect.fn("SessionInbox.projectAdmitted")(function* (
db: DatabaseService,
@@ -392,39 +451,11 @@ export const has = Effect.fn("SessionInbox.has")(function* (
})
const publishMutation = <A, E, R>(input: PendingRef, effect: Effect.Effect<A, E, R>) =>
serialized(input.sessionID, effect).pipe(Effect.asVoid)
export const cancel = Effect.fn("SessionInbox.cancel")((bus: Bus.Interface, input: PendingRef) =>
publishMutation(
input,
bus.publish(SessionEvent.InboxCancelled, {
sessionID: input.sessionID,
inboxID: input.id,
}),
),
)
export const steer = Effect.fn("SessionInbox.steer")((bus: Bus.Interface, input: PendingRef) =>
publishMutation(
input,
bus.publish(SessionEvent.InboxDeliveryChanged, {
sessionID: input.sessionID,
inboxID: input.id,
delivery: "steer",
}),
),
)
export const queue = Effect.fn("SessionInbox.queue")((bus: Bus.Interface, input: PendingRef) =>
publishMutation(
input,
bus.publish(SessionEvent.InboxDeliveryChanged, {
sessionID: input.sessionID,
inboxID: input.id,
delivery: "queue",
}),
),
)
serialized(input.sessionID, effect).pipe(
Effect.asVoid,
// Bus projectors abort their transaction through the defect channel.
Effect.catchDefect((defect) => (defect instanceof LifecycleConflict ? Effect.fail(defect) : Effect.die(defect))),
)
const publish = Effect.fn("SessionInbox.publish")(function* (
db: DatabaseService,
@@ -447,6 +478,7 @@ const publish = Effect.fn("SessionInbox.publish")(function* (
defect instanceof LifecycleConflict
? promotedFromMessage(db, sessionID, entry.id, entry.delivery).pipe(
Effect.flatMap((stored) => (stored !== undefined ? Effect.void : Effect.die(defect))),
Effect.orDie,
)
: Effect.die(defect),
),
+16
View File
@@ -0,0 +1,16 @@
export * as SessionPromptNode from "./prompt-node.js"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Image } from "../image.js"
import { PluginHooks } from "../plugin/hooks.js"
import { PluginSupervisor } from "../plugin/supervisor.js"
import { Skill } from "../skill.js"
import { SessionPrompt } from "./prompt.js"
// Keep the supervisor implementation out of the global Session import path.
export const node = makeLocationNode({
service: SessionPrompt.Service,
layer: SessionPrompt.layer,
deps: [FSUtil.node, PluginSupervisor.node, PluginHooks.node, Image.node, Skill.node],
})
+231
View File
@@ -0,0 +1,231 @@
export * as SessionPrompt from "./prompt.js"
import { Base64, FileAttachment, Prompt } from "@opencode-ai/schema/prompt"
import { PromptInput } from "@opencode-ai/schema/prompt-input"
import { SessionInbox } from "@opencode-ai/schema/session-inbox"
import type { Session } from "@opencode-ai/schema/session"
import type { SessionMessage } from "@opencode-ai/schema/session-message"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Context, Effect, Layer } from "effect"
import path from "path"
import { fileURLToPath } from "url"
import { Image } from "../image.js"
import { Mime } from "../mime.js"
import { PluginHooks } from "../plugin/hooks.js"
import { PluginSupervisor } from "../plugin/supervisor-service.js"
import { Skill } from "../skill.js"
import { AttachmentError, SkillNotFoundError } from "./error.js"
export type Input = {
text: string
files?: PromptInput.Prompt["files"]
agents?: PromptInput.Prompt["agents"]
skills?: PromptInput.Prompt["skills"]
metadata?: Record<string, unknown>
delivery?: SessionInbox.Delivery
}
export const make = Effect.fn("SessionPrompt.make")(function* () {
const fs = yield* FSUtil.Service
const plugins = yield* PluginSupervisor.Service
const hooks = yield* PluginHooks.Service
const image = yield* Image.Service
const skillService = yield* Skill.Service
const prepare = Effect.fn("SessionPrompt.prepare")(function* (request: {
sessionID: Session.ID
messageID: SessionMessage.ID
input: Input
}) {
yield* plugins.flush
const event = yield* hooks.trigger("session", "prompt", {
sessionID: request.sessionID,
messageID: request.messageID,
prompt: structuredClone({
text: request.input.text,
files: request.input.files?.slice(),
agents: request.input.agents?.slice(),
skills: request.input.skills?.slice(),
}),
metadata: structuredClone(request.input.metadata),
delivery: request.input.delivery ?? "steer",
})
const input = event.prompt
const files = input.files
? yield* Effect.forEach(input.files, materializeAttachment, { concurrency: 8 })
: undefined
const requested = input.skills
const selected = yield* Effect.gen(function* () {
if (!requested?.length) return undefined
const prepared = new Map<Skill.ID, Skill.Name>()
return yield* Effect.forEach(requested, (attachment) =>
Effect.gen(function* () {
const name = prepared.get(attachment.id)
if (name !== undefined) return { id: attachment.id, name, mention: attachment.mention }
const skill = yield* skillService.get(attachment.id)
if (!skill) return yield* new SkillNotFoundError({ skill: attachment.id })
prepared.set(skill.id, skill.name)
return {
id: skill.id,
name: skill.name,
text: (yield* Skill.prepare(fs, skill).pipe(Effect.orDie)).output,
mention: attachment.mention,
}
}),
)
})
return {
type: "user",
payload: SessionInbox.UserPayload.make({
...Prompt.make({
text: input.text,
agents: input.agents,
files,
skills: selected?.length ? selected : undefined,
}),
metadata: event.metadata,
}),
delivery: SessionInbox.Delivery.make(event.delivery),
} satisfies SessionInbox.Item
})
const materializeAttachment = Effect.fn("SessionPrompt.materializeAttachment")(function* (
input: PromptInput.FileAttachment,
) {
const resolved = input.uri.startsWith("data:")
? {
bytes: yield* decodeDataURL(input.uri),
source: { type: "inline" as const },
start: undefined,
end: undefined,
name: undefined,
mime: undefined,
}
: yield* readFileAttachment(input.uri)
if (resolved.bytes.byteLength > MAX_ATTACHMENT_BYTES)
return yield* new AttachmentError({
uri: input.uri,
message: `Attachment exceeds the ${MAX_ATTACHMENT_BYTES} byte limit: ${input.uri}`,
})
const mime = resolved.mime ?? Mime.detect(resolved.bytes)
const content =
mime === "text/plain" && resolved.start !== undefined
? Buffer.from(
Buffer.from(resolved.bytes)
.toString("utf8")
.split("\n")
.slice(resolved.start - 1, resolved.end)
.join("\n"),
)
: resolved.bytes
const normalized = yield* normalizeImageAttachment(input, Buffer.from(content).toString("base64"), mime)
return FileAttachment.create({
data: normalized.data,
mime: normalized.mime,
source: resolved.source,
name: input.name ?? resolved.name,
description: input.description,
mention: input.mention,
})
})
const normalizeImageAttachment = Effect.fn("SessionPrompt.normalizeImageAttachment")(function* (
input: PromptInput.FileAttachment,
data: string,
mime: string,
) {
if (!mime.startsWith("image/")) return { data: Base64.make(data), mime }
const label = input.name ?? (input.uri.startsWith("data:") ? "inline attachment" : input.uri)
const content = { uri: label, content: data, encoding: "base64" as const, mime }
const normalized = yield* image.normalize(label, content).pipe(
Effect.catchTag("Image.ResizerUnavailableError", () => Effect.succeed(content)),
Effect.mapError((error) => new AttachmentError({ uri: label, message: error.message })),
)
return { data: Base64.make(normalized.content), mime: normalized.mime }
})
const readFileAttachment = Effect.fn("SessionPrompt.readFileAttachment")(function* (uri: string) {
const url = yield* Effect.try({
try: () => new URL(uri),
catch: () => new AttachmentError({ uri, message: `Invalid attachment URI: ${uri}` }),
})
if (url.protocol !== "file:")
return yield* new AttachmentError({ uri, message: `Unsupported attachment URI: ${uri}` })
const start = positiveInt(url.searchParams.get("start"))
const end = positiveInt(url.searchParams.get("end"))
const target = yield* Effect.try({
try: () => {
url.search = ""
url.hash = ""
return fileURLToPath(url)
},
catch: () => new AttachmentError({ uri, message: `Invalid file URI: ${uri}` }),
})
const info = yield* fs
.stat(target)
.pipe(Effect.mapError(() => new AttachmentError({ uri, message: `Unable to read attachment: ${uri}` })))
if (info.type === "Directory") {
const entries = yield* fs
.readDirectoryEntries(target)
.pipe(Effect.mapError(() => new AttachmentError({ uri, message: `Unable to read attachment: ${uri}` })))
return {
bytes: Buffer.from(
entries
.filter((entry) => entry.type === "file" || entry.type === "directory")
.sort((a, b) => (a.type === b.type ? a.name.localeCompare(b.name) : a.type === "directory" ? -1 : 1))
.map((entry) => entry.name + (entry.type === "directory" ? path.sep : ""))
.join("\n"),
),
source: { type: "uri" as const, uri },
start: undefined,
end: undefined,
name: path.basename(target),
mime: "application/x-directory",
}
}
if (info.type !== "File") return yield* new AttachmentError({ uri, message: `Attachment is not a file: ${uri}` })
if (Number(info.size) > MAX_ATTACHMENT_BYTES)
return yield* new AttachmentError({
uri,
message: `Attachment exceeds the ${MAX_ATTACHMENT_BYTES} byte limit: ${uri}`,
})
const bytes = yield* fs
.readFile(target)
.pipe(Effect.mapError(() => new AttachmentError({ uri, message: `Unable to read attachment: ${uri}` })))
return { bytes, source: { type: "uri" as const, uri }, start, end, name: path.basename(target), mime: undefined }
})
return { prepare }
})
export type Interface = Effect.Success<ReturnType<typeof make>>
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionPrompt") {}
export const layer = Layer.effect(Service, make())
const MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024
function decodeDataURL(uri: string) {
return Effect.try({
try: () => {
const comma = uri.indexOf(",")
if (comma === -1) throw new Error("Invalid data URL")
const metadata = uri.slice(5, comma)
const payload = uri.slice(comma + 1)
if (!metadata.split(";").some((part) => part.toLowerCase() === "base64"))
return Buffer.from(decodeURIComponent(payload))
const bytes = Buffer.from(payload, "base64")
if (bytes.toString("base64") !== payload) throw new Error("Non-canonical base64")
return bytes
},
catch: () => new AttachmentError({ uri, message: "Invalid attachment data URL" }),
})
}
function positiveInt(value: string | null) {
if (value === null) return
const parsed = Number(value)
return Number.isInteger(parsed) && parsed > 0 ? parsed : undefined
}
+15
View File
@@ -0,0 +1,15 @@
export * as SessionRevertNode from "./revert-node.js"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { Bus } from "../bus.js"
import { Database } from "../database/database.js"
import { PluginSupervisor } from "../plugin/supervisor.js"
import { Snapshot } from "../snapshot.js"
import { SessionRevert } from "./revert.js"
// Keep the supervisor implementation out of the global Session import path.
export const node = makeLocationNode({
service: SessionRevert.Service,
layer: SessionRevert.layer,
deps: [Database.node, Bus.node, PluginSupervisor.node, Snapshot.node],
})
+76 -63
View File
@@ -1,28 +1,97 @@
export * as SessionRevert from "./revert.js"
import { and, asc, eq, gt } from "drizzle-orm"
import { Effect, Schema } from "effect"
import { Context, Effect, Layer, Schema } from "effect"
import { Database } from "../database/database.js"
import { Bus } from "../bus.js"
import { PluginSupervisor } from "../plugin/supervisor-service.js"
import { RelativePath } from "../schema.js"
import { Snapshot } from "../snapshot.js"
import { SessionEvent } from "./event.js"
import { MessageNotFoundError } from "./error.js"
import { SessionMessage } from "./message.js"
import { SessionSchema } from "./schema.js"
import { SessionMessageTable } from "./sql.js"
export class MessageNotFoundError extends Schema.TaggedError<MessageNotFoundError>()("Session.MessageNotFoundError", {
sessionID: SessionSchema.ID,
messageID: SessionMessage.ID,
}) {}
export { MessageNotFoundError }
interface BoundaryInput {
readonly sessionID: SessionSchema.ID
readonly messageID: SessionMessage.ID
}
const plan = Effect.fn("SessionRevert.plan")(function* (input: BoundaryInput) {
const db = (yield* Database.Service).db
export interface Interface {
readonly stage: (input: {
readonly session: SessionSchema.Info
readonly messageID: SessionMessage.ID
readonly files?: boolean
}) => Effect.Effect<SessionSchema.Revert, MessageNotFoundError | Snapshot.Error>
readonly clear: (session: SessionSchema.Info) => Effect.Effect<void, Snapshot.Error>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/SessionRevert") {}
export const make = Effect.fn("SessionRevert.make")(function* () {
const database = yield* Database.Service
const bus = yield* Bus.Service
const plugins = yield* PluginSupervisor.Service
const snapshot = yield* Snapshot.Service
const stage: Interface["stage"] = Effect.fn("SessionRevert.stage")(function* (input) {
yield* plugins.flush
const original = input.session.revert?.snapshot
? Snapshot.ID.make(input.session.revert.snapshot)
: yield* snapshot.capture()
const next = yield* plan(database.db, { sessionID: input.session.id, messageID: input.messageID })
const restore = new Map<RelativePath, Snapshot.ID>()
if (original) {
for (const file of input.session.revert?.files ?? []) restore.set(RelativePath.make(file.file), original)
}
if (input.files !== false) for (const [file, tree] of next) restore.set(file, tree)
if (restore.size) yield* snapshot.restore({ files: restore })
const paths = input.files === false ? [] : Array.from(next.keys())
const files = original
? yield* snapshot.diff({ from: original, to: (yield* snapshot.capture()) ?? original, paths })
: []
const revert = {
messageID: input.messageID,
snapshot: original,
files,
} satisfies SessionSchema.Info["revert"]
yield* bus.publish(SessionEvent.RevertEvent.Staged, {
sessionID: input.session.id,
revert,
})
return revert
})
const clear: Interface["clear"] = Effect.fn("SessionRevert.clear")(function* (session) {
yield* plugins.flush
if (!session.revert) return
const original = session.revert.snapshot ? Snapshot.ID.make(session.revert.snapshot) : undefined
if (original)
yield* snapshot.restore({
files: new Map((session.revert.files ?? []).map((file) => [RelativePath.make(file.file), original])),
})
yield* bus.publish(SessionEvent.RevertEvent.Cleared, {
sessionID: session.id,
})
})
return { stage, clear }
})
export const layer = Layer.effect(Service, make())
export const commit = Effect.fn("SessionRevert.commit")(function* (bus: Bus.Interface, session: SessionSchema.Info) {
if (!session.revert) return
yield* bus.publish(SessionEvent.RevertEvent.Committed, {
sessionID: session.id,
to: session.revert.messageID,
})
})
const plan = Effect.fn("SessionRevert.plan")(function* (db: Database.Interface["db"], input: BoundaryInput) {
const boundary = yield* db
.select({ seq: SessionMessageTable.seq })
.from(SessionMessageTable)
@@ -53,59 +122,3 @@ const plan = Effect.fn("SessionRevert.plan")(function* (input: BoundaryInput) {
}
return files
})
export const stage = Effect.fn("SessionRevert.stage")(function* (input: {
readonly session: SessionSchema.Info
readonly messageID: SessionMessage.ID
readonly files?: boolean
}) {
const snapshot = yield* Snapshot.Service
const bus = yield* Bus.Service
const original = input.session.revert?.snapshot
? Snapshot.ID.make(input.session.revert.snapshot)
: yield* snapshot.capture()
const next = yield* plan({ sessionID: input.session.id, messageID: input.messageID })
const restore = new Map<RelativePath, Snapshot.ID>()
if (original) {
for (const file of input.session.revert?.files ?? []) restore.set(RelativePath.make(file.file), original)
}
if (input.files !== false) for (const [file, tree] of next) restore.set(file, tree)
if (restore.size) yield* snapshot.restore({ files: restore })
const paths = input.files === false ? [] : Array.from(next.keys())
const files = original
? yield* snapshot.diff({ from: original, to: (yield* snapshot.capture()) ?? original, paths })
: []
const revert = {
messageID: input.messageID,
snapshot: original,
files,
} satisfies SessionSchema.Info["revert"]
yield* bus.publish(SessionEvent.RevertEvent.Staged, {
sessionID: input.session.id,
revert,
})
return revert
})
export const clear = Effect.fn("SessionRevert.clear")(function* (session: SessionSchema.Info) {
if (!session.revert) return
const snapshot = yield* Snapshot.Service
const original = session.revert.snapshot ? Snapshot.ID.make(session.revert.snapshot) : undefined
if (original)
yield* snapshot.restore({
files: new Map((session.revert.files ?? []).map((file) => [RelativePath.make(file.file), original])),
})
const bus = yield* Bus.Service
yield* bus.publish(SessionEvent.RevertEvent.Cleared, {
sessionID: session.id,
})
})
export const commit = Effect.fn("SessionRevert.commit")(function* (session: SessionSchema.Info) {
if (!session.revert) return
const bus = yield* Bus.Service
yield* bus.publish(SessionEvent.RevertEvent.Committed, {
sessionID: session.id,
to: session.revert.messageID,
})
})
+5 -1
View File
@@ -7,6 +7,8 @@ import type { Promotable } from "./inbox.js"
export interface Coordinator<Key, E, Reason = never> {
/** Snapshots keys with an execution owned by this coordinator. */
readonly active: Effect.Effect<ReadonlySet<Key>>
/** Checks ownership for one key, including cleanup and terminal settlement. */
readonly isActive: (key: Key) => Effect.Effect<boolean>
/** Starts an execution while idle, or joins the active execution and returns its exit. */
readonly run: (key: Key) => Effect.Effect<void, E>
/** Rings the doorbell: an idle key starts an execution; an active one drains again before settling. */
@@ -112,6 +114,8 @@ export const make = <Key, E, Reason = never>(options: {
Deferred.doneUnsafe(execution.done, exit)
}
const isActive = (key: Key) => Effect.sync(() => executions.has(key))
const run = (key: Key): Effect.Effect<void, E> =>
Effect.suspend(() => {
const execution = executions.get(key)
@@ -166,5 +170,5 @@ export const make = <Key, E, Reason = never>(options: {
return Deferred.await(execution.done).pipe(Effect.ignoreCause, Effect.andThen(awaitIdle(key)))
})
return { active: Effect.sync(() => new Set(executions.keys())), run, wake, interrupt, awaitIdle }
return { active: Effect.sync(() => new Set(executions.keys())), isActive, run, wake, interrupt, awaitIdle }
})
+416
View File
@@ -0,0 +1,416 @@
export * as Session from "./session.js"
import { DateTime, Effect, Fiber, Layer, Schema, Scope } from "effect"
import type { Agent } from "@opencode-ai/schema/agent"
import type { Model } from "@opencode-ai/schema/model"
import { Event } from "@opencode-ai/schema/event"
import { Bus } from "../bus.js"
import { Location } from "../location.js"
import { PluginSupervisor } from "../plugin/supervisor-service.js"
import { Shell } from "../shell.js"
import { ShellResult } from "../shell/result.js"
import {
BusyError,
CompactionConflictError,
InboxConflictError,
MessageIncompleteError,
MessageNotAssistantError,
MessageNotFoundError,
MessageToolIncompleteError,
NotFoundError,
PromptConflictError,
SyntheticConflictError,
} from "./error.js"
import { SessionEvent } from "./event.js"
import { SessionExecution } from "./execution.js"
import { SessionInbox } from "./inbox.js"
import { SessionMessage } from "./message.js"
import { SessionPrompt } from "./prompt.js"
import { SessionRevert } from "./revert.js"
import { SessionSchema } from "./schema.js"
import { SessionStore } from "./store.js"
export type Services = PluginSupervisor.Service | SessionPrompt.Service | SessionRevert.Service | Shell.Service
type PromptRequest = SessionPrompt.Input & {
id?: SessionMessage.ID
resume?: boolean
}
/**
* Build once in the host Scope: `const sessions = yield* Session.make(servicesFor)`.
* Use `sessions.forSession(id)` for handles that share host services and reload current state.
*/
export const make = Effect.fn("Session.make")(function* (servicesFor: (ref: Location.Ref) => Layer.Layer<Services>) {
const bus = yield* Bus.Service
const store = yield* SessionStore.Service
const execution = yield* SessionExecution.Service
const admission = yield* SessionInbox.Service
const scope = yield* Scope.Scope
const get = Effect.fn("Session.get")(function* (sessionID: SessionSchema.ID) {
const session = yield* store.get(sessionID)
if (!session) return yield* new NotFoundError({ sessionID })
return session
})
const message = Effect.fn("Session.message")(function* (sessionID: SessionSchema.ID, messageID: SessionMessage.ID) {
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 (
session.time.idle === undefined ||
input.idle > DateTime.toEpochMillis(session.time.idle) ||
(session.time.viewed !== undefined && DateTime.toEpochMillis(session.time.viewed) >= input.idle)
)
return
yield* bus.publish(SessionEvent.Viewed, { sessionID, idle: input.idle })
})
const rename = Effect.fn("Session.rename")(function* (sessionID: SessionSchema.ID, input: { title: string }) {
yield* get(sessionID)
yield* bus.publish(SessionEvent.Renamed, { sessionID, title: input.title })
})
const switchAgent = Effect.fn("Session.switchAgent")(function* (
sessionID: SessionSchema.ID,
input: { agent: Agent.ID },
) {
const session = yield* get(sessionID)
yield* bus.publish(SessionEvent.AgentSelected, { sessionID, agent: input.agent, previous: session.agent })
})
const switchModel = Effect.fn("Session.switchModel")(function* (
sessionID: SessionSchema.ID,
input: { model: Model.Ref },
) {
const session = yield* get(sessionID)
if (
session.model?.providerID === input.model.providerID &&
session.model.id === input.model.id &&
(session.model.variant ?? "default") === (input.model.variant ?? "default")
)
return
yield* bus.publish(SessionEvent.ModelSelected, { sessionID, model: input.model, previous: session.model })
})
const mutatePending = (
sessionID: SessionSchema.ID,
inboxID: SessionMessage.ID,
mutation: (input: {
readonly id: SessionMessage.ID
readonly sessionID: SessionSchema.ID
}) => Effect.Effect<void, SessionInbox.LifecycleConflict>,
) =>
mutation({ sessionID, id: inboxID }).pipe(
Effect.catchTag("SessionInbox.LifecycleConflict", () =>
Effect.gen(function* () {
yield* get(sessionID)
return yield* new InboxConflictError({ sessionID, inboxID })
}),
),
)
const inbox = Effect.fn("Session.inbox")(function* (sessionID: SessionSchema.ID) {
yield* get(sessionID)
return yield* admission.list(sessionID)
})
const cancelInbox = Effect.fn("Session.cancelInbox")(
(sessionID: SessionSchema.ID, inboxID: SessionMessage.ID) => mutatePending(sessionID, inboxID, admission.cancel),
Effect.uninterruptible,
)
const steerInbox = Effect.fn("Session.steerInbox")(function* (
sessionID: SessionSchema.ID,
inboxID: SessionMessage.ID,
) {
yield* mutatePending(sessionID, inboxID, admission.steer)
yield* execution.wake(sessionID)
}, Effect.uninterruptible)
const queueInbox = Effect.fn("Session.queueInbox")(
(sessionID: SessionSchema.ID, inboxID: SessionMessage.ID) => mutatePending(sessionID, inboxID, admission.queue),
Effect.uninterruptible,
)
const prompt = Effect.fn("Session.prompt")((sessionID: SessionSchema.ID, input: PromptRequest) =>
Effect.uninterruptibleMask((restore) =>
Effect.gen(function* () {
const session = yield* get(sessionID)
const messageID = input.id ?? SessionMessage.ID.create()
const admitted = yield* Effect.gen(function* () {
const existing = yield* admission.reconcile({
id: messageID,
sessionID: session.id,
type: "user",
delivery: input.delivery ?? "steer",
})
if (existing) return existing
const item = yield* restore(
SessionPrompt.Service.use((preparation) => preparation.prepare({ sessionID, messageID, input })).pipe(
Effect.provide(servicesFor(session.location)),
),
)
// Commit a staged revert only after preparation succeeds, before admitting new work.
if (session.revert) yield* SessionRevert.commit(bus, session)
return yield* admission.admit({
id: messageID,
sessionID: session.id,
item,
})
}).pipe(
Effect.catchTag("SessionInbox.LifecycleConflict", () => new PromptConflictError({ sessionID, messageID })),
)
if (input.resume !== false) yield* execution.wake(sessionID)
return admitted
}),
),
)
const shell = Effect.fn("Session.shell")(function* (
sessionID: SessionSchema.ID,
input: { id?: Event.ID; command: string },
) {
const session = yield* get(sessionID)
// The server owns completion recording even if the submitting client disconnects.
const running = yield* Effect.gen(function* () {
// Resolve shell services here without pinning Session events to this Location after a move.
const shell = yield* Effect.gen(function* () {
const plugins = yield* PluginSupervisor.Service
yield* plugins.flush
return yield* Shell.Service
}).pipe(Effect.provide(servicesFor(session.location)))
const started = yield* shell
.create({
command: input.command,
cwd: session.location.directory,
timeout: 0,
metadata: { sessionID, background: true },
})
.pipe(
Effect.tapError((error) =>
synthetic(sessionID, {
text: `User shell command failed to start:\n${input.command}\n\n${error.message}`,
description: input.command,
metadata: { source: "shell", state: "error" },
resume: false,
}),
),
Effect.orDie,
)
yield* bus.publish(
SessionEvent.Shell.Started,
{
sessionID,
shell: started,
},
{ id: input.id },
)
const terminal = yield* shell.result(started)
const preview = yield* shell
.output(started.id, { limit: SHELL_MAX_CAPTURE_BYTES })
.pipe(Effect.catchTag("Shell.NotFoundError", () => Effect.succeed(ShellResult.unavailable)))
yield* bus.publish(SessionEvent.Shell.Ended, {
sessionID,
shell: terminal.info,
output: preview,
})
yield* synthetic(sessionID, {
...ShellResult.userNotification(terminal),
resume: false,
}).pipe(
Effect.catchTag("Session.NotFoundError", () => Effect.void),
Effect.orDie,
)
}).pipe(Effect.forkIn(scope, { startImmediately: true }))
yield* Fiber.join(running)
})
const compact = Effect.fn("Session.compact")(function* (
sessionID: SessionSchema.ID,
input: { id?: SessionMessage.ID; delivery?: SessionInbox.Delivery },
) {
yield* get(sessionID)
const inputID = input.id ?? SessionMessage.ID.create()
const admitted = yield* admission
.admitCompaction({
id: inputID,
sessionID,
delivery: input.delivery ?? "steer",
})
.pipe(
Effect.catchTag("SessionInbox.LifecycleConflict", () => new CompactionConflictError({ sessionID, inputID })),
)
yield* execution.wake(sessionID)
return admitted
})
const wait = Effect.fn("Session.wait")(function* (sessionID: SessionSchema.ID) {
yield* get(sessionID)
yield* execution.awaitIdle(sessionID)
})
const resume = Effect.fn("Session.resume")(function* (sessionID: SessionSchema.ID) {
yield* get(sessionID)
yield* execution.resume(sessionID)
})
const synthetic = Effect.fn("Session.synthetic")(
(
sessionID: SessionSchema.ID,
input: {
id?: SessionMessage.ID
text: string
description?: string
metadata?: Record<string, unknown>
delivery?: SessionInbox.Delivery
resume?: boolean
},
) =>
Effect.uninterruptible(
Effect.gen(function* () {
yield* get(sessionID)
const inputID = input.id ?? SessionMessage.ID.create()
const admittedInput = {
type: "synthetic",
payload: SessionInbox.SyntheticPayload.make({
text: input.text,
description: input.description,
metadata: input.metadata,
}),
delivery: SessionInbox.Delivery.make(input.delivery ?? "steer"),
} satisfies SessionInbox.Item
const admitted = yield* admission
.admit({
id: inputID,
sessionID,
item: admittedInput,
})
.pipe(
Effect.catchTag(
"SessionInbox.LifecycleConflict",
() => new SyntheticConflictError({ sessionID, inputID }),
),
)
if (input.resume !== false && !(yield* get(sessionID)).revert) yield* execution.wake(sessionID)
return admitted
}),
),
)
const interrupt = Effect.fn("Session.interrupt")(
(sessionID: SessionSchema.ID, options?: { readonly continue?: boolean }) =>
Effect.uninterruptible(execution.interrupt(sessionID, options)),
)
const stage = Effect.fn("Session.revert.stage")(function* (
sessionID: SessionSchema.ID,
input: { messageID: SessionMessage.ID; files?: boolean },
) {
const session = yield* get(sessionID)
if (yield* execution.isActive(sessionID)) return yield* new BusyError({ sessionID })
return yield* SessionRevert.Service.use((revert) =>
revert.stage({ session, messageID: input.messageID, files: input.files }),
).pipe(Effect.provide(servicesFor(session.location)))
})
const clear = Effect.fn("Session.revert.clear")(function* (sessionID: SessionSchema.ID) {
const session = yield* get(sessionID)
if (yield* execution.isActive(sessionID)) return yield* new BusyError({ sessionID })
yield* SessionRevert.Service.use((revert) => revert.clear(session)).pipe(
Effect.provide(servicesFor(session.location)),
)
return yield* execution.wake(sessionID)
})
const commit = Effect.fn("Session.revert.commit")(function* (sessionID: SessionSchema.ID) {
const session = yield* get(sessionID)
if (yield* execution.isActive(sessionID)) return yield* new BusyError({ sessionID })
return yield* SessionRevert.commit(bus, session)
})
const revert = { stage, clear, commit }
const operations = {
get,
message,
updateMessage,
view,
rename,
switchAgent,
switchModel,
inbox,
prompt,
synthetic,
shell,
compact,
wait,
resume,
interrupt,
cancelInbox,
steerInbox,
queueInbox,
revert,
}
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)
const switchModel = operations.switchModel.bind(undefined, sessionID)
const inbox = operations.inbox.bind(undefined, sessionID)
const prompt = operations.prompt.bind(undefined, sessionID)
const synthetic = operations.synthetic.bind(undefined, sessionID)
const shell = operations.shell.bind(undefined, sessionID)
const compact = operations.compact.bind(undefined, sessionID)
const wait = operations.wait.bind(undefined, sessionID)
const resume = operations.resume.bind(undefined, sessionID)
const interrupt = operations.interrupt.bind(undefined, sessionID)
const cancelInbox = operations.cancelInbox.bind(undefined, sessionID)
const steerInbox = operations.steerInbox.bind(undefined, sessionID)
const queueInbox = operations.queueInbox.bind(undefined, sessionID)
const stage = operations.revert.stage.bind(undefined, sessionID)
const clear = operations.revert.clear.bind(undefined, sessionID)
const commit = operations.revert.commit.bind(undefined, sessionID)
const revert = { stage, clear, commit }
return {
id: sessionID,
get,
message,
updateMessage,
view,
rename,
switchAgent,
switchModel,
inbox,
prompt,
synthetic,
shell,
compact,
wait,
resume,
interrupt,
cancelInbox,
steerInbox,
queueInbox,
revert,
}
}
return { forSession }
})
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.
const SHELL_MAX_CAPTURE_BYTES = 1024 * 1024
+32 -10
View File
@@ -1,18 +1,40 @@
import { Bus } from "@opencode-ai/core/bus"
import { Image } from "@opencode-ai/core/image"
import { LocationServiceMap } from "@opencode-ai/core/location-service-map"
import type { LocationServices } from "@opencode-ai/core/location-services"
import { PluginHooks } from "@opencode-ai/core/plugin/hooks"
import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor-service"
import { SessionPrompt } from "@opencode-ai/core/session/prompt"
import { Skill } from "@opencode-ai/core/skill"
import type { Location } from "@opencode-ai/schema/location"
import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Effect, Layer, LayerMap } from "effect"
// Plain-prompt unit fixtures use virtual directories and need only the admission hook services.
export const promptLocationLayer = Layer.effect(
LocationServiceMap.Service,
LayerMap.make(
() =>
Layer.merge(
LayerNode.compile(PluginHooks.node),
Layer.succeed(PluginSupervisor.Service, { flush: Effect.void }),
) as Layer.Layer<LocationServices>,
// Plain-prompt unit fixtures use virtual directories and need only prompt preparation services.
export const promptLocationNode = makeGlobalNode({
service: LocationServiceMap.Service,
layer: Layer.effect(
LocationServiceMap.Service,
Effect.gen(function* () {
const bus = yield* Bus.Service
const fs = yield* FSUtil.Service
return yield* LayerMap.make(
(_ref: Location.Ref) =>
SessionPrompt.layer.pipe(
Layer.provideMerge(
Layer.mergeAll(
LayerNode.compile(LayerNode.group([PluginHooks.node, Image.node, Skill.node]), [
[Bus.node, Layer.succeed(Bus.Service, bus)],
]),
Layer.succeed(FSUtil.Service, fs),
Layer.succeed(PluginSupervisor.Service, { flush: Effect.void }),
),
),
) as Layer.Layer<LocationServices>,
)
}),
),
)
deps: [Bus.node, FSUtil.node],
})
+21 -8
View File
@@ -1,11 +1,24 @@
import { Database } from "@opencode-ai/core/database/database"
import { Project } from "@opencode-ai/core/project"
import { upsertProject } from "@opencode-ai/core/project/sql"
import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
import { Effect, Layer } from "effect"
export const globalProjectLayer = Layer.succeed(
Project.Service,
Project.Service.of({
list: () => Effect.succeed([]),
update: () => Effect.die("not implemented"),
resolve: (directory) => Effect.succeed({ id: Project.ID.global, directory, canonical: directory }),
}),
)
export const globalProjectNode = makeGlobalNode({
service: Project.Service,
layer: Layer.effect(
Project.Service,
Effect.gen(function* () {
const database = yield* Database.Service
return Project.Service.of({
list: () => Effect.succeed([]),
update: () => Effect.die("not implemented"),
resolve: (directory) => {
const project = { id: Project.ID.global, directory, canonical: directory }
return upsertProject(database.db, project).pipe(Effect.orDie, Effect.as(project))
},
})
}),
),
deps: [Database.node],
})
+2 -2
View File
@@ -21,7 +21,7 @@ import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model"
import { SessionStore } from "@opencode-ai/core/session/store"
import { Effect, Layer, LayerMap, Stream } from "effect"
import { testEffect } from "./lib/effect"
import { globalProjectLayer } from "./lib/project"
import { globalProjectNode } from "./lib/project"
const location = Location.Ref.make({ directory: AbsolutePath.make("/project") })
const model = LanguageModel.make({
@@ -66,7 +66,7 @@ const it = testEffect(
LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node, Session.node]),
[
[LocationServiceMap.node, locations],
[Project.node, globalProjectLayer],
[Project.node, globalProjectNode],
[SessionExecution.node, SessionExecution.noopLayer],
],
),
+5 -5
View File
@@ -35,8 +35,8 @@ import { Workspace } from "@opencode-ai/core/workspace"
import { Expected } from "./lib/session-message"
import { testEffect } from "./lib/effect"
import { LocationServiceMap } from "@opencode-ai/core/location-service-map"
import { promptLocationLayer } from "./fixture/prompt-location"
import { globalProjectLayer } from "./lib/project"
import { promptLocationNode } from "./fixture/prompt-location"
import { globalProjectNode } from "./lib/project"
import { tmpdirScoped } from "./fixture/tmpdir"
const it = testEffect(
@@ -52,8 +52,8 @@ const it = testEffect(
]),
[
[Bus.node, Bus.configured({ persist: true })],
[Project.node, globalProjectLayer],
[LocationServiceMap.node, promptLocationLayer],
[Project.node, globalProjectNode],
[LocationServiceMap.node, promptLocationNode],
[SessionExecution.node, SessionExecution.noopLayer],
],
),
@@ -73,7 +73,7 @@ const projectIt = testEffect(
[
[Bus.node, Bus.configured({ persist: true })],
// Project adoption needs plain-prompt admission, not live plugin/provider startup.
[LocationServiceMap.node, promptLocationLayer],
[LocationServiceMap.node, promptLocationNode],
[SessionExecution.node, SessionExecution.noopLayer],
],
),
+13 -3
View File
@@ -26,7 +26,9 @@ import { eq } from "drizzle-orm"
import { testEffect } from "./lib/effect"
const it = testEffect(
AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionStore.node, Job.node, KV.node, Session.node])),
AppNodeBuilder.build(
LayerNode.group([Database.node, Bus.node, SessionStore.node, SessionInbox.node, Job.node, KV.node, Session.node]),
),
)
describe("SessionExecution lifecycle", () => {
@@ -92,6 +94,8 @@ describe("SessionExecution lifecycle", () => {
: Deferred.succeed(interruptedRunning, undefined).pipe(Effect.andThen(Effect.never)),
)
const execution = Context.get(context, SessionExecution.Service)
const completedActive = execution.isActive(completed)
expect(yield* completedActive).toBe(false)
yield* execution.resume(interrupted).pipe(Effect.forkScoped)
const completing = yield* execution.resume(completed).pipe(Effect.forkIn(scope))
yield* Deferred.await(interruptedRunning)
@@ -99,17 +103,22 @@ describe("SessionExecution lifecycle", () => {
// The write-ahead claim exists WHILE the turns run — no shutdown hook involved.
expect(yield* claims(database)).toEqual({ [interrupted]: true, [completed]: true })
expect(yield* completedActive).toBe(true)
expect(yield* execution.isActive(interrupted)).toBe(true)
// A drain that finishes on its own releases its claim.
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(completing)
yield* execution.awaitIdle(completed)
expect((yield* claims(database))[completed]).toBe(false)
expect(yield* completedActive).toBe(false)
expect(yield* execution.isActive(interrupted)).toBe(true)
// Teardown interruption (graceful twin of an unclean death) preserves the claim
// for the next server start.
yield* Scope.close(scope, Exit.void)
expect((yield* claims(database))[interrupted]).toBe(true)
expect(yield* execution.isActive(interrupted)).toBe(false)
}),
)
@@ -146,6 +155,7 @@ describe("SessionExecution lifecycle", () => {
expect(yield* execution.interrupt(sessionID)).toBeFalse()
expect(yield* execution.active).not.toContain(sessionID)
expect(yield* execution.isActive(sessionID)).toBe(false)
}),
)
@@ -897,7 +907,7 @@ describe("SessionRestart background recovery", () => {
it.effect("retains a subagent completion marker when synthetic admission conflicts", () =>
Effect.gen(function* () {
const database = yield* Database.Service
const bus = yield* Bus.Service
const admission = yield* SessionInbox.Service
const jobs = yield* Job.Service
const sessions = yield* Session.Service
const parent = Session.ID.make("ses_completion_conflict_parent")
@@ -920,7 +930,7 @@ describe("SessionRestart background recovery", () => {
yield* jobs.background(child)
const marker = (yield* jobs.pendingBackground)[0]
if (!marker) return yield* Effect.die("background record missing")
yield* SessionInbox.admit(database.db, bus, {
yield* admission.admit({
id: marker.notificationID,
sessionID: parent,
item: { type: "user", payload: { text: "User input" }, delivery: "steer" },
@@ -33,7 +33,7 @@ import { tempLocationLayer } from "./fixture/location"
import { makeLocationNode } from "@opencode-ai/util/effect/app-node"
import { testEffect } from "./lib/effect"
import { permissionLayer } from "./lib/permission"
import { globalProjectLayer } from "./lib/project"
import { globalProjectNode } from "./lib/project"
import { executeTool, registerToolPlugin } from "./lib/tool"
const readToolNode = makeLocationNode({
@@ -74,7 +74,7 @@ const testLayer = AppNodeBuilder.build(
Image.node,
]),
[
[Project.node, globalProjectLayer],
[Project.node, globalProjectNode],
[SessionExecution.node, SessionExecution.noopLayer],
[Location.node, tempLocationLayer],
[Permission.node, permission],
+2 -2
View File
@@ -16,14 +16,14 @@ 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"
import { globalProjectLayer } from "./lib/project"
import { globalProjectNode } from "./lib/project"
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],
[Project.node, globalProjectNode],
[SessionExecution.node, SessionExecution.noopLayer],
],
),
@@ -23,7 +23,7 @@ import { Money } from "@opencode-ai/schema/money"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { tmpdirScoped } from "./fixture/tmpdir"
import { testEffect } from "./lib/effect"
import { globalProjectLayer } from "./lib/project"
import { globalProjectNode } from "./lib/project"
const active = new Set<Session.ID>()
const it = testEffect(
@@ -31,13 +31,14 @@ const it = testEffect(
LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node, Session.node]),
[
[Bus.node, Bus.configured({ persist: true })],
[Project.node, globalProjectLayer],
[Project.node, globalProjectNode],
[
SessionExecution.node,
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),
@@ -212,16 +213,53 @@ describe("Session.updateMessage", () => {
)
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 }))
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)
}),
)
+3 -3
View File
@@ -19,13 +19,13 @@ import { SessionStore } from "@opencode-ai/core/session/store"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { tmpdirScoped } from "./fixture/tmpdir"
import { testEffect } from "./lib/effect"
import { globalProjectLayer } from "./lib/project"
import { globalProjectNode } from "./lib/project"
const it = testEffect(
AppNodeBuilder.build(
LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node, Session.node]),
[
[Project.node, globalProjectLayer],
[Project.node, globalProjectNode],
[SessionExecution.node, SessionExecution.noopLayer],
],
),
@@ -40,7 +40,7 @@ const itWithUnavailableDestination = testEffect(
AppNodeBuilder.build(
LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node, Session.node]),
[
[Project.node, globalProjectLayer],
[Project.node, globalProjectNode],
[SessionExecution.node, SessionExecution.noopLayer],
[LocationServiceMap.node, unavailableLocations],
],
+976
View File
@@ -0,0 +1,976 @@
import { describe, expect } from "bun:test"
import { and, eq } from "drizzle-orm"
import { Cause, Context, DateTime, Deferred, Effect, Exit, Fiber, Layer, Scope } from "effect"
import { Agent } from "@opencode-ai/schema/agent"
import { Event } from "@opencode-ai/schema/event"
import { Location } from "@opencode-ai/schema/location"
import { Model } from "@opencode-ai/schema/model"
import { Money } from "@opencode-ai/schema/money"
import { Project } from "@opencode-ai/schema/project"
import { Provider } from "@opencode-ai/schema/provider"
import { ID, Info, Output } from "@opencode-ai/schema/shell"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Global } from "@opencode-ai/util/global"
import { Bus } from "../src/bus.js"
import { Database } from "../src/database/database.js"
import { EventTable } from "../src/event/sql.js"
import { Image } from "../src/image.js"
import { PluginHooks } from "../src/plugin/hooks.js"
import { PluginSupervisor } from "../src/plugin/supervisor-service.js"
import { ProjectTable } from "../src/project/sql.js"
import { AbsolutePath, RelativePath } from "../src/schema.js"
import { InboxConflictError, NotFoundError, PromptConflictError } from "../src/session/error.js"
import { SessionEvent } from "../src/session/event.js"
import { SessionExecution } from "../src/session/execution.js"
import { SessionInbox } from "../src/session/inbox.js"
import { SessionMessage } from "../src/session/message.js"
import { SessionPrompt } from "../src/session/prompt.js"
import { SessionProjector } from "../src/session/projector.js"
import { SessionRevert } from "../src/session/revert.js"
import { SessionRunCoordinator } from "../src/session/run-coordinator.js"
import { SessionSchema } from "../src/session/schema.js"
import { Session } from "../src/session/session.js"
import { SessionTable } from "../src/session/sql.js"
import { SessionStore } from "../src/session/store.js"
import { Shell } from "../src/shell.js"
import { Skill } from "../src/skill.js"
import { Snapshot } from "../src/snapshot.js"
import { tempGlobalLayer } from "./fixture/global"
import { testEffect } from "./lib/effect"
const it = testEffect(
LayerNode.compile(
LayerNode.group([
Database.node,
Bus.node,
SessionProjector.node,
SessionStore.node,
SessionInbox.node,
FSUtil.node,
]),
[
[Bus.node, Bus.configured({ persist: true })],
[Global.node, tempGlobalLayer],
],
),
)
const sessionID = SessionSchema.ID.make("ses_owned")
const otherID = SessionSchema.ID.make("ses_owned_other")
const source = Location.Ref.make({ directory: AbsolutePath.make("/project") })
const setup = Effect.fnUntraced(function* (options?: {
execution?: SessionExecution.Interface
shell?: Layer.Layer<Shell.Service>
snapshot?: (ref: Location.Ref) => Layer.Layer<Snapshot.Service>
}) {
const database = yield* Database.Service
const bus = yield* Bus.Service
const store = yield* SessionStore.Service
const fs = yield* FSUtil.Service
yield* database.db
.insert(ProjectTable)
.values({ id: Project.ID.global, worktree: source.directory, sandboxes: [] })
.run()
.pipe(Effect.orDie)
yield* Effect.forEach([sessionID, otherID], (id) =>
bus.publish(SessionEvent.Created, {
sessionID: id,
projectID: Project.ID.global,
location: source,
slug: "owned",
title: "Owned session",
version: "test",
}),
)
const hooks = yield* PluginHooks.Service.pipe(Effect.provide(LayerNode.compile(PluginHooks.node)))
const locations: Location.Ref[] = []
const flushes: Location.Ref[] = []
const wakes: Array<{ sessionID: SessionSchema.ID; pending: SessionMessage.ID[]; enqueued: number }> = []
const execution = SessionExecution.Service.of({
active: Effect.succeed(new Set<SessionSchema.ID>()),
isActive: () => Effect.succeed(false),
resume: () => Effect.void,
awaitIdle: () => Effect.void,
interrupt: () => Effect.succeed(false),
wake: (id) =>
Effect.gen(function* () {
const pending = yield* SessionInbox.list(database.db, id)
const events = yield* database.db
.select({ id: EventTable.id })
.from(EventTable)
.where(
and(
eq(EventTable.aggregate_id, id),
eq(EventTable.type, Bus.versionedType(SessionEvent.InboxEnqueued.type, 1)),
),
)
.all()
.pipe(Effect.orDie)
wakes.push({ sessionID: id, pending: pending.map((item) => item.id), enqueued: events.length })
}),
})
const services = Layer.mergeAll(
Layer.succeed(PluginHooks.Service, hooks),
Layer.mock(Image.Service, {}),
Layer.mock(Skill.Service, {}),
options?.shell ?? Layer.mock(Shell.Service, {}),
)
const servicesFor = (ref: Location.Ref): Layer.Layer<Session.Services> => {
locations.push(ref)
return Layer.merge(SessionRevert.layer, SessionPrompt.layer).pipe(
Layer.provideMerge(
Layer.mergeAll(
services,
options?.snapshot?.(ref) ?? Layer.mock(Snapshot.Service, {}),
Layer.succeed(PluginSupervisor.Service, {
flush: Effect.sync(() => {
flushes.push(ref)
}),
}),
),
),
Layer.provide(
Layer.mergeAll(
Layer.succeed(Database.Service, database),
Layer.succeed(Bus.Service, bus),
Layer.succeed(FSUtil.Service, fs),
),
),
Layer.fresh,
)
}
const sessions = yield* Session.make(servicesFor).pipe(
Effect.satisfiesServicesType<
Bus.Service | SessionStore.Service | SessionExecution.Service | SessionInbox.Service | Scope.Scope
>(),
Effect.provideService(SessionExecution.Service, options?.execution ?? execution),
)
return { sessions, hooks, locations, flushes, wakes, db: database.db, bus, store }
})
describe("Session-owned handles", () => {
it.live("owns state changes and message editing without caller services or Location acquisition", () =>
Effect.gen(function* () {
const fixture = yield* setup()
const handle = fixture.sessions.forSession(sessionID)
const model = { id: Model.ID.make("test-model"), providerID: Provider.ID.make("test-provider") }
const messageID = SessionMessage.ID.create()
yield* fixture.bus.publish(SessionEvent.Step.Started, {
sessionID,
assistantMessageID: messageID,
agent: Agent.ID.make("build"),
model: { ...model, id: Model.ID.make("initial-model") },
})
yield* fixture.bus.publish(SessionEvent.Step.Ended, {
sessionID,
assistantMessageID: messageID,
finish: "stop",
cost: Money.USD.zero,
tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
})
yield* fixture.db
.update(SessionTable)
.set({ time_idle: 0 })
.where(eq(SessionTable.id, sessionID))
.run()
.pipe(Effect.orDie)
const { rename, switchAgent, switchModel, view, message, updateMessage } = handle
yield* Effect.gen(function* () {
yield* rename({ title: "Renamed" })
yield* switchAgent({ agent: Agent.ID.make("review") })
yield* switchModel({ model })
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 })
}).pipe(Effect.satisfiesServicesType<never>(), Effect.setContext(Context.empty()))
const session = yield* handle.get()
expect(session).toMatchObject({ title: "Renamed", agent: "review", model })
expect(session.time.viewed && DateTime.toEpochMillis(session.time.viewed)).toBe(0)
expect(yield* fixture.sessions.forSession(otherID).message(messageID)).toBeUndefined()
expect((yield* fixture.sessions.forSession(otherID).get()).title).toBe("Owned session")
expect(fixture.locations).toEqual([])
expect(fixture.wakes).toEqual([])
const events = yield* fixture.db
.select()
.from(EventTable)
.where(eq(EventTable.aggregate_id, sessionID))
.all()
.pipe(Effect.orDie)
expect(events.filter((event) => event.type === Bus.versionedType(SessionEvent.Viewed.type, 1))).toHaveLength(1)
expect(
events.filter((event) => event.type === Bus.versionedType(SessionEvent.ModelSelected.type, 1)),
).toHaveLength(1)
}),
)
it.live("acquires Location only for new prompt preparation and persists before waking", () =>
Effect.gen(function* () {
const fixture = yield* setup()
const handle = fixture.sessions.forSession(sessionID)
const { get, prompt } = handle
expect(handle.id).toBe(sessionID)
expect((yield* get().pipe(Effect.satisfiesServicesType<never>())).location).toEqual(source)
const synthetic = yield* handle.synthetic({ text: "Background result", resume: false })
expect(fixture.locations).toEqual([])
expect(fixture.wakes).toEqual([])
const calls: string[] = []
yield* fixture.hooks.register("session", "prompt", (event) =>
Effect.sync(() => {
expect(fixture.flushes).toEqual([source])
calls.push(event.prompt.text)
event.prompt.text += " prepared"
}),
)
const first = yield* prompt({
id: SessionMessage.ID.make("msg_owned_prepared"),
text: "Original",
files: [{ uri: new URL("./session-owned.test.ts", import.meta.url).href }],
})
const retried = yield* fixture.sessions.forSession(sessionID).prompt({
id: first.id,
text: "Ignored retry",
files: [{ uri: "file:///missing-owned-retry" }],
delivery: "queue",
})
expect(retried).toEqual(first)
expect(first.payload.text).toBe("Original prepared")
expect(first.payload.files?.[0]?.mime).toBe("text/plain")
expect(Buffer.from(first.payload.files?.[0]?.data ?? "", "base64").toString()).toBe(
yield* Effect.promise(() => Bun.file(import.meta.path).text()),
)
expect(calls).toEqual(["Original"])
expect(fixture.locations).toEqual([source])
expect(fixture.flushes).toEqual([source])
expect(fixture.wakes).toEqual([
{ sessionID, pending: [synthetic.id, first.id], enqueued: 2 },
{ sessionID, pending: [synthetic.id, first.id], enqueued: 2 },
])
expect(yield* SessionInbox.find(fixture.db, first.id)).toEqual(first)
expect(yield* fixture.store.context(sessionID)).toEqual([])
}),
)
it.live("keeps the first admission across handles, including delivered retries and identity conflicts", () =>
Effect.gen(function* () {
const fixture = yield* setup()
const first = fixture.sessions.forSession(sessionID)
const second = fixture.sessions.forSession(sessionID)
const other = fixture.sessions.forSession(otherID)
const prompt = yield* first.prompt({ text: "Keep this", metadata: { source: "first" }, resume: false })
const retry = { id: prompt.id, text: "Ignore this", metadata: { source: "retry" }, resume: false }
expect(yield* second.prompt({ ...retry, delivery: "queue" })).toEqual(prompt)
const conflict = yield* other.prompt(retry).pipe(Effect.flip)
expect(conflict).toBeInstanceOf(PromptConflictError)
expect(conflict).toMatchObject({ _tag: "Session.PromptConflictError", sessionID: otherID, messageID: prompt.id })
expect(yield* second.synthetic(retry).pipe(Effect.flip)).toMatchObject({
_tag: "Session.SyntheticConflictError",
sessionID,
inputID: prompt.id,
})
const synthetic = yield* first.synthetic({ text: "Original completion", description: "Job", resume: false })
expect(yield* second.synthetic({ ...retry, id: synthetic.id })).toEqual(synthetic)
expect(yield* first.inbox()).toEqual([prompt, synthetic])
yield* SessionInbox.promote(fixture.db, fixture.bus, sessionID, "steer")
// Delivered identity must be recoverable from the message, without retained enqueue history.
yield* fixture.db
.delete(EventTable)
.where(
and(
eq(EventTable.aggregate_id, sessionID),
eq(EventTable.type, Bus.versionedType(SessionEvent.InboxEnqueued.type, 1)),
),
)
.run()
.pipe(Effect.orDie)
expect((yield* second.prompt({ ...retry, files: [{ uri: "file:///missing-owned-retry" }] })).payload).toEqual(
prompt.payload,
)
expect((yield* second.synthetic({ ...retry, id: synthetic.id })).payload).toEqual(synthetic.payload)
expect(yield* other.synthetic({ ...retry, id: synthetic.id }).pipe(Effect.flip)).toMatchObject({
_tag: "Session.SyntheticConflictError",
sessionID: otherID,
inputID: synthetic.id,
})
expect(yield* second.prompt({ ...retry, id: synthetic.id }).pipe(Effect.flip)).toMatchObject({
_tag: "Session.PromptConflictError",
sessionID,
messageID: synthetic.id,
})
expect(yield* second.inbox()).toEqual([])
expect(yield* fixture.store.context(sessionID)).toMatchObject([
{ id: prompt.id, text: "Keep this", metadata: { source: "first" } },
{ id: synthetic.id, text: "Original completion", description: "Job" },
])
expect(fixture.locations).toEqual([source])
}),
)
it.live("reads fresh placement through an existing handle after a projected move", () =>
Effect.gen(function* () {
const fixture = yield* setup()
const handle = fixture.sessions.forSession(sessionID)
yield* handle.prompt({ text: "Before move", resume: false })
const get = handle.get()
const prompt = handle.prompt({ text: "After move", resume: false })
const destination = Location.Ref.make({ directory: AbsolutePath.make("/project/moved") })
yield* fixture.bus.publish(SessionEvent.Moved, {
sessionID,
location: destination,
projectID: Project.ID.global,
subpath: RelativePath.make("moved"),
})
expect((yield* get).location).toEqual(destination)
expect(fixture.locations).toEqual([source])
yield* prompt
expect(fixture.locations).toEqual([source, destination])
expect(fixture.flushes).toEqual([source, destination])
expect((yield* fixture.sessions.forSession(otherID).get()).location).toEqual(source)
}),
)
it.live("keeps prompt wakes independent of shell work across handles", () =>
Effect.gen(function* () {
const blocked = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const started = Info.make({
id: ID.make("sh_owned"),
command: "echo owned",
cwd: source.directory,
shell: "sh",
file: "/project/shell.out",
status: "running",
metadata: { sessionID, background: true },
time: { started: 0 },
})
const fixture = yield* setup({
shell: Layer.mock(Shell.Service, {
create: (input) =>
Effect.sync(() => {
expect(input).toEqual({
command: started.command,
cwd: source.directory,
timeout: 0,
metadata: { sessionID, background: true },
})
return started
}),
result: () =>
Deferred.succeed(blocked, undefined).pipe(
Effect.andThen(Deferred.await(release)),
Effect.as({
info: Info.make({ ...started, status: "exited", exit: 0, time: { started: 0, completed: 1 } }),
capture: { output: "owned", truncated: false },
}),
),
output: () => Effect.succeed(Output.make({ output: "owned", cursor: 5, size: 5, truncated: false })),
}),
})
const shell = yield* fixture.sessions
.forSession(sessionID)
.shell({ id: Event.ID.make("evt_owned_shell"), command: started.command })
.pipe(Effect.forkScoped)
yield* Deferred.await(blocked)
const admitted = yield* fixture.sessions.forSession(sessionID).prompt({ text: "Admit while the shell runs" })
expect(yield* SessionInbox.find(fixture.db, admitted.id)).toEqual(admitted)
expect(fixture.wakes).toEqual([{ sessionID, pending: [admitted.id], enqueued: 1 }])
const other = yield* fixture.sessions.forSession(otherID).prompt({ text: "Independent Session" })
expect(fixture.wakes).toEqual([
{ sessionID, pending: [admitted.id], enqueued: 1 },
{ sessionID: otherID, pending: [other.id], enqueued: 1 },
])
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(shell)
expect(fixture.wakes).toEqual([
{ sessionID, pending: [admitted.id], enqueued: 1 },
{ sessionID: otherID, pending: [other.id], enqueued: 1 },
])
expect(yield* fixture.store.context(sessionID)).toMatchObject([
{ type: "shell", shellID: started.id, status: "exited", output: { output: "owned" } },
])
expect(yield* fixture.sessions.forSession(sessionID).inbox()).toMatchObject([
{ id: admitted.id, type: "user" },
{ type: "synthetic", payload: { metadata: { source: "shell", shellID: started.id, state: "completed" } } },
])
}),
)
it.live("allows a prompt hook to admit synthetic input through another handle for the same Session", () =>
Effect.gen(function* () {
const fixture = yield* setup()
const handle = fixture.sessions.forSession(sessionID)
const nested = fixture.sessions.forSession(sessionID)
yield* fixture.hooks.register("session", "prompt", (event) =>
Effect.gen(function* () {
expect(event.sessionID).toBe(sessionID)
yield* nested.synthetic({ text: "Admitted by hook", resume: false })
event.prompt.text += " prepared"
}).pipe(Effect.orDie),
)
const prompt = yield* handle.prompt({ text: "Original", resume: false })
expect(yield* handle.inbox()).toMatchObject([
{ type: "synthetic", payload: { text: "Admitted by hook" } },
{ id: prompt.id, type: "user", payload: { text: "Original prepared" } },
])
yield* SessionInbox.promote(fixture.db, fixture.bus, sessionID, "steer")
expect(yield* fixture.store.context(sessionID)).toMatchObject([
{ type: "synthetic", text: "Admitted by hook" },
{ type: "user", text: "Original prepared" },
])
expect(fixture.locations).toEqual([source])
}),
)
it.live("mutates only this handle's pending inbox and preserves public conflict tags", () =>
Effect.gen(function* () {
const fixture = yield* setup()
const handle = fixture.sessions.forSession(sessionID)
const second = fixture.sessions.forSession(sessionID)
const queued = yield* handle.synthetic({ text: "Queued", delivery: "queue", resume: false })
const steer = yield* handle.prompt({ text: "Steer", resume: false })
const compact = yield* handle.compact({ delivery: "queue" })
yield* second.steerInbox(queued.id)
yield* second.queueInbox(steer.id)
expect(yield* handle.inbox()).toMatchObject([
{ id: queued.id, delivery: "steer" },
{ id: steer.id, delivery: "queue" },
{ id: compact.id, type: "compaction", delivery: "queue" },
])
expect(fixture.wakes).toHaveLength(2)
expect(yield* fixture.sessions.forSession(otherID).cancelInbox(queued.id).pipe(Effect.flip)).toMatchObject({
_tag: "Session.InboxConflictError",
sessionID: otherID,
inboxID: queued.id,
})
yield* second.cancelInbox(compact.id)
const cancelled = yield* handle.cancelInbox(compact.id).pipe(Effect.flip)
expect(cancelled).toBeInstanceOf(InboxConflictError)
expect(cancelled).toMatchObject({ _tag: "Session.InboxConflictError", sessionID, inboxID: compact.id })
expect(yield* handle.compact({ id: steer.id }).pipe(Effect.flip)).toMatchObject({
_tag: "Session.CompactionConflictError",
sessionID,
inputID: steer.id,
})
expect(yield* SessionInbox.promote(fixture.db, fixture.bus, sessionID, "steer")).toBe(1)
expect(yield* second.queueInbox(queued.id).pipe(Effect.flip)).toMatchObject({
_tag: "Session.InboxConflictError",
sessionID,
inboxID: queued.id,
})
expect(yield* handle.inbox()).toMatchObject([{ id: steer.id, delivery: "queue" }])
yield* second.cancelInbox(steer.id)
expect(yield* handle.inbox()).toEqual([])
const missingID = SessionSchema.ID.make("ses_owned_missing")
const missing = yield* fixture.sessions.forSession(missingID).inbox().pipe(Effect.flip)
expect(missing).toBeInstanceOf(NotFoundError)
expect(missing).toMatchObject({ _tag: "Session.NotFoundError", sessionID: missingID })
expect(fixture.locations).toEqual([source])
}),
)
it.live("joins same-ID resumes without transferring execution ownership to a cancelled caller", () =>
Effect.gen(function* () {
const started = yield* Deferred.make<void>()
const release = yield* Deferred.make<void>()
const joining = yield* Deferred.make<void>()
const drains: SessionSchema.ID[] = []
const resumes: SessionSchema.ID[] = []
const interrupts: Array<{ sessionID: SessionSchema.ID; options?: { readonly continue?: boolean } }> = []
const coordinator = yield* SessionRunCoordinator.make<SessionSchema.ID, never>({
drain: (id) =>
Effect.sync(() => void drains.push(id)).pipe(
Effect.andThen(Deferred.succeed(started, undefined)),
Effect.andThen(Deferred.await(release)),
),
})
const fixture = yield* setup({
execution: SessionExecution.Service.of({
active: coordinator.active,
isActive: coordinator.isActive,
resume: (id) =>
Effect.gen(function* () {
resumes.push(id)
if (resumes.length === 2) yield* Deferred.succeed(joining, undefined)
yield* coordinator.run(id)
}),
wake: coordinator.wake,
awaitIdle: coordinator.awaitIdle,
interrupt: (id, options) =>
Effect.sync(() => void interrupts.push({ sessionID: id, options })).pipe(
Effect.andThen(coordinator.interrupt(id)),
),
}),
})
const first = yield* fixture.sessions.forSession(sessionID).resume().pipe(Effect.forkScoped)
yield* Deferred.await(started)
const second = yield* fixture.sessions.forSession(sessionID).resume().pipe(Effect.forkScoped)
yield* Deferred.await(joining)
yield* Fiber.interrupt(second)
const cancelled = yield* Fiber.await(second)
expect(Exit.isFailure(cancelled) && Cause.hasInterruptsOnly(cancelled.cause)).toBe(true)
expect(yield* coordinator.active).toEqual(new Set([sessionID]))
expect(drains).toEqual([sessionID])
yield* Deferred.succeed(release, undefined)
yield* Fiber.join(first)
yield* fixture.sessions.forSession(sessionID).wait()
expect(drains).toEqual([sessionID])
expect(yield* coordinator.active).toEqual(new Set())
expect(yield* fixture.sessions.forSession(sessionID).interrupt({ continue: true })).toBe(false)
expect(yield* fixture.sessions.forSession(sessionID).interrupt()).toBe(false)
expect(interrupts).toEqual([
{ sessionID, options: { continue: true } },
{ sessionID, options: undefined },
])
expect(fixture.locations).toEqual([])
}),
)
it.live("keeps preparation interruptible without admitting input or committing a staged revert", () =>
Effect.gen(function* () {
const fixture = yield* setup({
snapshot: () => Layer.mock(Snapshot.Service, { capture: () => Effect.undefined }),
})
const handle = fixture.sessions.forSession(sessionID)
const boundary = yield* handle.synthetic({ text: "Revert boundary", resume: false })
yield* SessionInbox.promote(fixture.db, fixture.bus, sessionID, "steer")
yield* handle.revert.stage({ messageID: boundary.id, files: false })
const entered = yield* Deferred.make<void>()
const hook = yield* fixture.hooks.register("session", "prompt", () =>
Deferred.succeed(entered, undefined).pipe(Effect.andThen(Effect.never)),
)
const submission = yield* handle.prompt({ text: "Cancelled before admission" }).pipe(Effect.forkScoped)
yield* Deferred.await(entered)
yield* Fiber.interrupt(submission)
const cancelled = yield* Fiber.await(submission)
expect(Exit.isFailure(cancelled) && Cause.hasInterruptsOnly(cancelled.cause)).toBe(true)
expect(yield* handle.inbox()).toEqual([])
expect((yield* handle.get()).revert?.messageID).toBe(boundary.id)
expect(yield* fixture.store.context(sessionID)).toMatchObject([{ id: boundary.id }])
expect(fixture.wakes).toEqual([])
yield* hook.dispose
yield* handle.revert.clear()
expect((yield* handle.get()).revert).toBeUndefined()
expect(yield* fixture.store.context(sessionID)).toMatchObject([{ id: boundary.id }])
yield* handle.revert.stage({ messageID: boundary.id, files: false })
const acquisitions = fixture.locations.length
yield* fixture.sessions.forSession(sessionID).revert.commit()
expect((yield* handle.get()).revert).toBeUndefined()
expect(yield* fixture.store.context(sessionID)).toEqual([])
expect(fixture.locations).toHaveLength(acquisitions)
}),
)
it.live("selects the destination's constructed revert operations after a move", () =>
Effect.gen(function* () {
const captures: Location.Ref[] = []
const fixture = yield* setup({
snapshot: (ref) =>
Layer.mock(Snapshot.Service, {
capture: () =>
Effect.sync(() => {
captures.push(ref)
return undefined
}),
}),
})
const handle = fixture.sessions.forSession(sessionID)
const boundary = yield* handle.synthetic({ text: "Revert boundary", resume: false })
yield* SessionInbox.promote(fixture.db, fixture.bus, sessionID, "steer")
yield* handle.revert.stage({ messageID: boundary.id, files: false })
const destination = Location.Ref.make({ directory: AbsolutePath.make("/project/moved") })
yield* fixture.bus.publish(SessionEvent.Moved, {
sessionID,
location: destination,
projectID: Project.ID.global,
subpath: RelativePath.make("moved"),
})
yield* handle.revert.stage({ messageID: boundary.id, files: false })
yield* handle.revert.clear()
expect(captures).toEqual([source, destination])
expect(fixture.locations).toEqual([source, destination, destination])
expect(fixture.flushes).toEqual([source, destination, destination])
expect((yield* handle.get()).revert).toBeUndefined()
}),
)
})
describe("SessionPrompt construction", () => {
it.live("captures preparation dependencies without admitting input and checks readiness on every call", () =>
Effect.gen(function* () {
const fixture = yield* setup()
const calls: string[] = []
yield* fixture.hooks.register("session", "prompt", (event) =>
Effect.sync(() => {
calls.push("hook")
event.prompt.text += " prepared"
}),
)
const { prepare } = yield* SessionPrompt.Service.pipe(
Effect.provide(
SessionPrompt.layer.pipe(
Layer.provide(
Layer.mergeAll(
Layer.succeed(PluginHooks.Service, fixture.hooks),
Layer.succeed(PluginSupervisor.Service, {
flush: Effect.sync(() => {
calls.push("ready")
}),
}),
Layer.mock(Image.Service, {}),
Layer.mock(Skill.Service, {}),
),
),
),
),
)
expect(calls).toEqual([])
const input = { text: "Original", files: [{ uri: new URL("./session-owned.test.ts", import.meta.url).href }] }
const request = { sessionID, messageID: SessionMessage.ID.create(), input }
const items = yield* Effect.forEach([0, 1], () => prepare(request)).pipe(
Effect.satisfiesServicesType<never>(),
Effect.setContext(Context.empty()),
)
expect(calls).toEqual(["ready", "hook", "ready", "hook"])
expect(items[0]).toEqual(items[1])
expect(items[0]).toMatchObject({ type: "user", payload: { text: "Original prepared" }, delivery: "steer" })
expect(items[0]?.payload.files?.[0]?.mime).toBe("text/plain")
expect(input.text).toBe("Original")
expect(yield* fixture.sessions.forSession(sessionID).inbox()).toEqual([])
expect(fixture.wakes).toEqual([])
}),
)
})
describe("SessionRevert construction", () => {
it.live("captures dependencies without work, then checks readiness on every stage and clear", () =>
Effect.gen(function* () {
const fixture = yield* setup()
const handle = fixture.sessions.forSession(sessionID)
const boundary = yield* handle.synthetic({ text: "Revert boundary", resume: false })
yield* SessionInbox.promote(fixture.db, fixture.bus, sessionID, "steer")
const calls: string[] = []
const revert = yield* SessionRevert.make().pipe(
Effect.provide(
Layer.merge(
Layer.succeed(PluginSupervisor.Service, {
flush: Effect.sync(() => {
calls.push("flush")
}),
}),
Layer.mock(Snapshot.Service, {
capture: () =>
Effect.sync(() => {
calls.push("capture")
return Snapshot.ID.make("captured-tree")
}),
diff: () =>
Effect.sync(() => {
calls.push("diff")
return []
}),
restore: () =>
Effect.sync(() => {
calls.push("restore")
}),
}),
),
),
)
expect(calls).toEqual([])
const unrelated = Layer.merge(Layer.mock(PluginSupervisor.Service, {}), Layer.mock(Snapshot.Service, {}))
const session = yield* handle.get()
yield* revert
.stage({ session, messageID: boundary.id, files: false })
.pipe(Effect.satisfiesServicesType<never>(), Effect.provide(unrelated))
expect(calls).toEqual(["flush", "capture", "capture", "diff"])
const staged = yield* handle.get()
expect(staged.revert?.snapshot).toBe(Snapshot.ID.make("captured-tree"))
yield* revert.clear(staged).pipe(Effect.satisfiesServicesType<never>(), Effect.provide(unrelated))
const cleared = yield* handle.get()
expect(cleared.revert).toBeUndefined()
yield* revert.clear(cleared).pipe(Effect.satisfiesServicesType<never>(), Effect.provide(unrelated))
expect(calls).toEqual(["flush", "capture", "capture", "diff", "flush", "restore", "flush"])
}),
)
})
describe("SessionInbox command contracts", () => {
it.live("captures the provided Inbox service when constructing Session", () =>
Effect.gen(function* () {
const admission = yield* SessionInbox.Service
const cancelled: SessionMessage.ID[] = []
const fixture = yield* setup().pipe(
Effect.provideService(
SessionInbox.Service,
SessionInbox.Service.of({
...admission,
cancel: (input) =>
admission.cancel(input).pipe(Effect.tap(() => Effect.sync(() => cancelled.push(input.id)))),
}),
),
)
const handle = fixture.sessions.forSession(sessionID)
const pending = yield* handle.synthetic({ text: "Pending", resume: false })
yield* handle.cancelInbox(pending.id).pipe(Effect.setContext(Context.empty()))
expect(cancelled).toEqual([pending.id])
expect(yield* handle.inbox()).toEqual([])
expect(fixture.wakes).toEqual([])
}),
)
it.live("captures the host dependencies for detached commands", () =>
Effect.gen(function* () {
const fixture = yield* setup()
const { list, admit, reconcile, admitCompaction, cancel, steer, queue } = yield* SessionInbox.Service
const other = yield* SessionInbox.make()
expect(yield* SessionInbox.list(fixture.db, sessionID)).toEqual([])
yield* Effect.gen(function* () {
expect(yield* list(sessionID)).toEqual([])
const user = yield* admit({
id: SessionMessage.ID.create(),
sessionID,
item: { type: "user", payload: { text: "Captured services" }, delivery: "queue" },
})
expect(yield* reconcile({ id: user.id, sessionID, type: "user", delivery: "queue" })).toEqual(user)
yield* steer({ id: user.id, sessionID })
yield* queue({ id: user.id, sessionID })
yield* cancel({ id: user.id, sessionID })
const [compaction, duplicate] = yield* Effect.all(
[
admitCompaction({ id: SessionMessage.ID.create(), sessionID, delivery: "queue" }),
other.admitCompaction({ id: SessionMessage.ID.create(), sessionID, delivery: "queue" }),
],
{ concurrency: "unbounded" },
)
expect(compaction).toEqual(duplicate)
yield* cancel({ id: compaction.id, sessionID })
expect(yield* list(sessionID)).toEqual([])
}).pipe(Effect.satisfiesServicesType<never>(), Effect.setContext(Context.empty()))
expect(yield* SessionInbox.list(fixture.db, sessionID)).toEqual([])
expect(fixture.wakes).toEqual([])
}),
)
it.live("returns checked user and synthetic admissions and typed pending or delivered conflicts", () =>
Effect.gen(function* () {
const fixture = yield* setup()
const admission = yield* SessionInbox.Service
const user = yield* admission
.admit({
id: SessionMessage.ID.create(),
sessionID,
item: { type: "user", payload: { text: "Keep user input" }, delivery: "steer" },
})
.pipe(
Effect.satisfiesSuccessType<SessionInbox.User>(),
Effect.satisfiesErrorType<SessionInbox.LifecycleConflict>(),
)
const synthetic = yield* admission
.admit({
id: SessionMessage.ID.create(),
sessionID,
item: { type: "synthetic", payload: { text: "Keep synthetic input" }, delivery: "steer" },
})
.pipe(Effect.satisfiesSuccessType<SessionInbox.Synthetic>())
yield* Effect.forEach([false, true], (delivered) =>
Effect.gen(function* () {
if (delivered) yield* SessionInbox.promote(fixture.db, fixture.bus, sessionID, "steer")
const reconciled = yield* admission
.reconcile({
id: user.id,
sessionID,
type: "user",
delivery: "steer",
})
.pipe(Effect.satisfiesSuccessType<SessionInbox.User | undefined>())
expect(reconciled).toMatchObject({
id: user.id,
sessionID,
type: "user",
payload: user.payload,
delivery: "steer",
})
if (!delivered) expect(reconciled).toEqual(user)
yield* Effect.forEach([user, synthetic], (original) =>
Effect.gen(function* () {
expect(
yield* admission.admit({
id: original.id,
sessionID,
item: { type: original.type, payload: { text: "Ignored retry" }, delivery: "queue" },
}),
).toMatchObject({ id: original.id, sessionID, type: original.type, payload: original.payload })
yield* Effect.forEach(
[
{ sessionID: otherID, type: original.type },
{ sessionID, type: original.type === "user" ? ("synthetic" as const) : ("user" as const) },
],
(conflict) =>
Effect.gen(function* () {
expect(
yield* admission
.reconcile({
...conflict,
id: original.id,
delivery: "steer",
})
.pipe(Effect.flip),
).toBeInstanceOf(SessionInbox.LifecycleConflict)
expect(
yield* admission
.admit({
id: original.id,
sessionID: conflict.sessionID,
item: { type: conflict.type, payload: { text: "Conflicting input" }, delivery: "steer" },
})
.pipe(Effect.flip),
).toMatchObject({ _tag: "SessionInbox.LifecycleConflict", id: original.id })
}),
)
}),
)
}),
)
expect(fixture.locations).toEqual([])
expect(fixture.wakes).toEqual([])
}),
)
it.live("checks the winner of concurrent admissions before returning it", () =>
Effect.gen(function* () {
const fixture = yield* setup()
const admission = yield* SessionInbox.Service
const other = yield* SessionInbox.make()
const id = SessionMessage.ID.create()
const requests = [
{ sessionID, item: { type: "user", payload: { text: "First" }, delivery: "steer" } },
{ sessionID, item: { type: "user", payload: { text: "Retry" }, delivery: "queue" } },
{ sessionID, item: { type: "synthetic", payload: { text: "Other type" }, delivery: "steer" } },
{ sessionID: otherID, item: { type: "user", payload: { text: "Other Session" }, delivery: "steer" } },
] satisfies Array<{ sessionID: SessionSchema.ID; item: SessionInbox.Item }>
const results = yield* Effect.forEach(
requests,
(request, index) => (index % 2 === 0 ? admission : other).admit({ id, ...request }).pipe(Effect.exit),
{ concurrency: "unbounded" },
)
const stored = yield* SessionInbox.find(fixture.db, id)
expect(stored).toBeDefined()
expect(results.some(Exit.isSuccess)).toBe(true)
results.forEach((result, index) => {
if (Exit.isSuccess(result)) {
expect(stored).toEqual(result.value)
expect(result.value.sessionID).toBe(requests[index]?.sessionID)
expect(result.value.type).toBe(requests[index]?.item.type)
return
}
expect(Cause.hasDies(result.cause)).toBe(false)
expect(Cause.hasFails(result.cause)).toBe(true)
})
expect(
(yield* SessionInbox.list(fixture.db, sessionID)).length +
(yield* SessionInbox.list(fixture.db, otherID)).length,
).toBe(1)
}),
)
it.live("exposes failed pending transitions as typed conflicts and rolls back their events", () =>
Effect.gen(function* () {
const fixture = yield* setup()
const admission = yield* SessionInbox.Service
const pending = yield* admission.admit({
id: SessionMessage.ID.create(),
sessionID,
item: { type: "user", payload: { text: "Pending" }, delivery: "queue" },
})
const input = { id: pending.id, sessionID }
yield* Effect.forEach([admission.cancel, admission.steer, admission.queue], (mutation) =>
Effect.gen(function* () {
expect(yield* mutation({ ...input, sessionID: otherID }).pipe(Effect.flip)).toMatchObject({
_tag: "SessionInbox.LifecycleConflict",
id: pending.id,
})
}),
)
yield* admission.steer(input)
expect(yield* admission.steer(input).pipe(Effect.flip)).toBeInstanceOf(SessionInbox.LifecycleConflict)
yield* admission.queue(input)
expect(yield* admission.queue(input).pipe(Effect.flip)).toBeInstanceOf(SessionInbox.LifecycleConflict)
yield* admission.cancel(input)
expect(yield* admission.cancel(input).pipe(Effect.flip)).toBeInstanceOf(SessionInbox.LifecycleConflict)
expect(yield* SessionInbox.list(fixture.db, sessionID)).toEqual([])
expect(
(yield* fixture.db
.select({ type: EventTable.type })
.from(EventTable)
.where(eq(EventTable.aggregate_id, sessionID))
.orderBy(EventTable.seq)
.all()
.pipe(Effect.orDie))
.filter((event) => event.type.startsWith("session.inbox."))
.map((event) => event.type),
).toEqual([
Bus.versionedType(SessionEvent.InboxEnqueued.type, 1),
Bus.versionedType(SessionEvent.InboxDeliveryChanged.type, 1),
Bus.versionedType(SessionEvent.InboxDeliveryChanged.type, 1),
Bus.versionedType(SessionEvent.InboxCancelled.type, 1),
])
}),
)
it.live("does not turn unrelated projector defects into conflicts", () =>
Effect.gen(function* () {
const fixture = yield* setup()
const admission = yield* SessionInbox.Service
const pending = yield* admission.admit({
id: SessionMessage.ID.create(),
sessionID,
item: { type: "user", payload: { text: "Pending" }, delivery: "queue" },
})
const defect = new Error("Projector failed")
yield* fixture.bus.project(SessionEvent.InboxEnqueued, () => Effect.die(defect))
yield* fixture.bus.project(SessionEvent.InboxCancelled, () => Effect.die(defect))
expect(
yield* admission
.admit({
id: SessionMessage.ID.create(),
sessionID,
item: { type: "user", payload: { text: "Rolled back" }, delivery: "steer" },
})
.pipe(Effect.catchDefect(Effect.succeed)),
).toBe(defect)
expect(yield* admission.cancel({ id: pending.id, sessionID }).pipe(Effect.catchDefect(Effect.succeed))).toBe(
defect,
)
expect(yield* SessionInbox.list(fixture.db, sessionID)).toEqual([pending])
}),
)
})
+5 -3
View File
@@ -32,7 +32,7 @@ import { testEffect } from "./lib/effect"
import { Snapshot } from "@opencode-ai/core/snapshot"
const it = testEffect(
AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionProjector.node]), [
AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionInbox.node]), [
[Bus.node, Bus.configured({ persist: true })],
]),
)
@@ -87,8 +87,9 @@ describe("SessionProjector", () => {
Effect.gen(function* () {
const db = yield* seedSession()
const bus = yield* Bus.Service
const inbox = yield* SessionInbox.Service
const inputID = SessionMessage.ID.make("msg_manual_compaction")
yield* SessionInbox.admitCompaction(db, bus, { id: inputID, sessionID, delivery: "queue" })
yield* inbox.admitCompaction({ id: inputID, sessionID, delivery: "queue" })
yield* bus.publish(SessionEvent.Compaction.Failed, {
sessionID,
@@ -290,8 +291,9 @@ describe("SessionProjector", () => {
Effect.gen(function* () {
const db = yield* seedSession()
const bus = yield* Bus.Service
const inbox = yield* SessionInbox.Service
const id = SessionMessage.ID.make("msg_admitted")
const admitted = yield* SessionInbox.admit(db, bus, {
const admitted = yield* inbox.admit({
id,
sessionID,
item: { type: "user", payload: { text: "promote me" }, delivery: "steer" },
+60 -30
View File
@@ -7,8 +7,11 @@ import { Database } from "@opencode-ai/core/database/database"
import { Agent } from "@opencode-ai/core/agent"
import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Bus } from "@opencode-ai/core/bus"
import { EventTable } from "@opencode-ai/core/event/sql"
import { Location } from "@opencode-ai/schema/location"
import { SessionEvent } from "@opencode-ai/core/session/event"
import { Model } from "@opencode-ai/core/model"
import { Provider } from "@opencode-ai/core/provider"
@@ -17,7 +20,9 @@ import { ProjectTable } from "@opencode-ai/core/project/sql"
import { AbsolutePath } from "@opencode-ai/core/schema"
import { Session } from "@opencode-ai/core/session"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { SessionPrompt } from "@opencode-ai/core/session/prompt"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionRevert } from "@opencode-ai/core/session/revert"
import { SessionExecution } from "@opencode-ai/core/session/execution"
import { SessionInbox } from "@opencode-ai/core/session/inbox"
import { SessionInboxTable, SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql"
@@ -28,6 +33,7 @@ import { Image } from "@opencode-ai/core/image"
import { PluginSupervisor } from "@opencode-ai/core/plugin/supervisor"
import { PluginHooks } from "@opencode-ai/core/plugin/hooks"
import { Snapshot } from "@opencode-ai/core/snapshot"
import { Skill } from "@opencode-ai/core/skill"
import { tmpdirScoped } from "./fixture/tmpdir"
import { testEffect } from "./lib/effect"
@@ -40,6 +46,7 @@ const execution = Layer.succeed(
SessionExecution.Service,
SessionExecution.Service.of({
active: Effect.sync(() => new Set(activeSessions)),
isActive: (sessionID) => Effect.sync(() => activeSessions.has(sessionID)),
resume: (sessionID) =>
Effect.sync(() => {
executionCalls.push(sessionID)
@@ -57,37 +64,60 @@ const execution = Layer.succeed(
awaitIdle: () => Effect.void,
}),
)
const locations = Layer.effect(
LocationServiceMap.Service,
LayerMap.make(
() =>
// These operations resolve Location services lazily and must wait for plugin-projected state.
// oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion
Layer.unwrap(
Effect.sync(() => {
let ready = false
return Layer.mergeAll(
LayerNode.compile(PluginHooks.node),
Layer.mock(Image.Service, {
normalize: (_resource, content) =>
ready
? Effect.succeed(content.content.length > 5 * 1024 * 1024 ? { ...content, content: "AA==" } : content)
: Effect.die(new Error("Image service used before plugins were ready")),
}),
Layer.mock(Snapshot.Service, {
capture: () =>
ready ? Effect.undefined : Effect.die(new Error("Snapshot used before plugins were ready")),
restore: () => (ready ? Effect.void : Effect.die(new Error("Snapshot used before plugins were ready"))),
}),
Layer.succeed(
PluginSupervisor.Service,
PluginSupervisor.Service.of({ flush: Effect.sync(() => (ready = true)) }),
),
)
}),
) as unknown as Layer.Layer<LocationServices>,
const locations = makeGlobalNode({
service: LocationServiceMap.Service,
layer: Layer.effect(
LocationServiceMap.Service,
Effect.gen(function* () {
const database = yield* Database.Service
const bus = yield* Bus.Service
const fs = yield* FSUtil.Service
const shared = Layer.mergeAll(
Layer.succeed(Database.Service, database),
Layer.succeed(Bus.Service, bus),
Layer.succeed(FSUtil.Service, fs),
)
return yield* LayerMap.make(
(_ref: Location.Ref) =>
// These operations resolve Location services lazily and must wait for plugin-projected state.
// oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion
Layer.suspend(() => {
let ready = false
return Layer.merge(SessionRevert.layer, SessionPrompt.layer).pipe(
Layer.provideMerge(
Layer.mergeAll(
LayerNode.compile(LayerNode.group([PluginHooks.node, Skill.node]), [
[Bus.node, Layer.succeed(Bus.Service, bus)],
]),
Layer.mock(Image.Service, {
normalize: (_resource, content) =>
ready
? Effect.succeed(
content.content.length > 5 * 1024 * 1024 ? { ...content, content: "AA==" } : content,
)
: Effect.die(new Error("Image service used before plugins were ready")),
}),
Layer.mock(Snapshot.Service, {
capture: () =>
ready ? Effect.undefined : Effect.die(new Error("Snapshot used before plugins were ready")),
restore: () =>
ready ? Effect.void : Effect.die(new Error("Snapshot used before plugins were ready")),
}),
Layer.succeed(
PluginSupervisor.Service,
PluginSupervisor.Service.of({ flush: Effect.sync(() => (ready = true)) }),
),
),
),
Layer.provide(shared),
Layer.fresh,
)
}) as unknown as Layer.Layer<LocationServices>,
)
}),
),
)
deps: [Database.node, Bus.node, FSUtil.node],
})
const it = testEffect(
AppNodeBuilder.build(
LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node, Session.node]),
+2 -2
View File
@@ -15,7 +15,7 @@ import { SessionStore } from "@opencode-ai/core/session/store"
import { SessionEnvironment } from "@opencode-ai/core/session/environment"
import { LocationServiceMap } from "@opencode-ai/core/location-services"
import { testEffect } from "./lib/effect"
import { globalProjectLayer } from "./lib/project"
import { globalProjectNode } from "./lib/project"
import { tmpdirScoped } from "./fixture/tmpdir"
const closed: Session.ID[] = []
@@ -39,7 +39,7 @@ const it = testEffect(
LocationServiceMap.node,
]),
[
[Project.node, globalProjectLayer],
[Project.node, globalProjectNode],
[SessionExecution.node, SessionExecution.noopLayer],
[SessionModelTransport.node, transport],
],
@@ -18,6 +18,7 @@ import { SessionExecution } from "@opencode-ai/core/session/execution"
import { SessionInbox } from "@opencode-ai/core/session/inbox"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionRevert } from "@opencode-ai/core/session/revert"
import { Snapshot } from "@opencode-ai/core/snapshot"
import { Money } from "@opencode-ai/schema/money"
import { LayerNode } from "@opencode-ai/util/effect/layer-node"
@@ -60,6 +61,9 @@ describe("Session.revert files", () => {
const created = yield* session.create({ location: { directory: AbsolutePath.make(directory) } })
const prompt = yield* session.prompt({ sessionID: created.id, text: "Rename the file", resume: false })
yield* SessionInbox.promote(database.db, bus, created.id, "steer")
const services = LocationServiceMap.Service.get(created.location)
const revert = yield* SessionRevert.Service.pipe(Effect.provide(services))
expect(yield* SessionRevert.Service.pipe(Effect.provide(services))).toBe(revert)
yield* Effect.gen(function* () {
const plugins = yield* PluginSupervisor.Service
@@ -73,22 +73,30 @@ describe("SessionRunCoordinator", () => {
Effect.andThen(Deferred.await(key === "first" ? firstGate : secondGate)),
),
})
const firstActive = coordinator.isActive("first")
expect(Array.from(yield* coordinator.active)).toEqual([])
expect(yield* firstActive).toBe(false)
const first = yield* coordinator.run("first").pipe(Effect.forkChild)
yield* Deferred.await(firstStarted)
expect(Array.from(yield* coordinator.active)).toEqual(["first"])
expect(yield* firstActive).toBe(true)
expect(yield* coordinator.isActive("second")).toBe(false)
const second = yield* coordinator.run("second").pipe(Effect.forkChild)
yield* Deferred.await(secondStarted)
expect(Array.from(yield* coordinator.active)).toEqual(["first", "second"])
expect(yield* coordinator.isActive("second")).toBe(true)
yield* Deferred.succeed(firstGate, undefined)
yield* Fiber.join(first)
expect(Array.from(yield* coordinator.active)).toEqual(["second"])
expect(yield* firstActive).toBe(false)
expect(yield* coordinator.isActive("second")).toBe(true)
yield* Deferred.succeed(secondGate, undefined)
yield* Fiber.join(second)
expect(Array.from(yield* coordinator.active)).toEqual([])
expect(yield* coordinator.isActive("second")).toBe(false)
}),
)
@@ -105,10 +113,12 @@ describe("SessionRunCoordinator", () => {
const failed = yield* coordinator.run("failure").pipe(Effect.exit)
expect(Exit.isFailure(failed) && Cause.hasFails(failed.cause)).toBeTrue()
expect(Array.from(yield* coordinator.active)).toEqual([])
expect(yield* coordinator.isActive("failure")).toBe(false)
const died = yield* coordinator.run("defect").pipe(Effect.exit)
expect(Exit.isFailure(died) && Cause.hasDies(died.cause)).toBeTrue()
expect(Array.from(yield* coordinator.active)).toEqual([])
expect(yield* coordinator.isActive("defect")).toBe(false)
expect(settled).toHaveLength(2)
}),
)
@@ -612,6 +622,7 @@ describe("SessionRunCoordinator", () => {
yield* coordinator.interrupt("session", "user")
expect(settled).toHaveLength(0)
expect(Array.from(yield* coordinator.active)).toEqual(["session"])
expect(yield* coordinator.isActive("session")).toBe(true)
// Repeating the interrupt during cleanup stays an immediate no-op.
yield* coordinator.interrupt("session", "user")
@@ -621,6 +632,7 @@ describe("SessionRunCoordinator", () => {
expect(settled).toEqual(["user"])
expect(yield* coordinator.active).toEqual(new Set())
expect(yield* coordinator.isActive("session")).toBe(false)
}),
)
@@ -636,6 +648,7 @@ describe("SessionRunCoordinator", () => {
yield* coordinator.wake("session")
yield* Deferred.await(settling)
expect(yield* coordinator.isActive("session")).toBe(true)
// The owner has exited; this wake lands on the settling execution's doorbell.
yield* coordinator.wake("session")
// The interrupt claims it: settle must not start a successor for the dead intent.
@@ -645,6 +658,7 @@ describe("SessionRunCoordinator", () => {
expect(drains).toBe(1)
expect(yield* coordinator.active).toEqual(new Set())
expect(yield* coordinator.isActive("session")).toBe(false)
}),
)
@@ -42,7 +42,7 @@ import { HttpClient, HttpClientResponse } from "effect/unstable/http"
import path from "node:path"
import { testEffect } from "./lib/effect"
import { LocationServiceMap } from "@opencode-ai/core/location-service-map"
import { promptLocationLayer } from "./fixture/prompt-location"
import { promptLocationNode } from "./fixture/prompt-location"
import { permissionLayer } from "./lib/permission"
import { agentHost, catalogHost, host } from "./plugin/host"
@@ -125,6 +125,7 @@ const execution = (llmClient: Layer.Layer<typeof LLMClient.Service>) =>
})
return SessionExecution.Service.of({
active: coordinator.active,
isActive: coordinator.isActive,
resume: coordinator.run,
wake: coordinator.wake,
interrupt: (sessionID) => coordinator.interrupt(sessionID),
@@ -155,7 +156,7 @@ const testLayer = (llmClient: Layer.Layer<typeof LLMClient.Service>) =>
]),
[
[Bus.node, Bus.configured({ persist: true })],
[LocationServiceMap.node, promptLocationLayer],
[LocationServiceMap.node, promptLocationNode],
[LayerNodePlatform.llmClient, llmClient],
[Permission.node, permission],
[Catalog.node, promptCatalog],
+16 -12
View File
@@ -81,7 +81,7 @@ import { TestClock } from "effect/testing"
import { HttpClientRequest, HttpClientResponse } from "effect/unstable/http"
import { asc, desc, eq, sql } from "drizzle-orm"
import { testEffect } from "./lib/effect"
import { promptLocationLayer } from "./fixture/prompt-location"
import { promptLocationNode } from "./fixture/prompt-location"
import { LocationServiceMap } from "@opencode-ai/core/location-service-map"
import { Expected } from "./lib/session-message"
import { permissionLayer } from "./lib/permission"
@@ -447,6 +447,7 @@ const layer = Layer.unwrap(
})
return SessionExecution.Service.of({
active: coordinator.active,
isActive: coordinator.isActive,
resume: coordinator.run,
wake: coordinator.wake,
interrupt: (sessionID) => coordinator.interrupt(sessionID),
@@ -461,6 +462,7 @@ const layer = Layer.unwrap(
Form.node,
SessionProjector.node,
SessionStore.node,
SessionInbox.node,
Agent.node,
Catalog.node,
Tool.node,
@@ -484,7 +486,7 @@ const layer = Layer.unwrap(
[
...replacements,
[Bus.node, Bus.configured({ persist: true })],
[LocationServiceMap.node, promptLocationLayer],
[LocationServiceMap.node, promptLocationNode],
[Catalog.node, promptCatalog],
[SessionExecution.node, execution],
],
@@ -516,6 +518,7 @@ const insertSession = (id: Session.ID) =>
const setup = Effect.gen(function* () {
const { db } = yield* Database.Service
const bus = yield* Bus.Service
const sessionInbox = yield* SessionInbox.Service
const agents = yield* Agent.Service
const catalog = yield* Catalog.Service
const hooks = yield* PluginHooks.Service
@@ -547,6 +550,7 @@ const setup = Effect.gen(function* () {
return Object.assign(state, {
db,
bus,
sessionInbox,
session,
llm,
requests: llm.requests,
@@ -1387,12 +1391,12 @@ describe("SessionRunnerLLM", () => {
s.systemLoadHook = Effect.sync(() => {
reads++
})
const compaction = yield* SessionInbox.admitCompaction(s.db, s.bus, {
const compaction = yield* s.sessionInbox.admitCompaction({
id: SessionMessage.ID.create(),
sessionID,
delivery: "queue",
})
yield* SessionInbox.admit(s.db, s.bus, {
yield* s.sessionInbox.admit({
id: SessionMessage.ID.create(),
sessionID,
item: {
@@ -1420,7 +1424,7 @@ describe("SessionRunnerLLM", () => {
scenario("delivers a queued move atomically at the idle boundary", function* (s) {
const inboxID = SessionMessage.ID.create()
yield* SessionInbox.admit(s.db, s.bus, {
yield* s.sessionInbox.admit({
id: inboxID,
sessionID,
item: {
@@ -1456,7 +1460,7 @@ describe("SessionRunnerLLM", () => {
const tools = yield* s.blockTools()
const run = yield* s.resume.pipe(Effect.forkChild)
yield* tools.started
yield* SessionInbox.admit(s.db, s.bus, {
yield* s.sessionInbox.admit({
id: SessionMessage.ID.create(),
sessionID,
item: {
@@ -1488,7 +1492,7 @@ describe("SessionRunnerLLM", () => {
const run = yield* s.resume.pipe(Effect.forkChild)
yield* tools.started
yield* s.session.prompt({ sessionID, text: "Queued for later", delivery: "queue", resume: false })
yield* SessionInbox.admit(s.db, s.bus, {
yield* s.sessionInbox.admit({
id: SessionMessage.ID.create(),
sessionID,
item: {
@@ -1521,12 +1525,12 @@ describe("SessionRunnerLLM", () => {
const stream = yield* s.llm.gate
const run = yield* runner.drain({ sessionID, force: false }).pipe(Effect.forkChild)
yield* stream.started
const compaction = yield* SessionInbox.admitCompaction(s.db, s.bus, {
const compaction = yield* s.sessionInbox.admitCompaction({
id: SessionMessage.ID.create(),
sessionID,
delivery: "queue",
})
yield* SessionInbox.admit(s.db, s.bus, {
yield* s.sessionInbox.admit({
id: SessionMessage.ID.create(),
sessionID,
item: {
@@ -3207,7 +3211,7 @@ describe("SessionRunnerLLM", () => {
const run = yield* runner.drain({ sessionID, force: false }).pipe(Effect.forkChild)
yield* stream.started
yield* SessionInbox.admit(s.db, s.bus, {
yield* s.sessionInbox.admit({
id: SessionMessage.ID.create(),
sessionID,
item: {
@@ -3319,7 +3323,7 @@ describe("SessionRunnerLLM", () => {
scenario("a steer-scoped drain runs a queued manual compaction next in line", function* (s) {
// Admit without waking so the steer-scoped drain below is the first consumer.
const compaction = yield* SessionInbox.admitCompaction(s.db, s.bus, {
const compaction = yield* s.sessionInbox.admitCompaction({
id: SessionMessage.ID.create(),
sessionID,
delivery: "queue",
@@ -3340,7 +3344,7 @@ describe("SessionRunnerLLM", () => {
scenario("a steer-scoped drain leaves a compaction parked behind a queued prompt", function* (s) {
yield* s.session.prompt({ sessionID, text: "Queue for later", delivery: "queue", resume: false })
const compaction = yield* SessionInbox.admitCompaction(s.db, s.bus, {
const compaction = yield* s.sessionInbox.admitCompaction({
id: SessionMessage.ID.create(),
sessionID,
delivery: "queue",
+1
View File
@@ -42,6 +42,7 @@ const executionLayer = Layer.effect(
})
return SessionExecution.Service.of({
active: coordinator.active,
isActive: coordinator.isActive,
resume: coordinator.run,
interrupt: (sessionID) => coordinator.interrupt(sessionID),
awaitIdle: coordinator.awaitIdle,
+34 -20
View File
@@ -4,7 +4,10 @@ import { Effect, Layer, LayerMap } 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/util/effect/layer-node"
import { makeGlobalNode } from "@opencode-ai/util/effect/app-node"
import { FSUtil } from "@opencode-ai/util/fs-util"
import { Bus } from "@opencode-ai/core/bus"
import { Image } from "@opencode-ai/core/image"
import { Location } from "@opencode-ai/core/location"
import { LocationServiceMap } from "@opencode-ai/core/location-service-map"
import type { LocationServices } from "@opencode-ai/core/location-services"
@@ -15,16 +18,15 @@ import { AbsolutePath } from "@opencode-ai/core/schema"
import { Session } from "@opencode-ai/core/session"
import { SessionExecution } from "@opencode-ai/core/session/execution"
import { SessionMessage } from "@opencode-ai/core/session/message"
import { SessionPrompt } from "@opencode-ai/core/session/prompt"
import { SessionProjector } from "@opencode-ai/core/session/projector"
import { SessionStore } from "@opencode-ai/core/session/store"
import { SessionInbox } from "@opencode-ai/core/session/inbox"
import { Skill } from "@opencode-ai/core/skill"
import { testEffect } from "./lib/effect"
import { globalProjectNode } from "./lib/project"
const location = Location.Ref.make({ directory: AbsolutePath.make("/project") })
const projects = Layer.mock(Project.Service, {
resolve: (directory) => Effect.succeed({ id: Project.ID.global, directory, canonical: directory }),
})
const info = Skill.Info.make({
id: Skill.ID.make("effect"),
name: Skill.Name.make("Effect"),
@@ -32,29 +34,41 @@ const info = Skill.Info.make({
location: AbsolutePath.make(path.resolve("/skills/effect.md")),
content: "Use Effect",
})
const skills = Layer.mergeAll(
LayerNode.compile(PluginHooks.node),
Layer.mock(Skill.Service, {
get: (id) => Effect.succeed(id === info.id ? info : undefined),
list: () => Effect.succeed([info]),
}),
Layer.succeed(PluginSupervisor.Service, { flush: Effect.void }),
)
const locations = Layer.effect(
LocationServiceMap.Service,
LayerMap.make(
() =>
// The skill endpoint only needs the location-scoped Skill service.
// oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion
skills as unknown as Layer.Layer<LocationServices>,
const locations = makeGlobalNode({
service: LocationServiceMap.Service,
layer: Layer.effect(
LocationServiceMap.Service,
Effect.gen(function* () {
const fs = yield* FSUtil.Service
const skills = SessionPrompt.layer.pipe(
Layer.provideMerge(
Layer.mergeAll(
LayerNode.compile(LayerNode.group([PluginHooks.node, Image.node])),
Layer.succeed(FSUtil.Service, fs),
Layer.mock(Skill.Service, {
get: (id) => Effect.succeed(id === info.id ? info : undefined),
list: () => Effect.succeed([info]),
}),
Layer.succeed(PluginSupervisor.Service, { flush: Effect.void }),
),
),
)
return yield* LayerMap.make(
(_ref: Location.Ref) =>
// These tests need skill activation and prompt preparation from the same location services.
// oxlint-disable-next-line typescript-eslint/no-unsafe-type-assertion
skills as unknown as Layer.Layer<LocationServices>,
)
}),
),
)
deps: [FSUtil.node],
})
const it = testEffect(
AppNodeBuilder.build(
LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node, Session.node]),
[
[LocationServiceMap.node, locations],
[Project.node, projects],
[Project.node, globalProjectNode],
[SessionExecution.node, SessionExecution.noopLayer],
],
),
+2 -2
View File
@@ -19,14 +19,14 @@ import { DateTime, Effect, Layer } from "effect"
import { asc, eq } from "drizzle-orm"
import { tmpdirScoped } from "./fixture/tmpdir"
import { testEffect } from "./lib/effect"
import { globalProjectLayer } from "./lib/project"
import { globalProjectNode } from "./lib/project"
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],
[Project.node, globalProjectNode],
[SessionExecution.node, SessionExecution.noopLayer],
],
),
+2 -4
View File
@@ -12,12 +12,10 @@ 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 { testEffect } from "./lib/effect"
import { globalProjectNode } from "./lib/project"
const location = Location.Ref.make({ directory: AbsolutePath.make("/project") })
const awaited: Session.ID[] = []
const projects = Layer.mock(Project.Service, {
resolve: (directory) => Effect.succeed({ id: Project.ID.global, directory, canonical: directory }),
})
const execution = Layer.mock(SessionExecution.Service, {
awaitIdle: (sessionID) => Effect.sync(() => awaited.push(sessionID)),
})
@@ -25,7 +23,7 @@ const it = testEffect(
AppNodeBuilder.build(
LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node, Session.node]),
[
[Project.node, projects],
[Project.node, globalProjectNode],
[SessionExecution.node, execution],
],
),
+1
View File
@@ -114,6 +114,7 @@ const executionNode = makeGlobalNode({
})
return SessionExecution.Service.of({
active: Effect.succeed(new Set()),
isActive: () => Effect.succeed(false),
resume: complete,
wake: () => Effect.void,
interrupt: () => Effect.succeed(false),
+1
View File
@@ -92,6 +92,7 @@ const executionNode = makeGlobalNode({
})
return SessionExecution.Service.of({
active: Effect.succeed(new Set()),
isActive: () => Effect.succeed(false),
resume: complete,
wake: () => Effect.void,
interrupt: () => Effect.succeed(false),
@@ -26,6 +26,7 @@ it.live("updates completed assistant message content through the session HTTP AP
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* () {
+29 -1
View File
@@ -6,7 +6,7 @@ Status: **Current semantic overview.** Protocol owns public operations, Schema o
`Session.prompt(...)` publishes one durable `session.inbox.enqueued` fact whose projection inserts one `session_inbox` row before advisory execution begins. An inbox item remains outside model-visible Session History until delivery. The `session.inbox.delivered` projection consumes the row and inserts a visible user or synthetic message atomically; compaction and move control items are consumed without becoming transcript messages.
Reusing a Session ID adopts the existing Session. While a user or synthetic item remains pending, reusing its ID reconciles only when Session, item type, complete payload, metadata, and delivery match; conflicting reuse fails. After delivery, retry reconciliation for those message-producing items uses the projected message and does not require enqueue history or the original delivery mode. Compaction and move controls retain operation-specific conflict behavior.
Reusing a Session ID adopts the existing Session. Reusing a user or synthetic inbox item ID is idempotent when Session and type match: the first admission wins, and retried payload, metadata, and delivery are ignored. After delivery, retry reconciliation uses the projected message and does not require enqueue history. Cross-Session and cross-type reuse fail. Compaction and move controls retain operation-specific conflict behavior.
`resume` controls scheduling, not durability:
@@ -22,6 +22,34 @@ Promoting new user input resets the selected agent's step allowance. A batch of
Manual compaction and Session movement use the same inbox as control items. Each request has its own inbox identity and delivery mode. A control item forms a delivery boundary so later steers do not cross it.
## Session Operations Own Admission Policy
Core's ID-addressed Session facade delegates prompt, synthetic, compaction, pending-input, shell, revert, execution controls, renaming, agent/model selection, viewed acknowledgements, and message lookup/editing to the ID-bound Session values in `session/session.ts`. Those operations own request idempotency, admission ordering, wake policy, and per-Session state-change rules. The facade supplies lazy Location services; the lower operations do not depend on `LocationServiceMap` or call back into the facade. Collection operations and host-infrastructure routing remain in the facade. Project resolution persists the owning Project; Session does not repeat that write.
The host acquires `Session.make` once within its host `Scope`, then selects ID-bound values through `forSession`. The factory captures that Scope for shell completion recording; it must outlive individual callers. The existing facade remains the Effect service; the lower factory does not introduce a second service or a Layer per Session ID.
```ts
Effect.gen(function* () {
const sessions = yield* Session.make((ref) => locations.get(ref))
const session = sessions.forSession(sessionID)
yield* session.prompt({ text: "Inspect the failing tests", resume: false })
})
```
References share operation implementations, host services, and the existing execution coordinator. A Session value retains its ID, not a cached projection or permanently selected runner. Obtaining or discarding a value does not start or interrupt execution. Admission stays independently callable while execution is active.
User shell commands start immediately as background work without waiting for model execution or other user shells. They do not suppress prompt wakeups. The shell operation forks completion recording into the captured host Scope with `startImmediately: true` and joins that fiber, so caller cancellation does not cancel recording; closing the host Scope does. Shell started and ended events retain output in one shell entry. Completion and startup-failure notifications are admitted as synthetic input with `resume: false`, without waking execution. Shell services are resolved at startup, while Session events remain outside that Location context so movement does not pin them to the old Location.
Location services are acquired only when an operation needs them. In particular, retry reconciliation happens before prompt preparation, so an already-admitted input skips hooks and attachment resolution. Execution continues to resolve placement independently at drain start and after movement.
`servicesFor` selects instance services from the saved placement. Each instance constructs `SessionPrompt.Service`, whose `prepare` method turns submitted input into a user inbox item without admitting it, committing a revert, or waking execution. It captures FSUtil, PluginSupervisor, PluginHooks, Image, and Skill; readiness is checked before hooks on every call. Prompt keeps early retry reconciliation outside preparation and invokes preparation interruptibly in the current instance. Lower Session does not depend directly on Database or FSUtil; its Location requirements are SessionPrompt, SessionRevert, Shell, and the PluginSupervisor still used by manual shell startup.
Each instance constructs its `SessionRevert.Service` through `SessionRevert.make`, capturing Database, Bus, PluginSupervisor, and Snapshot. Stage and clear check plugin readiness on each invocation and require no service provisioning inside their implementations. Session methods select the current instance for each operation, so an ID-bound Session does not retain a previous instance's snapshots after movement. Commit uses only the host's captured Bus and does not acquire an instance.
`SessionInbox.Service` is host-scoped. Its node depends on Database and Bus, and its layer uses `SessionInbox.make` to capture those services and construct admission and pending-input commands that take only domain inputs. Its `list` method supplies the normal pending-input read without exposing Database to Session. `Session.make` and the facade's move admission consume the registered service rather than constructing separate command objects. The service is not Session-ID scoped and does not own execution. Its commands retain the existing shared inbox serialization lock. Standalone query helpers, transaction-facing projectors, and runner delivery retain their explicit database/Bus inputs. The layer compiler is unchanged.
Inbox commands own identity and type checks and return typed `SessionInbox.LifecycleConflict` errors. Session operations translate these into their public operation-specific errors and decide whether to wake execution. The Bus/projector boundary still uses defects to abort invalid projections; Inbox translates only lifecycle conflicts, not unrelated defects. Pending-input mutation does not schedule execution itself: steering wakes after a successful mutation, while queueing and cancellation do not.
## Execution Is Process-Local
`SessionExecution` is process-global and keyed only by Session ID. At drain start it loads the Session, enters its Location through `LocationServiceMap`, and invokes the Location-scoped runner. The runner, model resolution, tools, permissions, plugins, and filesystem remain Location-scoped.