openAgent

1import { describe, expect } from "bun:test"
2import { DateTime, Effect, Fiber, Layer, Stream } from "effect"
3import { eq } from "drizzle-orm"
4import { Database } from "@opencode-ai/core/database/database"
5import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
6import { LayerNode } from "@opencode-ai/core/effect/layer-node"
7import { EventV2 } from "@opencode-ai/core/event"
8import { EventTable } from "@opencode-ai/core/event/sql"
9import { SessionEvent } from "@opencode-ai/core/session/event"
10import { Project } from "@opencode-ai/core/project"
11import { ProjectTable } from "@opencode-ai/core/project/sql"
12import { AbsolutePath } from "@opencode-ai/core/schema"
13import { SessionV2 } from "@opencode-ai/core/session"
14import { Prompt } from "@opencode-ai/core/session/prompt"
15import { SessionMessage } from "@opencode-ai/core/session/message"
16import { SessionProjector } from "@opencode-ai/core/session/projector"
17import { SessionExecution } from "@opencode-ai/core/session/execution"
18import { SessionInput } from "@opencode-ai/core/session/input"
19import { SessionInputTable, SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql"
20import { SessionStore } from "@opencode-ai/core/session/store"
21import { testEffect } from "./lib/effect"
22 
23const executionCalls: SessionV2.ID[] = []
24const interruptCalls: SessionV2.ID[] = []
25const wakeCalls: SessionV2.ID[] = []
26const activeSessions = new Set<SessionV2.ID>()
27const execution = Layer.succeed(
28  SessionExecution.Service,
29  SessionExecution.Service.of({
30    active: Effect.sync(() => new Set(activeSessions)),
31    resume: (sessionID) =>
32      Effect.sync(() => {
33        executionCalls.push(sessionID)
34      }),
35    interrupt: (sessionID) =>
36      Effect.sync(() => {
37        interruptCalls.push(sessionID)
38      }),
39    wake: (sessionID) =>
40      Effect.sync(() => {
41        wakeCalls.push(sessionID)
42      }),
43  }),
44)
45const it = testEffect(
46  AppNodeBuilder.build(
47    LayerNode.group([Database.node, EventV2.node, SessionProjector.node, SessionStore.node, SessionV2.node]),
48    [[SessionExecution.node, execution]],
49  ),
50)
51const sessionID = SessionV2.ID.make("ses_prompt_test")
52const messageID = SessionMessage.ID.create()
53 
54const setup = Effect.gen(function* () {
55  const { db } = yield* Database.Service
56  yield* db
57    .insert(ProjectTable)
58    .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
59    .onConflictDoNothing()
60    .run()
61    .pipe(Effect.orDie)
62  yield* db
63    .insert(SessionTable)
64    .values({
65      id: sessionID,
66      project_id: Project.ID.global,
67      slug: "test",
68      directory: "/project",
69      title: "test",
70      version: "test",
71    })
72    .onConflictDoNothing()
73    .run()
74    .pipe(Effect.orDie)
75})
76 
77const admitted = (id: SessionMessage.ID) => Database.Service.use(({ db }) => SessionInput.find(db, id))
78const admittedCount = Database.Service.use(({ db }) =>
79  db
80    .select()
81    .from(SessionInputTable)
82    .all()
83    .pipe(
84      Effect.orDie,
85      Effect.map((rows) => rows.length),
86    ),
87)
88const eventCount = (type: string) =>
89  Database.Service.use(({ db }) =>
90    db
91      .select()
92      .from(EventTable)
93      .where(eq(EventTable.type, type))
94      .all()
95      .pipe(
96        Effect.orDie,
97        Effect.map((rows) => rows.length),
98      ),
99  )
100 
101describe("SessionV2.prompt", () => {
102  it.effect("exposes the execution registry", () =>
103    Effect.gen(function* () {
104      activeSessions.add(sessionID)
105      expect(Array.from(yield* (yield* SessionV2.Service).active)).toEqual([sessionID])
106    }).pipe(Effect.ensuring(Effect.sync(() => activeSessions.clear()))),
107  )
108 
109  it.effect("delegates execution continuation through SessionExecution", () =>
110    Effect.gen(function* () {
111      yield* setup
112      const session = yield* SessionV2.Service
113      executionCalls.length = 0
114      wakeCalls.length = 0
115      yield* session.resume(sessionID)
116      expect(executionCalls).toEqual([sessionID])
117      expect(wakeCalls).toEqual([])
118    }),
119  )
120 
121  it.effect("delegates process-local interruption through SessionExecution", () =>
122    Effect.gen(function* () {
123      yield* setup
124      const session = yield* SessionV2.Service
125      interruptCalls.length = 0
126 
127      yield* session.interrupt(sessionID)
128      expect(interruptCalls).toEqual([sessionID])
129      expect(yield* session.messages({ sessionID })).toEqual([])
130    }),
131  )
132 
133  it.effect("delegates interruption without requiring a recorded Session", () =>
134    Effect.gen(function* () {
135      const session = yield* SessionV2.Service
136      interruptCalls.length = 0
137 
138      yield* session.interrupt(SessionV2.ID.make("ses_missing"))
139      expect(interruptCalls).toEqual([SessionV2.ID.make("ses_missing")])
140    }),
141  )
142 
143  it.effect("durably admits one user message before transcript promotion", () =>
144    Effect.gen(function* () {
145      yield* setup
146      const session = yield* SessionV2.Service
147 
148      const message = yield* session.prompt({
149        sessionID,
150        prompt: Prompt.make({ text: "Fix the failing tests" }),
151        resume: false,
152      })
153 
154      expect(message.prompt.text).toBe("Fix the failing tests")
155      expect(yield* session.messages({ sessionID })).toEqual([])
156      expect(yield* admitted(message.id)).toMatchObject({
157        id: message.id,
158        sessionID,
159        prompt: { text: "Fix the failing tests" },
160        delivery: "steer",
161      })
162    }),
163  )
164 
165  it.effect("resolves attachment MIME before admission", () =>
166    Effect.gen(function* () {
167      yield* setup
168      const session = yield* SessionV2.Service
169 
170      const message = yield* session.prompt({
171        sessionID,
172        prompt: {
173          text: "Inspect this image",
174          files: [{ uri: "data:image/png;base64,aGVsbG8=", name: "image.png" }],
175        },
176        resume: false,
177      })
178 
179      expect(message.prompt.files).toEqual([
180        { uri: "data:image/png;base64,aGVsbG8=", name: "image.png", mime: "image/png" },
181      ])
182      expect((yield* admitted(message.id))?.prompt.files).toEqual(message.prompt.files)
183    }),
184  )
185 
186  it.effect("streams durable Session events after an aggregate sequence", () =>
187    Effect.gen(function* () {
188      yield* setup
189      const session = yield* SessionV2.Service
190      const events = yield* EventV2.Service
191      const { db } = yield* Database.Service
192      const fiber = yield* session.events({ sessionID }).pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped)
193      yield* Effect.yieldNow
194 
195      yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false })
196      yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false })
197      yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER)
198      const streamed = Array.from(yield* Fiber.join(fiber))
199 
200      expect(streamed.map((event) => [event.durable?.seq, event.type])).toEqual([
201        [0, "session.next.prompt.admitted"],
202        [1, "session.next.prompt.admitted"],
203        [2, "session.next.prompted"],
204        [3, "session.next.prompted"],
205      ])
206      expect(
207        Array.from(
208          yield* session
209            .events({ sessionID, after: streamed[0]!.durable?.seq })
210            .pipe(Stream.take(1), Stream.runCollect),
211        ).map((event) => [event.durable?.seq, event.type]),
212      ).toEqual([[1, "session.next.prompt.admitted"]])
213    }),
214  )
215 
216  it.effect("resumes through a recorded message without appending another prompt", () =>
217    Effect.gen(function* () {
218      yield* setup
219      const session = yield* SessionV2.Service
220      const message = yield* session.prompt({
221        sessionID,
222        prompt: Prompt.make({ text: "Fix the failing tests" }),
223        resume: false,
224      })
225 
226      executionCalls.length = 0
227      wakeCalls.length = 0
228      yield* session.resume(sessionID)
229 
230      expect(yield* session.messages({ sessionID })).toEqual([])
231      expect(yield* admitted(message.id)).not.toHaveProperty("promotedSeq")
232      expect(executionCalls).toEqual([sessionID])
233      expect(wakeCalls).toEqual([])
234    }),
235  )
236 
237  it.effect("records distinct messages when the ID is omitted", () =>
238    Effect.gen(function* () {
239      yield* setup
240      const session = yield* SessionV2.Service
241      const input = { sessionID, prompt: Prompt.make({ text: "Fix the failing tests" }), resume: false }
242 
243      const first = yield* session.prompt(input)
244      const second = yield* session.prompt(input)
245 
246      expect(second.id).not.toBe(first.id)
247      expect(yield* session.messages({ sessionID })).toEqual([])
248      expect(yield* admittedCount).toBe(2)
249    }),
250  )
251 
252  it.effect("returns the original recorded message when the ID is retried", () =>
253    Effect.gen(function* () {
254      yield* setup
255      const session = yield* SessionV2.Service
256      const input = {
257        sessionID,
258        id: messageID,
259        prompt: Prompt.make({ text: "Fix the failing tests" }),
260        resume: false,
261      }
262 
263      const first = yield* session.prompt(input)
264      const retried = yield* session.prompt(input)
265 
266      expect(retried).toEqual(first)
267      expect(yield* session.messages({ sessionID })).toEqual([])
268      expect(yield* admittedCount).toBe(1)
269    }),
270  )
271 
272  it.effect("wakes execution when an exact prompt retry recovers a committed message", () =>
273    Effect.gen(function* () {
274      yield* setup
275      const session = yield* SessionV2.Service
276      const input = {
277        sessionID,
278        id: messageID,
279        prompt: Prompt.make({ text: "Recover committed prompt" }),
280        resume: false,
281      }
282      const first = yield* session.prompt(input)
283      wakeCalls.length = 0
284 
285      const retried = yield* session.prompt({ ...input, resume: true })
286 
287      expect(retried).toEqual(first)
288      expect(wakeCalls).toEqual([sessionID])
289    }),
290  )
291 
292  it.effect("rejects reuse of one ID with a different prompt", () =>
293    Effect.gen(function* () {
294      yield* setup
295      const session = yield* SessionV2.Service
296 
297      yield* session.prompt({
298        sessionID,
299        id: messageID,
300        prompt: Prompt.make({ text: "Fix the failing tests" }),
301      })
302      const failure = yield* session
303        .prompt({
304          sessionID,
305          id: messageID,
306          prompt: Prompt.make({ text: "Delete the failing tests" }),
307          resume: false,
308        })
309        .pipe(Effect.flip)
310 
311      expect(failure._tag).toBe("Session.PromptConflictError")
312      expect(yield* session.messages({ sessionID })).toHaveLength(0)
313      expect(yield* admittedCount).toBe(1)
314    }),
315  )
316 
317  it.effect("rejects reuse of one ID with a different delivery mode", () =>
318    Effect.gen(function* () {
319      yield* setup
320      const session = yield* SessionV2.Service
321 
322      yield* session.prompt({
323        id: messageID,
324        sessionID,
325        prompt: Prompt.make({ text: "Fix the failing tests" }),
326        resume: false,
327      })
328      const failure = yield* session
329        .prompt({
330          id: messageID,
331          sessionID,
332          prompt: Prompt.make({ text: "Fix the failing tests" }),
333          delivery: "queue",
334          resume: false,
335        })
336        .pipe(Effect.flip)
337 
338      expect(failure._tag).toBe("Session.PromptConflictError")
339    }),
340  )
341 
342  it.effect("returns one recorded message to concurrent exact retries", () =>
343    Effect.gen(function* () {
344      yield* setup
345      const session = yield* SessionV2.Service
346      const input = {
347        sessionID,
348        id: messageID,
349        prompt: Prompt.make({ text: "Fix the failing tests" }),
350        resume: false,
351      }
352 
353      const messages = yield* Effect.all([session.prompt(input), session.prompt(input)], { concurrency: "unbounded" })
354 
355      expect(messages[1]).toEqual(messages[0])
356      expect(yield* session.messages({ sessionID })).toEqual([])
357      expect(yield* admittedCount).toBe(1)
358      expect(yield* eventCount(EventV2.versionedType(SessionEvent.PromptAdmitted.type, 1))).toBe(1)
359    }),
360  )
361 
362  it.effect("promotes one message once under concurrent promotion attempts", () =>
363    Effect.gen(function* () {
364      yield* setup
365      const { db } = yield* Database.Service
366      const session = yield* SessionV2.Service
367      const events = yield* EventV2.Service
368      yield* session.prompt({ id: messageID, sessionID, prompt: Prompt.make({ text: "Promote once" }), resume: false })
369 
370      yield* Effect.all(
371        [
372          SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER),
373          SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER),
374        ],
375        { concurrency: "unbounded" },
376      )
377 
378      expect(yield* eventCount(EventV2.versionedType(SessionEvent.Prompted.type, 1))).toBe(1)
379      expect(yield* admitted(messageID)).toMatchObject({ promotedSeq: 1 })
380      expect(yield* session.messages({ sessionID })).toMatchObject([
381        { id: messageID, type: "user", text: "Promote once" },
382      ])
383    }),
384  )
385 
386  it.effect("promotes steers only through the captured inbox cutoff", () =>
387    Effect.gen(function* () {
388      yield* setup
389      const { db } = yield* Database.Service
390      const session = yield* SessionV2.Service
391      const events = yield* EventV2.Service
392      const first = yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Before cutoff" }), resume: false })
393      const cutoff = first.admittedSeq
394      const second = yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "After cutoff" }), resume: false })
395 
396      yield* SessionInput.promoteSteers(db, events, sessionID, cutoff)
397 
398      expect(yield* admitted(first.id)).toHaveProperty("promotedSeq")
399      expect(yield* admitted(second.id)).not.toHaveProperty("promotedSeq")
400    }),
401  )
402 
403  it.effect("reprojects pending inbox input without scheduling execution", () =>
404    Effect.gen(function* () {
405      yield* setup
406      const { db } = yield* Database.Service
407      const session = yield* SessionV2.Service
408      const events = yield* EventV2.Service
409      wakeCalls.length = 0
410      yield* session.prompt({
411        id: messageID,
412        sessionID,
413        prompt: Prompt.make({ text: "Replay pending" }),
414        resume: false,
415      })
416      const recorded = yield* db
417        .select()
418        .from(EventTable)
419        .where(eq(EventTable.aggregate_id, sessionID))
420        .all()
421        .pipe(Effect.orDie)
422 
423      yield* events.remove(sessionID)
424      yield* db.delete(SessionInputTable).where(eq(SessionInputTable.session_id, sessionID)).run().pipe(Effect.orDie)
425      yield* db
426        .delete(SessionMessageTable)
427        .where(eq(SessionMessageTable.session_id, sessionID))
428        .run()
429        .pipe(Effect.orDie)
430      yield* events.replayAll(
431        recorded.map((event) => ({
432          id: event.id,
433          aggregateID: event.aggregate_id,
434          seq: event.seq,
435          type: event.type,
436          data: event.data,
437        })),
438      )
439 
440      expect(yield* admitted(messageID)).toMatchObject({ id: messageID, prompt: { text: "Replay pending" } })
441      expect(yield* session.messages({ sessionID })).toEqual([])
442      expect(wakeCalls).toEqual([])
443    }),
444  )
445 
446  it.effect("returns an exact retry of a legacy projected prompt", () =>
447    Effect.gen(function* () {
448      yield* setup
449      const session = yield* SessionV2.Service
450      const events = yield* EventV2.Service
451      const prompt = Prompt.make({ text: "Historical prompt" })
452      yield* events.publish(SessionEvent.Prompted, {
453        sessionID,
454        messageID,
455        timestamp: yield* DateTime.now,
456        prompt,
457        delivery: "steer",
458      })
459 
460      const retried = yield* session.prompt({ id: messageID, sessionID, prompt, resume: false })
461 
462      expect(retried).toMatchObject({ id: messageID, prompt: { text: "Historical prompt" } })
463      expect(yield* admitted(messageID)).toHaveProperty("promotedSeq")
464    }),
465  )
466 
467  it.effect("returns an exact retry of a legacy projected queued prompt", () =>
468    Effect.gen(function* () {
469      yield* setup
470      const session = yield* SessionV2.Service
471      const events = yield* EventV2.Service
472      const prompt = Prompt.make({ text: "Historical queued prompt" })
473      yield* events.publish(SessionEvent.Prompted, {
474        sessionID,
475        messageID,
476        timestamp: yield* DateTime.now,
477        prompt,
478        delivery: "queue",
479      })
480 
481      const retried = yield* session.prompt({ id: messageID, sessionID, prompt, delivery: "queue", resume: false })
482 
483      expect(retried).toMatchObject({ id: messageID, prompt: { text: "Historical queued prompt" } })
484      expect(yield* admitted(messageID)).toMatchObject({ delivery: "queue" })
485    }),
486  )
487 
488  it.effect("rejects reuse of one globally unique message ID across sessions", () =>
489    Effect.gen(function* () {
490      yield* setup
491      const { db } = yield* Database.Service
492      const session = yield* SessionV2.Service
493      const other = SessionV2.ID.make("ses_prompt_other")
494      yield* db
495        .insert(SessionTable)
496        .values({
497          id: other,
498          project_id: Project.ID.global,
499          slug: "other",
500          directory: "/project",
501          title: "other",
502          version: "test",
503        })
504        .onConflictDoNothing()
505        .run()
506        .pipe(Effect.orDie)
507      const prompt = Prompt.make({ text: "Fix the failing tests" })
508 
509      yield* session.prompt({ id: messageID, sessionID, prompt, resume: false })
510      const failure = yield* session
511        .prompt({ id: messageID, sessionID: other, prompt, resume: false })
512        .pipe(Effect.flip)
513 
514      expect(failure).toMatchObject({ _tag: "Session.PromptConflictError", sessionID: other, messageID })
515    }),
516  )
517 
518  it.effect("rejects a prompt ID already used by visible Session history", () =>
519    Effect.gen(function* () {
520      yield* setup
521      const session = yield* SessionV2.Service
522      const events = yield* EventV2.Service
523      yield* events.publish(SessionEvent.Synthetic, {
524        sessionID,
525        messageID,
526        timestamp: yield* DateTime.now,
527        text: "Existing history",
528      })
529 
530      const failure = yield* session
531        .prompt({ id: messageID, sessionID, prompt: Prompt.make({ text: "Conflicting prompt" }), resume: false })
532        .pipe(Effect.flip)
533 
534      expect(failure).toMatchObject({ _tag: "Session.PromptConflictError", sessionID, messageID })
535      expect(yield* admitted(messageID)).toBeUndefined()
536    }),
537  )
538 
539  it.effect("starts execution by default after recording the prompt", () =>
540    Effect.gen(function* () {
541      yield* setup
542      const session = yield* SessionV2.Service
543      executionCalls.length = 0
544      wakeCalls.length = 0
545 
546      yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Run by default" }) })
547 
548      expect(executionCalls).toEqual([])
549      expect(wakeCalls).toEqual([sessionID])
550    }),
551  )
552 
553  it.effect("starts execution when resume is explicitly true", () =>
554    Effect.gen(function* () {
555      yield* setup
556      const session = yield* SessionV2.Service
557      executionCalls.length = 0
558      wakeCalls.length = 0
559 
560      yield* session.prompt({
561        sessionID,
562        prompt: Prompt.make({ text: "Run explicitly" }),
563        resume: true,
564      })
565 
566      expect(executionCalls).toEqual([])
567      expect(wakeCalls).toEqual([sessionID])
568    }),
569  )
570 
571  it.effect("only records the prompt when resume is false", () =>
572    Effect.gen(function* () {
573      yield* setup
574      const session = yield* SessionV2.Service
575      executionCalls.length = 0
576      wakeCalls.length = 0
577 
578      yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Do not run" }), resume: false })
579 
580      expect(executionCalls).toEqual([])
581      expect(wakeCalls).toEqual([])
582    }),
583  )
584})
585