session-projections.host.spec.ts 25 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597
  1. /**
  2. * Session Controller projection paths: the history tail page's
  3. * projections block reads the registry's watermark snapshot (asOfSeq = last
  4. * event seq, one consistent cut); loadOlder pages never carry the block; a
  5. * composition without the registry serves histories without it; a disposed
  6. * registration's key leaves subsequent responses; and every unit change is
  7. * pushed through the control stream.
  8. */
  9. import { describe, expect, it, vi } from 'vitest'
  10. import { mkdtemp, readFile, rm } from 'node:fs/promises'
  11. import { tmpdir } from 'node:os'
  12. import { join } from 'node:path'
  13. import { Context } from '@deepseek-ai/cordis'
  14. import { z } from 'zod'
  15. import AgentRegistry, { Inbox } from '@deepseek-ai/dsh-agent'
  16. import { AttachmentStore } from '@deepseek-ai/dsh-attachment'
  17. import { agentPresetProjectionDefinition } from '@deepseek-ai/dsh-agent-presets'
  18. import type { Agent } from '@deepseek-ai/dsh-agent'
  19. import { createUserMessage } from '@deepseek-ai/dsh-llm'
  20. import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
  21. import type { Session } from '@deepseek-ai/dsh-session'
  22. import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
  23. import type { ProjectionDefinition } from '@deepseek-ai/dsh-session-projection'
  24. import SessionProjectionCache, { projectionCacheDomainSpec } from '@deepseek-ai/dsh-session-projection-cache'
  25. import Storage from '@deepseek-ai/dsh-storage'
  26. import * as StorageDomain from '@deepseek-ai/dsh-storage-domain'
  27. import * as StorageJson from '@deepseek-ai/dsh-storage-json'
  28. import { SessionControlController } from '@deepseek-ai/dsh-api-session-controller/src/control.ts'
  29. import type { SessionControlFrame, SessionFollowFrame } from '@deepseek-ai/dsh-api-session-controller/types'
  30. import { createSessionTestRemote, type TestSessionRemote } from './test-remote.ts'
  31. declare module '@deepseek-ai/dsh-session-projection/types' {
  32. interface SessionProjectionStateMap {
  33. 'test/last-user': LastUserState
  34. 'test/internal-count': number
  35. 'test/private-prompt': string | null
  36. }
  37. interface SessionProjectionMap {
  38. 'test/last-user': { text: string } | null
  39. }
  40. }
  41. function request<P>(payload: P): P {
  42. return payload
  43. }
  44. function page(
  45. remote: TestSessionRemote,
  46. request: { sessionId: SessionId; throughSeq: number; beforeSeq?: number; maxMessages?: number },
  47. ) {
  48. return remote.page({
  49. address: { kind: 'session', sessionId: request.sessionId },
  50. throughSeq: request.throughSeq,
  51. ...(request.beforeSeq === undefined ? {} : { beforeSeq: request.beforeSeq }),
  52. ...(request.maxMessages === undefined ? {} : { maxMessages: request.maxMessages }),
  53. })
  54. }
  55. /** Read and close one snapshot-first follow generation. */
  56. async function opening(
  57. remote: TestSessionRemote,
  58. sessionId: SessionId,
  59. maxMessages?: number,
  60. ): Promise<Extract<SessionFollowFrame, { type: 'snapshot' }>> {
  61. const abort = new AbortController()
  62. const iterator = remote.follow({
  63. address: { kind: 'session', sessionId },
  64. ...(maxMessages === undefined ? {} : { maxMessages }),
  65. }, abort.signal)[Symbol.asyncIterator]()
  66. const first = await iterator.next()
  67. abort.abort()
  68. await iterator.return?.()
  69. if (first.done || first.value.type !== 'snapshot') throw new Error('follow did not open with a snapshot')
  70. return first.value
  71. }
  72. /** Whole-value unit folding the latest user/message text; null before the first. */
  73. type LastUserState = { text: string } | null
  74. const lastUserUnit = () => ({
  75. key: 'test/last-user',
  76. stateSchema: z.union([z.object({ text: z.string() }), z.null()]),
  77. init: () => null,
  78. apply: (state, event) => (event.type === 'user/message'
  79. ? { text: (event.data.content[0] as { text?: string }).text ?? '' }
  80. : state),
  81. wire: {
  82. viewSchema: z.union([z.object({ text: z.string() }), z.null()]),
  83. view: state => state,
  84. },
  85. stateVersion: 1,
  86. }) satisfies ProjectionDefinition<'test/last-user', LastUserState>
  87. const internalCountUnit = () => ({
  88. key: 'test/internal-count',
  89. stateSchema: z.number().int().nonnegative(),
  90. init: () => 0,
  91. apply: (state: number) => state + 1,
  92. stateVersion: 1,
  93. }) satisfies ProjectionDefinition<'test/internal-count', number>
  94. const privatePromptUnit = () => ({
  95. key: 'test/private-prompt',
  96. stateSchema: z.string().nullable(),
  97. init: () => null,
  98. apply: (state, event) => (event.type === 'user/message'
  99. ? (event.data.content[0] as { text?: string }).text ?? ''
  100. : state),
  101. stateVersion: 1,
  102. }) satisfies ProjectionDefinition<'test/private-prompt', string | null>
  103. async function harness(withRegistry: boolean): Promise<{ ctx: Context; session: Session }> {
  104. const ctx = new Context()
  105. await ctx.plugin(SessionStore)
  106. await ctx.plugin(AgentRegistry)
  107. if (withRegistry) await ctx.plugin(SessionProjectionRegistry)
  108. const session = ctx.sessions.create(undefined, { meta: { cwd: '/workspace' } })
  109. // The gateway reads both the session and durable inbox baseline.
  110. ctx.agents.register({ id: session.id, session, inbox: new Inbox(session, { inserted: () => {}, discarded: () => {}, claimed: () => {} }), status: 'idle', ctx } as Agent)
  111. return { ctx, session }
  112. }
  113. /** Append `count` user messages so the log has paginable message boundaries. */
  114. function seedMessages(session: Session, count: number): void {
  115. for (let i = 0; i < count; i++) {
  116. session.append('user/message', createUserMessage({
  117. content: [{ type: 'text', text: `m${i}` }],
  118. source: { kind: 'user' },
  119. }), { surfaceOp: 'append' })
  120. }
  121. }
  122. const remote = (ctx: Context) => createSessionTestRemote(ctx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
  123. describe('session.history projections block', () => {
  124. it('tracks pending and used model selections across repeated request headers', async () => {
  125. const { ctx, session } = await harness(true)
  126. remote(ctx)
  127. await new Promise(resolve => setTimeout(resolve, 0))
  128. const selected = { provider: 'p', model: 'next' }
  129. session.append('model/selection', selected)
  130. session.append('model/selection', selected)
  131. session.append('request/header', {
  132. header: { config: { provider: 'p', model: 'used' } }, reason: 'initial',
  133. })
  134. session.append('request/header', {
  135. header: { config: { provider: 'p', model: 'used' } }, reason: 'initial',
  136. })
  137. expect(ctx.sessionProjections.snapshot(session).values.modelSelection).toEqual({
  138. lastUsed: { provider: 'p', model: 'used' },
  139. next: selected,
  140. })
  141. session.append('request/header', {
  142. header: { config: selected }, reason: 'initial',
  143. })
  144. expect(ctx.sessionProjections.snapshot(session).values.modelSelection).toEqual({
  145. lastUsed: selected,
  146. next: selected,
  147. })
  148. })
  149. it('serves the unit value on the tail page with asOfSeq = last event seq', async () => {
  150. const { ctx, session } = await harness(true)
  151. ctx.sessionProjections.register(lastUserUnit())
  152. seedMessages(session, 3)
  153. const snapshot = await opening(remote(ctx), session.id)
  154. const { records, projections } = snapshot
  155. expect(projections.asOfSeq).toBe(session.seq - 1)
  156. expect(projections.values['test/last-user']).toEqual({ text: 'm2' })
  157. // asOfSeq IS the window tail: the last served event carries it.
  158. const last = records.at(-1)
  159. expect(last?.event.seq).toBe(projections.asOfSeq)
  160. })
  161. it('returns a complete current replacement cut on each follow generation', async () => {
  162. const { ctx, session } = await harness(true)
  163. ctx.sessionProjections.register(lastUserUnit())
  164. seedMessages(session, 2)
  165. const snapshot = await opening(remote(ctx), session.id)
  166. expect(snapshot.records.map(record => record.event.seq)).toEqual([0, 1])
  167. expect(snapshot.projections.asOfSeq).toBe(1)
  168. expect(snapshot.projections.values).toEqual(
  169. expect.objectContaining({ 'test/last-user': { text: 'm1' } }),
  170. )
  171. })
  172. it('projects an empty log at cursor -1', async () => {
  173. const { ctx, session } = await harness(true)
  174. ctx.sessionProjections.register(lastUserUnit())
  175. const snapshot = await opening(remote(ctx), session.id)
  176. expect(snapshot.records).toEqual([])
  177. expect(snapshot.projections.asOfSeq).toBe(-1)
  178. expect(snapshot.projections.values).toEqual(
  179. expect.objectContaining({ 'test/last-user': null }),
  180. )
  181. })
  182. it('publishes the attachments imageLimits as a constant unit while both seams are composed', async () => {
  183. const { ctx, session } = await harness(true)
  184. const limits = {
  185. maxImageBytes: 5 * 1024 * 1024,
  186. maxImagesPerMessage: 20,
  187. maxMessageImageBytes: 100 * 1024 * 1024,
  188. maxImagePixels: 40_000_000,
  189. maxImageDimension: 2000,
  190. mediaTypes: ['image/png'] as const,
  191. }
  192. await ctx.plugin(class extends AttachmentStore {
  193. readonly imageLimits = limits
  194. validateImage(): Promise<void> { return Promise.resolve() }
  195. saveImage(): Promise<never> { return Promise.reject(new Error('unused')) }
  196. readImage(): Promise<never> { return Promise.reject(new Error('unused')) }
  197. })
  198. const gateway = remote(ctx)
  199. await new Promise(resolve => setTimeout(resolve, 0))
  200. seedMessages(session, 2)
  201. const snapshot = await opening(gateway, session.id)
  202. expect(snapshot.projections.values['imageLimits']).toEqual(limits)
  203. // Constant unit: appending events must never broadcast an imageLimits projection.
  204. await new Promise(resolve => setTimeout(resolve, 0))
  205. const abort = new AbortController()
  206. const iterator = gateway.control(abort.signal)[Symbol.asyncIterator]()
  207. await iterator.next()
  208. const next = iterator.next()
  209. seedMessages(session, 1)
  210. await new Promise(resolve => setTimeout(resolve, 0))
  211. await expect(next).resolves.toMatchObject({
  212. done: false,
  213. value: { type: 'projection', key: 'sessionListMetadata' },
  214. })
  215. const extra = iterator.next()
  216. const quiet = Symbol('quiet')
  217. expect(await Promise.race([
  218. extra,
  219. new Promise<typeof quiet>(resolve => setTimeout(() => { resolve(quiet) }, 0)),
  220. ])).toBe(quiet)
  221. abort.abort()
  222. await expect(extra).resolves.toEqual({ done: true, value: undefined })
  223. })
  224. it('leaves the imageLimits key absent while no attachment service is composed', async () => {
  225. const { ctx, session } = await harness(true)
  226. seedMessages(session, 1)
  227. const snapshot = await opening(remote(ctx), session.id)
  228. expect('imageLimits' in snapshot.projections.values).toBe(false)
  229. })
  230. it('never carries the block on loadOlder pages (beforeSeq present)', async () => {
  231. const { ctx, session } = await harness(true)
  232. ctx.sessionProjections.register(lastUserUnit())
  233. seedMessages(session, 5)
  234. const older = await page(remote(ctx), request({
  235. sessionId: session.id, throughSeq: session.seq - 1, beforeSeq: 3, maxMessages: 2,
  236. }))
  237. expect(older.ok).toBe(true)
  238. if (!older.ok) throw new Error('unreachable')
  239. expect('projections' in older.value).toBe(false)
  240. })
  241. it('serves no block when the composition has no projection registry', async () => {
  242. const { ctx, session } = await harness(false)
  243. seedMessages(session, 2)
  244. const response = await page(remote(ctx), request({ sessionId: session.id, throughSeq: session.seq - 1 }))
  245. expect(response.ok).toBe(true)
  246. if (!response.ok) throw new Error('unreachable')
  247. expect('projections' in response.value).toBe(false)
  248. })
  249. it('never exposes a host-only unit through history, listing, or push frames', async () => {
  250. const { ctx, session } = await harness(true)
  251. ctx.sessionProjections.register(internalCountUnit())
  252. const proxy = remote(ctx)
  253. await new Promise(resolve => setTimeout(resolve, 0))
  254. const abort = new AbortController()
  255. const iterator = proxy.control(abort.signal)[Symbol.asyncIterator]()
  256. const baseline = await iterator.next()
  257. if (baseline.done || baseline.value.type !== 'baseline') {
  258. throw new Error('control stream ended before its baseline')
  259. }
  260. expect('test/internal-count' in (baseline.value.value.projections[session.id]?.values ?? {}))
  261. .toBe(false)
  262. seedMessages(session, 1)
  263. const changed = await iterator.next()
  264. expect(changed).toMatchObject({
  265. done: false,
  266. value: { type: 'projection', key: 'sessionListMetadata' },
  267. })
  268. abort.abort()
  269. await iterator.return?.()
  270. const history = await opening(proxy, session.id)
  271. expect('test/internal-count' in history.projections.values).toBe(false)
  272. const listing = await proxy.list(request({}))
  273. if (!listing.ok) throw new Error('listing failed')
  274. const row = listing.value.items.find(item => item.sessionId === session.id)
  275. expect('test/internal-count' in (row?.projections?.values ?? {})).toBe(false)
  276. })
  277. it('drops a disposed registration from subsequent tail pages (empty block, key absent)', async () => {
  278. const { ctx, session } = await harness(true)
  279. const dispose = ctx.sessionProjections.register(lastUserUnit())
  280. seedMessages(session, 1)
  281. const proxy = remote(ctx)
  282. const before = await opening(proxy, session.id)
  283. expect(before.projections.values['test/last-user']).toEqual({ text: 'm0' })
  284. dispose()
  285. const after = await opening(proxy, session.id)
  286. // The registry stays mounted; only the disposed key leaves while the
  287. // gateway-owned Session-list unit remains.
  288. expect(after.projections.asOfSeq).toBe(session.seq - 1)
  289. expect('test/last-user' in after.projections.values).toBe(false)
  290. expect(after.projections.values.sessionListMetadata).toEqual({
  291. blank: true,
  292. lastPromptAt: session.events.at(-1)?.time,
  293. })
  294. })
  295. it('removes the gateway-owned Session-list unit when the gateway fiber unloads', async () => {
  296. const { ctx, session } = await harness(true)
  297. expect('sessionListMetadata' in ctx.sessionProjections.snapshot(session).values).toBe(false)
  298. const fiber = ctx.plugin(Object.assign((gatewayCtx: Context) => {
  299. createSessionTestRemote(gatewayCtx, { defaultModelSelection: () => ({ provider: 'p', model: 'm' }), cwd: '/tmp' })
  300. }, { inject: ['sessions', 'agents', 'sessionProjections'] }))
  301. await fiber.await()
  302. await vi.waitFor(() => {
  303. expect(ctx.sessionProjections.snapshot(session).values.sessionListMetadata)
  304. .toEqual({ blank: true, lastPromptAt: null })
  305. })
  306. await fiber.dispose()
  307. expect('sessionListMetadata' in ctx.sessionProjections.snapshot(session).values).toBe(false)
  308. })
  309. })
  310. describe('session.list projections column', () => {
  311. it('serves every already-materialized wire value from the live registry without folding', async () => {
  312. const { ctx, session } = await harness(true)
  313. ctx.sessionProjections.register(lastUserUnit())
  314. const gateway = remote(ctx)
  315. await new Promise(resolve => setTimeout(resolve, 0))
  316. session.append('turn/start', { turn: 1 })
  317. seedMessages(session, 1)
  318. const response = await gateway.list(request({}))
  319. if (!response.ok) throw new Error('unreachable')
  320. const row = response.value.items.find(item => item.sessionId === session.id)
  321. expect(row?.projections?.values['test/last-user']).toEqual({ text: 'm0' })
  322. expect(row?.projections?.values.sessionListMetadata).toEqual({
  323. blank: false,
  324. lastPromptAt: session.events.at(-1)?.time,
  325. })
  326. expect(row?.projections?.asOfSeq).toBe(session.seq - 1)
  327. })
  328. it('lists the latest preset selected by a blank Session instead of its creation preset', async () => {
  329. const { ctx } = await harness(true)
  330. const session = ctx.sessions.create(SessionId('preset-list'), {
  331. meta: { cwd: '/workspace', agentPreset: 'standard' },
  332. })
  333. ctx.sessionProjections.register(agentPresetProjectionDefinition)
  334. const gateway = remote(ctx)
  335. await new Promise(resolve => setTimeout(resolve, 0))
  336. session.append('agent-preset/selected', { agentPreset: 'minimal' })
  337. const response = await gateway.list(request({}))
  338. if (!response.ok) throw new Error('unreachable')
  339. const row = response.value.items.find(item => item.sessionId === session.id)
  340. expect(row?.projections?.values.agentPreset).toBe('minimal')
  341. })
  342. it('omits an unmaterialized live projection instead of folding history for listing', async () => {
  343. const { ctx, session } = await harness(true)
  344. seedMessages(session, 1)
  345. const unit = lastUserUnit()
  346. const apply = vi.fn(unit.apply)
  347. ctx.sessionProjections.register({ ...unit, apply })
  348. const response = await remote(ctx).list(request({}))
  349. if (!response.ok) throw new Error('unreachable')
  350. const row = response.value.items.find(item => item.sessionId === session.id)
  351. expect(row).toBeDefined()
  352. expect('test/last-user' in (row?.projections?.values ?? {})).toBe(false)
  353. expect(apply).not.toHaveBeenCalled()
  354. })
  355. it('omits the column entirely when no registry is mounted', async () => {
  356. const { ctx, session } = await harness(false)
  357. seedMessages(session, 1)
  358. const response = await remote(ctx).list(request({}))
  359. if (!response.ok) throw new Error('unreachable')
  360. const row = response.value.items.find(item => item.sessionId === session.id)
  361. expect(row).toBeDefined()
  362. expect(row !== undefined && 'projections' in row).toBe(false)
  363. })
  364. it('serves every available cold projection hint from the cache with zero log loads', async () => {
  365. const { ctx } = await harness(true)
  366. const coldId = SessionId('session-cold-listing')
  367. const load = () => { throw new Error('list must not load event logs') }
  368. ctx.provide('sessionPersistence', {
  369. list: async () => [{ version: 0, id: coldId, createdAt: 5, cwd: '/tmp' }],
  370. locate: () => undefined,
  371. load,
  372. inspect: load,
  373. readFrom: load,
  374. } as never)
  375. ctx.provide('sessionProjectionCache', {
  376. // The carrier hands the listed header through as the identity witness.
  377. cachedSnapshot: (meta: { id: unknown; createdAt: number }) =>
  378. (meta.id === coldId && meta.createdAt === 5
  379. ? {
  380. asOfSeq: 7,
  381. values: {
  382. 'test/last-user': { text: 'cached' },
  383. sessionListMetadata: { blank: false, lastPromptAt: 6 },
  384. title: 'Cached title',
  385. },
  386. }
  387. : undefined),
  388. } as never)
  389. const response = await remote(ctx).list(request({}))
  390. if (!response.ok) throw new Error('unreachable')
  391. const row = response.value.items.find(item => item.sessionId === coldId)
  392. expect(row?.running).toBe(false)
  393. expect(row?.projections).toEqual({
  394. asOfSeq: 7,
  395. values: {
  396. 'test/last-user': { text: 'cached' },
  397. sessionListMetadata: { blank: false, lastPromptAt: 6 },
  398. title: 'Cached title',
  399. },
  400. })
  401. })
  402. it('keeps persisted host-only state out of a cold session.list response', async () => {
  403. const root = await mkdtemp(join(tmpdir(), 'dsh-api-projcache-'))
  404. const ctx = new Context()
  405. try {
  406. await ctx.plugin(Storage)
  407. await ctx.plugin(StorageJson, { root })
  408. await ctx.plugin(StorageDomain, { backend: 'json' })
  409. await ctx.plugin(SessionStore)
  410. await ctx.plugin(AgentRegistry)
  411. await ctx.plugin(SessionProjectionRegistry)
  412. ctx.sessionProjections.register(privatePromptUnit())
  413. await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 100, writeIntervalMs: 60_000 })
  414. const gateway = remote(ctx)
  415. await new Promise(resolve => setTimeout(resolve, 0))
  416. const id = SessionId('session-cold-host-state')
  417. const secret = 'private prompt text from the cache'
  418. let session: Session | undefined
  419. const owner = await ctx.plugin(Object.assign((sessionCtx: Context) => {
  420. session = sessionCtx.sessions.create(id, { meta: { createdAt: 5, cwd: '/workspace' } })
  421. }, { inject: ['sessions'] }))
  422. if (session === undefined) throw new Error('session was not created')
  423. session.append('turn/start', { turn: 1 })
  424. session.append('user/message', createUserMessage({
  425. content: [{ type: 'text', text: secret }],
  426. source: { kind: 'user' },
  427. }), { surfaceOp: 'append' })
  428. await ctx.sessionProjectionCache.write(session)
  429. const stored = await readFile(
  430. join(root, projectionCacheDomainSpec.name, 'sessions', `${id}.json`),
  431. 'utf8',
  432. )
  433. expect(stored).toContain(secret)
  434. const header = session.header
  435. await owner.dispose()
  436. expect(ctx.sessions.get(id)).toBeUndefined()
  437. ctx.provide('sessionPersistence', {
  438. list: async () => [header],
  439. locate: () => undefined,
  440. } as never)
  441. const response = await gateway.list(request({}))
  442. if (!response.ok) throw new Error('unreachable')
  443. const row = response.value.items.find(item => item.sessionId === id)
  444. expect(row?.projections?.values.sessionListMetadata).toMatchObject({ blank: false })
  445. expect('test/private-prompt' in (row?.projections?.values ?? {})).toBe(false)
  446. expect(JSON.stringify(row)).not.toContain(secret)
  447. } finally {
  448. await ctx.fiber.dispose()
  449. await rm(root, { recursive: true, force: true })
  450. }
  451. })
  452. it('cold rows without a cache plugin (or without a stored row) just lack the column', async () => {
  453. const { ctx } = await harness(true)
  454. const coldId = SessionId('session-cold-uncached')
  455. ctx.provide('sessionPersistence', {
  456. list: async () => [{ version: 0, id: coldId, createdAt: 5, cwd: '/tmp' }],
  457. locate: () => undefined,
  458. } as never)
  459. const response = await remote(ctx).list(request({}))
  460. if (!response.ok) throw new Error('unreachable')
  461. const row = response.value.items.find(item => item.sessionId === coldId)
  462. expect(row).toBeDefined()
  463. expect(row !== undefined && 'projections' in row).toBe(false)
  464. })
  465. it('a throwing column read degrades that row, never the listing', async () => {
  466. const { ctx, session } = await harness(true)
  467. ctx.sessionProjections.register({
  468. ...lastUserUnit(),
  469. wire: {
  470. viewSchema: z.union([z.object({ text: z.string() }), z.null()]),
  471. view: () => { throw new Error('unit exploded') },
  472. },
  473. })
  474. seedMessages(session, 1)
  475. const response = await remote(ctx).list(request({}))
  476. if (!response.ok) throw new Error('unreachable')
  477. const row = response.value.items.find(item => item.sessionId === session.id)
  478. expect(row).toBeDefined()
  479. expect(row !== undefined && 'projections' in row).toBe(false)
  480. })
  481. })
  482. describe('Session control projection frames', () => {
  483. /** Drain frames until `count` projection replacements arrive. */
  484. async function collect(
  485. iterable: AsyncIterable<SessionControlFrame>,
  486. count: number,
  487. abort: AbortController,
  488. ): Promise<SessionControlFrame[]> {
  489. const frames: SessionControlFrame[] = []
  490. for await (const frame of iterable) {
  491. frames.push(frame)
  492. if (frames.filter(candidate => candidate.type === 'projection').length >= count) abort.abort()
  493. }
  494. return frames
  495. }
  496. it('broadcasts a frame per changed unit with the causing seq, and none for same-reference applies', async () => {
  497. const { ctx, session } = await harness(true)
  498. ctx.sessionProjections.register(lastUserUnit())
  499. const proxy = remote(ctx)
  500. // The controller's onChanged subscription lives in an inject child whose
  501. // fiber activates asynchronously; yield until it lands before appending.
  502. await new Promise(resolve => setTimeout(resolve, 0))
  503. const abort = new AbortController()
  504. const stream = proxy.control(abort.signal)
  505. const collected = collect(stream, 5, abort)
  506. const now = vi.spyOn(Date, 'now').mockReturnValue(100)
  507. seedMessages(session, 1)
  508. now.mockReturnValue(200)
  509. session.append('turn/start', { turn: 1 })
  510. now.mockReturnValue(300)
  511. seedMessages(session, 1)
  512. now.mockRestore()
  513. const frames = await collected
  514. const pushes = frames.filter(
  515. (f): f is Extract<SessionControlFrame, { type: 'projection' }> =>
  516. f.type === 'projection' && f.key === 'test/last-user',
  517. )
  518. expect(pushes).toEqual([
  519. { type: 'projection', sessionId: session.id, key: 'test/last-user', value: { text: 'm0' }, seq: 0 },
  520. { type: 'projection', sessionId: session.id, key: 'test/last-user', value: { text: 'm0' }, seq: 2 },
  521. ])
  522. expect(frames.filter(
  523. (f): f is Extract<SessionControlFrame, { type: 'projection' }> =>
  524. f.type === 'projection' && f.key === 'sessionListMetadata',
  525. )).toEqual([
  526. { type: 'projection', sessionId: session.id, key: 'sessionListMetadata', value: { blank: true, lastPromptAt: 100 }, seq: 0 },
  527. { type: 'projection', sessionId: session.id, key: 'sessionListMetadata', value: { blank: false, lastPromptAt: 100 }, seq: 1 },
  528. { type: 'projection', sessionId: session.id, key: 'sessionListMetadata', value: { blank: false, lastPromptAt: 300 }, seq: 2 },
  529. ])
  530. // Frame seq aligns with the tail block's asOfSeq vocabulary (higher-seq-wins compatible).
  531. const tail = await opening(proxy, session.id)
  532. expect(tail.projections.asOfSeq).toBe(pushes.at(-1)?.seq)
  533. })
  534. it('emits no projection frames when the composition has no registry', async () => {
  535. const { ctx, session } = await harness(false)
  536. const control = new SessionControlController(ctx)
  537. const abort = new AbortController()
  538. const iterator = control.control(abort.signal)[Symbol.asyncIterator]()
  539. const baseline = await iterator.next()
  540. const next = iterator.next()
  541. seedMessages(session, 2)
  542. await new Promise(resolve => setTimeout(resolve, 0))
  543. abort.abort()
  544. if (baseline.done) throw new Error('Control stream ended before its baseline')
  545. expect(baseline.value.type).toBe('baseline')
  546. await expect(next).resolves.toEqual({ done: true, value: undefined })
  547. })
  548. })