发布 v0.2.28：优化热点读取缓存与上游认证检查

README.md

@@ -79,9 +79,11 @@ ID 和包名。浏览器只提交路径和公开元数据；8800 本地校验 RS
 插件安装接口安装。默认源为
 `https://git.samien.cn/samien/hermes-yaoyao.git#hermes-plugins/yaoyao`。
 
-当前发布版本：**Git `v0.2.27` / 夭夭 Web `0.2.27` / Hermes Dashboard 插件 `1.7.3`**。版本组合由仓库根目录的 `release.json` 唯一声明，并由 `npm run release:verify` 校验。
+当前发布版本：**Git `v0.2.28` / 夭夭 Web `0.2.28` / Hermes Dashboard 插件 `1.7.3`**。版本组合由仓库根目录的 `release.json` 唯一声明，并由 `npm run release:verify` 校验。
 
-本版将普通聊天和团队事件统一为 HTTP+SSE，增加服务端连接管理、近期事件补收、提交回执去重与权限隔离。8800 到 9119 的上游接口及 Python 插件保持不变。详见 [HTTP+SSE 接口与升级边界](docs/http-sse-realtime.md)。
+本版增加有界热点读取缓存、并发读取合并、30 秒上游状态/身份检查复用，以及写操作和实时事件驱动的缓存失效。客户端授权与 9119 实际请求鉴权仍然保留。详见 [读取缓存与一致性边界](docs/read-cache.md)。
+
+普通聊天和团队事件继续使用 [HTTP+SSE 接口](docs/http-sse-realtime.md)，不兼容旧客户端 WebSocket。8800 到 9119 的上游接口及 Python 插件保持不变。
 
 需要由自动化 Agent 部署或升级时，请直接使用 [Agent 安装手册](docs/agent-install.md)。其中包含固定版本校验、备份、同步、单一 Dashboard 重载与验证步骤。
 

docs/agent-install.md

@@ -6,14 +6,16 @@
 
 | 项目 | 版本 | 校验位置 |
 | --- | --- | --- |
-| Git 发布标签 | `v0.2.27` | `release.json` 的 `gitTag` 与 `git describe --tags --exact-match HEAD` |
-| 夭夭 Web | `0.2.27` | `release.json` 与 `package.json` 的 `version` |
+| Git 发布标签 | `v0.2.28` | `release.json` 的 `gitTag` 与 `git describe --tags --exact-match HEAD` |
+| 夭夭 Web | `0.2.28` | `release.json` 与 `package.json` 的 `version` |
 | Hermes Dashboard 插件 | `1.7.3` | `hermes-plugins/yaoyao/dashboard/manifest.json` 的 `version` |
 
 以下步骤以仓库根目录为工作目录。不要使用浮动分支替代发布标签。
 
 本版客户端只支持 HTTP+SSE，旧 WebSocket、票据和租约入口返回 410。升级服务端时必须同步更新 iOS 客户端；Web 静态资源随本版发布。上游 9119 与 Python 插件未迁移，插件版本仍为 1.7.3。
 
+v0.2.28 增加服务端热点读取缓存与 30 秒上游登录状态检查复用；实际请求仍携带凭据，客户端权限仍逐次校验。已有 HTTP+SSE 版 iOS 无需因本次缓存优化重新打包。缓存边界见 [读取缓存说明](read-cache.md)。
+
 ## 网络默认值与局域网策略
 
 通过 `service install` 安装的受管服务默认监听可信局域网，两个端口一起开放：
@@ -80,12 +82,12 @@ curl --fail --silent --show-error http://127.0.0.1:9119/api/auth/providers
 ## 1. 获取并校验指定发布版本
 
 ```bash
-RELEASE_VERSION=v0.2.27
+RELEASE_VERSION=v0.2.28
 git fetch --tags
 git checkout "$RELEASE_VERSION"
 
 test "$(git describe --tags --exact-match HEAD)" = "$RELEASE_VERSION"
-test "$(node -p \"require('./package.json').version\")" = "0.2.27"
+test "$(node -p \"require('./package.json').version\")" = "0.2.28"
 test "$(node -e \"console.log(require('./hermes-plugins/yaoyao/dashboard/manifest.json').version)\")" = "1.7.3"
 npm run release:verify
 ```
@@ -249,7 +251,7 @@ Git、发布目录、命令输出或日志。配置后检查系统管理中的 A
 
 ## 7. 通过受管服务进行配套升级
 
-首次按第 6 步安装 Web `0.2.27` 后，可在左下角 Agent 菜单打开“系统更新”。
+首次按第 6 步安装 Web `0.2.28` 后，可在左下角 Agent 菜单打开“系统更新”。
 检查更新只读取固定的 `HERMES_YAOYAO_RELEASE_SOURCE`，并锁定最新 `vX.Y.Z`
 标签解析到的 Git 提交；执行升级前仍会检查插件持久目录是否安全。
 

docs/read-cache.md

@@ -0,0 +1,34 @@
+# 8800 热点读取缓存
+
+本功能仅优化数据读取与上游认证检查，不改变模型生成速度，不迁移 9119 数据库，也不替代 iOS 本地缓存。
+
+## 上游认证租期
+
+`UpstreamServiceSession` 将成功的状态/身份检查复用 30 秒，并合并同一时间的并发检查。每个客户端请求仍执行 8800 的用户、角色或配对设备授权。
+
+上游返回 401 时仅对该请求重新认证并重试一次；并发 401 共用登录任务。不会因为网络超时或 5xx 自动重放写请求。服务账号配置变化会立即使租期及读取缓存失效；旧验证或登录响应不能覆盖新凭据。鉴权端点本身不进入读取缓存。
+
+## 读取策略
+
+| 数据 | 新鲜期 | 后台刷新宽限 |
+| --- | --- | --- |
+| Profile/身份列表 | 5 秒 | 15 秒 |
+| 模型选项 | 30 秒 | 60 秒 |
+| 会话列表、团队/话题列表 | 2 秒 | 无 |
+| 会话详情、最近消息、团队详情 | 1 秒 | 无 |
+| 已加载的较早消息页 | 10 秒 | 无 |
+
+仅缓存白名单 GET 的成功 JSON。容量上限为 64 MiB / 512 项，每项不超过 2 MiB，使用 LRU 淘汰；相同读取会合并。缓存键包含已验证的 8800 用户/授权版本或配对设备、上游 Cookie 的摘要、Profile、分页参数和请求头。缓存键不记录原始凭据。
+
+后台推送确认、命令回执、鉴权、搜索、未读确认、团队游标能力端点、增量 `afterSeq`、Range/条件读取以及错误响应不缓存。上游显式 `no-store` 或 `Set-Cookie` 响应不缓存。
+
+这是服务端应用数据缓存，而不是浏览器 HTTP 缓存。浏览器仍可保持 `cache: no-store`。客户端需要强制最新数据时可发送 `X-Yaoyao-Cache: bypass`；内部调用可指定 `cache: 'reload'`。
+
+## 一致性边界
+
+- HTTP 写入前后失效相应资源；聊天命令和实时事件在投递给客户端之前失效对应会话缓存。
+- 团队事件失效房间与列表缓存；获取新的团队 journal cursor 后清除较早的团队快照，保留 snapshot-after-anchor 顺序。
+- 断线重置会清除相关缓存。刷新期间发生失效时，较早响应不会重新填回缓存，后续请求不会加入过时的在途读取。
+- 缓存只负责读取；发送、停止、审批必须等待真实上游确认。没有事件通知的外部修改仍受上述短 TTL 约束，不能声称跨所有外部写入的强一致。
+
+`UpstreamClient.readCache.stats()` 提供进程内命中、未命中、合并、淘汰、内存及在途数量，用于测试和诊断，不公开缓存内容。

