session-create.test.ts 29 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762
  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. expect(
  251. yield* session.prompt({ id: forkContext[0].id, sessionID: forked.id, text: "First", resume: false }),
  252. ).toMatchObject({ id: forkContext[0].id, type: "user", data: { text: "First" } })
  253. yield* session.prompt({
  254. sessionID: parent.id,
  255. text: "Parent changed",
  256. resume: false,
  257. })
  258. yield* SessionPending.promote(db, bus, parent.id, "steer")
  259. yield* session.prompt({
  260. sessionID: forked.id,
  261. text: "Child continues",
  262. resume: false,
  263. })
  264. yield* SessionPending.promote(db, bus, forked.id, "steer")
  265. expect((yield* session.context(parent.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"])
  266. expect((yield* session.context(forked.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"])
  267. expect((yield* session.context(forked.id)).at(-1)).toMatchObject({ text: "Child continues" })
  268. expect(
  269. Array.from(yield* Stream.runCollect(logEvents(session, forked.id))).map(
  270. (event): number | undefined => event.durable?.seq,
  271. ),
  272. ).toEqual([0, 5, 6])
  273. expect(yield* SessionPending.find(db, admitted.id)).toBeUndefined()
  274. }),
  275. )
  276. it.effect("keeps a fork untitled when its parent is untitled", () =>
  277. Effect.gen(function* () {
  278. const session = yield* Session.Service
  279. const bus = yield* Bus.Service
  280. const { db } = yield* Database.Service
  281. const parent = yield* session.create({ location })
  282. yield* session.prompt({ sessionID: parent.id, text: "First", resume: false })
  283. yield* SessionPending.promote(db, bus, parent.id, "steer")
  284. const forked = yield* session.fork({ sessionID: parent.id, boundary: { type: "through" } })
  285. const row = yield* db.select().from(SessionTable).where(eq(SessionTable.id, forked.id)).get().pipe(Effect.orDie)
  286. expect(forked.title).toBeUndefined()
  287. expect(row?.title).toBeNull()
  288. }),
  289. )
  290. it.effect("rejects forking an empty session", () =>
  291. Effect.gen(function* () {
  292. const session = yield* Session.Service
  293. const parent = yield* session.create({ location })
  294. expect(
  295. yield* session.fork({ sessionID: parent.id, boundary: { type: "through" } }).pipe(Effect.flip),
  296. ).toMatchObject({ _tag: "Session.ForkEmptyError", sessionID: parent.id })
  297. }),
  298. )
  299. it.effect("forks before the selected boundary message", () =>
  300. Effect.gen(function* () {
  301. const session = yield* Session.Service
  302. const bus = yield* Bus.Service
  303. const { db } = yield* Database.Service
  304. const parent = yield* session.create({ location })
  305. const first = yield* session.prompt({
  306. sessionID: parent.id,
  307. text: "First",
  308. resume: false,
  309. })
  310. yield* SessionPending.promote(db, bus, parent.id, "steer")
  311. const second = yield* session.prompt({
  312. sessionID: parent.id,
  313. text: "Second",
  314. resume: false,
  315. })
  316. yield* SessionPending.promote(db, bus, parent.id, "steer")
  317. const assistantMessageID = SessionMessage.ID.create()
  318. const model = Model.Ref.make({ id: Model.ID.make("model"), providerID: Provider.ID.make("provider") })
  319. yield* bus.publish(SessionEvent.Step.Started, {
  320. sessionID: parent.id,
  321. assistantMessageID,
  322. agent: Agent.ID.make("build"),
  323. model,
  324. })
  325. yield* bus.publish(SessionEvent.Step.Ended, {
  326. sessionID: parent.id,
  327. assistantMessageID,
  328. finish: "stop",
  329. cost: Money.USD.make(0.75),
  330. tokens: { input: 6, output: 3, reasoning: 1, cache: { read: 2, write: 1 } },
  331. })
  332. const forked = yield* session.fork({
  333. sessionID: parent.id,
  334. boundary: { type: "before", messageID: second.id },
  335. })
  336. const beforeFirst = yield* session.fork({
  337. sessionID: parent.id,
  338. boundary: { type: "before", messageID: first.id },
  339. })
  340. const complete = yield* session.fork({ sessionID: parent.id, boundary: { type: "through" } })
  341. const context = yield* session.context(forked.id)
  342. const history = Array.from(yield* Stream.runCollect(logEvents(session, forked.id)))
  343. expect(forked.fork).toEqual({
  344. sessionID: parent.id,
  345. boundary: { type: "before", messageID: second.id },
  346. })
  347. expect(context).toMatchObject([{ text: "First" }])
  348. expect(context[0]?.id).not.toBe(first.id)
  349. expect(history[0]).toMatchObject({
  350. data: { boundary: { type: "before", messageID: second.id } },
  351. })
  352. expect(forked).toMatchObject({ cost: 0, tokens: { input: 0, output: 0, reasoning: 0 } })
  353. expect(yield* session.context(beforeFirst.id)).toEqual([])
  354. expect(beforeFirst).toMatchObject({ cost: 0, tokens: { input: 0, output: 0, reasoning: 0 } })
  355. expect(complete).toMatchObject({
  356. cost: 0,
  357. tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
  358. })
  359. }),
  360. )
  361. it.effect("returns the existing Session when one ID is reused with different create arguments", () =>
  362. Effect.gen(function* () {
  363. const session = yield* Session.Service
  364. const created = yield* session.create({ id, location })
  365. const changed = [
  366. { id, location: Location.Ref.make({ directory: AbsolutePath.make("/other") }) },
  367. { id, location, agent: Agent.ID.make("build") },
  368. {
  369. id,
  370. location,
  371. model: Model.Ref.make({ id: Model.ID.make("sonnet"), providerID: Provider.ID.anthropic }),
  372. },
  373. ]
  374. for (const input of changed) {
  375. expect(yield* session.create(input)).toEqual(created)
  376. }
  377. expect((yield* session.list()).data).toHaveLength(1)
  378. }),
  379. )
  380. it.effect("returns one recorded session to concurrent exact retries", () =>
  381. Effect.gen(function* () {
  382. const session = yield* Session.Service
  383. const input = { id, location }
  384. const created = yield* Effect.all([session.create(input), session.create(input)], { concurrency: "unbounded" })
  385. expect(created[1]).toEqual(created[0])
  386. expect((yield* session.list()).data).toEqual([created[0]])
  387. }),
  388. )
  389. it.effect("returns the current Session projection after updates", () =>
  390. Effect.gen(function* () {
  391. const session = yield* Session.Service
  392. const { db } = yield* Database.Service
  393. const input = { id, location }
  394. const created = yield* session.create(input)
  395. yield* db.update(SessionTable).set({ agent: "build" }).where(eq(SessionTable.id, id)).run().pipe(Effect.orDie)
  396. expect(yield* session.create(input)).toMatchObject({ id: created.id, agent: "build" })
  397. }),
  398. )
  399. it.effect("returns the current Session projection after projected updates", () =>
  400. Effect.gen(function* () {
  401. const session = yield* Session.Service
  402. const bus = yield* Bus.Service
  403. const input = { id, location }
  404. const created = yield* session.create(input)
  405. yield* bus.publish(SessionV1.Event.Updated, {
  406. sessionID: id,
  407. info: SessionV1.SessionInfo.make({
  408. id,
  409. slug: "updated",
  410. version: "test",
  411. projectID: created.projectID,
  412. directory: created.location.directory,
  413. title: "updated",
  414. agent: "build",
  415. time: { created: 0, updated: 1 },
  416. }),
  417. })
  418. expect(yield* session.create(input)).toMatchObject({ id, agent: "build" })
  419. }),
  420. )
  421. it.effect("persists creation through the existing legacy created event", () =>
  422. Effect.gen(function* () {
  423. const session = yield* Session.Service
  424. const { db } = yield* Database.Service
  425. const created = yield* session.create({ location })
  426. expect(
  427. yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, created.id)).all().pipe(Effect.orDie),
  428. ).toMatchObject([{ type: Bus.versionedType(SessionV1.Event.Created.type, 1) }])
  429. }),
  430. )
  431. it.effect("persists caller-ID creation through the existing created event", () =>
  432. Effect.gen(function* () {
  433. const session = yield* Session.Service
  434. const { db } = yield* Database.Service
  435. const created = yield* session.create({ id, location })
  436. expect(
  437. yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, created.id)).get().pipe(Effect.orDie),
  438. ).toMatchObject({
  439. data: { sessionID: id },
  440. })
  441. }),
  442. )
  443. it.effect("omits legacy creation rows from the Session event stream", () =>
  444. Effect.gen(function* () {
  445. const session = yield* Session.Service
  446. const bus = yield* Bus.Service
  447. const { db } = yield* Database.Service
  448. const created = yield* session.create({ location })
  449. yield* session.prompt({
  450. sessionID: created.id,
  451. text: "Hello",
  452. resume: false,
  453. })
  454. yield* SessionPending.promote(db, bus, created.id, "steer")
  455. expect(
  456. Array.from(yield* logEvents(session, created.id, true).pipe(Stream.take(2), Stream.runCollect)),
  457. ).toMatchObject([
  458. {
  459. durable: { seq: 1 },
  460. type: "session.input.admitted",
  461. data: { input: { type: "user", data: { text: "Hello" }, delivery: "steer" } },
  462. },
  463. { durable: { seq: 2 }, type: "session.input.promoted" },
  464. ])
  465. }),
  466. )
  467. it.effect("replays one prompt lifecycle into a fresh target database", () =>
  468. Effect.gen(function* () {
  469. const session = yield* Session.Service
  470. const sourceEvents = yield* Bus.Service
  471. const sourceDb = (yield* Database.Service).db
  472. const created = yield* session.create({ id: Session.ID.make("ses_fresh_target_replay"), location })
  473. const admitted = yield* session.prompt({
  474. sessionID: created.id,
  475. text: "Replay lifecycle",
  476. resume: false,
  477. })
  478. yield* SessionPending.promote(sourceDb, sourceEvents, created.id, "steer")
  479. const serialized = (yield* sourceDb
  480. .select()
  481. .from(EventTable)
  482. .where(eq(EventTable.aggregate_id, created.id))
  483. .orderBy(asc(EventTable.seq))
  484. .all()
  485. .pipe(Effect.orDie)).map((event) => ({
  486. id: event.id,
  487. created: DateTime.makeUnsafe(event.created),
  488. aggregateID: event.aggregate_id,
  489. seq: event.seq,
  490. type: event.type,
  491. data: event.data,
  492. }))
  493. const tmp = yield* Effect.acquireRelease(
  494. Effect.promise(() => tmpdir()),
  495. (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()),
  496. )
  497. const targetDatabase = Database.layer({ path: path.join(tmp.path, "target.sqlite") })
  498. const targetLayer = AppNodeBuilder.build(
  499. LayerNode.group([Database.node, Bus.node, SessionProjector.node, SessionStore.node]),
  500. [[Database.node, targetDatabase]],
  501. )
  502. yield* Effect.gen(function* () {
  503. const db = (yield* Database.Service).db
  504. const bus = yield* Bus.Service
  505. const store = yield* SessionStore.Service
  506. yield* db
  507. .insert(ProjectTable)
  508. .values({ id: Project.ID.global, worktree: location.directory, sandboxes: [] })
  509. .run()
  510. .pipe(Effect.orDie)
  511. expect(yield* store.get(created.id)).toBeUndefined()
  512. expect(yield* bus.replayAll(serialized.slice(0, 2))).toBe(created.id)
  513. expect(yield* SessionPending.find(db, admitted.id)).toMatchObject({
  514. id: admitted.id,
  515. sessionID: created.id,
  516. type: "user",
  517. data: { text: "Replay lifecycle" },
  518. delivery: "steer",
  519. })
  520. expect(yield* store.context(created.id)).toEqual([])
  521. expect(yield* bus.replayAll(serialized.slice(2))).toBe(created.id)
  522. expect(yield* SessionPending.find(db, admitted.id)).toBeUndefined()
  523. expect(yield* store.context(created.id)).toMatchObject([
  524. { id: admitted.id, type: "user", text: "Replay lifecycle" },
  525. ])
  526. expect(
  527. (yield* db
  528. .select()
  529. .from(EventTable)
  530. .where(eq(EventTable.aggregate_id, created.id))
  531. .orderBy(asc(EventTable.seq))
  532. .all()
  533. .pipe(Effect.orDie)).map((event) => [event.seq, event.type]),
  534. ).toEqual([
  535. [0, Bus.versionedType(SessionV1.Event.Created.type, 1)],
  536. [1, Bus.versionedType(SessionEvent.InputAdmitted.type, 1)],
  537. [2, Bus.versionedType(SessionEvent.InputPromoted.type, 1)],
  538. ])
  539. }).pipe(Effect.provide(Layer.fresh(targetLayer)))
  540. }),
  541. )
  542. it.effect("does not mask unrelated created projector defects", () =>
  543. Effect.gen(function* () {
  544. const session = yield* Session.Service
  545. const event = yield* Bus.Service
  546. const defect = new Error("unrelated projector defect")
  547. yield* event.project(SessionV1.Event.Created, () => Effect.die(defect))
  548. expect(yield* session.create({ id, location }).pipe(Effect.catchDefect(Effect.succeed))).toBe(defect)
  549. }),
  550. )
  551. it.live("runs a shell command and projects the started/ended shell message", () =>
  552. withTmp((directory) =>
  553. Effect.gen(function* () {
  554. const session = yield* Session.Service
  555. const created = yield* session.create({
  556. location: Location.Ref.make({ directory: AbsolutePath.make(directory) }),
  557. })
  558. yield* session.shell({ sessionID: created.id, command: "echo hello" })
  559. const messages = yield* session.messages({ sessionID: created.id, order: "asc" })
  560. const shell = messages.find((message): message is SessionMessage.Shell => message.type === "shell")
  561. expect(shell).toMatchObject({ type: "shell", command: "echo hello", status: "exited", exit: 0 })
  562. expect(shell?.output?.output).toContain("hello")
  563. expect(shell?.output?.truncated).toBe(false)
  564. expect(shell?.time.completed).toBeDefined()
  565. }),
  566. ),
  567. )
  568. it.live("still emits shell ended for a failing command", () =>
  569. withTmp((directory) =>
  570. Effect.gen(function* () {
  571. const session = yield* Session.Service
  572. const created = yield* session.create({
  573. location: Location.Ref.make({ directory: AbsolutePath.make(directory) }),
  574. })
  575. yield* session.shell({ sessionID: created.id, command: "false" })
  576. const messages = yield* session.messages({ sessionID: created.id, order: "asc" })
  577. const shell = messages.find((message): message is SessionMessage.Shell => message.type === "shell")
  578. expect(shell).toMatchObject({ type: "shell", command: "false", status: "exited" })
  579. expect(shell?.exit).not.toBe(0)
  580. expect(shell?.time.completed).toBeDefined()
  581. }),
  582. ),
  583. )
  584. it.effect("switches the selected agent through the durable Session event", () =>
  585. Effect.gen(function* () {
  586. const session = yield* Session.Service
  587. const created = yield* session.create({ location })
  588. yield* session.switchAgent({ sessionID: created.id, agent: Agent.ID.make("plan") })
  589. expect(yield* session.get(created.id)).toMatchObject({ agent: "plan" })
  590. expect(
  591. Array.from(yield* logEvents(session, created.id, true).pipe(Stream.take(1), Stream.runCollect)),
  592. ).toMatchObject([{ type: "session.agent.selected", data: { agent: "plan" } }])
  593. }),
  594. )
  595. it.effect("rejects an agent switch for a missing Session", () =>
  596. Effect.gen(function* () {
  597. const session = yield* Session.Service
  598. const missing = Session.ID.make("ses_missing_agent_switch")
  599. expect(
  600. yield* session.switchAgent({ sessionID: missing, agent: Agent.ID.make("plan") }).pipe(
  601. Effect.flip,
  602. Effect.map((error) => error._tag),
  603. ),
  604. ).toBe("Session.NotFoundError")
  605. }),
  606. )
  607. it.effect("switches the selected model through the durable Session event", () =>
  608. Effect.gen(function* () {
  609. const session = yield* Session.Service
  610. const created = yield* session.create({ location })
  611. const model = Model.Ref.make({
  612. id: Model.ID.make("sonnet"),
  613. providerID: Provider.ID.anthropic,
  614. variant: Model.VariantID.make("high"),
  615. })
  616. yield* session.switchModel({ sessionID: created.id, model })
  617. expect(yield* session.get(created.id)).toMatchObject({ model })
  618. const bus = Array.from(yield* logEvents(session, created.id, true).pipe(Stream.take(1), Stream.runCollect))
  619. expect(bus).toMatchObject([{ type: "session.model.selected" }])
  620. expect(bus[0]?.data).toEqual({ sessionID: created.id, model })
  621. }),
  622. )
  623. it.effect("ignores a model switch when the selected model is unchanged", () =>
  624. Effect.gen(function* () {
  625. const session = yield* Session.Service
  626. const created = yield* session.create({ location })
  627. const model = Model.Ref.make({ id: Model.ID.make("sonnet"), providerID: Provider.ID.anthropic })
  628. yield* session.switchModel({ sessionID: created.id, model })
  629. yield* session.switchModel({ sessionID: created.id, model })
  630. const { db } = yield* Database.Service
  631. expect(
  632. yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, created.id)).all().pipe(Effect.orDie),
  633. ).toHaveLength(2)
  634. expect(yield* session.get(created.id)).toMatchObject({ model })
  635. }),
  636. )
  637. it.effect("treats an omitted variant as the default variant", () =>
  638. Effect.gen(function* () {
  639. const session = yield* Session.Service
  640. const model = Model.Ref.make({ id: Model.ID.make("sonnet"), providerID: Provider.ID.anthropic })
  641. const created = yield* session.create({ location, model })
  642. yield* session.switchModel({
  643. sessionID: created.id,
  644. model: Model.Ref.make({ ...model, variant: Model.VariantID.make("default") }),
  645. })
  646. const { db } = yield* Database.Service
  647. expect(
  648. yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, created.id)).all().pipe(Effect.orDie),
  649. ).toHaveLength(1)
  650. }),
  651. )
  652. it.effect("rejects a model switch for a missing Session", () =>
  653. Effect.gen(function* () {
  654. const session = yield* Session.Service
  655. const missing = Session.ID.make("ses_missing_model_switch")
  656. expect(
  657. yield* session
  658. .switchModel({
  659. sessionID: missing,
  660. model: Model.Ref.make({ id: Model.ID.make("sonnet"), providerID: Provider.ID.anthropic }),
  661. })
  662. .pipe(
  663. Effect.flip,
  664. Effect.map((error) => error._tag),
  665. ),
  666. ).toBe("Session.NotFoundError")
  667. }),
  668. )
  669. })