openAgent

1import { describe, expect, test } from "bun:test"
2import { LayerNode } from "@opencode-ai/core/effect/layer-node"
3import { SessionV1 } from "@opencode-ai/core/v1/session"
4import { SessionProjector } from "@opencode-ai/core/session/projector"
5import { Effect, Option } from "effect"
6import { Session as SessionNs } from "@/session/session"
7import { MessageV2 } from "../../src/session/message-v2"
8import { MessageID, PartID, type SessionID } from "../../src/session/schema"
9 
10import { NotFoundError } from "@/storage/storage"
11import { testEffect } from "../lib/effect"
12import { ProviderV2 } from "@opencode-ai/core/provider"
13import { ModelV2 } from "@opencode-ai/core/model"
14 
15const it = testEffect(LayerNode.compile(LayerNode.group([SessionNs.node, MessageV2.node, SessionProjector.node])))
16 
17const withSession = <A, E, R>(
18  fn: (input: { session: SessionNs.Interface; sessionID: SessionID }) => Effect.Effect<A, E, R>,
19) =>
20  Effect.acquireUseRelease(
21    Effect.gen(function* () {
22      const session = yield* SessionNs.Service
23      const created = yield* session.create({})
24      return { session, sessionID: created.id }
25    }),
26    fn,
27    (input) => input.session.remove(input.sessionID).pipe(Effect.ignore),
28  )
29 
30// Helper functions using Effect.gen
31const fill = Effect.fn("Test.fill")(function* (
32  sessionID: SessionID,
33  count: number,
34  time = (i: number) => Date.now() + i,
35) {
36  const session = yield* SessionNs.Service
37  const ids = [] as MessageID[]
38  for (let i = 0; i < count; i++) {
39    const id = MessageID.ascending()
40    ids.push(id)
41    yield* session.updateMessage({
42      id,
43      sessionID,
44      role: "user",
45      time: { created: time(i) },
46      agent: "test",
47      model: { providerID: "test", modelID: "test" },
48      tools: {},
49      mode: "",
50    } as unknown as SessionV1.Info)
51    yield* session.updatePart({
52      id: PartID.ascending(),
53      sessionID,
54      messageID: id,
55      type: "text",
56      text: `m${i}`,
57    })
58  }
59  return ids
60})
61 
62const addUser = Effect.fn("Test.addUser")(function* (sessionID: SessionID, text?: string) {
63  const session = yield* SessionNs.Service
64  const id = MessageID.ascending()
65  yield* session.updateMessage({
66    id,
67    sessionID,
68    role: "user",
69    time: { created: Date.now() },
70    agent: "test",
71    model: { providerID: "test", modelID: "test" },
72    tools: {},
73    mode: "",
74  } as unknown as SessionV1.Info)
75  if (text) {
76    yield* session.updatePart({
77      id: PartID.ascending(),
78      sessionID,
79      messageID: id,
80      type: "text",
81      text,
82    })
83  }
84  return id
85})
86 
87const addAssistant = Effect.fn("Test.addAssistant")(function* (
88  sessionID: SessionID,
89  parentID: MessageID,
90  opts?: { summary?: boolean; finish?: string; error?: SessionV1.Assistant["error"] },
91) {
92  const session = yield* SessionNs.Service
93  const id = MessageID.ascending()
94  yield* session.updateMessage({
95    id,
96    sessionID,
97    role: "assistant",
98    time: { created: Date.now() },
99    parentID,
100    modelID: ModelV2.ID.make("test"),
101    providerID: ProviderV2.ID.make("test"),
102    mode: "",
103    agent: "default",
104    path: { cwd: "/", root: "/" },
105    cost: 0,
106    tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } },
107    summary: opts?.summary,
108    finish: opts?.finish,
109    error: opts?.error,
110  } as unknown as SessionV1.Info)
111  return id
112})
113 
114const addCompactionPart = Effect.fn("Test.addCompactionPart")(function* (
115  sessionID: SessionID,
116  messageID: MessageID,
117  tailStartID?: MessageID,
118) {
119  const session = yield* SessionNs.Service
120  yield* session.updatePart({
121    id: PartID.ascending(),
122    sessionID,
123    messageID,
124    type: "compaction",
125    auto: true,
126    tail_start_id: tailStartID,
127  } as any)
128})
129 
130describe("MessageV2.page", () => {
131  it.instance("returns page result", () =>
132    withSession(({ sessionID }) =>
133      Effect.gen(function* () {
134        yield* fill(sessionID, 2)
135 
136        const result = yield* MessageV2.page({ sessionID, limit: 10 })
137        expect(result).toBeDefined()
138        expect(result.items).toBeArray()
139      }),
140    ),
141  )
142 
143  it.instance("pages backward with opaque cursors", () =>
144    withSession(({ sessionID }) =>
145      Effect.gen(function* () {
146        const ids = yield* fill(sessionID, 6)
147 
148        const a = yield* MessageV2.page({ sessionID, limit: 2 })
149        expect(a.items.map((item) => item.info.id)).toEqual(ids.slice(-2))
150        expect(a.items.every((item) => item.parts.length === 1)).toBe(true)
151        expect(a.more).toBe(true)
152        expect(a.cursor).toBeTruthy()
153 
154        const b = yield* MessageV2.page({ sessionID, limit: 2, before: a.cursor! })
155        expect(b.items.map((item) => item.info.id)).toEqual(ids.slice(-4, -2))
156        expect(b.more).toBe(true)
157        expect(b.cursor).toBeTruthy()
158 
159        const c = yield* MessageV2.page({ sessionID, limit: 2, before: b.cursor! })
160        expect(c.items.map((item) => item.info.id)).toEqual(ids.slice(0, 2))
161        expect(c.more).toBe(false)
162        expect(c.cursor).toBeUndefined()
163      }),
164    ),
165  )
166 
167  it.instance("returns items in chronological order within a page", () =>
168    withSession(({ sessionID }) =>
169      Effect.gen(function* () {
170        const ids = yield* fill(sessionID, 4)
171 
172        const result = yield* MessageV2.page({ sessionID, limit: 4 })
173        expect(result.items.map((item) => item.info.id)).toEqual(ids)
174      }),
175    ),
176  )
177 
178  it.instance("returns empty items for session with no messages", () =>
179    withSession(({ sessionID }) =>
180      Effect.gen(function* () {
181        const result = yield* MessageV2.page({ sessionID, limit: 10 })
182        expect(result.items).toEqual([])
183        expect(result.more).toBe(false)
184        expect(result.cursor).toBeUndefined()
185      }),
186    ),
187  )
188 
189  it.instance("fails with NotFoundError for non-existent session", () =>
190    Effect.gen(function* () {
191      const fake = "non-existent-session" as SessionID
192      const error = yield* Effect.flip(MessageV2.page({ sessionID: fake, limit: 10 }))
193      expect(error).toBeInstanceOf(NotFoundError)
194      expect(error.message).toBe(`Session not found: ${fake}`)
195    }),
196  )
197 
198  it.instance("handles exact limit boundary", () =>
199    withSession(({ sessionID }) =>
200      Effect.gen(function* () {
201        const ids = yield* fill(sessionID, 3)
202 
203        const result = yield* MessageV2.page({ sessionID, limit: 3 })
204        expect(result.items.map((item) => item.info.id)).toEqual(ids)
205        expect(result.more).toBe(false)
206        expect(result.cursor).toBeUndefined()
207      }),
208    ),
209  )
210 
211  it.instance("limit of 1 returns single newest message", () =>
212    withSession(({ sessionID }) =>
213      Effect.gen(function* () {
214        const ids = yield* fill(sessionID, 5)
215 
216        const result = yield* MessageV2.page({ sessionID, limit: 1 })
217        expect(result.items).toHaveLength(1)
218        expect(result.items[0].info.id).toBe(ids[ids.length - 1])
219        expect(result.more).toBe(true)
220      }),
221    ),
222  )
223 
224  it.instance("hydrates multiple parts per message", () =>
225    withSession(({ session, sessionID }) =>
226      Effect.gen(function* () {
227        const [id] = yield* fill(sessionID, 1)
228 
229        yield* session.updatePart({
230          id: PartID.ascending(),
231          sessionID,
232          messageID: id,
233          type: "text",
234          text: "extra",
235        })
236 
237        const result = yield* MessageV2.page({ sessionID, limit: 10 })
238        expect(result.items).toHaveLength(1)
239        expect(result.items[0].parts).toHaveLength(2)
240      }),
241    ),
242  )
243 
244  it.instance("accepts cursors from fractional timestamps", () =>
245    withSession(({ sessionID }) =>
246      Effect.gen(function* () {
247        const ids = yield* fill(sessionID, 4, (i: number) => 1000.5 + i)
248 
249        const a = yield* MessageV2.page({ sessionID, limit: 2 })
250        const b = yield* MessageV2.page({ sessionID, limit: 2, before: a.cursor! })
251 
252        expect(a.items.map((item) => item.info.id)).toEqual(ids.slice(-2))
253        expect(b.items.map((item) => item.info.id)).toEqual(ids.slice(0, 2))
254      }),
255    ),
256  )
257 
258  it.instance("messages with same timestamp are ordered by id", () =>
259    withSession(({ sessionID }) =>
260      Effect.gen(function* () {
261        const ids = yield* fill(sessionID, 4, () => 1000)
262 
263        const a = yield* MessageV2.page({ sessionID, limit: 2 })
264        expect(a.items.map((item) => item.info.id)).toEqual(ids.slice(-2))
265        expect(a.more).toBe(true)
266 
267        const b = yield* MessageV2.page({ sessionID, limit: 2, before: a.cursor! })
268        expect(b.items.map((item) => item.info.id)).toEqual(ids.slice(0, 2))
269        expect(b.more).toBe(false)
270      }),
271    ),
272  )
273 
274  it.instance("does not return messages from other sessions", () =>
275    Effect.gen(function* () {
276      const session = yield* SessionNs.Service
277      const a = yield* session.create({})
278      const b = yield* session.create({})
279      yield* fill(a.id, 3)
280      yield* fill(b.id, 2)
281 
282      const resultA = yield* MessageV2.page({ sessionID: a.id, limit: 10 })
283      const resultB = yield* MessageV2.page({ sessionID: b.id, limit: 10 })
284      expect(resultA.items).toHaveLength(3)
285      expect(resultB.items).toHaveLength(2)
286      expect(resultA.items.every((item) => item.info.sessionID === a.id)).toBe(true)
287      expect(resultB.items.every((item) => item.info.sessionID === b.id)).toBe(true)
288 
289      yield* session.remove(a.id)
290      yield* session.remove(b.id)
291    }),
292  )
293 
294  it.instance("large limit returns all messages without cursor", () =>
295    withSession(({ sessionID }) =>
296      Effect.gen(function* () {
297        const ids = yield* fill(sessionID, 10)
298 
299        const result = yield* MessageV2.page({ sessionID, limit: 100 })
300        expect(result.items).toHaveLength(10)
301        expect(result.items.map((item) => item.info.id)).toEqual(ids)
302        expect(result.more).toBe(false)
303        expect(result.cursor).toBeUndefined()
304      }),
305    ),
306  )
307})
308 
309describe("MessageV2.stream", () => {
310  it.instance("yields items newest first", () =>
311    withSession(({ sessionID }) =>
312      Effect.gen(function* () {
313        const ids = yield* fill(sessionID, 5)
314 
315        const items = yield* MessageV2.stream(sessionID)
316        expect(items.map((item) => item.info.id)).toEqual(ids.slice().reverse())
317      }),
318    ),
319  )
320 
321  it.instance("yields nothing for empty session", () =>
322    withSession(({ sessionID }) =>
323      Effect.gen(function* () {
324        const items = yield* MessageV2.stream(sessionID)
325        expect(items).toHaveLength(0)
326      }),
327    ),
328  )
329 
330  it.instance("yields single message", () =>
331    withSession(({ sessionID }) =>
332      Effect.gen(function* () {
333        const ids = yield* fill(sessionID, 1)
334 
335        const items = yield* MessageV2.stream(sessionID)
336        expect(items).toHaveLength(1)
337        expect(items[0].info.id).toBe(ids[0])
338      }),
339    ),
340  )
341 
342  it.instance("hydrates parts for each yielded message", () =>
343    withSession(({ sessionID }) =>
344      Effect.gen(function* () {
345        yield* fill(sessionID, 3)
346 
347        const items = yield* MessageV2.stream(sessionID)
348        for (const item of items) {
349          expect(item.parts).toHaveLength(1)
350          expect(item.parts[0].type).toBe("text")
351        }
352      }),
353    ),
354  )
355 
356  it.instance("handles sets exceeding internal page size", () =>
357    withSession(({ sessionID }) =>
358      Effect.gen(function* () {
359        const ids = yield* fill(sessionID, 60)
360 
361        const items = yield* MessageV2.stream(sessionID)
362        expect(items).toHaveLength(60)
363        expect(items[0].info.id).toBe(ids[ids.length - 1])
364        expect(items[59].info.id).toBe(ids[0])
365      }),
366    ),
367  )
368 
369  it.instance("returns an Effect", () =>
370    withSession(({ sessionID }) =>
371      Effect.gen(function* () {
372        yield* fill(sessionID, 1)
373 
374        const result = yield* MessageV2.stream(sessionID)
375        expect(result).toHaveLength(1)
376      }),
377    ),
378  )
379})
380 
381describe("MessageV2.parts", () => {
382  it.instance("returns parts for a message", () =>
383    withSession(({ sessionID }) =>
384      Effect.gen(function* () {
385        const [id] = yield* fill(sessionID, 1)
386 
387        const result = yield* MessageV2.parts(id)
388        expect(result).toHaveLength(1)
389        expect(result[0].type).toBe("text")
390        expect((result[0] as SessionV1.TextPart).text).toBe("m0")
391      }),
392    ),
393  )
394 
395  it.instance("returns empty array for message with no parts", () =>
396    withSession(({ sessionID }) =>
397      Effect.gen(function* () {
398        const id = yield* addUser(sessionID)
399 
400        const result = yield* MessageV2.parts(id)
401        expect(result).toEqual([])
402      }),
403    ),
404  )
405 
406  it.instance("returns multiple parts in order", () =>
407    withSession(({ session, sessionID }) =>
408      Effect.gen(function* () {
409        const [id] = yield* fill(sessionID, 1)
410 
411        yield* session.updatePart({
412          id: PartID.ascending(),
413          sessionID,
414          messageID: id,
415          type: "text",
416          text: "second",
417        })
418        yield* session.updatePart({
419          id: PartID.ascending(),
420          sessionID,
421          messageID: id,
422          type: "text",
423          text: "third",
424        })
425 
426        const result = yield* MessageV2.parts(id)
427        expect(result).toHaveLength(3)
428        expect((result[0] as SessionV1.TextPart).text).toBe("m0")
429        expect((result[1] as SessionV1.TextPart).text).toBe("second")
430        expect((result[2] as SessionV1.TextPart).text).toBe("third")
431      }),
432    ),
433  )
434 
435  it.instance("returns empty for non-existent message id", () =>
436    Effect.gen(function* () {
437      yield* SessionNs.Service
438      const result = yield* MessageV2.parts(MessageID.ascending())
439      expect(result).toEqual([])
440    }),
441  )
442 
443  it.instance("parts contain sessionID and messageID", () =>
444    withSession(({ sessionID }) =>
445      Effect.gen(function* () {
446        const [id] = yield* fill(sessionID, 1)
447 
448        const result = yield* MessageV2.parts(id)
449        expect(result[0].sessionID).toBe(sessionID)
450        expect(result[0].messageID).toBe(id)
451      }),
452    ),
453  )
454})
455 
456describe("MessageV2.get", () => {
457  it.instance("returns message with hydrated parts", () =>
458    withSession(({ sessionID }) =>
459      Effect.gen(function* () {
460        const [id] = yield* fill(sessionID, 1)
461 
462        const result = yield* MessageV2.get({ sessionID, messageID: id })
463        expect(result.info.id).toBe(id)
464        expect(result.info.sessionID).toBe(sessionID)
465        expect(result.info.role).toBe("user")
466        expect(result.parts).toHaveLength(1)
467        expect((result.parts[0] as SessionV1.TextPart).text).toBe("m0")
468      }),
469    ),
470  )
471 
472  it.instance("fails with NotFoundError for non-existent message", () =>
473    withSession(({ sessionID }) =>
474      Effect.gen(function* () {
475        const messageID = MessageID.ascending()
476        const error = yield* Effect.flip(MessageV2.get({ sessionID, messageID }))
477        expect(error).toBeInstanceOf(NotFoundError)
478        expect(error.message).toBe(`Message not found: ${messageID}`)
479      }),
480    ),
481  )
482 
483  it.instance("scopes by session id", () =>
484    Effect.gen(function* () {
485      const session = yield* SessionNs.Service
486      const a = yield* session.create({})
487      const b = yield* session.create({})
488      const [id] = yield* fill(a.id, 1)
489 
490      const error = yield* Effect.flip(MessageV2.get({ sessionID: b.id, messageID: id }))
491      expect(error).toBeInstanceOf(NotFoundError)
492      expect(error.message).toBe(`Message not found: ${id}`)
493      const result = yield* MessageV2.get({ sessionID: a.id, messageID: id })
494      expect(result.info.id).toBe(id)
495 
496      yield* session.remove(a.id)
497      yield* session.remove(b.id)
498    }),
499  )
500 
501  it.instance("returns message with multiple parts", () =>
502    withSession(({ session, sessionID }) =>
503      Effect.gen(function* () {
504        const [id] = yield* fill(sessionID, 1)
505 
506        yield* session.updatePart({
507          id: PartID.ascending(),
508          sessionID,
509          messageID: id,
510          type: "text",
511          text: "extra",
512        })
513 
514        const result = yield* MessageV2.get({ sessionID, messageID: id })
515        expect(result.parts).toHaveLength(2)
516      }),
517    ),
518  )
519 
520  it.instance("returns assistant message with correct role", () =>
521    withSession(({ session, sessionID }) =>
522      Effect.gen(function* () {
523        const uid = yield* addUser(sessionID, "hello")
524        const aid = yield* addAssistant(sessionID, uid)
525 
526        yield* session.updatePart({
527          id: PartID.ascending(),
528          sessionID,
529          messageID: aid,
530          type: "text",
531          text: "response",
532        })
533 
534        const result = yield* MessageV2.get({ sessionID, messageID: aid })
535        expect(result.info.role).toBe("assistant")
536        expect(result.parts).toHaveLength(1)
537        expect((result.parts[0] as SessionV1.TextPart).text).toBe("response")
538      }),
539    ),
540  )
541 
542  it.instance("returns message with zero parts", () =>
543    withSession(({ sessionID }) =>
544      Effect.gen(function* () {
545        const id = yield* addUser(sessionID)
546 
547        const result = yield* MessageV2.get({ sessionID, messageID: id })
548        expect(result.info.id).toBe(id)
549        expect(result.parts).toEqual([])
550      }),
551    ),
552  )
553})
554 
555describe("Session.messages", () => {
556  it.instance("returns all messages in chronological order across pages", () =>
557    withSession(({ session, sessionID }) =>
558      Effect.gen(function* () {
559        const ids = yield* fill(sessionID, 55)
560        const result = yield* session.messages({ sessionID })
561        expect(result.map((item) => item.info.id)).toEqual(ids)
562      }),
563    ),
564  )
565 
566  it.instance("fails with NotFoundError for non-existent session", () =>
567    Effect.gen(function* () {
568      const session = yield* SessionNs.Service
569      const fake = "non-existent-session" as SessionID
570      const error = yield* Effect.flip(session.messages({ sessionID: fake }))
571      expect(error).toBeInstanceOf(NotFoundError)
572      expect(error.message).toBe(`Session not found: ${fake}`)
573    }),
574  )
575})
576 
577describe("Session.findMessage", () => {
578  it.instance("searches newest-first", () =>
579    withSession(({ session, sessionID }) =>
580      Effect.gen(function* () {
581        const ids = yield* fill(sessionID, 3)
582        const result = yield* session.findMessage(sessionID, () => true)
583        expect(Option.isSome(result) ? result.value.info.id : undefined).toBe(ids.at(-1))
584      }),
585    ),
586  )
587 
588  it.instance("fails with NotFoundError for non-existent session", () =>
589    Effect.gen(function* () {
590      const session = yield* SessionNs.Service
591      const fake = "non-existent-session" as SessionID
592      const error = yield* Effect.flip(session.findMessage(fake, () => true))
593      expect(error).toBeInstanceOf(NotFoundError)
594      expect(error.message).toBe(`Session not found: ${fake}`)
595    }),
596  )
597})
598 
599describe("MessageV2.filterCompacted", () => {
600  it.instance("returns all messages when no compaction", () =>
601    withSession(({ sessionID }) =>
602      Effect.gen(function* () {
603        const ids = yield* fill(sessionID, 5)
604 
605        const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
606        expect(result).toHaveLength(5)
607        // reversed from newest-first to chronological
608        expect(result.map((item) => item.info.id)).toEqual(ids)
609      }),
610    ),
611  )
612 
613  it.instance("stops at compaction boundary and returns chronological order", () =>
614    withSession(({ session, sessionID }) =>
615      Effect.gen(function* () {
616        // Chronological: u1(+compaction part), a1(summary, parentID=u1), u2, a2
617        // Stream (newest first): a2, u2, a1(adds u1 to completed), u1(in completed + compaction) -> break
618        const u1 = yield* addUser(sessionID, "first question")
619        const a1 = yield* addAssistant(sessionID, u1, { summary: true, finish: "end_turn" })
620        yield* session.updatePart({
621          id: PartID.ascending(),
622          sessionID,
623          messageID: a1,
624          type: "text",
625          text: "summary",
626        })
627        yield* addCompactionPart(sessionID, u1)
628 
629        const u2 = yield* addUser(sessionID, "new question")
630        const a2 = yield* addAssistant(sessionID, u2)
631        yield* session.updatePart({
632          id: PartID.ascending(),
633          sessionID,
634          messageID: a2,
635          type: "text",
636          text: "new response",
637        })
638 
639        const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
640        // Includes compaction boundary: u1, a1, u2, a2
641        expect(result[0].info.id).toBe(u1)
642        expect(result.length).toBe(4)
643      }),
644    ),
645  )
646 
647  it.live("handles empty iterable", () =>
648    Effect.sync(() => {
649      const result = MessageV2.filterCompacted([])
650      expect(result).toEqual([])
651    }),
652  )
653 
654  it.instance("does not break on compaction part without matching summary", () =>
655    withSession(({ sessionID }) =>
656      Effect.gen(function* () {
657        const u1 = yield* addUser(sessionID, "hello")
658        yield* addCompactionPart(sessionID, u1)
659        yield* addUser(sessionID, "world")
660 
661        const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
662        expect(result).toHaveLength(2)
663      }),
664    ),
665  )
666 
667  it.instance("skips assistant with error even if marked as summary", () =>
668    withSession(({ sessionID }) =>
669      Effect.gen(function* () {
670        const u1 = yield* addUser(sessionID, "hello")
671        yield* addCompactionPart(sessionID, u1)
672 
673        const error = new SessionV1.APIError({
674          message: "boom",
675          isRetryable: true,
676        }).toObject() as SessionV1.Assistant["error"]
677        yield* addAssistant(sessionID, u1, { summary: true, finish: "end_turn", error })
678        yield* addUser(sessionID, "retry")
679 
680        const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
681        // Error assistant doesn't add to completed, so compaction boundary never triggers
682        expect(result).toHaveLength(3)
683      }),
684    ),
685  )
686 
687  it.instance("skips assistant without finish even if marked as summary", () =>
688    withSession(({ sessionID }) =>
689      Effect.gen(function* () {
690        const u1 = yield* addUser(sessionID, "hello")
691        yield* addCompactionPart(sessionID, u1)
692 
693        // summary=true but no finish
694        yield* addAssistant(sessionID, u1, { summary: true })
695        yield* addUser(sessionID, "next")
696 
697        const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
698        expect(result).toHaveLength(3)
699      }),
700    ),
701  )
702 
703  it.instance("retains original tail when compaction stores tail_start_id", () =>
704    withSession(({ session, sessionID }) =>
705      Effect.gen(function* () {
706        const u1 = yield* addUser(sessionID, "first")
707        const a1 = yield* addAssistant(sessionID, u1, { finish: "end_turn" })
708        yield* session.updatePart({
709          id: PartID.ascending(),
710          sessionID,
711          messageID: a1,
712          type: "text",
713          text: "first reply",
714        })
715 
716        const u2 = yield* addUser(sessionID, "second")
717        const a2 = yield* addAssistant(sessionID, u2, { finish: "end_turn" })
718        yield* session.updatePart({
719          id: PartID.ascending(),
720          sessionID,
721          messageID: a2,
722          type: "text",
723          text: "second reply",
724        })
725 
726        const c1 = yield* addUser(sessionID)
727        yield* addCompactionPart(sessionID, c1, u2)
728        const s1 = yield* addAssistant(sessionID, c1, { summary: true, finish: "end_turn" })
729        yield* session.updatePart({
730          id: PartID.ascending(),
731          sessionID,
732          messageID: s1,
733          type: "text",
734          text: "summary",
735        })
736 
737        const u3 = yield* addUser(sessionID, "third")
738        const a3 = yield* addAssistant(sessionID, u3, { finish: "end_turn" })
739        yield* session.updatePart({
740          id: PartID.ascending(),
741          sessionID,
742          messageID: a3,
743          type: "text",
744          text: "third reply",
745        })
746 
747        const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
748 
749        expect(result.map((item) => item.info.id)).toEqual([c1, s1, u2, a2, u3, a3])
750      }),
751    ),
752  )
753 
754  it.instance("fork remaps compaction tail_start_id for filterCompacted", () =>
755    Effect.gen(function* () {
756      const session = yield* SessionNs.Service
757      const created = yield* session.create({})
758 
759      const u1 = yield* addUser(created.id, "first")
760      const a1 = yield* addAssistant(created.id, u1, { finish: "end_turn" })
761      yield* session.updatePart({
762        id: PartID.ascending(),
763        sessionID: created.id,
764        messageID: a1,
765        type: "text",
766        text: "first reply",
767      })
768 
769      const u2 = yield* addUser(created.id, "second")
770      const a2 = yield* addAssistant(created.id, u2, { finish: "end_turn" })
771      yield* session.updatePart({
772        id: PartID.ascending(),
773        sessionID: created.id,
774        messageID: a2,
775        type: "text",
776        text: "second reply",
777      })
778 
779      const c1 = yield* addUser(created.id)
780      yield* addCompactionPart(created.id, c1, u2)
781      const s1 = yield* addAssistant(created.id, c1, { summary: true, finish: "end_turn" })
782      yield* session.updatePart({
783        id: PartID.ascending(),
784        sessionID: created.id,
785        messageID: s1,
786        type: "text",
787        text: "summary",
788      })
789 
790      const u3 = yield* addUser(created.id, "third")
791      const a3 = yield* addAssistant(created.id, u3, { finish: "end_turn" })
792      yield* session.updatePart({
793        id: PartID.ascending(),
794        sessionID: created.id,
795        messageID: a3,
796        type: "text",
797        text: "third reply",
798      })
799 
800      const parentFiltered = MessageV2.filterCompacted(yield* MessageV2.stream(created.id))
801      expect(parentFiltered.map((item) => item.info.id)).toEqual([c1, s1, u2, a2, u3, a3])
802 
803      const forked = yield* session.fork({ sessionID: created.id })
804      const childFiltered = MessageV2.filterCompacted(yield* MessageV2.stream(forked.id))
805      expect(childFiltered).toHaveLength(parentFiltered.length)
806 
807      const tailPart = childFiltered.flatMap((m) => m.parts).find((p) => p.type === "compaction")
808      expect(tailPart?.type).toBe("compaction")
809      if (!tailPart || tailPart.type !== "compaction") throw new Error("Expected forked compaction part")
810      expect(tailPart.tail_start_id).toBeDefined()
811      expect(childFiltered.some((m) => m.info.id === tailPart.tail_start_id)).toBe(true)
812 
813      yield* session.remove(forked.id)
814      yield* session.remove(created.id)
815    }),
816  )
817 
818  it.instance("retains an assistant tail when compaction starts inside a turn", () =>
819    withSession(({ session, sessionID }) =>
820      Effect.gen(function* () {
821        const u1 = yield* addUser(sessionID, "first")
822        const a1 = yield* addAssistant(sessionID, u1, { finish: "end_turn" })
823        yield* session.updatePart({
824          id: PartID.ascending(),
825          sessionID,
826          messageID: a1,
827          type: "text",
828          text: "first reply",
829        })
830 
831        const u2 = yield* addUser(sessionID, "second")
832        const a2 = yield* addAssistant(sessionID, u2, { finish: "end_turn" })
833        yield* session.updatePart({
834          id: PartID.ascending(),
835          sessionID,
836          messageID: a2,
837          type: "text",
838          text: "second reply",
839        })
840        const a3 = yield* addAssistant(sessionID, u2, { finish: "end_turn" })
841        yield* session.updatePart({
842          id: PartID.ascending(),
843          sessionID,
844          messageID: a3,
845          type: "text",
846          text: "tail reply",
847        })
848 
849        const c1 = yield* addUser(sessionID)
850        yield* addCompactionPart(sessionID, c1, a3)
851        const s1 = yield* addAssistant(sessionID, c1, { summary: true, finish: "end_turn" })
852        yield* session.updatePart({
853          id: PartID.ascending(),
854          sessionID,
855          messageID: s1,
856          type: "text",
857          text: "summary",
858        })
859 
860        const u3 = yield* addUser(sessionID, "third")
861        const a4 = yield* addAssistant(sessionID, u3, { finish: "end_turn" })
862        yield* session.updatePart({
863          id: PartID.ascending(),
864          sessionID,
865          messageID: a4,
866          type: "text",
867          text: "third reply",
868        })
869 
870        const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
871 
872        expect(result.map((item) => item.info.id)).toEqual([c1, s1, a3, u3, a4])
873      }),
874    ),
875  )
876 
877  it.instance("prefers latest compaction boundary when repeated compactions exist", () =>
878    withSession(({ session, sessionID }) =>
879      Effect.gen(function* () {
880        const u1 = yield* addUser(sessionID, "first")
881        const a1 = yield* addAssistant(sessionID, u1, { finish: "end_turn" })
882        yield* session.updatePart({
883          id: PartID.ascending(),
884          sessionID,
885          messageID: a1,
886          type: "text",
887          text: "first reply",
888        })
889 
890        const u2 = yield* addUser(sessionID, "second")
891        const a2 = yield* addAssistant(sessionID, u2, { finish: "end_turn" })
892        yield* session.updatePart({
893          id: PartID.ascending(),
894          sessionID,
895          messageID: a2,
896          type: "text",
897          text: "second reply",
898        })
899 
900        const c1 = yield* addUser(sessionID)
901        yield* addCompactionPart(sessionID, c1, u2)
902        const s1 = yield* addAssistant(sessionID, c1, { summary: true, finish: "end_turn" })
903        yield* session.updatePart({
904          id: PartID.ascending(),
905          sessionID,
906          messageID: s1,
907          type: "text",
908          text: "summary one",
909        })
910 
911        const u3 = yield* addUser(sessionID, "third")
912        const a3 = yield* addAssistant(sessionID, u3, { finish: "end_turn" })
913        yield* session.updatePart({
914          id: PartID.ascending(),
915          sessionID,
916          messageID: a3,
917          type: "text",
918          text: "third reply",
919        })
920 
921        const c2 = yield* addUser(sessionID)
922        yield* addCompactionPart(sessionID, c2, u3)
923        const s2 = yield* addAssistant(sessionID, c2, { summary: true, finish: "end_turn" })
924        yield* session.updatePart({
925          id: PartID.ascending(),
926          sessionID,
927          messageID: s2,
928          type: "text",
929          text: "summary two",
930        })
931 
932        const u4 = yield* addUser(sessionID, "fourth")
933        const a4 = yield* addAssistant(sessionID, u4, { finish: "end_turn" })
934        yield* session.updatePart({
935          id: PartID.ascending(),
936          sessionID,
937          messageID: a4,
938          type: "text",
939          text: "fourth reply",
940        })
941 
942        const result = MessageV2.filterCompacted(yield* MessageV2.stream(sessionID))
943 
944        expect(result.map((item) => item.info.id)).toEqual([c2, s2, u3, a3, u4, a4])
945      }),
946    ),
947  )
948 
949  test("works with array input", () => {
950    // filterCompacted accepts any Iterable, not just generators
951    const id = MessageID.ascending()
952    const items: SessionV1.WithParts[] = [
953      {
954        info: {
955          id,
956          sessionID: "s1",
957          role: "user",
958          time: { created: 1 },
959          agent: "test",
960          model: { providerID: "test", modelID: "test" },
961        } as unknown as SessionV1.Info,
962        parts: [{ type: "text", text: "hello" }] as unknown as SessionV1.Part[],
963      },
964    ]
965    const result = MessageV2.filterCompacted(items)
966    expect(result).toHaveLength(1)
967    expect(result[0].info.id).toBe(id)
968  })
969})
970 
971describe("MessageV2.cursor", () => {
972  test("encode/decode roundtrip", () => {
973    const input = { id: MessageID.ascending(), time: 1234567890 }
974    const encoded = MessageV2.cursor.encode(input)
975    const decoded = MessageV2.cursor.decode(encoded)
976    expect(decoded.id).toBe(input.id)
977    expect(decoded.time).toBe(input.time)
978  })
979 
980  test("encode/decode with fractional time", () => {
981    const input = { id: MessageID.ascending(), time: 1234567890.5 }
982    const encoded = MessageV2.cursor.encode(input)
983    const decoded = MessageV2.cursor.decode(encoded)
984    expect(decoded.time).toBe(1234567890.5)
985  })
986 
987  test("encoded cursor is base64url", () => {
988    const encoded = MessageV2.cursor.encode({ id: MessageID.ascending(), time: 0 })
989    expect(encoded).toMatch(/^[A-Za-z0-9_-]+$/)
990  })
991})
992 
993describe("MessageV2 consistency", () => {
994  it.instance("page hydration matches get for each message", () =>
995    withSession(({ sessionID }) =>
996      Effect.gen(function* () {
997        yield* fill(sessionID, 3)
998 
999        const paged = yield* MessageV2.page({ sessionID, limit: 10 })
1000        for (const item of paged.items) {
1001          const got = yield* MessageV2.get({ sessionID, messageID: item.info.id as MessageID })
1002          expect(got.info).toEqual(item.info)
1003          expect(got.parts).toEqual(item.parts)
1004        }
1005      }),
1006    ),
1007  )
1008 
1009  it.instance("parts from get match standalone parts call", () =>
1010    withSession(({ sessionID }) =>
1011      Effect.gen(function* () {
1012        const [id] = yield* fill(sessionID, 1)
1013 
1014        const got = yield* MessageV2.get({ sessionID, messageID: id })
1015        const standalone = yield* MessageV2.parts(id)
1016        expect(got.parts).toEqual(standalone)
1017      }),
1018    ),
1019  )
1020 
1021  it.instance("stream collects same messages as exhaustive page iteration", () =>
1022    withSession(({ sessionID }) =>
1023      Effect.gen(function* () {
1024        yield* fill(sessionID, 7)
1025 
1026        const streamed = yield* MessageV2.stream(sessionID)
1027 
1028        const paged = [] as SessionV1.WithParts[]
1029        let cursor: string | undefined
1030        while (true) {
1031          const result = yield* MessageV2.page({ sessionID, limit: 3, before: cursor })
1032          for (let i = result.items.length - 1; i >= 0; i--) {
1033            paged.push(result.items[i])
1034          }
1035          if (!result.more || !result.cursor) break
1036          cursor = result.cursor
1037        }
1038 
1039        expect(streamed.map((m) => m.info.id)).toEqual(paged.map((m) => m.info.id))
1040      }),
1041    ),
1042  )
1043 
1044  it.instance("filterCompacted of full stream returns same as Array.from when no compaction", () =>
1045    withSession(({ sessionID }) =>
1046      Effect.gen(function* () {
1047        yield* fill(sessionID, 4)
1048 
1049        const stream = yield* MessageV2.stream(sessionID)
1050        const filtered = MessageV2.filterCompacted(stream)
1051        const all = stream.toReversed()
1052 
1053        expect(filtered.map((m) => m.info.id)).toEqual(all.map((m) => m.info.id))
1054      }),
1055    ),
1056  )
1057})
1058