session-projections.host.spec.ts 26 KB

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