1
0

session-create.test.ts 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742
  1. import { describe, expect } from "bun:test"
  2. import path from "path"
  3. import { DateTime, Effect, Layer, Stream } from "effect"
  4. import { Money } from "@opencode-ai/schema/money"
  5. import { Agent } from "@opencode-ai/core/agent"
  6. import { asc, eq } from "drizzle-orm"
  7. import { Database } from "@opencode-ai/core/database/database"
  8. import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
  9. import { LayerNode } from "@opencode-ai/util/effect/layer-node"
  10. import { Bus } from "@opencode-ai/core/bus"
  11. import { EventTable } from "@opencode-ai/core/event/sql"
  12. import { Location } from "@opencode-ai/core/location"
  13. import { Model } from "@opencode-ai/core/model"
  14. import { Project } from "@opencode-ai/core/project"
  15. import { ProjectTable } from "@opencode-ai/core/project/sql"
  16. import { Provider } from "@opencode-ai/core/provider"
  17. import { AbsolutePath, RelativePath } from "@opencode-ai/core/schema"
  18. import { Session } from "@opencode-ai/core/session"
  19. import { SessionMessage } from "@opencode-ai/core/session/message"
  20. import { SessionProjector } from "@opencode-ai/core/session/projector"
  21. import { SessionExecution } from "@opencode-ai/core/session/execution"
  22. import { SessionPending } from "@opencode-ai/core/session/pending"
  23. import { SessionEvent } from "@opencode-ai/core/session/event"
  24. import { SessionTable } from "@opencode-ai/core/session/sql"
  25. import { SessionStore } from "@opencode-ai/core/session/store"
  26. import { Workspace } from "@opencode-ai/core/workspace"
  27. import { testEffect } from "./lib/effect"
  28. import { tmpdir } from "./fixture/tmpdir"
  29. const projects = Layer.succeed(
  30. Project.Service,
  31. Project.Service.of({
  32. list: () => Effect.succeed([]),
  33. resolve: (directory) => Effect.succeed({ id: Project.ID.global, directory, canonical: directory }),
  34. directories: () => Effect.succeed([]),
  35. }),
  36. )
  37. const it = testEffect(
  38. AppNodeBuilder.build(
  39. LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node, Session.node]),
  40. [
  41. [Bus.node, Bus.configured({ persist: true })],
  42. [Project.node, projects],
  43. [SessionExecution.node, SessionExecution.noopLayer],
  44. ],
  45. ),
  46. )
  47. const location = Location.Ref.make({ directory: AbsolutePath.make("/project") })
  48. const id = Session.ID.create()
  49. /** Public session events from a `log` read, without synced markers. */
  50. const logEvents = (session: Session.Interface, sessionID: Session.ID, follow?: boolean) =>
  51. session
  52. .log({ sessionID, follow })
  53. .pipe(Stream.filter((item): item is SessionEvent.DurableEvent => !Bus.isSynced(item)))
  54. const assertCreateInputTypes = (session: Session.Interface) => {
  55. // @ts-expect-error location or parentID is required.
  56. session.create({})
  57. // @ts-expect-error child sessions inherit their parent's location.
  58. session.create({ parentID: Session.ID.create(), location })
  59. }
  60. void assertCreateInputTypes
  61. function withTmp<A, E, R>(f: (directory: string) => Effect.Effect<A, E, R>) {
  62. return Effect.acquireRelease(
  63. Effect.promise(() => tmpdir()),
  64. (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
  65. ).pipe(Effect.flatMap((tmp) => f(tmp.path)))
  66. }
  67. describe("Session.create", () => {
  68. it.effect("persists a missing title until one is generated or supplied", () =>
  69. Effect.gen(function* () {
  70. const session = yield* Session.Service
  71. const { db } = yield* Database.Service
  72. const created = yield* session.create({ location })
  73. const row = yield* db.select().from(SessionTable).where(eq(SessionTable.id, created.id)).get().pipe(Effect.orDie)
  74. const event = yield* db
  75. .select({ data: EventTable.data })
  76. .from(EventTable)
  77. .where(eq(EventTable.aggregate_id, created.id))
  78. .get()
  79. .pipe(Effect.orDie)
  80. expect(created.title).toBeUndefined()
  81. expect(row?.title).toBeNull()
  82. expect(event?.data).not.toHaveProperty("title")
  83. expect((yield* session.create({ location, title: "Explicit title" })).title).toBe("Explicit title")
  84. }),
  85. )
  86. it.effect("creates a fresh projected session when the ID is omitted", () =>
  87. Effect.gen(function* () {
  88. const session = yield* Session.Service
  89. const first = yield* session.create({ location })
  90. const second = yield* session.create({ location })
  91. expect(second.id).not.toBe(first.id)
  92. expect((yield* session.list()).data).toHaveLength(2)
  93. }),
  94. )
  95. it.effect("returns the original session when the ID is retried", () =>
  96. Effect.gen(function* () {
  97. const session = yield* Session.Service
  98. const input = { id, location }
  99. const first = yield* session.create(input)
  100. const retried = yield* session.create(input)
  101. expect(retried).toEqual(first)
  102. expect((yield* session.list()).data).toEqual([first])
  103. }),
  104. )
  105. it.effect("stores supplied immutable create attributes", () =>
  106. Effect.gen(function* () {
  107. const session = yield* Session.Service
  108. const workspaceID = Workspace.ID.make("wrk_test")
  109. const model = Model.Ref.make({
  110. id: Model.ID.make("sonnet"),
  111. providerID: Provider.ID.anthropic,
  112. variant: Model.VariantID.make("fast"),
  113. })
  114. expect(
  115. yield* session.create({
  116. location: Location.Ref.make({ directory: location.directory, workspaceID }),
  117. agent: Agent.ID.make("build"),
  118. model,
  119. }),
  120. ).toMatchObject({ location: { directory: location.directory, workspaceID }, agent: "build", model })
  121. }),
  122. )
  123. it.effect("inherits location from an existing parent when omitted", () =>
  124. Effect.gen(function* () {
  125. const session = yield* Session.Service
  126. const parent = yield* session.create({ location })
  127. const child = yield* session.create({ parentID: parent.id, title: "child" })
  128. expect(child).toMatchObject({ parentID: parent.id, location })
  129. }),
  130. )
  131. it.effect("rejects child creation when the parent does not exist", () =>
  132. Effect.gen(function* () {
  133. const session = yield* Session.Service
  134. const missing = Session.ID.create()
  135. expect(yield* Effect.flip(session.create({ parentID: missing, title: "child" }))).toEqual(
  136. new Session.NotFoundError({ sessionID: missing }),
  137. )
  138. }),
  139. )
  140. it.effect("filters root sessions before applying the page limit", () =>
  141. Effect.gen(function* () {
  142. const session = yield* Session.Service
  143. const { db } = yield* Database.Service
  144. const staleRoot = yield* session.create({ location, title: "stale root" })
  145. const root = yield* session.create({ location, title: "root" })
  146. const children = yield* Effect.forEach(Array.from({ length: 60 }), (_, index) =>
  147. session.create({ parentID: root.id, title: `child ${index}` }),
  148. )
  149. yield* Effect.forEach(children, (item, index) =>
  150. db
  151. .update(SessionTable)
  152. .set({ time_created: index + 100, time_updated: index + 20_000 })
  153. .where(eq(SessionTable.id, item.id))
  154. .run(),
  155. )
  156. yield* db
  157. .update(SessionTable)
  158. .set({ time_created: 2, time_updated: 5_000 })
  159. .where(eq(SessionTable.id, staleRoot.id))
  160. .run()
  161. yield* db
  162. .update(SessionTable)
  163. .set({ time_created: 1, time_updated: 10_000 })
  164. .where(eq(SessionTable.id, root.id))
  165. .run()
  166. const page = yield* session.list({ directory: location.directory, parentID: null, limit: 1, order: "desc" })
  167. expect(page.data.map((item) => item.id)).toEqual([root.id])
  168. }),
  169. )
  170. it.effect("orders sessions by their latest prompt", () =>
  171. Effect.gen(function* () {
  172. const session = yield* Session.Service
  173. const { db } = yield* Database.Service
  174. const active = yield* session.create({ location, title: "active" })
  175. const newer = yield* session.create({ location, title: "newer" })
  176. yield* db
  177. .update(SessionTable)
  178. .set({ time_created: -2, time_updated: -2 })
  179. .where(eq(SessionTable.id, active.id))
  180. .run()
  181. yield* db
  182. .update(SessionTable)
  183. .set({ time_created: -1, time_updated: -1 })
  184. .where(eq(SessionTable.id, newer.id))
  185. .run()
  186. yield* session.prompt({ sessionID: active.id, text: "continue", resume: false })
  187. expect((yield* session.list()).data.map((item) => item.id)).toEqual([active.id, newer.id])
  188. }),
  189. )
  190. it.effect("filters direct child sessions by parent ID", () =>
  191. Effect.gen(function* () {
  192. const session = yield* Session.Service
  193. const parent = yield* session.create({ location, title: "parent" })
  194. const child = yield* session.create({ parentID: parent.id, title: "child" })
  195. yield* session.create({ location, title: "other root" })
  196. const page = yield* session.list({ parentID: parent.id })
  197. expect(page.data.map((item) => item.id)).toEqual([child.id])
  198. }),
  199. )
  200. it.effect("filters project sessions by subpath", () =>
  201. Effect.gen(function* () {
  202. const session = yield* Session.Service
  203. const { db } = yield* Database.Service
  204. const root = yield* session.create({ location, title: "root" })
  205. const nested = yield* session.create({ location, title: "nested" })
  206. yield* db.update(SessionTable).set({ path: "packages/tui" }).where(eq(SessionTable.id, nested.id)).run()
  207. const page = yield* session.list({
  208. project: Project.ID.global,
  209. subpath: RelativePath.make("packages/tui"),
  210. parentID: null,
  211. })
  212. expect(page.data.map((item) => item.id)).toEqual([nested.id])
  213. expect(page.data.map((item) => item.id)).not.toContain(root.id)
  214. }),
  215. )
  216. it.effect("forks a session by replaying a durable fork event into copied projected rows", () =>
  217. Effect.gen(function* () {
  218. const session = yield* Session.Service
  219. const bus = yield* Bus.Service
  220. const { db } = yield* Database.Service
  221. const parent = yield* session.create({ location, title: "Parent" })
  222. const admitted = yield* session.prompt({
  223. sessionID: parent.id,
  224. text: "First",
  225. resume: false,
  226. })
  227. yield* SessionPending.promote(db, bus, parent.id, "steer")
  228. yield* session.synthetic({ sessionID: parent.id, text: "parent note", resume: false })
  229. yield* SessionPending.promote(db, bus, parent.id, "steer")
  230. const forked = yield* session.fork({ sessionID: parent.id, boundary: { type: "through" } })
  231. const parentContext = yield* session.context(parent.id)
  232. const forkContext = yield* session.context(forked.id)
  233. const history = Array.from(yield* Stream.runCollect(logEvents(session, forked.id)))
  234. expect(forked).toMatchObject({ title: "Parent (fork #1)", fork: { sessionID: parent.id } })
  235. expect(forked.parentID).toBeUndefined()
  236. expect(forkContext).toMatchObject([
  237. { type: "user", text: "First" },
  238. { type: "synthetic", text: "parent note" },
  239. ])
  240. expect(forkContext.map((message) => message.id)).not.toEqual(parentContext.map((message) => message.id))
  241. expect(history).toHaveLength(1)
  242. expect(history[0]).toMatchObject({
  243. type: "session.forked",
  244. durable: { seq: 0 },
  245. data: { sessionID: forked.id, parentID: parent.id },
  246. })
  247. expect(yield* SessionPending.find(db, forkContext[0].id)).toBeUndefined()
  248. expect(yield* SessionPending.find(db, forkContext[1].id)).toBeUndefined()
  249. expect(
  250. yield* session.prompt({ id: forkContext[0].id, sessionID: forked.id, text: "First", resume: false }),
  251. ).toMatchObject({ id: forkContext[0].id, type: "user", data: { text: "First" } })
  252. yield* session.prompt({
  253. sessionID: parent.id,
  254. text: "Parent changed",
  255. resume: false,
  256. })
  257. yield* SessionPending.promote(db, bus, parent.id, "steer")
  258. yield* session.prompt({
  259. sessionID: forked.id,
  260. text: "Child continues",
  261. resume: false,
  262. })
  263. yield* SessionPending.promote(db, bus, forked.id, "steer")
  264. expect((yield* session.context(parent.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"])
  265. expect((yield* session.context(forked.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"])
  266. expect((yield* session.context(forked.id)).at(-1)).toMatchObject({ text: "Child continues" })
  267. expect(
  268. Array.from(yield* Stream.runCollect(logEvents(session, forked.id))).map(
  269. (event): number | undefined => event.durable?.seq,
  270. ),
  271. ).toEqual([0, 5, 6])
  272. expect(yield* SessionPending.find(db, admitted.id)).toBeUndefined()
  273. }),
  274. )
  275. it.effect("keeps a fork untitled when its parent is untitled", () =>
  276. Effect.gen(function* () {
  277. const session = yield* Session.Service
  278. const bus = yield* Bus.Service
  279. const { db } = yield* Database.Service
  280. const parent = yield* session.create({ location })
  281. yield* session.prompt({ sessionID: parent.id, text: "First", resume: false })
  282. yield* SessionPending.promote(db, bus, parent.id, "steer")
  283. const forked = yield* session.fork({ sessionID: parent.id, boundary: { type: "through" } })
  284. const row = yield* db.select().from(SessionTable).where(eq(SessionTable.id, forked.id)).get().pipe(Effect.orDie)
  285. expect(forked.title).toBeUndefined()
  286. expect(row?.title).toBeNull()
  287. }),
  288. )
  289. it.effect("rejects forking an empty session", () =>
  290. Effect.gen(function* () {
  291. const session = yield* Session.Service
  292. const parent = yield* session.create({ location })
  293. expect(
  294. yield* session.fork({ sessionID: parent.id, boundary: { type: "through" } }).pipe(Effect.flip),
  295. ).toMatchObject({ _tag: "Session.ForkEmptyError", sessionID: parent.id })
  296. }),
  297. )
  298. it.effect("forks before the selected boundary message", () =>
  299. Effect.gen(function* () {
  300. const session = yield* Session.Service
  301. const bus = yield* Bus.Service
  302. const { db } = yield* Database.Service
  303. const parent = yield* session.create({ location })
  304. const first = yield* session.prompt({
  305. sessionID: parent.id,
  306. text: "First",
  307. resume: false,
  308. })
  309. yield* SessionPending.promote(db, bus, parent.id, "steer")
  310. const second = yield* session.prompt({
  311. sessionID: parent.id,
  312. text: "Second",
  313. resume: false,
  314. })
  315. yield* SessionPending.promote(db, bus, parent.id, "steer")
  316. const assistantMessageID = SessionMessage.ID.create()
  317. const model = Model.Ref.make({ id: Model.ID.make("model"), providerID: Provider.ID.make("provider") })
  318. yield* bus.publish(SessionEvent.Step.Started, {
  319. sessionID: parent.id,
  320. assistantMessageID,
  321. agent: Agent.ID.make("build"),
  322. model,
  323. })
  324. yield* bus.publish(SessionEvent.Step.Ended, {
  325. sessionID: parent.id,
  326. assistantMessageID,
  327. finish: "stop",
  328. cost: Money.USD.make(0.75),
  329. tokens: { input: 6, output: 3, reasoning: 1, cache: { read: 2, write: 1 } },
  330. })
  331. const forked = yield* session.fork({
  332. sessionID: parent.id,
  333. boundary: { type: "before", messageID: second.id },
  334. })
  335. const beforeFirst = yield* session.fork({
  336. sessionID: parent.id,
  337. boundary: { type: "before", messageID: first.id },
  338. })
  339. const complete = yield* session.fork({ sessionID: parent.id, boundary: { type: "through" } })
  340. const context = yield* session.context(forked.id)
  341. const history = Array.from(yield* Stream.runCollect(logEvents(session, forked.id)))
  342. expect(forked.fork).toEqual({
  343. sessionID: parent.id,
  344. boundary: { type: "before", messageID: second.id },
  345. })
  346. expect(context).toMatchObject([{ text: "First" }])
  347. expect(context[0]?.id).not.toBe(first.id)
  348. expect(history[0]).toMatchObject({
  349. data: { boundary: { type: "before", messageID: second.id } },
  350. })
  351. expect(forked).toMatchObject({ cost: 0, tokens: { input: 0, output: 0, reasoning: 0 } })
  352. expect(yield* session.context(beforeFirst.id)).toEqual([])
  353. expect(beforeFirst).toMatchObject({ cost: 0, tokens: { input: 0, output: 0, reasoning: 0 } })
  354. expect(complete).toMatchObject({
  355. cost: 0,
  356. tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
  357. })
  358. }),
  359. )
  360. it.effect("returns the existing Session when one ID is reused with different create arguments", () =>
  361. Effect.gen(function* () {
  362. const session = yield* Session.Service
  363. const created = yield* session.create({ id, location })
  364. const changed = [
  365. { id, location: Location.Ref.make({ directory: AbsolutePath.make("/other") }) },
  366. { id, location, agent: Agent.ID.make("build") },
  367. {
  368. id,
  369. location,
  370. model: Model.Ref.make({ id: Model.ID.make("sonnet"), providerID: Provider.ID.anthropic }),
  371. },
  372. ]
  373. for (const input of changed) {
  374. expect(yield* session.create(input)).toEqual(created)
  375. }
  376. expect((yield* session.list()).data).toHaveLength(1)
  377. }),
  378. )
  379. it.effect("returns one recorded session to concurrent exact retries", () =>
  380. Effect.gen(function* () {
  381. const session = yield* Session.Service
  382. const input = { id, location }
  383. const created = yield* Effect.all([session.create(input), session.create(input)], { concurrency: "unbounded" })
  384. expect(created[1]).toEqual(created[0])
  385. expect((yield* session.list()).data).toEqual([created[0]])
  386. }),
  387. )
  388. it.effect("returns the current Session projection after updates", () =>
  389. Effect.gen(function* () {
  390. const session = yield* Session.Service
  391. const { db } = yield* Database.Service
  392. const input = { id, location }
  393. const created = yield* session.create(input)
  394. yield* db.update(SessionTable).set({ agent: "build" }).where(eq(SessionTable.id, id)).run().pipe(Effect.orDie)
  395. expect(yield* session.create(input)).toMatchObject({ id: created.id, agent: "build" })
  396. }),
  397. )
  398. it.effect("persists creation through the current created event", () =>
  399. Effect.gen(function* () {
  400. const session = yield* Session.Service
  401. const { db } = yield* Database.Service
  402. const created = yield* session.create({ location })
  403. expect(
  404. yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, created.id)).all().pipe(Effect.orDie),
  405. ).toMatchObject([{ type: Bus.versionedType(SessionEvent.Created.type, 1) }])
  406. }),
  407. )
  408. it.effect("persists caller-ID creation through the existing created event", () =>
  409. Effect.gen(function* () {
  410. const session = yield* Session.Service
  411. const { db } = yield* Database.Service
  412. const created = yield* session.create({ id, location })
  413. expect(
  414. yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, created.id)).get().pipe(Effect.orDie),
  415. ).toMatchObject({
  416. data: { sessionID: id },
  417. })
  418. }),
  419. )
  420. it.effect("includes current creation rows in the Session event stream", () =>
  421. Effect.gen(function* () {
  422. const session = yield* Session.Service
  423. const bus = yield* Bus.Service
  424. const { db } = yield* Database.Service
  425. const created = yield* session.create({ location })
  426. yield* session.prompt({
  427. sessionID: created.id,
  428. text: "Hello",
  429. resume: false,
  430. })
  431. yield* SessionPending.promote(db, bus, created.id, "steer")
  432. expect(
  433. Array.from(yield* logEvents(session, created.id, true).pipe(Stream.take(3), Stream.runCollect)),
  434. ).toMatchObject([
  435. { durable: { seq: 0 }, type: "session.created" },
  436. {
  437. durable: { seq: 1 },
  438. type: "session.input.admitted",
  439. data: { input: { type: "user", data: { text: "Hello" }, delivery: "steer" } },
  440. },
  441. { durable: { seq: 2 }, type: "session.input.promoted" },
  442. ])
  443. }),
  444. )
  445. it.effect("replays one prompt lifecycle into a fresh target database", () =>
  446. Effect.gen(function* () {
  447. const session = yield* Session.Service
  448. const sourceEvents = yield* Bus.Service
  449. const sourceDb = (yield* Database.Service).db
  450. const created = yield* session.create({ id: Session.ID.make("ses_fresh_target_replay"), location })
  451. const admitted = yield* session.prompt({
  452. sessionID: created.id,
  453. text: "Replay lifecycle",
  454. resume: false,
  455. })
  456. yield* SessionPending.promote(sourceDb, sourceEvents, created.id, "steer")
  457. const serialized = (yield* sourceDb
  458. .select()
  459. .from(EventTable)
  460. .where(eq(EventTable.aggregate_id, created.id))
  461. .orderBy(asc(EventTable.seq))
  462. .all()
  463. .pipe(Effect.orDie)).map((event) => ({
  464. id: event.id,
  465. created: DateTime.makeUnsafe(event.created),
  466. aggregateID: event.aggregate_id,
  467. seq: event.seq,
  468. type: event.type,
  469. data: event.data,
  470. }))
  471. const tmp = yield* Effect.acquireRelease(
  472. Effect.promise(() => tmpdir()),
  473. (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
  474. )
  475. const targetDatabase = Database.layer({ path: path.join(tmp.path, "target.sqlite") })
  476. const targetLayer = AppNodeBuilder.build(
  477. LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node]),
  478. [
  479. [Database.node, targetDatabase],
  480. [Bus.node, Bus.configured({ persist: true })],
  481. ],
  482. )
  483. yield* Effect.gen(function* () {
  484. const db = (yield* Database.Service).db
  485. const bus = yield* Bus.Service
  486. const store = yield* SessionStore.Service
  487. yield* db
  488. .insert(ProjectTable)
  489. .values({ id: Project.ID.global, worktree: location.directory, sandboxes: [] })
  490. .run()
  491. .pipe(Effect.orDie)
  492. expect(yield* store.get(created.id)).toBeUndefined()
  493. expect(yield* bus.replayAll(serialized.slice(0, 2))).toBe(created.id)
  494. expect(yield* SessionPending.find(db, admitted.id)).toMatchObject({
  495. id: admitted.id,
  496. sessionID: created.id,
  497. type: "user",
  498. data: { text: "Replay lifecycle" },
  499. delivery: "steer",
  500. })
  501. expect(yield* store.context(created.id)).toEqual([])
  502. expect(yield* bus.replayAll(serialized.slice(2))).toBe(created.id)
  503. expect(yield* SessionPending.find(db, admitted.id)).toBeUndefined()
  504. expect(yield* store.context(created.id)).toMatchObject([
  505. { id: admitted.id, type: "user", text: "Replay lifecycle" },
  506. ])
  507. expect(
  508. (yield* db
  509. .select()
  510. .from(EventTable)
  511. .where(eq(EventTable.aggregate_id, created.id))
  512. .orderBy(asc(EventTable.seq))
  513. .all()
  514. .pipe(Effect.orDie)).map((event) => [event.seq, event.type]),
  515. ).toEqual([
  516. [0, Bus.versionedType(SessionEvent.Created.type, 1)],
  517. [1, Bus.versionedType(SessionEvent.InputAdmitted.type, 1)],
  518. [2, Bus.versionedType(SessionEvent.InputPromoted.type, 1)],
  519. ])
  520. }).pipe(Effect.provide(Layer.fresh(targetLayer)))
  521. }),
  522. )
  523. it.effect("does not mask unrelated created projector defects", () =>
  524. Effect.gen(function* () {
  525. const session = yield* Session.Service
  526. const event = yield* Bus.Service
  527. const defect = new Error("unrelated projector defect")
  528. yield* event.project(SessionEvent.Created, () => Effect.die(defect))
  529. expect(yield* session.create({ id, location }).pipe(Effect.catchDefect(Effect.succeed))).toBe(defect)
  530. }),
  531. )
  532. it.live("runs a shell command and projects the started/ended shell message", () =>
  533. withTmp((directory) =>
  534. Effect.gen(function* () {
  535. const session = yield* Session.Service
  536. const created = yield* session.create({
  537. location: Location.Ref.make({ directory: AbsolutePath.make(directory) }),
  538. })
  539. yield* session.shell({ sessionID: created.id, command: "echo hello" })
  540. const messages = yield* session.messages({ sessionID: created.id, order: "asc" })
  541. const shell = messages.find((message): message is SessionMessage.Shell => message.type === "shell")
  542. expect(shell).toMatchObject({ type: "shell", command: "echo hello", status: "exited", exit: 0 })
  543. expect(shell?.output?.output).toContain("hello")
  544. expect(shell?.output?.truncated).toBe(false)
  545. expect(shell?.time.completed).toBeDefined()
  546. }),
  547. ),
  548. )
  549. it.live("still emits shell ended for a failing command", () =>
  550. withTmp((directory) =>
  551. Effect.gen(function* () {
  552. const session = yield* Session.Service
  553. const created = yield* session.create({
  554. location: Location.Ref.make({ directory: AbsolutePath.make(directory) }),
  555. })
  556. yield* session.shell({ sessionID: created.id, command: "false" })
  557. const messages = yield* session.messages({ sessionID: created.id, order: "asc" })
  558. const shell = messages.find((message): message is SessionMessage.Shell => message.type === "shell")
  559. expect(shell).toMatchObject({ type: "shell", command: "false", status: "exited" })
  560. expect(shell?.exit).not.toBe(0)
  561. expect(shell?.time.completed).toBeDefined()
  562. }),
  563. ),
  564. )
  565. it.effect("switches the selected agent through the durable Session event", () =>
  566. Effect.gen(function* () {
  567. const session = yield* Session.Service
  568. const created = yield* session.create({ location })
  569. yield* session.switchAgent({ sessionID: created.id, agent: Agent.ID.make("plan") })
  570. expect(yield* session.get(created.id)).toMatchObject({ agent: "plan" })
  571. expect(
  572. Array.from(yield* logEvents(session, created.id, true).pipe(Stream.drop(1), Stream.take(1), Stream.runCollect)),
  573. ).toMatchObject([{ type: "session.agent.selected", data: { agent: "plan" } }])
  574. }),
  575. )
  576. it.effect("rejects an agent switch for a missing Session", () =>
  577. Effect.gen(function* () {
  578. const session = yield* Session.Service
  579. const missing = Session.ID.make("ses_missing_agent_switch")
  580. expect(
  581. yield* session.switchAgent({ sessionID: missing, agent: Agent.ID.make("plan") }).pipe(
  582. Effect.flip,
  583. Effect.map((error) => error._tag),
  584. ),
  585. ).toBe("Session.NotFoundError")
  586. }),
  587. )
  588. it.effect("switches the selected model through the durable Session event", () =>
  589. Effect.gen(function* () {
  590. const session = yield* Session.Service
  591. const created = yield* session.create({ location })
  592. const model = Model.Ref.make({
  593. id: Model.ID.make("sonnet"),
  594. providerID: Provider.ID.anthropic,
  595. variant: Model.VariantID.make("high"),
  596. })
  597. yield* session.switchModel({ sessionID: created.id, model })
  598. expect(yield* session.get(created.id)).toMatchObject({ model })
  599. const bus = Array.from(
  600. yield* logEvents(session, created.id, true).pipe(Stream.drop(1), Stream.take(1), Stream.runCollect),
  601. )
  602. expect(bus).toMatchObject([{ type: "session.model.selected" }])
  603. expect(bus[0]?.data).toEqual({ sessionID: created.id, model })
  604. }),
  605. )
  606. it.effect("ignores a model switch when the selected model is unchanged", () =>
  607. Effect.gen(function* () {
  608. const session = yield* Session.Service
  609. const created = yield* session.create({ location })
  610. const model = Model.Ref.make({ id: Model.ID.make("sonnet"), providerID: Provider.ID.anthropic })
  611. yield* session.switchModel({ sessionID: created.id, model })
  612. yield* session.switchModel({ sessionID: created.id, model })
  613. const { db } = yield* Database.Service
  614. expect(
  615. yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, created.id)).all().pipe(Effect.orDie),
  616. ).toHaveLength(2)
  617. expect(yield* session.get(created.id)).toMatchObject({ model })
  618. }),
  619. )
  620. it.effect("treats an omitted variant as the default variant", () =>
  621. Effect.gen(function* () {
  622. const session = yield* Session.Service
  623. const model = Model.Ref.make({ id: Model.ID.make("sonnet"), providerID: Provider.ID.anthropic })
  624. const created = yield* session.create({ location, model })
  625. yield* session.switchModel({
  626. sessionID: created.id,
  627. model: Model.Ref.make({ ...model, variant: Model.VariantID.make("default") }),
  628. })
  629. const { db } = yield* Database.Service
  630. expect(
  631. yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, created.id)).all().pipe(Effect.orDie),
  632. ).toHaveLength(1)
  633. }),
  634. )
  635. it.effect("rejects a model switch for a missing Session", () =>
  636. Effect.gen(function* () {
  637. const session = yield* Session.Service
  638. const missing = Session.ID.make("ses_missing_model_switch")
  639. expect(
  640. yield* session
  641. .switchModel({
  642. sessionID: missing,
  643. model: Model.Ref.make({ id: Model.ID.make("sonnet"), providerID: Provider.ID.anthropic }),
  644. })
  645. .pipe(
  646. Effect.flip,
  647. Effect.map((error) => error._tag),
  648. ),
  649. ).toBe("Session.NotFoundError")
  650. }),
  651. )
  652. })