deepseek-harness/packages/api/session-controller/tests/queue-store.client.spec.ts

/** Inbox projection delivery and queue-operation transport. */

import { describe, expect, it } from 'vitest'
import { createUserMessage } from '@deepseek-ai/dsh-llm'
import type { InboxState } from '@deepseek-ai/dsh-agent/types'
import type { ContentBlock } from '@deepseek-ai/dsh-llm/types'
import type { SessionControlFrame } from '@deepseek-ai/dsh-api-session-controller/types'
import type { SessionId } from '@deepseek-ai/dsh-session/types'
import { SessionManager } from '../src/client/sessions/manager.ts'
import { FakeApiClient, fakeRemote } from './fake-api.client.ts'

const SID = 'fk-q1' as SessionId
const text = (value: string): ContentBlock[] => [{ type: 'text', text: value }]

let nextSeq = 1

function message(label: string, body: string) {
  return createUserMessage({
    content: text(body),
    source: { kind: 'user', rpcId: `rpc-${label}` } as never,
  })
}

function inboxFrame(value: InboxState): Extract<SessionControlFrame, { type: 'projection' }> {
  return {
    type: 'projection',
    sessionId: SID,
    key: 'inbox',
    seq: nextSeq++,
    value: value as never,
  }
}

describe('Inbox projection intake', () => {
  it('stores the complete Agent-owned value without adding queue state to the Session snapshot', () => {
    const manager = new SessionManager(fakeRemote(new FakeApiClient()))
    const queued = message('queued', 'later')
    const steering = message('steering', 'now')
    const value = { 'next-turn': [queued], 'next-step': [steering] }

    manager.handleControlFrame(inboxFrame(value))
    const session = manager.get(SID)

    expect(session.projections.faceOf('inbox').getSnapshot()).toEqual(value)
    expect(session.getSnapshot()).not.toHaveProperty('queue')
  })

  it('drops a projection that claims events beyond a reconnect baseline', () => {
    const manager = new SessionManager(fakeRemote(new FakeApiClient()))
    manager.handleControlFrame(inboxFrame({
      'next-turn': [message('stale', 'stale')],
      'next-step': [],
    }))

    manager.handleControlFrame({
      type: 'baseline',
      value: {
        jobs: {},
        projections: { [SID]: { asOfSeq: 0, values: {} } },
      },
    })

    expect(manager.get(SID).projections.faceOf('inbox').getSnapshot()).toBeUndefined()
  })

  it('retains only the highest-seq value received before Session materialization', () => {
    const manager = new SessionManager(fakeRemote(new FakeApiClient()))
    manager.handleControlFrame(inboxFrame({
      'next-turn': [message('old', 'old')],
      'next-step': [],
    }))
    const latest = {
      'next-turn': [message('latest', 'latest')],
      'next-step': [],
    }
    manager.handleControlFrame(inboxFrame(latest))

    expect(manager.get(SID).projections.faceOf('inbox').getSnapshot()).toEqual(latest)
  })
})

describe('queue operation transport', () => {
  it('does not mutate the Inbox projection before the Host publishes its committed value', async () => {
    const api = new FakeApiClient()
    const manager = new SessionManager(fakeRemote(api))
    const pending = message('pending', 'before')
    const initial = { 'next-turn': [pending], 'next-step': [] }
    manager.handleControlFrame(inboxFrame(initial))
    const session = manager.get(SID)

    await expect(session.updateQueue(pending.id, { kind: 'edit', content: text('after') }))
      .resolves.toEqual({ ok: true, value: { accepted: true } })
    expect(api.callsOf('session.updateQueue')).toEqual([{
      sessionId: SID,
      itemId: pending.id,
      action: { kind: 'edit', content: text('after') },
    }])
    expect(session.projections.faceOf('inbox').getSnapshot()).toBe(initial)
  })
})