openAgent

1import { describe, expect } from "bun:test"
2import { DateTime, Effect, Schema } from "effect"
3import { asc, eq, sql } from "drizzle-orm"
4import { Database } from "@opencode-ai/core/database/database"
5import { LayerNode } from "@opencode-ai/core/effect/layer-node"
6import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder"
7import { EventV2 } from "@opencode-ai/core/event"
8import { EventTable } from "@opencode-ai/core/event/sql"
9import { ModelV2 } from "@opencode-ai/core/model"
10import { Project } from "@opencode-ai/core/project"
11import { ProjectTable } from "@opencode-ai/core/project/sql"
12import { ProviderV2 } from "@opencode-ai/core/provider"
13import { AbsolutePath } from "@opencode-ai/core/schema"
14import { SessionV2 } from "@opencode-ai/core/session"
15import { SessionEvent } from "@opencode-ai/core/session/event"
16import { SessionMessage } from "@opencode-ai/core/session/message"
17import { Prompt } from "@opencode-ai/core/session/prompt"
18import { SessionMessageUpdater } from "@opencode-ai/core/session/message-updater"
19import { SessionProjector } from "@opencode-ai/core/session/projector"
20import { SessionExecution } from "@opencode-ai/core/session/execution"
21import { SessionInput } from "@opencode-ai/core/session/input"
22import { SessionInputTable, SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql"
23import { testEffect } from "./lib/effect"
24import { Snapshot } from "@opencode-ai/core/snapshot"
25import { Location } from "@opencode-ai/core/location"
26 
27const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, SessionProjector.node])))
28const sessionsLayer = AppNodeBuilder.build(SessionV2.node, [[SessionExecution.node, SessionExecution.noopLayer]])
29const sessionID = SessionV2.ID.make("ses_projector_test")
30const created = DateTime.makeUnsafe(0)
31const model = { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") }
32const encodeMessage = Schema.encodeSync(SessionMessage.Message)
33 
34const assistantRow = (
35  id: SessionMessage.ID,
36  seq: number,
37  time: { created: DateTime.Utc; completed?: DateTime.Utc } = { created },
38) => {
39  const {
40    id: _,
41    type,
42    ...data
43  } = encodeMessage(SessionMessage.Assistant.make({ id, type: "assistant", agent: "build", model, content: [], time }))
44  return { id, session_id: sessionID, type, seq, time_created: DateTime.toEpochMillis(time.created), data }
45}
46 
47describe("SessionProjector", () => {
48  it.effect("projects moved sessions without the transitional context epoch table", () =>
49    Effect.gen(function* () {
50      const { db } = yield* Database.Service
51      const events = yield* EventV2.Service
52      yield* db
53        .insert(ProjectTable)
54        .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
55        .run()
56      yield* db
57        .insert(SessionTable)
58        .values({
59          id: sessionID,
60          project_id: Project.ID.global,
61          slug: "test",
62          directory: "/project",
63          title: "test",
64          version: "test",
65        })
66        .run()
67      yield* db.run(sql`DROP TABLE session_context_epoch`)
68 
69      yield* events.publish(SessionEvent.Moved, {
70        sessionID,
71        timestamp: DateTime.makeUnsafe(1),
72        location: Location.Ref.make({ directory: AbsolutePath.make("/project/subdir") }),
73      })
74 
75      expect(yield* db.select({ directory: SessionTable.directory }).from(SessionTable).get()).toEqual({
76        directory: "/project/subdir",
77      })
78    }),
79  )
80 
81  it.effect("projects staged, cleared, and committed reverts", () =>
82    Effect.gen(function* () {
83      const db = (yield* Database.Service).db
84      yield* db
85        .insert(ProjectTable)
86        .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
87        .run()
88      yield* db
89        .insert(SessionTable)
90        .values({
91          id: sessionID,
92          project_id: Project.ID.global,
93          slug: "test",
94          directory: "/project",
95          title: "test",
96          version: "test",
97        })
98        .run()
99      const boundary = SessionMessage.ID.make("msg_boundary")
100      yield* db
101        .insert(SessionMessageTable)
102        .values([assistantRow(boundary, 1), assistantRow(SessionMessage.ID.make("msg_later"), 2)])
103        .run()
104      const events = yield* EventV2.Service
105      yield* events.publish(SessionEvent.RevertEvent.Staged, {
106        sessionID,
107        timestamp: DateTime.makeUnsafe(1),
108        revert: { messageID: boundary, snapshot: Snapshot.ID.make("tree"), diff: "patch", files: [] },
109      })
110      expect((yield* db.select({ revert: SessionTable.revert }).from(SessionTable).get())?.revert).toMatchObject({
111        messageID: boundary,
112        snapshot: "tree",
113        files: [],
114      })
115      yield* events.publish(SessionEvent.RevertEvent.Cleared, { sessionID, timestamp: DateTime.makeUnsafe(2) })
116      expect((yield* db.select({ revert: SessionTable.revert }).from(SessionTable).get())?.revert).toBeNull()
117      yield* events.publish(SessionEvent.RevertEvent.Staged, {
118        sessionID,
119        timestamp: DateTime.makeUnsafe(3),
120        revert: { messageID: boundary, files: [] },
121      })
122      yield* events.publish(SessionEvent.RevertEvent.Committed, {
123        sessionID,
124        messageID: boundary,
125        timestamp: DateTime.makeUnsafe(4),
126      })
127      expect(
128        (yield* db.select({ id: SessionMessageTable.id }).from(SessionMessageTable).all()).map((row) => row.id),
129      ).toEqual([boundary])
130    }),
131  )
132 
133  it.effect("orders projected messages and context by durable aggregate sequence", () =>
134    Effect.gen(function* () {
135      const { db } = yield* Database.Service
136      yield* db
137        .insert(ProjectTable)
138        .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
139        .run()
140        .pipe(Effect.orDie)
141      yield* db
142        .insert(SessionTable)
143        .values({
144          id: sessionID,
145          project_id: Project.ID.global,
146          slug: "test",
147          directory: "/project",
148          title: "test",
149          version: "test",
150        })
151        .run()
152        .pipe(Effect.orDie)
153      const events = yield* EventV2.Service
154 
155      yield* events.publish(
156        SessionEvent.Prompted,
157        {
158          sessionID,
159          messageID: SessionMessage.ID.make("msg_first"),
160          timestamp: created,
161          prompt: Prompt.make({ text: "first" }),
162          delivery: "steer",
163        },
164        { id: EventV2.ID.make("evt_z") },
165      )
166      yield* events.publish(
167        SessionEvent.Prompted,
168        {
169          sessionID,
170          messageID: SessionMessage.ID.make("msg_second"),
171          timestamp: created,
172          prompt: Prompt.make({ text: "second" }),
173          delivery: "steer",
174        },
175        { id: EventV2.ID.make("evt_a") },
176      )
177 
178      const sessions = yield* SessionV2.Service
179      const firstPage = yield* sessions.messages({ sessionID, limit: 1, order: "asc" })
180      expect(firstPage.map((message) => (message.type === "user" ? message.text : message.type))).toEqual(["first"])
181      const secondPage = yield* sessions.messages({
182        sessionID,
183        limit: 1,
184        order: "asc",
185        cursor: { id: firstPage[0]!.id, direction: "next" },
186      })
187      expect(secondPage.map((message) => (message.type === "user" ? message.text : message.type))).toEqual(["second"])
188      expect(
189        (yield* sessions.messages({
190          sessionID,
191          limit: 1,
192          order: "asc",
193          cursor: { id: secondPage[0]!.id, direction: "previous" },
194        })).map((message) => (message.type === "user" ? message.text : message.type)),
195      ).toEqual(["first"])
196      expect(
197        (yield* sessions.context(sessionID)).map((message) => (message.type === "user" ? message.text : message.type)),
198      ).toEqual(["first", "second"])
199    }).pipe(Effect.provide(sessionsLayer)),
200  )
201 
202  it.effect("marks an inbox row promoted with the Prompted event sequence", () =>
203    Effect.gen(function* () {
204      const { db } = yield* Database.Service
205      yield* db
206        .insert(ProjectTable)
207        .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
208        .run()
209        .pipe(Effect.orDie)
210      yield* db
211        .insert(SessionTable)
212        .values({
213          id: sessionID,
214          project_id: Project.ID.global,
215          slug: "test",
216          directory: "/project",
217          title: "test",
218          version: "test",
219        })
220        .run()
221        .pipe(Effect.orDie)
222      const events = yield* EventV2.Service
223      const id = SessionMessage.ID.make("msg_admitted")
224      const admitted = yield* SessionInput.admit(db, events, {
225        id,
226        sessionID,
227        prompt: Prompt.make({ text: "promote me" }),
228        delivery: "steer",
229      })
230      if (!admitted) return yield* Effect.die("Prompt admission failed")
231 
232      const event = yield* events.publish(SessionEvent.Prompted, {
233        sessionID,
234        timestamp: admitted.timeCreated,
235        messageID: id,
236        prompt: Prompt.make({ text: "promote me" }),
237        delivery: "steer",
238      })
239 
240      expect(
241        yield* db.select().from(SessionInputTable).where(eq(SessionInputTable.id, id)).get().pipe(Effect.orDie),
242      ).toMatchObject({ promoted_seq: event.durable?.seq })
243    }),
244  )
245 
246  it.effect("projects durable context messages supported by the updater", () =>
247    Effect.gen(function* () {
248      const { db } = yield* Database.Service
249      yield* db
250        .insert(ProjectTable)
251        .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
252        .run()
253        .pipe(Effect.orDie)
254      yield* db
255        .insert(SessionTable)
256        .values({
257          id: sessionID,
258          project_id: Project.ID.global,
259          slug: "test",
260          directory: "/project",
261          title: "test",
262          version: "test",
263        })
264        .run()
265        .pipe(Effect.orDie)
266      const events = yield* EventV2.Service
267 
268      yield* events.publish(SessionEvent.AgentSwitched, {
269        sessionID,
270        messageID: SessionMessage.ID.create(),
271        timestamp: created,
272        agent: "build",
273      })
274      yield* events.publish(SessionEvent.ModelSwitched, {
275        sessionID,
276        messageID: SessionMessage.ID.create(),
277        timestamp: created,
278        model,
279      })
280      yield* events.publish(SessionEvent.Synthetic, {
281        sessionID,
282        messageID: SessionMessage.ID.create(),
283        timestamp: created,
284        text: "synthetic context",
285      })
286      yield* events.publish(SessionEvent.Shell.Started, {
287        sessionID,
288        messageID: SessionMessage.ID.create(),
289        timestamp: created,
290        callID: "shell-1",
291        command: "pwd",
292      })
293      yield* events.publish(SessionEvent.Shell.Ended, {
294        sessionID,
295        timestamp: DateTime.makeUnsafe(1),
296        callID: "shell-1",
297        output: "/project",
298      })
299      const compactionID = SessionMessage.ID.create()
300      yield* events.publish(SessionEvent.Compaction.Started, {
301        sessionID,
302        messageID: compactionID,
303        timestamp: created,
304        reason: "manual",
305      })
306      yield* events.publish(SessionEvent.Compaction.Delta, {
307        sessionID,
308        messageID: compactionID,
309        timestamp: created,
310        text: "partial",
311      })
312      expect(
313        yield* db
314          .select({ id: EventTable.id })
315          .from(EventTable)
316          .where(eq(EventTable.type, SessionEvent.Compaction.Delta.type))
317          .all()
318          .pipe(Effect.orDie),
319      ).toEqual([])
320      expect(
321        yield* db
322          .select({ id: SessionMessageTable.id })
323          .from(SessionMessageTable)
324          .where(eq(SessionMessageTable.type, "compaction"))
325          .all()
326          .pipe(Effect.orDie),
327      ).toEqual([])
328      yield* events.publish(SessionEvent.Compaction.Ended, {
329        sessionID,
330        messageID: compactionID,
331        timestamp: DateTime.makeUnsafe(1),
332        reason: "manual",
333        text: "summary",
334        recent: "recent context",
335      })
336 
337      const rows = yield* db
338        .select()
339        .from(SessionMessageTable)
340        .where(eq(SessionMessageTable.session_id, sessionID))
341        .orderBy(asc(SessionMessageTable.seq))
342        .all()
343        .pipe(Effect.orDie)
344      const messages = rows.map((row) =>
345        Schema.decodeUnknownSync(SessionMessage.Message)({ ...row.data, id: row.id, type: row.type }),
346      )
347 
348      expect(messages.map((message) => message.type)).toEqual([
349        "agent-switched",
350        "model-switched",
351        "synthetic",
352        "shell",
353        "compaction",
354      ])
355      expect(messages.find((message) => message.type === "shell")).toMatchObject({
356        output: "/project",
357        time: { completed: DateTime.makeUnsafe(1) },
358      })
359      expect(messages.find((message) => message.type === "compaction")).toMatchObject({
360        summary: "summary",
361        recent: "recent context",
362      })
363      expect(
364        yield* db.select().from(SessionTable).where(eq(SessionTable.id, sessionID)).get().pipe(Effect.orDie),
365      ).toMatchObject({
366        agent: "build",
367        model,
368        time_updated: DateTime.toEpochMillis(created),
369      })
370    }),
371  )
372 
373  it.effect("rejects distinct creator events that reuse one projected message ID", () =>
374    Effect.gen(function* () {
375      const { db } = yield* Database.Service
376      yield* db
377        .insert(ProjectTable)
378        .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
379        .run()
380        .pipe(Effect.orDie)
381      yield* db
382        .insert(SessionTable)
383        .values({
384          id: sessionID,
385          project_id: Project.ID.global,
386          slug: "test",
387          directory: "/project",
388          title: "test",
389          version: "test",
390        })
391        .run()
392        .pipe(Effect.orDie)
393      const events = yield* EventV2.Service
394      const id = SessionMessage.ID.make("msg_creator_collision")
395 
396      yield* events.publish(SessionEvent.Synthetic, { sessionID, messageID: id, timestamp: created, text: "keep me" })
397      const exit = yield* events
398        .publish(SessionEvent.Step.Started, {
399          sessionID,
400          assistantMessageID: id,
401          timestamp: created,
402          agent: "build",
403          model,
404        })
405        .pipe(Effect.exit)
406 
407      expect(exit._tag).toBe("Failure")
408      expect(
409        yield* db.select().from(SessionMessageTable).where(eq(SessionMessageTable.id, id)).get().pipe(Effect.orDie),
410      ).toMatchObject({ type: "synthetic" })
411    }),
412  )
413 
414  it.effect("does not revive a stale incomplete in-memory assistant projection", () =>
415    Effect.gen(function* () {
416      const stale = SessionMessage.Assistant.make({
417        id: SessionMessage.ID.make("msg_assistant_stale"),
418        type: "assistant",
419        agent: "build",
420        model,
421        content: [],
422        time: { created },
423      })
424      const completed = SessionMessage.Assistant.make({
425        id: SessionMessage.ID.make("msg_assistant_completed"),
426        type: "assistant",
427        agent: "build",
428        model,
429        content: [],
430        time: { created: DateTime.makeUnsafe(1), completed: DateTime.makeUnsafe(2) },
431      })
432 
433      expect(
434        yield* SessionMessageUpdater.memory({ messages: [stale, completed] }).getCurrentAssistant(),
435      ).toBeUndefined()
436    }),
437  )
438 
439  it.effect("updates only the newest incomplete assistant projection", () =>
440    Effect.gen(function* () {
441      const { db } = yield* Database.Service
442      yield* db
443        .insert(ProjectTable)
444        .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
445        .run()
446        .pipe(Effect.orDie)
447      yield* db
448        .insert(SessionTable)
449        .values({
450          id: sessionID,
451          project_id: Project.ID.global,
452          slug: "test",
453          directory: "/project",
454          title: "test",
455          version: "test",
456        })
457        .run()
458        .pipe(Effect.orDie)
459      yield* db
460        .insert(SessionMessageTable)
461        .values([
462          assistantRow(SessionMessage.ID.make("msg_assistant_1"), 0),
463          assistantRow(SessionMessage.ID.make("msg_assistant_2"), 1),
464        ])
465        .run()
466        .pipe(Effect.orDie)
467 
468      const service = yield* EventV2.Service
469      yield* service.publish(SessionEvent.Step.Ended, {
470        sessionID,
471        timestamp: DateTime.makeUnsafe(1),
472        assistantMessageID: SessionMessage.ID.make("msg_assistant_2"),
473        finish: "stop",
474        cost: 0,
475        tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
476      })
477 
478      const rows = yield* db
479        .select()
480        .from(SessionMessageTable)
481        .where(eq(SessionMessageTable.session_id, sessionID))
482        .orderBy(asc(SessionMessageTable.id))
483        .all()
484        .pipe(Effect.orDie)
485      const messages = rows.map((row) =>
486        Schema.decodeUnknownSync(SessionMessage.Message)({ ...row.data, id: row.id, type: row.type }),
487      )
488      expect(messages[0]).not.toHaveProperty("time.completed")
489      expect(messages[1]).toMatchObject({
490        type: "assistant",
491        finish: "stop",
492        time: { completed: DateTime.makeUnsafe(1) },
493      })
494    }),
495  )
496 
497  it.effect("does not revive a stale incomplete assistant projection", () =>
498    Effect.gen(function* () {
499      const { db } = yield* Database.Service
500      yield* db
501        .insert(ProjectTable)
502        .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] })
503        .run()
504        .pipe(Effect.orDie)
505      yield* db
506        .insert(SessionTable)
507        .values({
508          id: sessionID,
509          project_id: Project.ID.global,
510          slug: "test",
511          directory: "/project",
512          title: "test",
513          version: "test",
514        })
515        .run()
516        .pipe(Effect.orDie)
517      yield* db
518        .insert(SessionMessageTable)
519        .values([
520          assistantRow(SessionMessage.ID.make("msg_assistant_stale"), 0),
521          assistantRow(SessionMessage.ID.make("msg_assistant_completed"), 1, {
522            created: DateTime.makeUnsafe(1),
523            completed: DateTime.makeUnsafe(2),
524          }),
525        ])
526        .run()
527        .pipe(Effect.orDie)
528 
529      const service = yield* EventV2.Service
530      yield* service.publish(SessionEvent.Text.Started, {
531        sessionID,
532        assistantMessageID: SessionMessage.ID.make("msg_assistant_completed"),
533        timestamp: DateTime.makeUnsafe(3),
534        textID: "text-stale",
535      })
536 
537      const rows = yield* db
538        .select()
539        .from(SessionMessageTable)
540        .where(eq(SessionMessageTable.session_id, sessionID))
541        .orderBy(asc(SessionMessageTable.id))
542        .all()
543        .pipe(Effect.orDie)
544      const messages = rows.map((row) =>
545        Schema.decodeUnknownSync(SessionMessage.Message)({ ...row.data, id: row.id, type: row.type }),
546      )
547      expect(messages).toEqual([
548        SessionMessage.Assistant.make({
549          id: SessionMessage.ID.make("msg_assistant_completed"),
550          type: "assistant",
551          agent: "build",
552          model,
553          content: [SessionMessage.AssistantText.make({ type: "text", id: "text-stale", text: "" })],
554          time: { created: DateTime.makeUnsafe(1), completed: DateTime.makeUnsafe(2) },
555        }),
556        SessionMessage.Assistant.make({
557          id: SessionMessage.ID.make("msg_assistant_stale"),
558          type: "assistant",
559          agent: "build",
560          model,
561          content: [],
562          time: { created },
563        }),
564      ])
565    }),
566  )
567})
568