package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "hermes-yaoyao",
-  "version": "0.2.27",
+  "version": "0.2.28",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "hermes-yaoyao",
-      "version": "0.2.27",
+      "version": "0.2.28",
       "license": "BSL-1.1",
       "dependencies": {
         "@koa/bodyparser": "^5.0.0",

package.json

@@ -1,6 +1,6 @@
 {
   "name": "hermes-yaoyao",
-  "version": "0.2.27",
+  "version": "0.2.28",
   "private": true,
   "license": "BSL-1.1",
   "description": "夭夭 Web：直连 Hermes 9119 的聊天、团队、文件库与产物工作台",

release.json

@@ -1,7 +1,7 @@
 {
   "schemaVersion": 1,
-  "releaseVersion": "0.2.27",
-  "webVersion": "0.2.27",
+  "releaseVersion": "0.2.28",
+  "webVersion": "0.2.28",
   "pluginVersion": "1.7.3",
-  "gitTag": "v0.2.27"
+  "gitTag": "v0.2.28"
 }

src/server/app.ts

@@ -292,7 +292,10 @@ export function createApplication(options: ApplicationOptions = {}): Application
         }
       }
     }
-    await next()
+    const user = ctx.state.localUser as { id: string } | undefined
+    if (user) {
+      await upstream.withReadScope(`user:${user.id}:${auth.pushAuthorizationVersion(user.id)}`, ctx.get('x-yaoyao-cache') === 'bypass', next)
+    } else await next()
   })
 
   const router = createApiRouter({
@@ -344,6 +347,7 @@ export function createApplication(options: ApplicationOptions = {}): Application
     chatPushJobs,
     groupPushEvents,
     close: () => {
+      upstream.readCache.close()
       realtime.close()
       chatPushJobs.stop()
       groupPushEvents.stop()

src/server/localAuth.ts

@@ -449,43 +449,98 @@ export class LocalAuthStore {
 export class UpstreamServiceSession {
   readonly #jar = new CookieJar('')
   #loginTask: Promise<void> | undefined
+  #ensureTask: Promise<void> | undefined
+  #validatedUntil = 0
+  #credentialFingerprint: string | undefined
+  #generation = 0
+  readonly #now: () => number
+  readonly #validationTTL: number
 
   constructor(
     readonly client: UpstreamClient,
     readonly credentials: () => UpstreamCredentials | undefined,
-  ) {}
+    options: { now?: () => number; validationTTL?: number } = {},
+  ) {
+    this.#now = options.now ?? Date.now
+    this.#validationTTL = options.validationTTL ?? 30_000
+    client.setReauthenticationHandler(this.#jar, async () => {
+      this.#syncCredentials()
+      this.#validatedUntil = 0
+      await this.#login(this.#generation)
+    })
+  }
 
   get jar(): CookieJar { return this.#jar }
 
   async ensure(): Promise<void> {
-    const status = await this.client.request('/api/status', this.#jar)
+    this.#syncCredentials()
+    if (this.#validatedUntil > this.#now()) return
+    if (this.#ensureTask) return this.#ensureTask
+    const generation = this.#generation
+    const task = this.#validate(generation).then(() => {
+      this.#checkGeneration(generation)
+      this.#validatedUntil = this.#now() + this.#validationTTL
+    }).finally(() => { if (this.#ensureTask === task) this.#ensureTask = undefined })
+    this.#ensureTask = task
+    return task
+  }
+
+  invalidateAuthentication(): void { this.#validatedUntil = 0; this.client.invalidateReads() }
+
+  #syncCredentials(): void {
+    const fingerprint = createHash('sha256').update(JSON.stringify(this.credentials() ?? null)).digest('hex')
+    if (this.#credentialFingerprint !== undefined && fingerprint !== this.#credentialFingerprint) {
+      this.#generation++
+      this.#validatedUntil = 0; this.#ensureTask = undefined; this.#loginTask = undefined
+      this.#jar.replace(new CookieJar())
+      this.client.invalidateReads()
+    }
+    this.#credentialFingerprint = fingerprint
+  }
+
+  #checkGeneration(generation: number): void {
+    this.#syncCredentials()
+    if (generation !== this.#generation) throw new HttpError(409, '上游凭据已变化，请重试读取', 'upstream_credentials_changed')
+  }
+
+  async #validate(generation: number): Promise<void> {
+    // Probe into a temporary jar. A delayed old probe must not overwrite freshly
+    // rotated service credentials/cookies in the live jar.
+    const probe = new CookieJar(this.#jar.header)
+    const status = await this.client.request('/api/status', probe)
     if (status.status < 200 || status.status >= 300) throw new HttpError(502, '9119 状态不可用', 'upstream_unavailable')
     const parsed = JSON.parse(status.body.toString('utf8')) as Record<string, unknown>
-    if (parsed.auth_required !== true && parsed.authRequired !== true) return
-    const me = await this.client.request('/api/auth/me', this.#jar)
-    if (me.status >= 200 && me.status < 300) return
-    if (!this.#loginTask) this.#loginTask = this.#login().finally(() => { this.#loginTask = undefined })
-    await this.#loginTask
+    this.#checkGeneration(generation)
+    if (parsed.auth_required !== true && parsed.authRequired !== true) { this.#jar.replace(probe); return }
+    const me = await this.client.request('/api/auth/me', probe)
+    this.#checkGeneration(generation)
+    if (me.status >= 200 && me.status < 300) { this.#jar.replace(probe); return }
+    if (me.status !== 401 && me.status !== 403) throw new HttpError(502, '无法验证 9119 服务会话', 'upstream_auth_unavailable')
+    await this.#login(generation)
   }
 
   async request(path: string, options: UpstreamRequestOptions = {}): Promise<UpstreamResponse> {
     await this.ensure()
-    let response = await this.client.request(path, this.#jar, options)
-    if (response.status === 401) {
-      await this.#login()
-      response = await this.client.request(path, this.#jar, options)
-    }
-    return response
+    return this.client.request(path, this.#jar, options)
   }
 
   async verify(credentials: UpstreamCredentials): Promise<void> {
     await this.#authenticate(credentials, new CookieJar(''))
   }
 
-  async #login(): Promise<void> {
+  async #login(generation: number): Promise<void> {
+    if (this.#loginTask) return this.#loginTask
     const credentials = this.credentials()
     if (!credentials) throw new HttpError(503, '尚未配置 9119 服务账号', 'upstream_credentials_required')
-    await this.#authenticate(credentials, this.#jar)
+    const jar = new CookieJar()
+    const task = this.#authenticate({ ...credentials }, jar).then(() => {
+      this.#checkGeneration(generation)
+      this.#jar.replace(jar)
+      this.client.invalidateReads()
+      this.#validatedUntil = this.#now() + this.#validationTTL
+    }).finally(() => { if (this.#loginTask === task) this.#loginTask = undefined })
+    this.#loginTask = task
+    return task
   }
 
   async #authenticate(credentials: UpstreamCredentials, jar: CookieJar): Promise<void> {

src/server/realtimeApi.ts

@@ -19,7 +19,10 @@ export class RealtimeAPI {
   constructor(readonly config: ServerConfig, readonly auth: LocalAuthStore, readonly csrf: CsrfProtection,
     readonly pairings: NodePairingStore, readonly upstream: UpstreamClient, readonly session: UpstreamServiceSession,
     readonly push?: { coordinator: PushEventCoordinator; resolver: ChatNotificationResolver }) {
-    this.broker = new RealtimeBroker(config.home)
+    this.broker = new RealtimeBroker(config.home, Date.now, change => {
+      upstream.observeRealtime(change)
+      if (change.kind === 'reset') session.invalidateAuthentication()
+    })
   }
   private principal(ctx: Koa.Context, device?: string): RealtimePrincipal {
     let key: string

src/server/realtimeBroker.ts

@@ -5,6 +5,7 @@ import { checkedChatFrame, CHAT_MAX_PAYLOAD, GroupFrameValidator } from './realt
 import { RealtimeReceipts, type CommandReceipt } from './realtimeReceipts.js'
 
 type Frame = Record<string, any>
+export interface RealtimeActivity { kind: 'command' | 'event' | 'group' | 'reset'; name: string; sessionId?: string; roomId?: string }
 export interface RealtimePrincipal {
   key: string
   upstreamKey: string
@@ -45,7 +46,7 @@ export class RealtimeBroker {
   private receipts: RealtimeReceipts
   private timer: ReturnType<typeof setInterval>
   private closed = false
-  constructor(home: string, readonly now = Date.now) {
+  constructor(home: string, readonly now = Date.now, readonly onActivity: (activity: RealtimeActivity) => void = () => {}) {
     this.receipts = new RealtimeReceipts(home)
     this.timer = setInterval(() => this.sweep(), 5_000)
     this.timer.unref()
@@ -197,7 +198,10 @@ export class RealtimeBroker {
       if (!c.principal.valid()) throw new HttpError(401, 'Authentication expired', 'authentication_required')
       const previouslyActive = route?.active ?? false
       if (route && (method === 'prompt.submit' || method === 'session.steer')) route.active = true
+      const activity: RealtimeActivity = { kind: 'command', name: method, sessionId: route?.stored }
+      this.onActivity(activity)
       const response = await this.rpc(u, method, p, f => c.principal.observeEvent?.(JSON.stringify(f)), f => c.principal.observeCommand?.(JSON.stringify(f)))
+        .finally(() => this.onActivity(activity))
       if (response.error && route && (method === 'prompt.submit' || method === 'session.steer')) route.active = previouslyActive
       if (!response.error && route && (method === 'prompt.submit' || method === 'session.steer')) route.active = true
       if ((opening || method === 'session.branch') && response.result) {
@@ -320,6 +324,7 @@ export class RealtimeBroker {
     if (!p || typeof p.type !== 'string') return
     const sid = String(p.session_id ?? '')
     const r = u.byRuntime.get(sid)
+    this.onActivity({ kind: 'event', name: p.type, sessionId: r?.stored })
     if (sid && !r) {
       const size = Buffer.byteLength(JSON.stringify(f))
       if (u.earlyBytes + size > 1024 * 1024) { u.early.clear(); u.earlyBytes = 0; this.reset(u, 'unmapped_event_overflow'); return }
@@ -371,6 +376,7 @@ export class RealtimeBroker {
     }
   }
   private reset(u: Upstream, reason: string): void {
+    this.onActivity({ kind: 'reset', name: reason })
     for (const c of this.channels.values()) if (c.kind === 'chat' && c.principal.upstreamKey === u.principal.upstreamKey) this.emit(c, 'reset', { reason })
   }
   private async connectGroup(c: RealtimeChannel): Promise<void> {
@@ -384,13 +390,14 @@ export class RealtimeBroker {
         try {
           if (!validator.accept(data, binary)) return
           const frame = JSON.parse(data.toString()) as Frame
+          this.onActivity({ kind: 'group', name: String(frame.event ?? frame.type), roomId: frame.roomId })
           if (frame.type === 'group.ready') { clearTimeout(timer); resolve() }
           if (frame.type === 'group.event') c.group!.cursor = frame.cursor
           this.emit(c, 'frame', frame)
         } catch { this.emit(c, 'reset', { reason: 'group_gap' }); ws.close() }
       })
       ws.on('error', () => { clearTimeout(timer); reject(new Error('Group connection failed')) })
-      ws.on('close', () => { clearTimeout(timer); reject(new Error('Group disconnected')); if (this.channels.has(c.id)) this.emit(c, 'reset', { reason: 'group_disconnected' }) })
+      ws.on('close', () => { clearTimeout(timer); reject(new Error('Group disconnected')); this.onActivity({ kind: 'group', name: 'group_disconnected' }); if (this.channels.has(c.id)) this.emit(c, 'reset', { reason: 'group_disconnected' }) })
     })
   }
   private emit(c: RealtimeChannel, event: StreamEntry['event'], value: Frame): void {

src/server/routes.ts

@@ -1589,7 +1589,7 @@ export function createApiRouter(dependencies: RouteDependencies): Router {
         'x-mime-type': mimeType,
       }
     }
-    const response = await dependencies.upstream.request(path, jar, {
+    const response = await dependencies.upstream.withReadScope(`device:${ctx.params.deviceID}`, ctx.get('x-yaoyao-cache') === 'bypass', () => dependencies.upstream.request(path, jar, {
       method: ctx.method,
       search: new URLSearchParams(ctx.querystring),
       body: ['GET', 'HEAD'].includes(ctx.method) || isWorkerAttachment
@@ -1598,7 +1598,7 @@ export function createApiRouter(dependencies: RouteDependencies): Router {
       headers: requestHeaders,
       clientAddress: path === '/api/auth/ws-ticket'
         ? undefined : ctx.req.socket.remoteAddress,
-    })
+    }))
     dependencies.pairings.updateCookies(ctx.params.deviceID, jar.header)
     sendUpstreamResponse(ctx, response, jar)
   })

src/server/upstream.ts

@@ -1,4 +1,6 @@
 import type Koa from 'koa'
+import { createHash } from 'node:crypto'
+import { readCacheContext, readPolicy, mutationTags, UpstreamReadCache } from './upstreamReadCache.js'
 import { parse } from 'cookie'
 import { HttpError } from './errors.js'
 import { appendSetCookies, CSRF_COOKIE } from './security.js'
@@ -70,6 +72,11 @@ export class CookieJar {
     return [...this.#values.entries()].map(([name, value]) => `${name}=${value}`).join('; ')
   }
 
+  replace(other: CookieJar): void {
+    this.#values.clear(); this.#browserCookies.clear()
+    for (const [name, value] of other.#values) this.#values.set(name, value)
+  }
+
   get browserCookies(): string[] {
     return [...this.#browserCookies.values()]
   }
@@ -88,6 +95,7 @@ export class UpstreamHttpError extends HttpError {
 }
 
 export interface UpstreamRequestOptions {
+  cache?: 'reload'
   method?: string
   search?: URLSearchParams
   body?: unknown
@@ -98,13 +106,84 @@ export interface UpstreamRequestOptions {
 }
 
 export class UpstreamClient {
+  readonly readCache: UpstreamReadCache
+  private reauthentication = new WeakMap<CookieJar, () => Promise<void>>()
   constructor(
     readonly baseURL: URL,
     readonly fetchImpl: typeof fetch = fetch,
     readonly publicSecure = false,
-  ) {}
+    now: () => number = Date.now,
+  ) { this.readCache = new UpstreamReadCache(now) }
+
+  withReadScope<T>(scope: string, fresh: boolean, run: () => T): T {
+    return readCacheContext.run({ scope, fresh }, run)
+  }
+  invalidateReads(tags?: readonly string[]): void { this.readCache.invalidate(tags) }
+  observeRealtime(change: { kind: 'command' | 'event' | 'group' | 'reset'; name: string; sessionId?: string; roomId?: string }): void {
+    if (change.kind === 'reset') { this.invalidateReads(['sessions', 'groups', 'profiles', 'models']); return }
+    if (change.kind === 'group') {
+      if (change.name === 'group.heartbeat' || change.name === 'group.ready') return
+      this.invalidateReads(change.roomId ? ['group-list', `group:${change.roomId}`] : ['groups'])
+      return
+    }
+    if (/^(profiles\.(configure|set_asset|changed)|pet\.changed|models\.changed|model\.)/.test(change.name)) {
+      this.invalidateReads(['profiles', 'models']); return
+    }
+    if (change.name === 'sessions.changed') { this.invalidateReads(['sessions']); return }
+    if (change.kind === 'command' && !/^(prompt\.submit|session\.(create|resume|branch|close|steer|interrupt)|config\.set|approval\.respond|clarify\.respond)$/.test(change.name)) return
+    if (change.kind === 'event' && !/^(message\.|session\.|config\.|compression\.|run\.|error$)/.test(change.name)) return
+    const tags = change.sessionId ? [`session:${change.sessionId}`] : ['sessions']
+    if (change.kind === 'command' || !/\.delta$/.test(change.name)) tags.push('session-list')
+    if (change.name === 'config.set') tags.push('models', 'profiles')
+    this.invalidateReads(tags)
+  }
+  setReauthenticationHandler(jar: CookieJar, renew: () => Promise<void>): void { this.reauthentication.set(jar, renew) }
+
+  async request(path: string, jar: CookieJar, options: UpstreamRequestOptions = {}): Promise<UpstreamResponse> {
+    const method = options.method ?? 'GET'
+    const tags = ['GET', 'HEAD'].includes(method) ? undefined : mutationTags(path)
+    if (tags) this.invalidateReads(tags)
+    const context = readCacheContext.getStore()
+    const forcedTags = method === 'GET' && (context?.fresh || options.cache === 'reload') ? readPolicy(path, options.search)?.tags : undefined
+    if (forcedTags) this.invalidateReads(forcedTags)
+    const load = async () => {
+      const sentCookies = jar.header
+      let response = await this.fetchResponse(path, jar, options)
+      if (response.status === 401 || response.status === 403) this.invalidateReads()
+      const renew = this.reauthentication.get(jar)
+      if (response.status === 401 && renew && !path.startsWith('/api/auth/') && !path.startsWith('/auth/')) {
+        // A parallel request may already have rotated the service cookies. Never replay
+        // a whole route handler (it may have performed several writes); repair this one 401 only.
+        if (jar.header === sentCookies) await renew()
+        response = await this.fetchResponse(path, jar, options)
+      }
+      return response
+    }
+    try {
+      const headers = new Headers(options.headers)
+      const policy = method === 'GET' && context && !context.fresh && options.cache !== 'reload'
+        && !['range', 'if-none-match', 'if-modified-since', 'if-match', 'if-unmodified-since'].some(name => headers.has(name))
+        ? readPolicy(path, options.search) : undefined
+      let response: UpstreamResponse
+      if (policy && context) {
+        const search = new URLSearchParams(options.search); search.sort()
+        const key = createHash('sha256').update(JSON.stringify([
+          context.scope, this.baseURL.href, path, search.toString(), jar.header ?? '',
+          [...headers.entries()].sort(([a], [b]) => a.localeCompare(b)), options.clientAddress ?? '', options.maxResponseBytes ?? null,
+        ])).digest('hex')
+        response = await this.readCache.read(key, policy, load)
+      } else response = await load()
+      // Snapshot-after-anchor ordering: no room snapshot fetched before this fresh
+      // journal cursor may be reused to bootstrap a stream starting at that cursor.
+      if (path === '/api/plugins/yaoyao/v1/capabilities' && response.status === 200) this.invalidateReads(['groups'])
+      return response
+    } finally {
+      if (tags) this.invalidateReads(tags)
+      if (forcedTags) this.invalidateReads(forcedTags)
+    }
+  }
 
-  async request(
+  private async fetchResponse(
     path: string,
     jar: CookieJar,
     options: UpstreamRequestOptions = {},

src/server/upstreamReadCache.ts

@@ -0,0 +1,135 @@
+import { AsyncLocalStorage } from 'node:async_hooks'
+import { HttpError } from './errors.js'
+import type { UpstreamResponse } from './upstream.js'
+
+export interface ReadCacheContext { scope: string; fresh?: boolean }
+export const readCacheContext = new AsyncLocalStorage<ReadCacheContext>()
+export interface ReadPolicy { tags: string[]; freshMs: number; staleMs: number }
+interface Entry { value: UpstreamResponse; policy: ReadPolicy; savedAt: number; bytes: number }
+interface Flight { promise: Promise<UpstreamResponse>; tags: string[]; invalidated: boolean }
+
+/** Application-data cache. Only authenticated HTTP request scopes opt in; background
+ * recovery, authentication and command acknowledgement reads remain authoritative. */
+export class UpstreamReadCache {
+  private entries = new Map<string, Entry>()
+  private flights = new Map<string, Flight>()
+  private bytes = 0
+  private loading = 0
+  private closed = false
+  private counters = { hits: 0, misses: 0, staleHits: 0, coalesced: 0, evictions: 0 }
+  constructor(readonly now = Date.now, readonly maxBytes = 64 * 1024 * 1024,
+    readonly maxEntries = 512, readonly maxEntryBytes = 2 * 1024 * 1024) {}
+
+  stats() { return { ...this.counters, entries: this.entries.size, bytes: this.bytes, inFlight: this.loading } }
+
+  async read(key: string, policy: ReadPolicy, load: () => Promise<UpstreamResponse>): Promise<UpstreamResponse> {
+    const entry = this.entries.get(key)
+    if (entry) {
+      const age = this.now() - entry.savedAt
+      if (age < policy.freshMs) {
+        this.counters.hits++; this.touch(key, entry)
+        return copy(entry.value)
+      }
+      if (age < policy.freshMs + policy.staleMs) {
+        this.counters.staleHits++; this.touch(key, entry)
+        if (!this.flights.has(key) && this.loading < 128) void this.refresh(key, policy, load).catch(() => {})
+        return copy(entry.value)
+      }
+      this.remove(key)
+    }
+    const flight = this.flights.get(key)
+    if (flight) { this.counters.coalesced++; return copy(await flight.promise) }
+    this.counters.misses++
+    return copy(await this.refresh(key, policy, load))
+  }
+
+  invalidate(tags?: readonly string[]): void {
+    const matches = (entryTags: string[]) => !tags || entryTags.some(tag => tags.includes(tag))
+    for (const [key, entry] of this.entries) if (matches(entry.policy.tags)) this.remove(key)
+    for (const [key, flight] of this.flights) if (matches(flight.tags)) {
+      flight.invalidated = true
+      this.flights.delete(key) // A newer request must not join an obsolete snapshot.
+    }
+  }
+
+  close(): void { this.closed = true; this.invalidate() }
+
+  private refresh(key: string, policy: ReadPolicy, load: () => Promise<UpstreamResponse>): Promise<UpstreamResponse> {
+    if (this.loading >= 128) return Promise.reject(new HttpError(429, 'Too many pending reads', 'read_capacity'))
+    this.loading++
+    const flight: Flight = { tags: policy.tags, invalidated: false, promise: Promise.resolve(undefined as never) }
+    // Defer load until the flight is registered, so synchronous invalidations cannot be lost.
+    flight.promise = Promise.resolve().then(load).then(value => {
+      if (!flight.invalidated && !this.closed && cacheable(value) && value.body.byteLength <= this.maxEntryBytes) {
+        this.remove(key)
+        const bytes = value.body.byteLength + key.length * 2 + 512
+        this.entries.set(key, { value: copy(value), policy, savedAt: this.now(), bytes })
+        this.bytes += bytes
+        while (this.entries.size > this.maxEntries || this.bytes > this.maxBytes) {
+          this.remove(this.entries.keys().next().value!); this.counters.evictions++
+        }
+      }
+      return value
+    }).finally(() => {
+      this.loading--
+      if (this.flights.get(key) === flight) this.flights.delete(key)
+    })
+    this.flights.set(key, flight)
+    return flight.promise
+  }
+
+  private touch(key: string, entry: Entry): void { this.entries.delete(key); this.entries.set(key, entry) }
+  private remove(key: string): void {
+    const entry = this.entries.get(key)
+    if (entry) { this.bytes -= entry.bytes; this.entries.delete(key) }
+  }
+}
+
+function copy(value: UpstreamResponse): UpstreamResponse {
+  return { status: value.status, headers: new Headers(value.headers), body: Buffer.from(value.body) }
+}
+function cacheable(value: UpstreamResponse): boolean {
+  return value.status === 200
+    && (value.headers.get('content-type') ?? '').includes('application/json')
+    && !/\bno-store\b/i.test(value.headers.get('cache-control') ?? '')
+    && value.headers.get('vary') !== '*'
+    && !value.headers.has('set-cookie')
+}
+
+export function readPolicy(path: string, search?: URLSearchParams): ReadPolicy | undefined {
+  if (path === '/api/profiles' || path === '/api/plugins/yaoyao/profiles') {
+    return { tags: ['profiles'], freshMs: 5_000, staleMs: 15_000 }
+  }
+  if (path === '/api/model/options') return { tags: ['models'], freshMs: 30_000, staleMs: 60_000 }
+  if (path === '/api/sessions' || path === '/api/profiles/sessions') {
+    return { tags: ['sessions', 'session-list'], freshMs: 2_000, staleMs: 0 }
+  }
+  const session = /^\/api\/sessions\/([^/]+)(?:\/messages)?$/.exec(path)
+  if (session && !['search', 'unread', 'prune', 'export'].includes(session[1]!)) {
+    const oldPage = Number(search?.get('offset') ?? 0) > 0
+    return { tags: ['sessions', `session:${decodeURIComponent(session[1]!)}`], freshMs: oldPage ? 10_000 : 1_000, staleMs: 0 }
+  }
+  const base = '/api/plugins/yaoyao/v1'
+  if ([`${base}/rooms`, `${base}/topics`, `${base}/topics/pinned`].includes(path)) {
+    return { tags: ['groups', 'group-list'], freshMs: 2_000, staleMs: 0 }
+  }
+  const room = /^\/api\/plugins\/yaoyao\/v1\/rooms\/([^/]+)(?:\/(messages|topics))?$/.exec(path)
+  if (room) {
+    // Incremental reconciliation reads must never stop at a cached old tail.
+    if (search?.has('afterSeq')) return undefined
+    return { tags: ['groups', `group:${room[1]}`], freshMs: search?.has('beforeSeq') ? 10_000 : 1_000, staleMs: 0 }
+  }
+  return undefined
+}
+
+export function mutationTags(path: string): string[] | undefined {
+  if (path.startsWith('/auth/') || path.startsWith('/api/auth/')) return undefined
+  const session = /^\/api\/sessions\/([^/]+)/.exec(path)
+  if (session) return ['session-list', `session:${decodeURIComponent(session[1]!)}`]
+  if (path.startsWith('/api/sessions')) return ['sessions']
+  const room = /^\/api\/plugins\/yaoyao\/v1\/rooms\/([^/]+)/.exec(path)
+  if (room) return ['group-list', `group:${room[1]}`]
+  if (path.startsWith('/api/plugins/yaoyao/v1/')) return ['groups']
+  if (/^\/api\/(profiles|model|config|env|providers|dashboard\/agent-plugins)/.test(path)) return ['profiles', 'models', 'sessions', 'groups']
+  return undefined
+}

tests/server/app.test.ts

@@ -357,8 +357,8 @@ describe('8800 BFF', () => {
       .expect(200)
     expect(records.map((entry) => entry.path)).toEqual([
       '/api/status', '/api/auth/me', '/api/status',
-      '/api/status', '/api/auth/me', '/api/profiles',
-      '/api/status', '/api/auth/me', '/api/plugins/yaoyao/profiles',
+      '/api/profiles',
+      '/api/plugins/yaoyao/profiles',
     ])
     expect(response.body).toMatchObject({
       authRequired: true,

tests/server/readCacheRoutes.test.ts

@@ -0,0 +1,56 @@
+import { mkdtempSync } from 'node:fs'
+import { tmpdir } from 'node:os'
+import { join } from 'node:path'
+import request from 'supertest'
+import { afterEach, describe, expect, it } from 'vitest'
+import { createApplication, type ApplicationRuntime } from '../../src/server/app.js'
+import type { ServerConfig } from '../../src/server/config.js'
+
+const runtimes: ApplicationRuntime[] = []
+afterEach(() => { for (const runtime of runtimes.splice(0)) runtime.close() })
+
+describe('authenticated read cache routes', () => {
+  it('hits cached reads, honors fresh reads and never bypasses logout authentication', async () => {
+    const home = mkdtempSync(join(tmpdir(), 'yaoyao-read-cache-routes-'))
+    const config: ServerConfig = { host: '127.0.0.1', port: 8800, upstream: new URL('http://127.0.0.1:19119'),
+      allowedHosts: new Set(), home, mediaRoot: home, attachmentsRoot: home, imagesRoot: home, mediaOwner: 'test',
+      allowInsecureLan: false, insecureLan: false, production: false }
+    let sessionReads = 0, statusReads = 0, currentTitle = 'original'
+    const runtime = createApplication({ config, fetchImpl: (async (input, init) => {
+      const path = new URL(String(input)).pathname
+      if (path === '/api/status') { statusReads++; return Response.json({ auth_required: false }) }
+      if (path === '/api/profiles' || path === '/api/plugins/yaoyao/profiles') return Response.json({ profiles: [{ name: 'default', is_default: true }] })
+      if (path === '/api/sessions' && (init?.method ?? 'GET') === 'GET') { sessionReads++; return Response.json({ sessions: [{ id: 'one', title: currentTitle }] }) }
+      if (path === '/api/sessions/one' && init?.method === 'PATCH') { currentTitle = 'renamed'; return Response.json({ ok: true }) }
+      return Response.json({ ok: true })
+    }) as typeof fetch })
+    runtimes.push(runtime)
+    const agent = request.agent(runtime.app.callback()), origin = 'http://127.0.0.1:8800'
+    const start = await agent.get('/api/app/bootstrap').set('Host', '127.0.0.1:8800').expect(200)
+    const login = await agent.post('/api/app/login').set('Host', '127.0.0.1:8800').set('Origin', origin)
+      .set('X-CSRF-Token', start.body.csrfToken).send({ username: 'admin', password: 'admin' }).expect(200)
+    await agent.put('/api/app/account/credentials').set('Host', '127.0.0.1:8800').set('Origin', origin)
+      .set('X-CSRF-Token', login.body.csrfToken).send({ currentPassword: 'admin', newPassword: 'fixture-password', username: 'owner' }).expect(200)
+    const statusBefore = statusReads
+    const first = await agent.get('/api/sessions').set('Host', '127.0.0.1:8800').expect(200)
+    const statusAfterColdRead = statusReads
+    const second = await agent.get('/api/sessions').set('Host', '127.0.0.1:8800').expect(200)
+    expect(second.body).toEqual(first.body)
+    expect(sessionReads).toBe(1)
+    expect(statusAfterColdRead - statusBefore).toBeLessThanOrEqual(1)
+    expect(statusReads).toBe(statusAfterColdRead)
+    currentTitle = 'external-change'
+    const fresh = await agent.get('/api/sessions').set('Host', '127.0.0.1:8800').set('X-Yaoyao-Cache', 'bypass').expect(200)
+    expect(fresh.body.sessions[0].title).toBe('external-change')
+    const after = await agent.get('/api/sessions').set('Host', '127.0.0.1:8800').expect(200)
+    expect(after.body.sessions[0].title).toBe('external-change')
+    expect(sessionReads).toBe(3)
+    await agent.patch('/api/sessions/one').set('Host', '127.0.0.1:8800').send({ title: 'renamed' }).expect(200)
+    const renamed = await agent.get('/api/sessions').set('Host', '127.0.0.1:8800').expect(200)
+    expect(renamed.body.sessions[0].title).toBe('renamed')
+    expect(sessionReads).toBe(4)
+    await agent.post('/auth/logout').set('Host', '127.0.0.1:8800').send({}).expect(200)
+    await agent.get('/api/sessions').set('Host', '127.0.0.1:8800').expect(401)
+    expect(sessionReads).toBe(4)
+  })
+})

tests/server/realtimeBroker.test.ts

@@ -4,7 +4,7 @@ import { tmpdir } from 'node:os'
 import { once } from 'node:events'
 import { WebSocketServer, type WebSocket } from 'ws'
 import { afterEach, describe, expect, it, vi } from 'vitest'
-import { RealtimeBroker, type RealtimePrincipal, type StreamEntry } from '../../src/server/realtimeBroker.js'
+import { RealtimeBroker, type RealtimeActivity, type RealtimePrincipal, type StreamEntry } from '../../src/server/realtimeBroker.js'
 import { RealtimeReceipts } from '../../src/server/realtimeReceipts.js'
 
 const cleanup: Array<() => void> = []
@@ -38,11 +38,12 @@ async function fixture() {
     })
   })
   let now = Date.now()
-  const broker = new RealtimeBroker(home, () => now)
+  const activities: RealtimeActivity[] = []
+  const broker = new RealtimeBroker(home, () => now, change => activities.push(change))
   cleanup.push(() => broker.close())
   const principal = (key: string): RealtimePrincipal => ({ key, upstreamKey: 'one-service', paired: false,
     valid: () => true, url: async () => new URL(`ws://127.0.0.1:${(ws.address() as any).port}`) })
-  return { home, broker, principal, commands, count: () => count, advance: (n: number) => { now += n },
+  return { home, broker, principal, commands, activities, count: () => count, advance: (n: number) => { now += n },
     disconnect: () => peer!.terminate(), truncate: () => { truncate = true }, dropPrompt: () => { dropPrompt = true },
     emit: (type: string, payload: unknown) => {
       const params = { type, session_id: 'runtime-1', seq: ++seq, payload }; history.push(params)
@@ -161,4 +162,20 @@ describe('realtime broker', () => {
     expect((await f.broker.command(c, 'lost-ack', prompt)).state).toBe('unknown')
     expect(f.commands.filter(f => f.method === 'prompt.submit')).toHaveLength(1)
   })
+  it('invalidates read snapshots before/after mutation and before delivering live events', async () => {
+    const f = await fixture(), c = await f.broker.create(f.principal('alice'), 'chat')
+    await f.broker.command(c, 'open', resume)
+    await f.broker.command(c, 'send', { method: 'prompt.submit', params: { session_id: 'runtime-1', text: 'once' } })
+    expect(f.activities.filter(c => c.kind === 'command' && c.name === 'prompt.submit')).toEqual([
+      { kind: 'command', name: 'prompt.submit', sessionId: 'stored-1' },
+      { kind: 'command', name: 'prompt.submit', sessionId: 'stored-1' },
+    ])
+    const seen: StreamEntry[] = []
+    f.broker.subscribe(c, undefined, entry => {
+      if (entry.data.includes('message.delta')) expect(f.activities.at(-1)).toMatchObject({ kind: 'event', name: 'message.delta', sessionId: 'stored-1' })
+      seen.push(entry)
+    })
+    f.emit('message.delta', { text: 'fresh' })
+    await vi.waitFor(() => expect(seen.some(e => e.data.includes('fresh'))).toBe(true))
+  })
 })

tests/server/upstreamReadCache.test.ts

@@ -0,0 +1,124 @@
+// @vitest-environment node
+import { describe, expect, it, vi } from 'vitest'
+import { CookieJar, UpstreamClient, type UpstreamResponse } from '../../src/server/upstream.js'
+import { readPolicy, UpstreamReadCache } from '../../src/server/upstreamReadCache.js'
+
+function value(n: number, status = 200): UpstreamResponse {
+  return { status, headers: new Headers({ 'content-type': 'application/json' }), body: Buffer.from(JSON.stringify({ n })) }
+}
+function gate<T>() { let resolve!: (v: T) => void; const promise = new Promise<T>(r => { resolve = r }); return { promise, resolve } }
+const policy = { tags: ['sessions', 'session:one'], freshMs: 1000, staleMs: 0 }
+
+describe('bounded upstream read cache', () => {
+  it('coalesces cold reads, copies responses and expires fresh entries', async () => {
+    let now = 0
+    const cache = new UpstreamReadCache(() => now), load = vi.fn(async () => value(1))
+    const [a, b] = await Promise.all([cache.read('one', policy, load), cache.read('one', policy, load)])
+    expect(load).toHaveBeenCalledTimes(1)
+    a.body.fill(0); a.headers.set('private', 'changed')
+    expect(b.body.toString()).toBe('{"n":1}')
+    expect((await cache.read('one', policy, load)).headers.has('private')).toBe(false)
+    now = 1001
+    await cache.read('one', policy, load)
+    expect(load).toHaveBeenCalledTimes(2)
+    expect(cache.stats()).toMatchObject({ hits: 1, coalesced: 1 })
+  })
+  it('refreshes stale metadata once in the background, without extending failed freshness', async () => {
+    let now = 0
+    const cache = new UpstreamReadCache(() => now), p = { ...policy, staleMs: 1000 }
+    await cache.read('one', p, async () => value(1))
+    now = 1001
+    const pending = gate<UpstreamResponse>(), load = vi.fn(() => pending.promise)
+    expect((await cache.read('one', p, load)).body.toString()).toContain('1')
+    await cache.read('one', p, load)
+    expect(load).toHaveBeenCalledTimes(1)
+    pending.resolve(value(2))
+    await vi.waitFor(() => expect(cache.stats().inFlight).toBe(0))
+    expect((await cache.read('one', p, load)).body.toString()).toContain('2')
+    now = 4000
+    await expect(cache.read('one', p, async () => { throw new Error('offline') })).rejects.toThrow('offline')
+  })
+  it('does not refill from a read that overlapped invalidation', async () => {
+    const cache = new UpstreamReadCache(), pending = gate<UpstreamResponse>()
+    const old = cache.read('one', policy, () => pending.promise)
+    cache.invalidate(['session:one'])
+    await cache.read('one', policy, async () => value(2))
+    pending.resolve(value(1)); await old
+    const fetched = await cache.read('one', policy, async () => value(3))
+    expect(fetched.body.toString()).toContain('2')
+  })
+  it('enforces both byte and entry limits and never caches errors or no-store', async () => {
+    const cache = new UpstreamReadCache(Date.now, 2000, 2, 1024)
+    for (let i = 0; i < 10; i++) await cache.read(String(i), policy, async () => value(i))
+    expect(cache.stats().entries).toBeLessThanOrEqual(2)
+    expect(cache.stats().bytes).toBeLessThanOrEqual(2000)
+    const error = vi.fn(async () => value(4, 401))
+    await cache.read('error', policy, error); await cache.read('error', policy, error)
+    expect(error).toHaveBeenCalledTimes(2)
+    const noStore = vi.fn(async () => { const v = value(4); v.headers.set('cache-control', 'no-store'); return v })
+    await cache.read('private', policy, noStore); await cache.read('private', policy, noStore)
+    expect(noStore).toHaveBeenCalledTimes(2)
+  })
+  it('does not cache critical or cursor-dependent endpoints', () => {
+    for (const path of ['/api/status', '/api/auth/me', '/api/auth/providers', '/api/config', '/api/sessions/search', '/api/session-unread', '/api/plugins/yaoyao/v1/capabilities']) {
+      expect(readPolicy(path)).toBeUndefined()
+    }
+    expect(readPolicy('/api/plugins/yaoyao/v1/rooms/room/messages', new URLSearchParams('afterSeq=5'))).toBeUndefined()
+  })
+})
+
+describe('scoped upstream read-through cache', () => {
+  function fixture() {
+    const fetch = vi.fn<typeof globalThis.fetch>(async () => Response.json({ value: fetch.mock.calls.length }))
+    const client = new UpstreamClient(new URL('http://localhost:9119'), fetch)
+    const jar = new CookieJar('service=one')
+    const read = (scope = 'alice', profile = 'a', options = {}) => client.withReadScope(scope, false,
+      () => client.request('/api/sessions/same/messages', jar, { search: new URLSearchParams({ profile }), ...options }))
+    return { client, jar, fetch, read }
+  }
+  it('isolates user/profile/cookie/headers while merging identical reads', async () => {
+    const f = fixture()
+    await Promise.all([f.read(), f.read(), f.read()]); expect(f.fetch).toHaveBeenCalledTimes(1)
+    await f.read('bob'); await f.read('alice', 'b')
+    expect(f.fetch).toHaveBeenCalledTimes(3)
+    f.jar.replace(new CookieJar('service=two')); await f.read()
+    expect(f.fetch).toHaveBeenCalledTimes(4)
+    await f.read('alice', 'a', { headers: { 'x-custom-scope': 'other' } })
+    expect(f.fetch).toHaveBeenCalledTimes(5)
+  })
+  it('bypasses background, explicitly fresh and range reads', async () => {
+    const f = fixture()
+    await f.read(); await f.read()
+    expect(f.fetch).toHaveBeenCalledTimes(1)
+    await f.client.request('/api/sessions/same/messages', f.jar)
+    expect(f.fetch).toHaveBeenCalledTimes(2)
+    await f.client.request('/api/sessions/same/messages', f.jar)
+    expect(f.fetch).toHaveBeenCalledTimes(3)
+    await f.read('alice', 'a', { cache: 'reload' })
+    expect(f.fetch).toHaveBeenCalledTimes(4)
+    expect(new Headers({ range: 'bytes=0-4' }).has('range')).toBe(true)
+    await f.read('alice', 'a', { headers: { range: 'bytes=0-4' } })
+    expect(f.fetch).toHaveBeenCalledTimes(5)
+  })
+  it('invalidates reads on writes and broker events without invalidating unrelated sessions', async () => {
+    const f = fixture()
+    await f.read()
+    await f.client.withReadScope('alice', false, () => f.client.request('/api/sessions/other/messages', f.jar))
+    f.client.observeRealtime({ kind: 'event', name: 'message.delta', sessionId: 'same' })
+    await f.read()
+    await f.client.withReadScope('alice', false, () => f.client.request('/api/sessions/other/messages', f.jar))
+    expect(f.fetch).toHaveBeenCalledTimes(3)
+    await f.client.request('/api/sessions/same', f.jar, { method: 'PATCH', body: { title: 'new' } })
+    await f.read()
+    expect(f.fetch).toHaveBeenCalledTimes(5)
+  })
+  it('invalidates earlier group snapshots after a fresh journal anchor', async () => {
+    const f = fixture()
+    const rooms = () => f.client.withReadScope('alice', false, () => f.client.request('/api/plugins/yaoyao/v1/rooms', f.jar))
+    await rooms(); await rooms()
+    expect(f.fetch).toHaveBeenCalledTimes(1)
+    await f.client.request('/api/plugins/yaoyao/v1/capabilities', f.jar)
+    await rooms()
+    expect(f.fetch).toHaveBeenCalledTimes(3)
+  })
+})

tests/server/upstreamServiceSession.test.ts

@@ -0,0 +1,99 @@
+// @vitest-environment node
+import { describe, expect, it, vi } from 'vitest'
+import { UpstreamServiceSession } from '../../src/server/localAuth.js'
+import { CookieJar, UpstreamClient } from '../../src/server/upstream.js'
+
+describe('upstream service validation lease', () => {
+  function fixture() {
+    let now = 0, expire = false
+    const paths: string[] = []
+    const fetch = vi.fn<typeof globalThis.fetch>(async (input, init) => {
+      const path = new URL(String(input)).pathname; paths.push(path)
+      if (path === '/api/status') return Response.json({ auth_required: true })
+      if (path === '/api/auth/me') return Response.json({ user_id: 'service' })
+      if (path === '/api/auth/providers') return Response.json({ providers: [] })
+      if (path === '/auth/password-login') return Response.json({ ok: true }, { headers: { 'set-cookie': 'session=new; Path=/' } })
+      if (expire && new Headers(init?.headers).get('cookie') === 'session=old') return Response.json({ error: 'expired' }, { status: 401 })
+      return Response.json({ path })
+    })
+    const client = new UpstreamClient(new URL('http://localhost:9119'), fetch, false, () => now)
+    const session = new UpstreamServiceSession(client, () => ({ username: 'service', password: 'fixture' }), { now: () => now })
+    session.jar.replace(new CookieJar('session=old'))
+    return { session, client, fetch, paths, advance: (n: number) => { now += n }, expire: () => { expire = true } }
+  }
+  it('validates once for concurrent requests and rechecks only after thirty seconds', async () => {
+    const f = fixture()
+    await Promise.all(Array.from({ length: 10 }, (_, i) => f.session.request(`/api/sessions/${i}`)))
+    expect(f.paths.filter(p => p === '/api/status')).toHaveLength(1)
+    expect(f.paths.filter(p => p === '/api/auth/me')).toHaveLength(1)
+    await f.session.request('/api/sessions/a')
+    expect(f.paths).toHaveLength(13)
+    f.advance(30_001); await f.session.request('/api/sessions/b')
+    expect(f.paths.filter(p => p === '/api/status')).toHaveLength(2)
+  })
+  it('reduces repeated protected reads to zero upstream calls on a warm hit', async () => {
+    const f = fixture()
+    const read = () => f.client.withReadScope('alice', false, () => f.session.request('/api/sessions'))
+    await Promise.all([read(), read(), read()])
+    expect(f.paths).toEqual(['/api/status', '/api/auth/me', '/api/sessions'])
+    await read(); await read()
+    expect(f.paths).toHaveLength(3)
+  })
+  it('repairs parallel 401s once, including raw withJar-style reads', async () => {
+    const f = fixture(); await f.session.ensure(); f.expire()
+    await Promise.all([f.client.request('/api/sessions/a', f.session.jar), f.client.request('/api/sessions/b', f.session.jar)])
+    expect(f.paths.filter(p => p === '/auth/password-login')).toHaveLength(1)
+    expect(f.paths.filter(p => p === '/api/sessions/a')).toHaveLength(2)
+    expect(f.paths.filter(p => p === '/api/sessions/b')).toHaveLength(2)
+    expect(f.session.jar.header).toBe('session=new')
+  })
+  it('never replays a write on a timeout or 500 response', async () => {
+    const f = fixture(); await f.session.ensure()
+    f.fetch.mockImplementation(async () => Response.json({ error: 'failed' }, { status: 500 }))
+    const before = f.fetch.mock.calls.length
+    expect((await f.session.request('/api/sessions/s', { method: 'POST', body: { text: 'once' } })).status).toBe(500)
+    expect(f.fetch.mock.calls.length - before).toBe(1)
+  })
+  it('invalidates validation and cached data on configured credential changes', async () => {
+    const f = fixture()
+    let password = 'first'
+    const session = new UpstreamServiceSession(f.client, () => ({ username: 'service', password }), { now: () => 0 })
+    await session.ensure()
+    await f.client.withReadScope('alice', false, () => session.request('/api/sessions'))
+    expect(f.client.readCache.stats().entries).toBe(1)
+    password = 'second'; await session.ensure()
+    expect(f.paths.filter(p => p === '/api/status')).toHaveLength(2)
+    expect(f.client.readCache.stats().entries).toBe(0)
+  })
+  it('does not memoize failed validation', async () => {
+    const f = fixture()
+    f.fetch.mockResolvedValueOnce(Response.json({ error: 'offline' }, { status: 503 }))
+    await expect(f.session.ensure()).rejects.toThrow('9119')
+    await f.session.ensure()
+    expect(f.fetch).toHaveBeenCalledTimes(3)
+  })
+  it('cannot overwrite new credentials with a late old login response', async () => {
+    let username = 'old', started = false
+    let release!: () => void
+    const blocked = new Promise<void>(r => { release = r })
+    const fetch = vi.fn<typeof globalThis.fetch>(async (input, init) => {
+      const path = new URL(String(input)).pathname
+      if (path === '/api/status') return Response.json({ auth_required: true })
+      if (path === '/api/auth/me') return Response.json({}, { status: 401 })
+      if (path === '/api/auth/providers') return Response.json({ providers: [] })
+      const body = JSON.parse(String(init?.body))
+      if (body.username === 'old') { started = true; await blocked }
+      return Response.json({ ok: true }, { headers: { 'set-cookie': `session=${body.username}; Path=/` } })
+    })
+    const client = new UpstreamClient(new URL('http://localhost:9119'), fetch)
+    const session = new UpstreamServiceSession(client, () => ({ username, password: 'fixture' }))
+    const old = session.ensure().then(() => undefined, error => error)
+    await vi.waitFor(() => expect(started).toBe(true))
+    username = 'new'
+    await session.ensure()
+    expect(session.jar.header).toBe('session=new')
+    release()
+    expect(await old).toMatchObject({ code: 'upstream_credentials_changed' })
+    expect(session.jar.header).toBe('session=new')
+  })
+})