mirror of
https://github.com/anomalyco/opencode.git
synced 2026-08-30 05:12:23 +00:00
6e954f75ee
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.
976 lines
42 KiB
TypeScript
976 lines
42 KiB
TypeScript
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])
|
|
}),
|
|
)
|
|
})
|