session-create.test.ts 29 KB

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