修复归档文件授权与登出后的 Bot 事件连接

src/server/fileAccess.ts

@@ -50,7 +50,7 @@ export async function fileAccessWorkingDirectory(upstream: FileHost, profile: st
   return path
 }
 
-export async function authorizeFileRead(config: ServerConfig, upstream: FileHost, raw: string, profile = 'default', directoryRequest = false): Promise<string> {
+export async function authorizeFileRead(config: Pick<ServerConfig, 'home'>, upstream: FileHost, raw: string, profile = 'default', directoryRequest = false): Promise<string> {
   const path = serverFilePath(raw, false)
   if (!path) throw new HttpError(400, '文件路径无效', 'invalid_file_path')
   const policy = readFileAccess(config.home)

src/server/workspaceAssets.ts

@@ -2,16 +2,19 @@ import { randomUUID, createHash } from 'node:crypto'
 import { mkdirSync, writeFileSync } from 'node:fs'
 import { join, basename } from 'node:path'
 import {HttpError} from './errors.js'
+import { authorizeFileRead } from './fileAccess.js'
 import { lookup } from 'mime-types'
 import { messageFileReferences, type MessageFileOrigin } from '../shared/messageFiles.js'
 import { WorkspaceStore } from './workspaceStore.js'
-import { WorkspaceNodes } from './workspaceGateway.js'
+import { WorkspaceNodes, type WorkspaceNode } from './workspaceGateway.js'
 import type { WorkspaceAgent, WorkspaceFile, WorkspaceMessage } from '../shared/workspace.js'
 export interface StoredWorkspaceFile extends WorkspaceFile, MessageFileOrigin {
   path: string
   digest?: string
   sourceNodeId?:string
 }
+const archiveKey = (nodeId: string, profile: string, messageId: string, path: string) =>
+  createHash('sha256').update(JSON.stringify([nodeId, profile, messageId, path])).digest('hex')
 export function publicFile(f: StoredWorkspaceFile): WorkspaceFile {
   const { path: _path, digest: _digest, messageFileSource: _source, ...result } = f
   return result
@@ -83,15 +86,39 @@ export class WorkspaceAssets {
     readonly nodes: WorkspaceNodes,
     readonly home: string,
   ) {}
+  private requireCapacity(owner: string, messageId: string, bytes: number): void {
+    const count = Number(this.store.db.prepare("SELECT count(*) AS n FROM workspace_entities WHERE owner=? AND kind='file' AND json_extract(data,'$.messageId')=?").get(owner, messageId)?.n ?? 0)
+    if (bytes > 25 * 1024 * 1024 || count >= 8)
+      throw new HttpError(413, '每轮最多 8 个产物，每个不超过 25 MiB', 'artifact_limit')
+    const used = Number(this.store.db.prepare("SELECT coalesce(sum(json_extract(data,'$.size')),0) AS n FROM workspace_entities WHERE kind='file'").get()?.n ?? 0)
+    if (used + bytes > 2 * 1024 * 1024 * 1024) throw new HttpError(413, '产物存储空间已达上限', 'artifact_quota')
+  }
+  /** Legacy archives lost nodeId, but their immutable archive key includes it.
+   * Restore only a unique authorized match; never assume a file came from local. */
+  recoverFileSource(owner: string, file: StoredWorkspaceFile): StoredWorkspaceFile {
+    if (file.sourceNodeId || !file.sourcePath || !file.profile || !file.messageId) return file
+    const records = Number(this.store.db.prepare("SELECT count(*) AS n FROM workspace_entities WHERE owner=? AND kind='file' AND json_extract(data,'$.sourcePath')=? AND json_extract(data,'$.profile')=? AND json_extract(data,'$.messageId')=?").get(owner, file.sourcePath, file.profile, file.messageId)?.n ?? 0)
+    if (records !== 1) return file
+    const candidates = ['local', ...this.store.list<WorkspaceNode>(owner, 'node').map(node => node.id)]
+    let revokedMatch = false
+    const matches = [...new Set(candidates)].filter(nodeId => {
+      if (!this.store.get(owner, 'archived-path', archiveKey(nodeId, file.profile!, file.messageId!, file.sourcePath!))) return false
+      try { this.nodes.requireSource(owner, { nodeId, profile: file.profile! }) }
+      catch { revokedMatch = true; return false }
+      return true
+    })
+    if (revokedMatch || matches.length !== 1) return file
+    const recovered = { ...file, sourceNodeId: matches[0]! }
+    this.store.put(owner, 'file', file.id, recovered)
+    return recovered
+  }
   publish(owner:string,message:WorkspaceMessage,agent:WorkspaceAgent,name:string,bytes:Buffer):WorkspaceFile {
     const digest=createHash('sha256').update(bytes).digest('hex')
     name=basename(name).replace(/[\\/\u0000-\u001f\u007f]/g,'_').slice(0,240)||'artifact'
     const files=this.store.list<StoredWorkspaceFile>(owner,'file')
     const existing=files.find(file=>file.messageId===message.id&&file.name===name&&file.digest===digest)
     if(existing)return publicFile(existing)
-    if(bytes.length>25*1024*1024||files.filter(file=>file.messageId===message.id).length>=8)throw new HttpError(413,'每轮最多 8 个产物，每个不超过 25 MiB','artifact_limit')
-    const used=Number(this.store.db.prepare("SELECT coalesce(sum(json_extract(data,'$.size')),0) AS n FROM workspace_entities WHERE kind='file'").get()?.n??0)
-    if(used+bytes.length>2*1024*1024*1024)throw new HttpError(413,'产物存储空间已达上限','artifact_quota')
+    this.requireCapacity(owner, message.id, bytes.length)
     const dir=join(this.home,'workspace-files',owner);mkdirSync(dir,{recursive:true,mode:0o700})
     const path=join(dir,digest)
     try{writeFileSync(path,bytes,{mode:0o600,flag:'wx'})}catch(error){if((error as NodeJS.ErrnoException).code!=='EEXIST')throw error}
@@ -143,19 +170,26 @@ export class WorkspaceAssets {
       }
     }
     for (const path of [...paths].slice(0, 8)) {
-      const key = createHash('sha256')
-        .update(JSON.stringify([nodeId, profile, messageId, path]))
-        .digest('hex')
+      const key = archiveKey(nodeId, profile, messageId, path)
       if (this.store.get(owner, 'archived-path', key) || this.pending.has(`${owner}:${key}`))
         continue
       this.pending.add(`${owner}:${key}`)
       try {
-        const response = await (remoteAgentId ? this.nodes.targetForAgent(owner,{nodeId,remoteAgentId}) : this.nodes.target(owner,nodeId))
-          .session.request('/api/files/download', {
-            search: new URLSearchParams({ profile, path }),
-            maxResponseBytes: 25 * 1024 * 1024,
-          })
+        this.nodes.requireSource(owner, { nodeId, profile })
+        this.requireCapacity(owner, messageId, 0)
+        const target = remoteAgentId ? this.nodes.targetForAgent(owner,{nodeId,remoteAgentId}) : this.nodes.target(owner,nodeId)
+        const authorizedPath = await authorizeFileRead({ home: this.home }, target.session, path, profile)
+        this.nodes.requireSource(owner, { nodeId, profile })
+        const response = await target.session.request('/api/files/download', {
+          search: new URLSearchParams({ profile, path: authorizedPath }),
+          maxResponseBytes: 25 * 1024 * 1024,
+        })
         if (response.status !== 200 || this.stopped) continue
+        // A download may outlive an assignment or directory-policy change.
+        this.nodes.requireSource(owner, { nodeId, profile })
+        if (await authorizeFileRead({ home: this.home }, target.session, path, profile) !== authorizedPath || this.stopped) continue
+        this.nodes.requireSource(owner, { nodeId, profile })
+        this.requireCapacity(owner, messageId, response.body.length)
         const dir = join(this.home, 'workspace-files', owner)
         mkdirSync(dir, { recursive: true, mode: 0o700 })
         const digest = createHash('sha256').update(response.body).digest('hex')
@@ -177,6 +211,7 @@ export class WorkspaceAssets {
           size: response.body.length,
           sender,
           profile,
+          sourceNodeId: nodeId,
           conversationId,
           messageId,
           createdAt: Date.now(),

src/server/workspaceRoutes.ts

@@ -56,15 +56,19 @@ export function workspaceRouter(
         .get(user, id)?.n ?? 0,
     )
   const fileRecord = (user: string, id: string): StoredWorkspaceFile => {
+    const authorize = (file: StoredWorkspaceFile) => {
+      file = assets.recoverFileSource(user, file)
+      if ((file.sourcePath || file.sourceNodeId) && (!file.sourceNodeId || !file.profile)) throw new HttpError(403, '归档文件来源无法确认，请重新交付文件', 'archived_file_source_unknown')
+      if (file.sourceNodeId && file.profile) nodes.requireSource(user, { nodeId: file.sourceNodeId, profile: file.profile })
+      return file
+    }
     if (/^\d+$/.test(id)) {
       const row = store.db
         .prepare("SELECT data FROM workspace_entities WHERE owner=? AND kind='file' AND rowid=?")
         .get(user, Number(id))
-      if (row) {const file:StoredWorkspaceFile=JSON.parse(String(row.data));if(file.sourceNodeId&&file.profile)nodes.requireSource(user,{nodeId:file.sourceNodeId,profile:file.profile});return file}
+      if (row) return authorize(JSON.parse(String(row.data)))
     }
-    const file=store.require<StoredWorkspaceFile>(user,'file',id)
-    if(file.sourceNodeId&&file.profile)nodes.requireSource(user,{nodeId:file.sourceNodeId,profile:file.profile})
-    return file
+    return authorize(store.require<StoredWorkspaceFile>(user,'file',id))
   }
   router.get('/api/app/capabilities', (ctx) => {
     owner(ctx)

src/server/workspaceSync.ts

@@ -29,6 +29,7 @@ export function workspaceDetail(store: WorkspaceStore, runtime: WorkspaceRuntime
 /** Durable replay followed by live updates. A slow reader reconnects from its last applied cursor. */
 export function streamWorkspace(ctx: Koa.Context, store: WorkspaceStore, auth: LocalAuthStore): void {
   const owner = auth.require(ctx).id, version = auth.pushAuthorizationVersion(owner)
+  const cookie = ctx.get('cookie')
   const patches = store.messagePatchesEnabled && ctx.query.format === 'patch-v1'
   const res = ctx.res
   const raw = ctx.get('last-event-id') || String(ctx.query.after ?? '0')
@@ -38,7 +39,9 @@ export function streamWorkspace(ctx: Koa.Context, store: WorkspaceStore, auth: L
   const pending = new Map<string, WorkspaceEvent>()
   const knownMessages = new Set<string>()
   const valid = () => {
-    try { return auth.require(ctx).id === owner && auth.isUserActive(owner) && auth.pushAuthorizationVersion(owner) === version }
+    // Koa's localUser is a request snapshot; a long-lived stream must check the
+    // captured session itself so logout and expiry revoke this connection.
+    try { return auth.currentFromCookieHeader(cookie)?.id === owner && auth.isUserActive(owner) && auth.pushAuthorizationVersion(owner) === version }
     catch { return false }
   }
   const close = () => {

tests/server/messageFiles.test.ts

@@ -8,12 +8,14 @@ import { WorkspaceAssets, type StoredWorkspaceFile } from '../../src/server/work
 import type { WorkspaceNodes } from '../../src/server/workspaceGateway'
 import type { WorkspaceMessage, WorkspaceConversation } from '../../src/shared/workspace'
 import { messageFileReferences, nativeMessageFileText } from '../../src/shared/messageFiles'
+import { saveFileAccess } from '../../src/server/fileAccess'
 
 it('archives only message references and reconciles legacy associations without deleting bytes or logs', async () => {
   const home = mkdtempSync(join(tmpdir(), 'yaoyao-message-files-'))
   const store = new WorkspaceStore(home)
   const request = vi.fn(async (_route: string, _options: { search: URLSearchParams }) => ({ status: 200, body: Buffer.from('report'), headers: new Headers({ 'content-type': 'text/plain' }) }))
-  const assets = new WorkspaceAssets(store, { target: () => ({ session: { request } }) } as unknown as WorkspaceNodes, home)
+  saveFileAccess(home, { mode: 'all', folders: [] })
+  const assets = new WorkspaceAssets(store, { requireSource: () => {}, target: () => ({ session: { request } }) } as unknown as WorkspaceNodes, home)
   try {
     const agent = store.createAgent('owner', { name: 'Agent', profile: 'default' })
     const conversation = store.list<WorkspaceConversation>('owner', 'conversation')[0]!

tests/server/workspaceAssetsAuthorization.test.ts

@@ -0,0 +1,149 @@
+// @vitest-environment node
+import { createHash } from 'node:crypto'
+import { existsSync, mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import type Koa from 'koa'
+import { afterEach, expect, it, vi } from 'vitest'
+import { WorkspaceAssets, type StoredWorkspaceFile } from '../../src/server/workspaceAssets'
+import { WorkspaceStore } from '../../src/server/workspaceStore'
+import { workspaceRouter } from '../../src/server/workspaceRoutes'
+import { saveFileAccess } from '../../src/server/fileAccess'
+import { HttpError } from '../../src/server/errors'
+import type { WorkspaceNodes } from '../../src/server/workspaceGateway'
+import type { WorkspaceRuntime } from '../../src/server/workspaceRuntime'
+import type { UploadStore } from '../../src/server/uploads'
+import type { LocalAuthStore } from '../../src/server/localAuth'
+import type { PushCoordinator } from '../../src/server/pushCoordinator'
+
+const fixtures: Array<{ home: string; store: WorkspaceStore; assets: WorkspaceAssets }> = []
+afterEach(() => {
+  for (const { home, store, assets } of fixtures.splice(0)) {
+    assets.close(); store.close(); rmSync(home, { recursive: true, force: true })
+  }
+})
+const response = (value: unknown) => ({ status: 200, body: Buffer.from(JSON.stringify(value)), headers: new Headers() })
+const downloaded = () => ({ status: 200, body: Buffer.from('fixture-report'), headers: new Headers({ 'content-type': 'text/plain' }) })
+const keyFor = (nodeId: string) => createHash('sha256').update(JSON.stringify([nodeId, 'alpha', 'message', '/remote/work/report.txt'])).digest('hex')
+function fixture() {
+  const home = mkdtempSync(join(tmpdir(), 'yaoyao-assets-auth-')), store = new WorkspaceStore(home)
+  let allowed = true
+  const request = vi.fn(async (route: string, options: { search?: URLSearchParams }) => {
+    if (route === '/api/config') return response({ terminal: { cwd: '/remote/work' } })
+    if (route === '/api/files') {
+      const path = options.search!.get('path')!
+      return response({ path, entries: [{ name: 'report.txt', path: path + '/report.txt' }, { name: 'escape.txt', path: '/private/escape.txt' }] })
+    }
+    if (route === '/api/files/download') return downloaded()
+    throw new Error('Unexpected mocked request: ' + route)
+  })
+  const requireSource = vi.fn((owner: string, _source: { nodeId: string; profile: string }) => {
+    if (owner !== 'owner' || !allowed) throw new HttpError(403, 'Fixture source revoked', 'agent_source_forbidden')
+  })
+  const nodes = { requireSource, target: () => ({ session: { request } }) } as unknown as WorkspaceNodes
+  const assets = new WorkspaceAssets(store, nodes, home)
+  const router = workspaceRouter(store, {} as WorkspaceRuntime, nodes, assets, {} as UploadStore,
+    { require: () => ({ id: 'owner' }) } as LocalAuthStore, {} as PushCoordinator)
+  const handler = router.stack.find(layer => layer.path === '/api/app/files/:id/download')!.stack[0]!
+  const download = async (id: string) => {
+    const ctx = { params: { id }, get: () => '', set() {} } as unknown as Koa.Context
+    await handler(ctx, async () => {})
+    const bytes: Buffer[] = []
+    for await (const chunk of ctx.body as NodeJS.ReadableStream) bytes.push(Buffer.from(chunk))
+    return Buffer.concat(bytes).toString()
+  }
+  fixtures.push({ home, store, assets })
+  return { home, store, assets, request, requireSource, download, revoke: () => { allowed = false },
+    files: () => store.list<StoredWorkspaceFile>('owner', 'file'),
+    archive: (path = '/remote/work/report.txt', messageId = 'message') =>
+      assets.archiveText('owner', `[report](${path})`, 'local', 'alpha', 'conversation', messageId) }
+}
+
+it('archives authorized references with their source and checks it for UUID and library-row downloads', async () => {
+  const f = fixture()
+  await f.archive()
+  const file = f.files()[0]!
+  expect(file).toMatchObject({ sourceNodeId: 'local', profile: 'alpha', sourcePath: '/remote/work/report.txt' })
+  expect(await f.download(file.id)).toBe('fixture-report')
+  const row = f.store.db.prepare("SELECT rowid AS n FROM workspace_entities WHERE owner='owner' AND kind='file' AND id=?").get(file.id)!
+  f.revoke()
+  for (const id of [file.id, String(row.n)]) await expect(f.download(id)).rejects.toMatchObject({ status: 403, code: 'agent_source_forbidden' })
+})
+
+it('refuses directory and symlink escapes before making a download request', async () => {
+  const f = fixture()
+  await f.archive('/outside/report.txt')
+  await f.archive('/remote/work/escape.txt')
+  expect(f.request.mock.calls.some(([route]) => route === '/api/files/download')).toBe(false)
+  expect(f.files()).toEqual([])
+  expect(existsSync(join(f.home, 'workspace-files'))).toBe(false)
+})
+
+it('refuses an unassigned source before any upstream request', async () => {
+  const f = fixture(); f.revoke()
+  await f.archive()
+  expect(f.request).not.toHaveBeenCalled()
+  expect(f.files()).toEqual([])
+})
+
+it.each(['source', 'directory', 'quota'] as const)('does not persist bytes if %s authorization changes during a download', async change => {
+  const f = fixture()
+  saveFileAccess(f.home, { mode: 'folders', folders: ['/extra'] })
+  let finish!: (value: ReturnType<typeof downloaded>) => void
+  let started!: () => void
+  const pending = new Promise<ReturnType<typeof downloaded>>(resolve => { finish = resolve })
+  const downloading = new Promise<void>(resolve => { started = resolve })
+  const original = f.request.getMockImplementation()!
+  f.request.mockImplementation((route, options) => {
+    if (route === '/api/files/download') { started(); return pending }
+    return original(route, options)
+  })
+  const archive = f.archive('/extra/report.txt')
+  await downloading
+  if (change === 'source') f.revoke()
+  else if (change === 'directory') saveFileAccess(f.home, { mode: 'folders', folders: [] })
+  else f.store.put('owner', 'file', 'full-quota', { id: 'full-quota', size: 2 * 1024 * 1024 * 1024 })
+  finish(downloaded()); await archive
+  expect(f.files()).toHaveLength(change === 'quota' ? 1 : 0)
+  expect(existsSync(join(f.home, 'workspace-files'))).toBe(false)
+  expect(f.store.list('owner', 'archived-path')).toEqual([])
+})
+
+it.each(['quota', 'message-files'] as const)('enforces the existing %s limit for automatic archives', async limit => {
+  const f = fixture()
+  for (let i = 0; i < (limit === 'quota' ? 1 : 8); i++)
+    f.store.put('owner', 'file', `existing-${i}`, { id: `existing-${i}`, messageId: limit === 'quota' ? 'older-message' : 'message', size: limit === 'quota' ? 2 * 1024 * 1024 * 1024 : 1 })
+  await f.archive()
+  expect(f.files()).toHaveLength(limit === 'quota' ? 1 : 8)
+  expect(existsSync(join(f.home, 'workspace-files'))).toBe(false)
+})
+
+it('recovers a uniquely authorized legacy source from its archive key without downloading again', async () => {
+  const f = fixture(); await f.archive()
+  const file = f.files()[0]!
+  delete file.sourceNodeId; f.store.put('owner', 'file', file.id, file)
+  f.request.mockClear()
+  expect(await f.download(file.id)).toBe('fixture-report')
+  expect(f.files()[0]!.sourceNodeId).toBe('local')
+  expect(f.request).not.toHaveBeenCalled()
+})
+
+it.each(['unknown', 'ambiguous', 'ambiguous-revoked', 'duplicate-record', 'revoked'] as const)('denies a %s legacy archive instead of guessing its source', async reason => {
+  const f = fixture(); await f.archive()
+  const file = f.files()[0]!
+  delete file.sourceNodeId; f.store.put('owner', 'file', file.id, file)
+  if (reason === 'unknown') f.store.remove('owner', 'archived-path', keyFor('local'))
+  if (reason === 'ambiguous' || reason === 'ambiguous-revoked') {
+    f.store.put('owner', 'node', 'other-node', { id: 'other-node' })
+    f.store.put('owner', 'archived-path', keyFor('other-node'), true)
+    if (reason === 'ambiguous-revoked') f.requireSource.mockImplementation((owner, source: { nodeId: string }) => {
+      if (owner !== 'owner' || source.nodeId !== 'local') throw new HttpError(403, 'Fixture source revoked', 'agent_source_forbidden')
+    })
+  }
+  if (reason === 'duplicate-record') f.store.put('owner', 'file', 'duplicate', { ...file, id: 'duplicate' })
+  if (reason === 'revoked') f.revoke()
+  f.request.mockClear()
+  await expect(f.download(file.id)).rejects.toMatchObject({ status: 403, code: 'archived_file_source_unknown' })
+  expect(f.files()[0]!.sourceNodeId).toBeUndefined()
+  expect(f.request).not.toHaveBeenCalled()
+})

tests/server/workspaceRoutes.test.ts

@@ -10,6 +10,7 @@ import { createApplication, type ApplicationRuntime } from '../../src/server/app
 import { loadServerConfig } from '../../src/server/config'
 import { LocalAuthStore, type LocalUser } from '../../src/server/localAuth'
 import { WorkspaceAssets } from '../../src/server/workspaceAssets'
+import { saveFileAccess } from '../../src/server/fileAccess'
 import { WorkspaceTranscriptStore } from '../../src/client/components/workspace/transcriptStore'
 import { SharedComputers, type SharedComputer } from '../../src/server/sharedComputers'
 import type { WorkspaceAgent, WorkspaceConversation } from '../../src/shared/workspace'
@@ -500,6 +501,7 @@ describe('application workspace HTTP contract', () => {
   })
 
   it('archives native-chat uploads through the standard file API with user ownership', async () => {
+    saveFileAccess(home, { mode: 'all', folders: [] })
     const assets = new WorkspaceAssets(runtime.workspace, runtime.workspaceRuntime.nodes, home)
     await assets.archiveText(
       'first',

tests/server/workspaceSync.test.ts

@@ -13,7 +13,7 @@ import type { Server } from 'node:http'
 import type { WorkspaceConversation, WorkspaceMessage } from '../../src/shared/workspace'
 
 let home: string, store: WorkspaceStore, server: Server, version: number
-const auth = { require: () => ({ id: 'owner' }), isUserActive: () => true, pushAuthorizationVersion: () => version } as unknown as LocalAuthStore
+const auth = { require: () => ({ id: 'owner' }), currentFromCookieHeader: () => ({ id: 'owner' }), isUserActive: () => true, pushAuthorizationVersion: () => version } as unknown as LocalAuthStore
 beforeEach(async () => {
   home = mkdtempSync(join(tmpdir(), 'workspace-sync-')); store = new WorkspaceStore(home); version = 1
   const app = new Koa(); app.use(ctx => streamWorkspace(ctx, store, auth))

tests/server/workspaceSyncAuthorization.test.ts

@@ -0,0 +1,72 @@
+// @vitest-environment node
+import { EventEmitter } from 'node:events'
+import { mkdtempSync, rmSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import type Koa from 'koa'
+import { afterEach, beforeEach, expect, it, vi } from 'vitest'
+import { LocalAuthStore } from '../../src/server/localAuth'
+import { WorkspaceStore } from '../../src/server/workspaceStore'
+import { streamWorkspace } from '../../src/server/workspaceSync'
+
+class Response extends EventEmitter {
+  writableEnded = false
+  destroyed = false
+  writableLength = 0
+  writes: string[] = []
+  writeHead() {}
+  flushHeaders() {}
+  write(value: string) { this.writes.push(value); return true }
+  end() { this.writableEnded = true; this.emit('close') }
+}
+function context(cookie = ''): Koa.Context {
+  const headers: Record<string, string | string[]> = {}
+  return { state: {}, response: { headers }, get: (name: string) => name === 'cookie' ? cookie : '',
+    set: (name: string, value: string | string[]) => { headers[name.toLowerCase()] = value } } as unknown as Koa.Context
+}
+const issuedCookie = (ctx: Koa.Context) => (ctx.response.headers['set-cookie'] as string[])[0]!.split(';')[0]!
+let home: string, store: WorkspaceStore, auth: LocalAuthStore, cookie: string, owner: string, res: Response
+beforeEach(() => {
+  vi.useFakeTimers()
+  home = mkdtempSync(join(tmpdir(), 'yaoyao-sse-auth-')); store = new WorkspaceStore(home); auth = new LocalAuthStore(home)
+  const setup = context(); owner = auth.setupAdmin(setup, 'fixture-user', 'fixture-password').id; cookie = issuedCookie(setup)
+  res = new Response()
+  const ctx = context(cookie)
+  // Mirror the authentication middleware's cached request identity.
+  ctx.state.localUser = auth.require(ctx); ctx.query = {}; ctx.res = res as unknown as Koa.Context['res']
+  streamWorkspace(ctx, store, auth)
+})
+afterEach(() => { res.end(); store.close(); rmSync(home, { recursive: true, force: true }); vi.useRealTimers() })
+
+it('closes the cached-user stream and sends no further event after logout', () => {
+  const version = auth.pushAuthorizationVersion(owner)
+  auth.logout(context(cookie))
+  expect(auth.pushAuthorizationVersion(owner)).toBe(version)
+  expect(auth.currentFromCookieHeader(cookie)).toBeUndefined()
+  store.event(owner, 'conversation.changed', { id: 'after-logout' })
+  expect(res.writableEnded).toBe(true)
+  expect(res.writes.join('')).not.toContain('after-logout')
+  expect(store.changes.listenerCount(owner)).toBe(0)
+})
+
+it('closes an idle expired-session stream on its next heartbeat', () => {
+  vi.setSystemTime(Date.now() + 31 * 24 * 60 * 60 * 1000)
+  vi.advanceTimersByTime(15_000)
+  expect(res.writableEnded).toBe(true)
+  expect(store.changes.listenerCount(owner)).toBe(0)
+})
+
+it('keeps a different valid login stream open when another session logs out', () => {
+  const other = context(); auth.login(other, 'fixture-user', 'fixture-password')
+  auth.logout(context(issuedCookie(other)))
+  store.event(owner, 'conversation.changed', { id: 'still-authorized' })
+  expect(res.writableEnded).toBe(false)
+  expect(res.writes.join('')).toContain('still-authorized')
+})
+
+it('still closes on account password revocation', () => {
+  auth.changeCredentials(context(cookie), 'fixture-password', 'fixture-password-changed')
+  store.event(owner, 'conversation.changed', { id: 'after-password-change' })
+  expect(res.writableEnded).toBe(true)
+  expect(res.writes.join('')).not.toContain('after-password-change')
+})