mirror of
https://github.com/deepseek-ai/deepseek-harness.git
synced 2026-11-09 20:07:43 +08:00
Merge pull request #5537 from deepseek-harness/perf/recent-trace-hotspot
perf(api): time-slice Session list work
This commit is contained in:
Turtle
@@ -0,0 +1,22 @@
|
||||
# Bilingual-pair consistency record for 2026-09-30-session-list-host-fairness.md (docs/i18n/README.md): per heading
|
||||
# section, a hash of its English and Chinese blocks outside code blocks and generated regions.
|
||||
# After editing either side, bring the other along and re-record with:
|
||||
# pnpm run verify-translation-pairing --write .agents/notes/implemented/bug-fix/2026-09-30-session-list-host-fairness.md
|
||||
/agent-note-time-sliced-session-list-work:
|
||||
en: 0e8f0cce65f887bc
|
||||
zh: 1f3b669c1eef811f
|
||||
/agent-note-time-sliced-session-list-work/problem:
|
||||
en: 5ee132d1d44f463b
|
||||
zh: 45589b49771d49e1
|
||||
/agent-note-time-sliced-session-list-work/decision:
|
||||
en: 95b04ae405a5299f
|
||||
zh: 53f1d8bf5e29e3ca
|
||||
/agent-note-time-sliced-session-list-work/measurement-and-calibration:
|
||||
en: e12fe17a32ede555
|
||||
zh: 6b16a72863073e47
|
||||
/agent-note-time-sliced-session-list-work/alternatives-considered:
|
||||
en: a2f441a550c0ed7e
|
||||
zh: 05e8bfd584217103
|
||||
/agent-note-time-sliced-session-list-work/consequences:
|
||||
en: 5fd10d559a120ae6
|
||||
zh: a56f45b77ce1f4e6
|
||||
@@ -0,0 +1,69 @@
|
||||
# Agent Note: Time-sliced Session-list work
|
||||
|
||||
Status: implemented
|
||||
|
||||
English | [中文](2026-09-30-session-list-host-fairness.zh.md)
|
||||
|
||||
## Problem
|
||||
|
||||
Large cached projection states can monopolize the Host while a Session list is assembled. Yielding after every row avoids that batch stall but adds thousands of event-loop round trips when a list contains many cheap rows. Row count alone is not a useful scheduling budget.
|
||||
|
||||
## Decision
|
||||
|
||||
[ApiSessionList](../../../../packages/api/session-controller/src/list.ts) uses one monotonic deadline for classification and summary generation. It yields only after elapsed work exhausts the configured `listWorkSliceMs`, then starts a new deadline after the wait. The optional deployment setting defaults to 16 positive integral milliseconds. Small lists have no forced final yield or no-op asynchronous helper. Cancellation is checked per row, after each yield, and before returning; no partial list is returned.
|
||||
|
||||
Every row remains synchronous. Live rows precede queued cold rows before the stable activity sort; the list promises no cross-Session atomic cut. A cold row already queued before an attachment remains cached, and Client mutation replay plus [cached/sequenced precedence](../architecture/2026-09-19-projection-cache-listing-identity-and-cached-rows.md) reconcile the newer live state. No projection data or validation is omitted.
|
||||
|
||||
## Measurement and calibration
|
||||
|
||||
The [production-entry benchmark](../../../../benchmarks/session-corpus/projection-list.bench.ts) measures the real Session Controller, query engine, persistence, projection registry/cache and storage domain. Persisted fixtures use `generationLogPath`, `toHeaderLine` and `compressZstdFrame` from the persistence owner. All content is deterministic generated data. Setup and explicit GC are outside timing; complete results and JSON remain reachable during memory sampling.
|
||||
|
||||
| Workload | Sessions | State observations | Request records | State JSON |
|
||||
|---|---:|---:|---:|---:|
|
||||
| Cheap live | 3,000 | 0 | 0 | 277,890 bytes |
|
||||
| Modest persisted | 50 | 4,925 | 641 | 1,813,852 bytes |
|
||||
| Heavy persisted | 300 | 300,000 | 37,575 | 111,255,440 bytes |
|
||||
|
||||
The cheap live case avoids disk enumeration masking scheduling cost. Persisted row counts use 50/75/100/175% strata of 100 or 1,000 generated observations. Each fresh child measures one first call and three repeats with a single fixed-cost immediate probe per call. A separate fifth call re-arms at most one pending probe to measure the worst queue delay over the complete list. Full-probe timings are not throughput evidence: per-row yielding would otherwise execute proportionally more diagnostic callbacks. Yield counts exclude post-measurement GC helpers. Peak RSS includes setup and the entire process lifetime.
|
||||
|
||||
Local Node 26.5.0/macOS arm64 comparisons alternate implementations and use three fresh children per workload. Repeat metrics first take each child's median, then the median across children. Queue delay takes the maximum in each separate diagnostic call, then the median across children.
|
||||
|
||||
| Repeated metric | Synchronous | Per-row yield | 16 ms slices |
|
||||
|---|---:|---:|---:|
|
||||
| Cheap live: list + JSON | 61.60 ms | 111.65 ms | 65.68 ms |
|
||||
| Cheap live: yield calls | 0 | 3,000 | 3 |
|
||||
| Cheap live: maximum queue delay | 54.35 ms | 0.54 ms | 16.54 ms |
|
||||
| Modest persisted: list + JSON | 19.76 ms | 18.23 ms | 18.26 ms |
|
||||
| Modest persisted: yield calls | 0 | 50 | 0 |
|
||||
| Heavy persisted: list + JSON | 253.22 ms | 291.36 ms | 232.18 ms |
|
||||
| Heavy persisted: yield calls | 0 | 300 | 10 |
|
||||
| Heavy persisted: maximum queue delay | 172.39 ms | 0.80 ms | 16.67 ms |
|
||||
|
||||
For cheap lists, slicing removes 99.9% of per-row yields and reduces measured repeat time by 41.2%; the synchronous reference remains slightly faster. Heavy-list wall time varies substantially on the shared host, so no universal throughput speedup is asserted. Retained heap stays about 35 MB for cheap lists and 186–188 MB for heavy lists, without a decoded-state cache.
|
||||
|
||||
| Raw child aggregates | Synchronous | Per-row yield | 16 ms slices |
|
||||
|---|---|---|---|
|
||||
| Cheap repeat ms | [63.10, 61.60, 59.08] | [108.19, 115.61, 111.65] | [65.68, 71.75, 65.17] |
|
||||
| Cheap maximum queue ms | [112.69, 54.35, 49.67] | [0.54, 0.76, 0.41] | [16.41, 16.54, 41.15] |
|
||||
| Heavy repeat ms | [277.06, 253.22, 236.22] | [291.36, 302.76, 282.01] | [212.35, 232.18, 304.38] |
|
||||
| Heavy maximum queue ms | [297.76, 172.39, 164.17] | [0.56, 1.34, 0.80] | [16.39, 16.67, 16.76] |
|
||||
|
||||
Separate 4/8/16 ms experiments measured cheap-list repeat medians of 58.39/58.57/52.34 ms and heavy-list medians of 279.81/253.85/216.67 ms. Their heavy-list maximum queued-work delay medians were 5.19/8.45/16.70 ms. The 16 ms default chooses fewer round trips while retaining a short work target; deployments can select a smaller budget.
|
||||
|
||||
CI uses elapsed queue-delay and coarse throughput limits, not an exact batch size. The shared scale 2 and headroom 1.25 turn a 30 ms queue reference allowance into 75 ms, leaving room for one-row overshoot and scheduling variance while rejecting an unyielded heavy batch. Throughput reference allowances are 50/500/150 ms for modest/heavy/cheap, producing 125/1,250/375 ms bounds. These are safety ceilings, not claims of fine-grained non-regression. The earlier Linux Node 24.21 EPYC run measured 102.16/779.57 ms first calls for modest/heavy cases; the larger heavy allowance accounts for the observed hosted/reference ratio exceeding 2. Heap uses 240 MiB and only variance headroom, giving 300 MiB.
|
||||
|
||||
The existing [3,000-Session corpus lane](../testing/2026-09-28-session-corpus-performance.md) remains the stored-corpus throughput authority. Its earlier per-row CI run measured first-call samples 3,205.3/3,166.2/3,165.1 ms and repeat samples 2,643.5/2,577.8/2,589.6 ms, within its existing calibrated limits. The projection benchmark adds a distinct cheap-live and large-state dimension rather than replacing that corpus. On the same local machine, the stored-corpus per-row/sliced first medians are 885.5/859.4 ms and repeat medians are 700.3/692.1 ms; that small difference is within ordinary run variation, with no observed throughput regression. The new three-workload file takes 23.98 s locally, including fixture setup and all nine children; its previously measured two-workload CI version took 36.80 s on EPYC Node 24.21.0.
|
||||
|
||||
## Alternatives considered
|
||||
|
||||
**Unconditional per-row yielding.** It minimizes individual queue waits but charges an event-loop round trip for every cheap row. The many-live-Session comparison exposes that cost; a fixed number of rows would still ignore variable row cost.
|
||||
|
||||
**Larger or smaller time slices.** Smaller slices offer shorter queue waits at more scheduling cost. Larger slices reduce switching but extend the period other Host work waits. Configuration exposes this tradeoff without changing row semantics.
|
||||
|
||||
**Caching decoded state or wire views.** Decoded-state reuse retains another graph per stored row and changes schema-evaluation ownership. Wire views can depend on current runtime state. Scheduling needs neither cache and preserves all schema checks and dynamic views.
|
||||
|
||||
## Consequences
|
||||
|
||||
[Controlled-clock tests](../../../../packages/api/session-controller/tests/list-scheduling.host.spec.ts) require cheap rows to remain in one synchronous slice, budget exhaustion and post-wait reset to govern yields, and cancellation to stop work even when no yield is due. They also preserve stable ties and attachment/removal/status interleavings. The time-based benchmark guard rejects unyielded or first-only-yield heavy work without forbidding efficient batches.
|
||||
|
||||
A single row, provider enumeration, GC, final sorting and serialization can exceed the target. Concurrent lists and operating-system scheduling also affect latency. The budget is not a hard global bound, and these measurements do not claim browser paint or model/network latency improvements.
|
||||
@@ -0,0 +1,69 @@
|
||||
# Agent Note: 按时间片执行 Session 列表工作
|
||||
|
||||
Status: implemented
|
||||
|
||||
[English](2026-09-30-session-list-host-fairness.md) | 中文
|
||||
|
||||
## 问题
|
||||
|
||||
较大的缓存投影状态会在组装 Session 列表时独占 Host。每行都让出可以避免整批停顿，却会在包含大量低开销行的列表上增加数千次事件循环往返。行数本身不是合适的调度预算。
|
||||
|
||||
## 决策
|
||||
|
||||
[ApiSessionList](../../../../packages/api/session-controller/src/list.ts) 对分类和摘要生成共享一个单调时钟截止时间。仅当经过的工作时间耗尽配置的 `listWorkSliceMs` 时才让出，并在等待结束后建立新的截止时间。这个可选部署设置默认为 16 毫秒，要求正整数。小列表没有强制的最后一次让出，也不经过空操作异步 helper。每行、每次让出之后以及返回前检查取消；不会返回部分列表。
|
||||
|
||||
每行仍同步计算。live 行先于已进入队列的 cold 行，随后按活动时间稳定排序；列表不承诺跨 Session 的原子 cut。接入前已经排入队列的冷行仍是缓存行，Client 变更重放与 [cached/sequenced 优先级](../architecture/2026-09-19-projection-cache-listing-identity-and-cached-rows.zh.md)协调较新的 live 状态。不省略任何投影数据或校验。
|
||||
|
||||
## 测量与校准
|
||||
|
||||
[生产入口基准](../../../../benchmarks/session-corpus/projection-list.bench.ts)测量真实 Session Controller、查询引擎、持久化、投影注册表与缓存，以及 storage domain。持久化夹具使用持久化拥有方的 `generationLogPath`、`toHeaderLine` 和 `compressZstdFrame`。全部内容都是确定性生成数据。准备和显式 GC 不计入耗时；内存采样时完整结果与 JSON 仍可达。
|
||||
|
||||
| 负载 | Session 数 | 状态观测数 | 请求记录数 | 状态 JSON |
|
||||
|---|---:|---:|---:|---:|
|
||||
| 低开销驻留 | 3,000 | 0 | 0 | 277,890 bytes |
|
||||
| 适中持久化 | 50 | 4,925 | 641 | 1,813,852 bytes |
|
||||
| 大状态持久化 | 300 | 300,000 | 37,575 | 111,255,440 bytes |
|
||||
|
||||
低开销驻留场景避免磁盘枚举掩盖调度成本。持久化行按 100 或 1,000 条生成观测的 50/75/100/175% 分层。每个全新子进程测量一次首次调用和三次重复调用，每次只使用一个固定成本的 immediate 探针。独立的第五次调用以最多一个待处理探针反复采样，测量完整列表中的最差排队延迟。完整探针模式的耗时不作为吞吐量证据，否则逐行让出会按比例执行更多诊断回调。yield 次数不包含测量后 GC helper 的调用。峰值 RSS 包含准备阶段及整个进程生命周期。
|
||||
|
||||
本地比较使用 Node 26.5.0/macOS arm64，交替运行实现，每种负载使用三个全新子进程。重复指标先取每个子进程的中位数，再取子进程间的中位数。排队延迟先取独立诊断调用的最大值，再取子进程间的中位数。
|
||||
|
||||
| 重复读取指标 | 同步 | 逐行让出 | 16 ms 时间片 |
|
||||
|---|---:|---:|---:|
|
||||
| 低开销驻留：列表与 JSON | 61.60 ms | 111.65 ms | 65.68 ms |
|
||||
| 低开销驻留：yield 次数 | 0 | 3,000 | 3 |
|
||||
| 低开销驻留：最大排队延迟 | 54.35 ms | 0.54 ms | 16.54 ms |
|
||||
| 适中持久化：列表与 JSON | 19.76 ms | 18.23 ms | 18.26 ms |
|
||||
| 适中持久化：yield 次数 | 0 | 50 | 0 |
|
||||
| 大状态持久化：列表与 JSON | 253.22 ms | 291.36 ms | 232.18 ms |
|
||||
| 大状态持久化：yield 次数 | 0 | 300 | 10 |
|
||||
| 大状态持久化：最大排队延迟 | 172.39 ms | 0.80 ms | 16.67 ms |
|
||||
|
||||
在低开销列表中，时间片去掉了 99.9% 的逐行 yield，并使测得的重复耗时下降 41.2%；同步参考仍略快。共享 Host 上大状态列表的墙钟耗时波动明显，因此不宣称普遍提高吞吐量。常驻堆在低开销场景约为 35 MB，在大状态场景约为 186–188 MB，没有解码状态缓存。
|
||||
|
||||
| 原始子进程聚合值 | 同步 | 逐行让出 | 16 ms 时间片 |
|
||||
|---|---|---|---|
|
||||
| 低开销重复 ms | [63.10, 61.60, 59.08] | [108.19, 115.61, 111.65] | [65.68, 71.75, 65.17] |
|
||||
| 低开销最大排队 ms | [112.69, 54.35, 49.67] | [0.54, 0.76, 0.41] | [16.41, 16.54, 41.15] |
|
||||
| 大状态重复 ms | [277.06, 253.22, 236.22] | [291.36, 302.76, 282.01] | [212.35, 232.18, 304.38] |
|
||||
| 大状态最大排队 ms | [297.76, 172.39, 164.17] | [0.56, 1.34, 0.80] | [16.39, 16.67, 16.76] |
|
||||
|
||||
独立的 4/8/16 ms 实验测得，低开销列表重复耗时中位数分别为 58.39/58.57/52.34 ms，大状态列表为 279.81/253.85/216.67 ms。大状态列表最大排队延迟的中位数分别为 5.19/8.45/16.70 ms。16 ms 默认值在保持较短工作目标的同时减少往返；部署可以选择更小的预算。
|
||||
|
||||
CI 使用经过时间的排队延迟与粗粒度吞吐量上限，而不是精确批大小。共享的 2 倍时间系数和 1.25 倍余量将 30 ms 排队参考余量转换为 75 ms，为单行超时和调度波动留出空间，同时拒绝没有让出的大状态批次。适中、大状态、低开销三种负载的吞吐量参考余量分别为 50/500/150 ms，得到 125/1,250/375 ms 上限。这些是安全上限，不证明细粒度不回退。先前 Linux Node 24.21 EPYC 运行测得适中和大状态场景首次调用为 102.16/779.57 ms；较大的大状态余量考虑了实际 hosted/reference 比值超过 2。堆使用 240 MiB，并只乘波动余量，得到 300 MiB。
|
||||
|
||||
已有的 [3,000 Session 会话集合测试](../testing/2026-09-28-session-corpus-performance.zh.md)仍是持久化集合吞吐量的依据。先前逐行版本的 CI 首次调用样本为 3,205.3/3,166.2/3,165.1 ms，重复调用为 2,643.5/2,577.8/2,589.6 ms，均在既有校准上限内。投影基准补充低开销驻留和大状态维度，不替代该集合。在同一本地机器上，持久化集合的逐行/时间片首次中位数为 885.5/859.4 ms，重复中位数为 700.3/692.1 ms；这点差异处于普通运行波动内，未观察到吞吐量回退。新的三负载文件在本地耗时 23.98 s，包含夹具准备和全部九个子进程；先前测得的双负载 CI 版本在 EPYC Node 24.21.0 上耗时 36.80 s。
|
||||
|
||||
## 考虑过的替代方案
|
||||
|
||||
**无条件逐行让出。** 它尽量缩短单次排队等待，却对每个低开销行都收取一次事件循环往返成本。大量驻留 Session 的比较暴露了这项成本；固定行数同样忽略了行成本差异。
|
||||
|
||||
**更大或更小的时间片。** 更小的时间片缩短排队等待，但增加调度成本。更大的时间片减少切换，却延长其他 Host 工作的等待。配置显式提供这一取舍，不改变行语义。
|
||||
|
||||
**缓存解码状态或 wire view。** 解码状态复用会为每份存储行保留另一份对象图，并改变 schema 执行的所有权。wire view 可能依赖当前运行状态。调度不需要这两类缓存，且保留全部 schema 校验和动态视图。
|
||||
|
||||
## 后果
|
||||
|
||||
[受控时钟测试](../../../../packages/api/session-controller/tests/list-scheduling.host.spec.ts)要求低开销行保持在同一同步时间片内，由预算耗尽和等待后重置决定让出，并在无需让出时也能响应取消。测试还保留了稳定并列排序和接入、移除、状态变更的交错行为。按时间判断的基准保护会拒绝不让出或仅首次让出的大状态工作，同时允许高效批处理。
|
||||
|
||||
单行、提供方枚举、GC、最终排序和序列化都可能超出目标。并发列表和操作系统调度也会影响延迟。预算不是全局硬上限，这些测量也不宣称改善浏览器绘制或模型、网络延迟。
|
||||
@@ -43,7 +43,8 @@
|
||||
"@deepseek-ai/dsh-tools": "workspace:*",
|
||||
"@deepseek-ai/dsh-token-meter": "workspace:*",
|
||||
"@deepseek-ai/dsh-typert-protocol": "workspace:*",
|
||||
"@deepseek-ai/dsh-util-values": "workspace:*"
|
||||
"@deepseek-ai/dsh-util-values": "workspace:*",
|
||||
"zod": "^4.4.3"
|
||||
},
|
||||
"peerDependencies": {
|
||||
"@deepseek-ai/cordis": "workspace:~"
|
||||
|
||||
@@ -0,0 +1,85 @@
|
||||
/** Session-list Host queue responsiveness; no browser input, transport, or paint is measured. */
|
||||
|
||||
import { mkdtemp, rm } from 'node:fs/promises'
|
||||
import { cpus, tmpdir } from 'node:os'
|
||||
import { join } from 'node:path'
|
||||
import { expect, it } from 'vitest'
|
||||
import { runBuiltBenchmarkWorker } from '../support/built-worker.ts'
|
||||
import { ciTimeBudget, PERFORMANCE_BUDGET_HEADROOM } from '../support/calibration.ts'
|
||||
import type { ProjectionListReport } from './projection-list.worker.ts'
|
||||
|
||||
const WORKER = join(import.meta.dirname, '..', '.dsh-build', 'session-corpus', 'projection-list.worker.js')
|
||||
const ATTEMPTS = 3
|
||||
const WORKER_TIMEOUT_MS = 120_000
|
||||
/** Coarse reference-machine throughput allowances, before CI scaling and variance headroom. */
|
||||
const LIST_REFERENCE_MS = { modest: 50, tail: 500, cheap: 150 } as const
|
||||
/** Queue-delay reference allowance above the measured 16 ms work slices and one-row overshoot. */
|
||||
const CALLBACK_REFERENCE_MS = 30
|
||||
/** Reference retained-heap allowance; memory receives variance headroom but no CPU scaling. */
|
||||
const RETAINED_HEAP_REFERENCE_BYTES = 240 * 1024 * 1024
|
||||
|
||||
function median(values: readonly number[]): number {
|
||||
return [...values].sort((left, right) => left - right)[Math.floor(values.length / 2)] as number
|
||||
}
|
||||
|
||||
async function run(workload: ProjectionListReport['workload']): Promise<ProjectionListReport> {
|
||||
const root = await mkdtemp(join(tmpdir(), 'dsh-projection-list-bench-'))
|
||||
try {
|
||||
const outcome = await runBuiltBenchmarkWorker<ProjectionListReport>({
|
||||
worker: WORKER, args: [root, workload], timeoutMs: WORKER_TIMEOUT_MS, exposeGc: true,
|
||||
})
|
||||
if (outcome.timedOut || outcome.signal !== null || outcome.exitCode !== 0 || outcome.report === undefined) {
|
||||
throw new Error(`projection-list ${workload} failed: exit=${String(outcome.exitCode)}, signal=${String(outcome.signal)}, `
|
||||
+ `timedOut=${String(outcome.timedOut)}\n${outcome.stderr.trim().split('\n').slice(-20).join('\n')}`)
|
||||
}
|
||||
return outcome.report
|
||||
} finally {
|
||||
// The shared launcher resolves only after child close, including timeout kills.
|
||||
await rm(root, { recursive: true, force: true })
|
||||
}
|
||||
}
|
||||
|
||||
it.each(['modest', 'tail', 'cheap'] as const)('serves the %s projection list without monopolizing the Host queue', async (workload) => {
|
||||
const reports: ProjectionListReport[] = []
|
||||
for (let attempt = 0; attempt < ATTEMPTS; attempt++) reports.push(await run(workload))
|
||||
const firstMs = reports.map(report => report.samples[0]!.listAndJsonMs)
|
||||
const repeatMs = reports.map(report => median(report.samples.slice(1).map(sample => sample.listAndJsonMs)))
|
||||
const heapBytes = reports.map(report => Math.max(...[...report.samples, report.responsiveness].map(sample => sample.memory.heapUsedBytes)))
|
||||
const medians = {
|
||||
firstListAndJsonMs: median(firstMs),
|
||||
repeatListAndJsonMs: median(repeatMs),
|
||||
firstListMs: median(reports.map(report => report.samples[0]!.listMs)),
|
||||
repeatListMs: median(reports.map(report => median(report.samples.slice(1).map(sample => sample.listMs)))),
|
||||
firstListAndJsonCpuMs: median(reports.map(report => report.samples[0]!.listAndJsonCpuMs)),
|
||||
repeatListAndJsonCpuMs: median(reports.map(report => median(report.samples.slice(1).map(sample => sample.listAndJsonCpuMs)))),
|
||||
retainedHeapBytes: median(heapBytes),
|
||||
worstCallbackDelayMs: median(reports.map(report => report.responsiveness.maxCallbackDelayMs)),
|
||||
repeatYieldCalls: median(reports.map(report => median(report.samples.slice(1).map(sample => sample.yieldCalls)))),
|
||||
firstCallbackDelayMs: median(reports.map(report => report.samples[0]!.callbackDelayMs)),
|
||||
repeatCallbackDelayMs: median(reports.map(report => median(report.samples.slice(1).map(sample => sample.callbackDelayMs)))),
|
||||
firstEventLoopMaxMs: median(reports.map(report => report.samples[0]!.eventLoopMaxMs)),
|
||||
repeatEventLoopMaxMs: median(reports.map(report => Math.max(...report.samples.slice(1).map(sample => sample.eventLoopMaxMs)))),
|
||||
}
|
||||
const budgets = {
|
||||
listAndJsonMs: ciTimeBudget(LIST_REFERENCE_MS[workload]),
|
||||
callbackDelayMs: ciTimeBudget(CALLBACK_REFERENCE_MS),
|
||||
retainedHeapBytes: Math.ceil(RETAINED_HEAP_REFERENCE_BYTES * PERFORMANCE_BUDGET_HEADROOM),
|
||||
}
|
||||
console.log(JSON.stringify({ benchmark: `session-corpus/projection-list-${workload}`,
|
||||
reports, medians, budgets, environment: { node: process.version, platform: process.platform, arch: process.arch, cpu: cpus()[0]?.model } }))
|
||||
|
||||
for (const report of reports) {
|
||||
expect(report.workload).toBe(workload)
|
||||
expect(report.fixture.sessions).toBe({ modest: 50, tail: 300, cheap: 3_000 }[workload])
|
||||
expect(report.fixture.baseRows).toBe({ modest: 100, tail: 1_000, cheap: 0 }[workload])
|
||||
expect(report.samples).toHaveLength(4)
|
||||
for (const sample of [...report.samples, report.responsiveness]) {
|
||||
expect(sample.items).toBe(report.fixture.sessions)
|
||||
expect(sample.wireViews).toBe(report.fixture.sessions)
|
||||
}
|
||||
}
|
||||
expect(medians.worstCallbackDelayMs).toBeLessThanOrEqual(budgets.callbackDelayMs)
|
||||
expect(medians.firstListAndJsonMs).toBeLessThanOrEqual(budgets.listAndJsonMs)
|
||||
expect(medians.repeatListAndJsonMs).toBeLessThanOrEqual(budgets.listAndJsonMs)
|
||||
expect(medians.retainedHeapBytes).toBeLessThanOrEqual(budgets.retainedHeapBytes)
|
||||
}, ATTEMPTS * WORKER_TIMEOUT_MS + 30_000)
|
||||
@@ -0,0 +1,309 @@
|
||||
/** Plain-Node Session-list measurements with large persisted projections and small wire summaries. */
|
||||
|
||||
import assert from 'node:assert/strict'
|
||||
import { mkdir, writeFile } from 'node:fs/promises'
|
||||
import { join } from 'node:path'
|
||||
import { monitorEventLoopDelay, performance } from 'node:perf_hooks'
|
||||
import { scheduler, setTimeout as delay } from 'node:timers/promises'
|
||||
import { Context } from '@deepseek-ai/cordis'
|
||||
import AgentRegistry from '@deepseek-ai/dsh-agent'
|
||||
import SessionController from '@deepseek-ai/dsh-api-session-controller'
|
||||
import SessionStore, { SESSION_FORMAT_VERSION, SessionId, SessionLogOffset, type SessionHeader } from '@deepseek-ai/dsh-session'
|
||||
import JsonlSessionPersistence from '@deepseek-ai/dsh-session-persistence-jsonl'
|
||||
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
|
||||
import SessionProjectionCache, { projectionCacheDomainSpec } from '@deepseek-ai/dsh-session-projection-cache'
|
||||
import SqliteSessionQueryEngine from '@deepseek-ai/dsh-session-query-sqlite'
|
||||
import Storage from '@deepseek-ai/dsh-storage'
|
||||
import * as StorageDomain from '@deepseek-ai/dsh-storage-domain'
|
||||
import * as StorageJson from '@deepseek-ai/dsh-storage-json'
|
||||
import { z } from 'zod'
|
||||
import { generationLogPath, sessionDir, toHeaderLine } from '../../packages/session/session-persistence-jsonl/src/format.ts'
|
||||
import { compressZstdFrame } from '../../packages/session/session-persistence-jsonl/src/zstd.ts'
|
||||
import { assertBuiltBenchmarkRuntime } from '../support/built-worker.ts'
|
||||
|
||||
const KEY = 'syntheticProjectionList'
|
||||
const CWD = '/synthetic-projection-benchmark'
|
||||
const BASE_TIME = 1_700_000_000_000
|
||||
const ROW_PERCENTAGES = [50, 75, 100, 175] as const
|
||||
const WORKLOADS = {
|
||||
modest: { sessions: 50, baseRows: 100, live: false },
|
||||
tail: { sessions: 300, baseRows: 1_000, live: false },
|
||||
cheap: { sessions: 3_000, baseRows: 0, live: true },
|
||||
} as const
|
||||
|
||||
const stateSchema = z.object({
|
||||
sessionOrdinal: z.number().int(),
|
||||
events: z.array(z.object({
|
||||
ordinal: z.number().int().nonnegative(),
|
||||
category: z.enum(['input', 'output', 'tool']),
|
||||
metrics: z.object({ tokens: z.number().int(), bytes: z.number().int(), elapsedMs: z.number().int() }),
|
||||
content: z.array(z.object({ kind: z.enum(['text', 'metadata']), text: z.string() })),
|
||||
tags: z.array(z.string()),
|
||||
})),
|
||||
requests: z.array(z.object({
|
||||
ordinal: z.number().int(), firstRow: z.number().int(), lastRow: z.number().int(),
|
||||
route: z.object({ provider: z.string(), model: z.string(), options: z.record(z.string(), z.union([z.string(), z.number(), z.boolean()])) }),
|
||||
usage: z.object({ inputTokens: z.number().int(), outputTokens: z.number().int(), cache: z.object({ hit: z.number().int(), miss: z.number().int() }) }),
|
||||
references: z.array(z.object({ ordinal: z.number().int(), weight: z.number() })),
|
||||
})),
|
||||
totals: z.object({ rows: z.number().int(), requests: z.number().int(), tokens: z.number().int() }),
|
||||
})
|
||||
const viewSchema = z.object({ sessionOrdinal: z.number().int(), rows: z.number().int(), requests: z.number().int(), tokens: z.number().int() })
|
||||
type ProjectionState = z.infer<typeof stateSchema>
|
||||
|
||||
declare module '@deepseek-ai/dsh-session-projection/types' {
|
||||
interface SessionProjectionStateMap { syntheticProjectionList: ProjectionState }
|
||||
interface SessionProjectionMap { syntheticProjectionList: z.infer<typeof viewSchema> }
|
||||
}
|
||||
|
||||
interface Memory {
|
||||
readonly heapUsedBytes: number
|
||||
readonly rssBytes: number
|
||||
/** Process lifetime peak, including fixture creation and Host mounting. */
|
||||
readonly peakRssBytes: number
|
||||
}
|
||||
|
||||
interface ListSample {
|
||||
readonly listMs: number
|
||||
readonly listAndJsonMs: number
|
||||
readonly listCpuMs: number
|
||||
readonly listAndJsonCpuMs: number
|
||||
readonly callbackDelayMs: number
|
||||
readonly maxCallbackDelayMs: number
|
||||
readonly maxViewsPerBatch: number
|
||||
readonly yieldCalls: number
|
||||
readonly eventLoopMaxMs: number
|
||||
readonly items: number
|
||||
readonly wireViews: number
|
||||
readonly responseJsonBytes: number
|
||||
readonly retainedHeapDeltaBytes: number
|
||||
readonly memory: Memory
|
||||
}
|
||||
|
||||
/** One fresh Host's fixture dimensions, retained memory, and first plus three repeated list samples. */
|
||||
export interface ProjectionListReport {
|
||||
readonly workload: keyof typeof WORKLOADS
|
||||
readonly fixture: { sessions: number; baseRows: number; totalRows: number; totalRequests: number; stateJsonBytes: number }
|
||||
readonly baselineMemory: Memory
|
||||
readonly samples: readonly ListSample[]
|
||||
readonly responsiveness: ListSample
|
||||
}
|
||||
|
||||
interface ViewObservation {
|
||||
calls: number
|
||||
onView: (() => void) | undefined
|
||||
}
|
||||
|
||||
function headerOf(ordinal: number): SessionHeader {
|
||||
return { version: SESSION_FORMAT_VERSION, id: SessionId(`projection-bench-${ordinal}`),
|
||||
createdAt: BASE_TIME + ordinal, cwd: CWD, isSeeded: false, delegationDepth: 0 }
|
||||
}
|
||||
|
||||
function makeState(sessionOrdinal: number, baseRows: number): ProjectionState {
|
||||
const count = Math.floor(baseRows * ROW_PERCENTAGES[sessionOrdinal % ROW_PERCENTAGES.length]! / 100)
|
||||
const events: ProjectionState['events'] = Array.from({ length: count }, (_, ordinal) => ({
|
||||
ordinal, category: (['input', 'output', 'tool'] as const)[ordinal % 3]!,
|
||||
metrics: { tokens: 32 + ordinal % 97, bytes: 128 + ordinal % 257, elapsedMs: ordinal % 41 },
|
||||
content: [
|
||||
{ kind: 'text', text: `Generated observation ${sessionOrdinal}:${ordinal}. ` + 'Synthetic text. '.repeat(4 + ordinal % 4) },
|
||||
{ kind: 'metadata', text: `bucket-${ordinal % 11}` },
|
||||
],
|
||||
tags: [`group-${sessionOrdinal % 7}`, `kind-${ordinal % 3}`],
|
||||
}))
|
||||
const requests = Array.from({ length: Math.ceil(count / 8) }, (_, ordinal) => ({
|
||||
ordinal, firstRow: ordinal * 8, lastRow: Math.min(count - 1, ordinal * 8 + 7),
|
||||
route: { provider: 'synthetic', model: 'generated-model', options: { temperature: 0, streaming: true, tier: 'fixed' } },
|
||||
usage: { inputTokens: 100 + ordinal % 500, outputTokens: 30 + ordinal % 200, cache: { hit: ordinal % 100, miss: 20 } },
|
||||
references: Array.from({ length: Math.min(8, count - ordinal * 8) }, (_, index) => ({ ordinal: ordinal * 8 + index, weight: (index + 1) / 8 })),
|
||||
}))
|
||||
return { sessionOrdinal, events, requests,
|
||||
totals: { rows: count, requests: requests.length, tokens: events.reduce((sum, event) => sum + event.metrics.tokens, 0) } }
|
||||
}
|
||||
|
||||
async function mountStorage(ctx: Context, root: string): Promise<void> {
|
||||
await ctx.plugin(Storage)
|
||||
await ctx.plugin(StorageJson, { root: join(root, 'storages') })
|
||||
await ctx.plugin(StorageDomain, { backend: 'json' })
|
||||
}
|
||||
|
||||
async function seed(root: string, workload: typeof WORKLOADS[keyof typeof WORKLOADS]): Promise<ProjectionListReport['fixture']> {
|
||||
const ctx = new Context()
|
||||
const fixture = { ...workload, totalRows: 0, totalRequests: 0, stateJsonBytes: 0 }
|
||||
try {
|
||||
if (workload.live) {
|
||||
for (let ordinal = 0; ordinal < workload.sessions; ordinal++) {
|
||||
fixture.stateJsonBytes += Buffer.byteLength(JSON.stringify(makeState(ordinal, workload.baseRows)))
|
||||
}
|
||||
return fixture
|
||||
}
|
||||
await mountStorage(ctx, root)
|
||||
const domain = await ctx.storageDomain.open(projectionCacheDomainSpec)
|
||||
try {
|
||||
const table = domain.table('sessions')
|
||||
for (let ordinal = 0; ordinal < workload.sessions; ordinal++) {
|
||||
const header = headerOf(ordinal)
|
||||
const directory = sessionDir(join(root, 'sessions'), header.cwd, header.id)
|
||||
await mkdir(directory, { recursive: true })
|
||||
await writeFile(generationLogPath(join(root, 'sessions'), header.cwd, header.id, SESSION_FORMAT_VERSION, 'zstd'),
|
||||
await compressZstdFrame(JSON.stringify(toHeaderLine(header)) + '\n'), { flag: 'wx' })
|
||||
const state = makeState(ordinal, workload.baseRows)
|
||||
await table.put(header.id, {
|
||||
identity: { formatVersion: header.version, createdAt: header.createdAt, cwd: header.cwd, isSeeded: false, inheritedEventCount: SessionLogOffset(0) },
|
||||
rows: {
|
||||
[KEY]: { ver: 1, seq: -1, val: state },
|
||||
sessionListMetadata: { ver: 1, seq: -1, val: { blank: true, lastPromptAt: null } },
|
||||
},
|
||||
})
|
||||
fixture.totalRows += state.events.length
|
||||
fixture.totalRequests += state.requests.length
|
||||
fixture.stateJsonBytes += Buffer.byteLength(JSON.stringify(state))
|
||||
}
|
||||
} finally {
|
||||
await domain.close()
|
||||
}
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
return fixture
|
||||
}
|
||||
|
||||
async function mountHost(ctx: Context, root: string, workload: typeof WORKLOADS[keyof typeof WORKLOADS], observation: ViewObservation, listWorkSliceMs?: number): Promise<SessionController> {
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
await ctx.plugin(SessionProjectionRegistry)
|
||||
ctx.sessionProjections.register({
|
||||
key: KEY, stateVersion: 1, stateSchema,
|
||||
init: header => makeState(Number(header.id.slice('projection-bench-'.length)), workload.baseRows),
|
||||
apply: state => state,
|
||||
wire: {
|
||||
viewSchema,
|
||||
view(state) {
|
||||
observation.calls++
|
||||
observation.onView?.()
|
||||
return { sessionOrdinal: state.sessionOrdinal, ...state.totals }
|
||||
},
|
||||
},
|
||||
})
|
||||
if (workload.live) {
|
||||
for (let ordinal = 0; ordinal < workload.sessions; ordinal++) {
|
||||
const header = headerOf(ordinal)
|
||||
ctx.sessions.create(header.id, { meta: header })
|
||||
}
|
||||
}
|
||||
await ctx.plugin(JsonlSessionPersistence, { root: join(root, 'sessions'), compression: 'zstd' })
|
||||
await mountStorage(ctx, root)
|
||||
await ctx.plugin(SessionProjectionCache, { writeEveryEvents: 200, writeIntervalMs: 5_000 })
|
||||
await ctx.plugin(SqliteSessionQueryEngine, { path: ':memory:', openAt: 'never' })
|
||||
// No Agents, uploads, model calls, or workspace actions occur on this list path.
|
||||
const dispose = (): void => {}
|
||||
ctx.provide('typert', { lookups: { configure: () => dispose }, contexts: { configureHost: () => dispose } } as never)
|
||||
ctx.provide('fileUploads', { registerAgentResolver: () => dispose } as never)
|
||||
ctx.provide('agentDefaultModel', { currentSelection: () => ({ provider: 'synthetic', model: 'generated-model' }) } as never)
|
||||
ctx.provide('workspaceRegistry', { list: () => [], archivedSessionIds: [] } as never)
|
||||
return new SessionController(ctx, { nativeOpen: false, ...(listWorkSliceMs === undefined ? {} : { listWorkSliceMs }) }, { canOpenPath: () => false })
|
||||
}
|
||||
|
||||
async function memory(): Promise<Memory> {
|
||||
const gc = (globalThis as typeof globalThis & { gc?: () => void }).gc
|
||||
assert.ok(gc, 'projection-list worker requires --expose-gc')
|
||||
gc()
|
||||
await scheduler.yield()
|
||||
gc()
|
||||
const usage = process.memoryUsage()
|
||||
return { heapUsedBytes: usage.heapUsed, rssBytes: usage.rss, peakRssBytes: process.resourceUsage().maxRSS * 1024 }
|
||||
}
|
||||
|
||||
async function measure(controller: SessionController, observation: ViewObservation, sessions: number, baseline: Memory, fullProbe = false): Promise<ListSample> {
|
||||
const histogram = monitorEventLoopDelay({ resolution: 1 })
|
||||
let immediate: NodeJS.Immediate | undefined
|
||||
const initialViews = observation.calls
|
||||
const yieldDescriptor = Object.getOwnPropertyDescriptor(scheduler, 'yield')
|
||||
const yieldWork = scheduler.yield.bind(scheduler)
|
||||
let yieldCalls = 0
|
||||
Object.defineProperty(scheduler, 'yield', {
|
||||
configurable: true,
|
||||
value: () => { yieldCalls++; return yieldWork() },
|
||||
})
|
||||
let queued: Promise<void> | undefined
|
||||
let callbackDelayMs = 0
|
||||
let maxCallbackDelayMs = 0
|
||||
let maxViewsPerBatch = 0
|
||||
observation.onView = () => {
|
||||
if (immediate !== undefined) return
|
||||
if (!fullProbe) observation.onView = undefined
|
||||
const beforeView = observation.calls - 1
|
||||
const start = performance.now()
|
||||
queued = new Promise<void>((resolve) => {
|
||||
immediate = setImmediate(() => {
|
||||
const elapsed = performance.now() - start
|
||||
if (maxViewsPerBatch === 0) callbackDelayMs = elapsed
|
||||
maxCallbackDelayMs = Math.max(maxCallbackDelayMs, elapsed)
|
||||
maxViewsPerBatch = Math.max(maxViewsPerBatch, observation.calls - beforeView)
|
||||
immediate = undefined
|
||||
resolve()
|
||||
})
|
||||
})
|
||||
}
|
||||
histogram.enable()
|
||||
try {
|
||||
// Arm the histogram before work, and keep it enabled through the final blocking segment.
|
||||
await delay(2)
|
||||
histogram.reset()
|
||||
const cpu = process.cpuUsage()
|
||||
const start = performance.now()
|
||||
const result = await controller.list({}, new AbortController().signal)
|
||||
const listDone = performance.now()
|
||||
const listYieldCalls = yieldCalls
|
||||
const listCpu = process.cpuUsage(cpu)
|
||||
const json = JSON.stringify(result)
|
||||
const jsonDone = performance.now()
|
||||
const jsonCpu = process.cpuUsage(cpu)
|
||||
assert.equal(result.items.length, sessions)
|
||||
assert.equal(observation.calls - initialViews, sessions)
|
||||
assert.ok(result.items.every(item => item.projections?.values[KEY] !== undefined))
|
||||
// No more views can arrive after list completion; drain the final armed batch.
|
||||
await queued
|
||||
await delay(2)
|
||||
const eventLoopMaxMs = histogram.max / 1_000_000
|
||||
histogram.disable()
|
||||
const endpoint = await memory()
|
||||
// Both endpoint values remain reachable while GC measures the retained heap.
|
||||
assert.ok(json.length > result.items.length)
|
||||
return { listMs: listDone - start, listAndJsonMs: jsonDone - start,
|
||||
listCpuMs: (listCpu.user + listCpu.system) / 1000, listAndJsonCpuMs: (jsonCpu.user + jsonCpu.system) / 1000,
|
||||
callbackDelayMs, maxCallbackDelayMs, maxViewsPerBatch, yieldCalls: listYieldCalls, eventLoopMaxMs, items: result.items.length, wireViews: observation.calls - initialViews,
|
||||
responseJsonBytes: Buffer.byteLength(json), retainedHeapDeltaBytes: endpoint.heapUsedBytes - baseline.heapUsedBytes, memory: endpoint }
|
||||
} finally {
|
||||
observation.onView = undefined
|
||||
if (immediate !== undefined) clearImmediate(immediate)
|
||||
histogram.disable()
|
||||
if (yieldDescriptor === undefined) Reflect.deleteProperty(scheduler, 'yield')
|
||||
else Object.defineProperty(scheduler, 'yield', yieldDescriptor)
|
||||
}
|
||||
}
|
||||
|
||||
assertBuiltBenchmarkRuntime(import.meta.url, {
|
||||
controller: import.meta.resolve('@deepseek-ai/dsh-api-session-controller'),
|
||||
registry: import.meta.resolve('@deepseek-ai/dsh-session-projection'),
|
||||
cache: import.meta.resolve('@deepseek-ai/dsh-session-projection-cache'),
|
||||
persistence: import.meta.resolve('@deepseek-ai/dsh-session-persistence-jsonl'),
|
||||
})
|
||||
const [root, workload, sliceArgument] = process.argv.slice(2)
|
||||
if (root === undefined || (workload !== 'modest' && workload !== 'tail' && workload !== 'cheap')) throw new Error('usage: projection-list.worker.js <private-root> <modest|tail|cheap> [work-slice-ms]')
|
||||
const listWorkSliceMs = sliceArgument === undefined ? undefined : Number(sliceArgument)
|
||||
if (listWorkSliceMs !== undefined && (!Number.isSafeInteger(listWorkSliceMs) || listWorkSliceMs < 1)) throw new Error('work-slice-ms must be a positive integer')
|
||||
const ctx = new Context()
|
||||
try {
|
||||
const fixture = await seed(root, WORKLOADS[workload])
|
||||
const observation: ViewObservation = { calls: 0, onView: undefined }
|
||||
const controller = await mountHost(ctx, root, WORKLOADS[workload], observation, listWorkSliceMs)
|
||||
assert.equal(observation.calls, 0, 'setup must not warm wire views')
|
||||
const baselineMemory = await memory()
|
||||
const samples: ListSample[] = []
|
||||
for (let call = 0; call < 4; call++) samples.push(await measure(controller, observation, fixture.sessions, baselineMemory))
|
||||
const responsiveness = await measure(controller, observation, fixture.sessions, baselineMemory, true)
|
||||
const report: ProjectionListReport = { workload, fixture, baselineMemory, samples, responsiveness }
|
||||
console.log(JSON.stringify(report))
|
||||
} finally {
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
@@ -49,7 +49,10 @@ export default defineConfig([
|
||||
},
|
||||
{
|
||||
...shared,
|
||||
entry: { 'session-corpus.worker': 'session-corpus/session-corpus.worker.ts' },
|
||||
entry: {
|
||||
'session-corpus.worker': 'session-corpus/session-corpus.worker.ts',
|
||||
'projection-list.worker': 'session-corpus/projection-list.worker.ts',
|
||||
},
|
||||
outDir: '.dsh-build/session-corpus',
|
||||
clean: true,
|
||||
tsconfig: 'tsconfig.host.json',
|
||||
|
||||
@@ -229,6 +229,8 @@ export interface Config {
|
||||
export interface Config {
|
||||
/** Override platform desktop-opener detection. */
|
||||
readonly nativeOpen?: boolean
|
||||
/** Positive integral milliseconds of list work before yielding between complete rows. */
|
||||
readonly listWorkSliceMs?: number
|
||||
}
|
||||
```
|
||||
<!-- END GENERATED config-catalog:@deepseek-ai/dsh-api-session-controller -->
|
||||
|
||||
@@ -231,6 +231,8 @@ export interface Config {
|
||||
export interface Config {
|
||||
/** Override platform desktop-opener detection. */
|
||||
readonly nativeOpen?: boolean
|
||||
/** Positive integral milliseconds of list work before yielding between complete rows. */
|
||||
readonly listWorkSliceMs?: number
|
||||
}
|
||||
```
|
||||
<!-- END GENERATED config-catalog:@deepseek-ai/dsh-api-session-controller -->
|
||||
|
||||
@@ -799,7 +799,7 @@ inspect( sessionId: SessionId, signal?: AbortSignal, ): Promise<SessionInspectio
|
||||
/**
|
||||
* Read all visible Session rows without resuming an Agent.
|
||||
* @param _request - reserved empty list request.
|
||||
* @param signal - cancellation for persistence reads.
|
||||
* @param signal - cancellation for persistence reads and summary generation.
|
||||
* @returns visible Session summaries ordered by activity.
|
||||
*/
|
||||
@Remote('list') async list(_request: SessionListRequest, signal: AbortSignal): Promise<SessionListValue>
|
||||
|
||||
@@ -803,7 +803,7 @@ inspect( sessionId: SessionId, signal?: AbortSignal, ): Promise<SessionInspectio
|
||||
/**
|
||||
* Read all visible Session rows without resuming an Agent.
|
||||
* @param _request - reserved empty list request.
|
||||
* @param signal - cancellation for persistence reads.
|
||||
* @param signal - cancellation for persistence reads and summary generation.
|
||||
* @returns visible Session summaries ordered by activity.
|
||||
*/
|
||||
@Remote('list') async list(_request: SessionListRequest, signal: AbortSignal): Promise<SessionListValue>
|
||||
|
||||
@@ -15,8 +15,8 @@
|
||||
en: 1fdf707322fb6996
|
||||
zh: 0204f7e9d4aa3cce
|
||||
/session-controller/use-this-package:
|
||||
en: b1a66e01de414924
|
||||
zh: 7c01f66b94b01ff9
|
||||
en: 77552723167e5d8e
|
||||
zh: 1d77a657c707190a
|
||||
/session-controller/client-references:
|
||||
en: 41bdda1ac32df558
|
||||
zh: aa5d0e325924066b
|
||||
@@ -24,8 +24,8 @@
|
||||
en: 266ab2b1e2fec542
|
||||
zh: b3450ae28a1e0baf
|
||||
/session-controller/configuration:
|
||||
en: 5c4c4224074e38a8
|
||||
zh: e0b0edac65c88ff4
|
||||
en: 5ae2b340c630897f
|
||||
zh: 1ffbd61ecf302ec2
|
||||
/session-controller/model-experience:
|
||||
en: 30b917bf822bb4df
|
||||
zh: b0b023b5b526c317
|
||||
@@ -33,8 +33,8 @@
|
||||
en: 22fc0c9dbedd3bb6
|
||||
zh: fad040ca55fd0e0a
|
||||
/session-controller/known-limitations-and-deferred-work:
|
||||
en: e896bfcaf38b4f6c
|
||||
zh: 55a4eba838f166de
|
||||
en: 5db0f9f69a8be926
|
||||
zh: 6ba1cc4a3850fcbc
|
||||
/session-controller/known-limitations-and-deferred-work/dev-note:
|
||||
en: 68e0dfe4588c5b88
|
||||
zh: db51b435611bded1
|
||||
|
||||
@@ -33,6 +33,8 @@ The Client journal validates current Session event envelopes before publishing f
|
||||
|
||||
Each endpoint states its activation policy. List reads only stored headers and projection-cache rows: it never calls per-session stat or opens a cold Session body. A current-format cache identity may supply every list hint; a lifecycle-matching predecessor cache may supply only its version-compatible title as a stale display fact, never as an authoritative fold seed. Search, attachment, history pages, log following, skill discovery, and workspace-path opening can inspect persistence without activating an Agent; `canOpenWorkspacePath()` reports native-opening availability without addressing a Session. Cancellation requires live state; queue mutation, model, rename, prompt, and file-reference operations may resolve or resume an ordinary Session. Prompt rejects content with neither non-whitespace text nor an attachment before resolving the Agent or appending Session events; queue edits accept only non-empty text content. Prompt admission consumes opaque receipts from the injected [`fileUploads`](../../client/file-upload/README.md) Host service and resolves every same-Agent receipt before sending the complete ordered content list through `ctx.attachments`. Prompt retries whose `requestId` is already queued or logged return the original acceptance without inserting another message. Create and fork are the only operations that create a new Agent directly. The service applies one preset-aware resume policy and subagent ownership fence to its own methods and to the Typert Agent and Session lookups used by other Remote namespaces. Queue mutation has one narrow exception: a live child whose current projected identity is continuable and comes from its own non-seed suffix accepts the ordinary Edit, Remove, and QueueDock Steer actions across both inbox destinations. One-shot, missing, unknown, corrupt, seed-only, or cold children remain rejected without resume. The skill catalog uses a live Agent when present or the recorded preset's standing scope when cold, so listing never starts an Agent. The authenticated delivery routes use `workspaceDesktop()` for the serving Host name and file-manager behavior. `openWorkspacePath({ path, action: "reveal" })` delegates file-manager navigation to the native adapter; omitting `action` opens the file-type association, including HTML and SVG. Both operations require the composed filesystem to map the requested Host path to the same canonical process path; unmapped remote paths are refused before a native command runs. `session.projections` reads one complete baseline through a live-preferred Session observation without activating an Agent. It returns null for a missing Session and serves any registered projection keys. The Client exposes shared values and explicit-read state through `projectionsBySession`; domains select their own keys. Session-list summaries carry `agentAvailable`, updated through existing summary and status events independently of durable projections. Initial reads and live projection frames use the same sequence ordering.
|
||||
|
||||
Host list generation yields between complete rows only when its configured work slice is exhausted; cheap remaining work returns without a forced yield. Request cancellation is checked per row and after each yield and rejects without partial results. Each row is synchronous; the list does not promise one cross-Session snapshot. A Session already queued as cold remains a cached row if it attaches during a yield: Client in-flight mutation replay corrects its availability, and sequenced projections supersede cached hints.
|
||||
|
||||
Client list rows and resident Sessions use the current `sessionListMetadata` projection to reject stale blank-session hints; recency is the later of the summary timestamp and the projected last user prompt. When SessionManager creates an instance before its list row arrives, it reconciles blankness with metadata already retained for that Session. An opened conversation is therefore not reused by New Session even when an older list response still marks it blank.
|
||||
|
||||
Explicit-id `session.create` adopts a live Session or resumes a persisted Session while retaining its writer lock. Writer contention returns `session/writer-held`; callers may try another blank without suppressing unrelated failures. `session.list` includes persisted blanks using cached metadata, without opening cold log bodies.
|
||||
@@ -92,6 +94,7 @@ A successful `selectModel` response acknowledges the Session-local selection wit
|
||||
| Field | Default | Meaning |
|
||||
|---|---:|---|
|
||||
| `nativeOpen` | platform-detected | Whether Session workspace paths can be handed to a native desktop opener |
|
||||
| `listWorkSliceMs` | `16` | Positive integral list-work time slice in milliseconds; checks occur between complete rows |
|
||||
|
||||
The generated [configuration catalog](../../../docs/config-catalog.md#deepseek-aidsh-api-session-controller) is the exhaustive source for accepted fields and their JSDoc.
|
||||
|
||||
@@ -110,6 +113,7 @@ No direct effect; model requests remain owned by the Agent and LLM packages.
|
||||
|
||||
<a id="known-limitations-and-deferred-work"></a>
|
||||
|
||||
- A single summary, the query provider's own enumeration, final sorting, and response serialization can exceed the list work-slice target; it is not a hard Host latency bound.
|
||||
- The image byte cap does not validate decoded dimensions or pixel count.
|
||||
- A failed follow resumption remains visible to the caller instead of retrying indefinitely.
|
||||
- The raw browser upload is one streaming HTTP request without resumable offsets; a retry sends the file again from byte zero.
|
||||
|
||||
@@ -33,6 +33,8 @@ Client journal 在发布 follow 快照、live entry 或历史页之前验证当
|
||||
|
||||
每个 endpoint 都声明自己的激活策略。列表只读取持久化 header 与 projection cache row，绝不调用逐 Session stat 或打开冷 Session body。当前格式 cache identity 可以提供全部列表 hint；生命周期匹配的 predecessor cache 只能提供版本兼容的 title，作为可能过时的展示事实，绝不能作为权威 fold seed。搜索、附件、历史页、日志跟随、skill 发现和工作区路径打开可以在不激活 Agent 的情况下检查 persistence；`canOpenWorkspacePath()` 无需指定 Session 即可报告原生打开能力。取消要求 live 状态；queue 变更、模型、重命名、prompt 和文件引用操作可以解析或恢复普通 Session。提示词会在解析 Agent 或追加 Session 事件前，拒绝既没有非空白文本也没有附件的 content；queue edit 只接受非空文本 content。prompt 准入从注入的 [`fileUploads`](../../client/file-upload/README.zh.md) Host 服务取得不透明凭证，在把完整有序内容列表交给 `ctx.attachments` 前解析每个属于同一 Agent 的凭证。`requestId` 已进入 queue 或日志时，prompt 重试直接返回原来的接受结果，不会重复插入消息。只有 create 与 fork 会直接创建新 Agent。该服务把同一套感知 preset 的恢复策略和 subagent ownership fence 同时用于自身方法，以及其他 Remote namespace 使用的 Typert Agent 与 Session lookup。Queue 变更只有一个狭窄例外：当前 projection identity 为 continuable 且来自自身非 seed suffix 的在线 child，可以在两个 inbox 目标上使用普通 Edit、Remove 与 QueueDock Steer action。One-shot、缺失、未知、损坏、仅含 seed identity 或冷 child 继续被拒绝，且不会恢复。skill 目录优先使用已有 live Agent，否则使用所记录 preset 的常驻 scope，因此列表查询绝不会启动 Agent。经过鉴权的文件交付路由通过 `workspaceDesktop()` 获取提供服务的 Host 名称和文件管理器行为。`openWorkspacePath({ path, action: "reveal" })` 将文件管理器导航委托给原生适配器；省略 `action` 时按文件类型关联打开，包括 HTML 和 SVG。两种操作都要求当前文件系统将请求的 Host 路径映射到同一个规范进程路径；无法映射的远端路径会在执行原生命令之前被拒绝。 `session.projections` 通过一次 live-preferred Session observation 读取完整基线，不激活 Agent。Session 不存在时返回 null，并可提供任意已注册的 projection key。Client 通过 `projectionsBySession` 暴露共享值和显式读取状态，由领域选择自身的 key。Session 列表摘要携带 `agentAvailable`，通过已有摘要与状态事件更新，与持久化 projection 相互独立。初始读取与实时 projection 帧使用相同的序号排序规则。
|
||||
|
||||
Host 仅在配置的工作时间片耗尽时，于完整条目之间让出事件循环；剩余工作较少时直接返回，不强制让出。每行及每次让出之后检查请求取消，取消时拒绝返回部分结果。每行同步计算；列表不承诺跨 Session 的统一快照。已经进入冷态队列的 Session 若在让出期间接入，仍返回缓存行：Client 对进行中请求的变更重放会纠正其可用性，带序号的投影会覆盖缓存提示。
|
||||
|
||||
Client 列表行和驻留 Session 使用当前 `sessionListMetadata` 投影纠正过期的空白会话提示；最近活动时间取摘要时间戳与投影中最后一次用户提示词时间的较晚值。当 SessionManager 在列表行到达前创建实例时，会使用已保留的该 Session 元数据对账空白状态。因此，即使旧列表响应仍将已有对话标为空白，新会话操作也不会复用已经打开过的对话。
|
||||
|
||||
显式 ID 的 `session.create` 会收养活动 Session，或恢复持久化 Session 并持续持有其写锁。写锁争用返回 `session/writer-held`；调用方可以尝试其他空白会话，同时保留其他失败。`session.list` 根据缓存元数据列出持久化空白会话，不打开冷日志正文。
|
||||
@@ -92,6 +94,7 @@ GUI 模型选择要求确切提供方／模型对出现在可用目录中；不
|
||||
| 字段 | 默认值 | 含义 |
|
||||
|---|---:|---|
|
||||
| `nativeOpen` | 平台探测 | 是否能把 Session 工作区路径交给原生桌面打开器 |
|
||||
| `listWorkSliceMs` | `16` | 列表工作时间片，单位为毫秒，取正整数；在完整条目之间检查 |
|
||||
|
||||
生成的[配置目录](../../../docs/config-catalog.zh.md#deepseek-aidsh-api-session-controller)是所有受支持字段及其 JSDoc 的完整来源。
|
||||
|
||||
@@ -110,6 +113,7 @@ GUI 模型选择要求确切提供方／模型对出现在可用目录中；不
|
||||
|
||||
<a id="known-limitations-and-deferred-work"></a>
|
||||
|
||||
- 单个摘要、查询提供方自身的枚举、最终排序和响应序列化都可能超出列表时间片目标；它不是 Host 延迟的硬上限。
|
||||
- 图片字节上限不校验解码后的尺寸或像素数。
|
||||
- follow 恢复失败会对调用方可见，而不会无限重试。
|
||||
- 浏览器原始字节上传使用一次不带断点续传偏移的流式 HTTP 请求；重试会从第零字节重新传输整个文件。
|
||||
|
||||
@@ -79,8 +79,13 @@ declare module '@deepseek-ai/cordis' {
|
||||
export interface Config {
|
||||
/** Override platform desktop-opener detection. */
|
||||
readonly nativeOpen?: boolean
|
||||
/** Positive integral milliseconds of list work before yielding between complete rows. */
|
||||
readonly listWorkSliceMs?: number
|
||||
}
|
||||
|
||||
/** Deployment policy after schema defaults have been applied. */
|
||||
type ResolvedConfig = Config & { readonly listWorkSliceMs: number }
|
||||
|
||||
/** Host integrations replaceable by direct unit tests. */
|
||||
export interface SessionControllerInternals {
|
||||
/** Native default-application handoff. */
|
||||
@@ -111,8 +116,9 @@ export class SessionController extends TypertRemoteService {
|
||||
'workspaceRegistry',
|
||||
]
|
||||
|
||||
static Config: z<Config> = z.object({
|
||||
static Config: z<Config, ResolvedConfig> = z.object({
|
||||
nativeOpen: z.boolean(),
|
||||
listWorkSliceMs: z.natural().min(1).default(16),
|
||||
})
|
||||
|
||||
private readonly agents: ApiSessionAgentController
|
||||
@@ -129,11 +135,12 @@ export class SessionController extends TypertRemoteService {
|
||||
|
||||
/**
|
||||
* @param ctx - Host context containing the Session capability assembly.
|
||||
* @param config - native-opener deployment policy.
|
||||
* @param config - native-opener and list-scheduling deployment policy.
|
||||
* @param internals - host integrations replaceable by direct unit tests.
|
||||
*/
|
||||
constructor(ctx: Context, config: Config, internals: SessionControllerInternals = {}) {
|
||||
super(ctx, 'sessionController', { namespace: 'session' })
|
||||
const resolved = SessionController.Config(config)
|
||||
installModelSelectionProjection(ctx)
|
||||
this.agents = new ApiSessionAgentController(ctx)
|
||||
this.commands = new SessionCommandController(ctx, this.agents, process.cwd())
|
||||
@@ -149,7 +156,7 @@ export class SessionController extends TypertRemoteService {
|
||||
await Promise.allSettled([...this.promotions])
|
||||
}, 'session-controller.promotions')
|
||||
this.history = new SessionHistoryController(ctx, (observation) => { this.promote(observation) })
|
||||
this.listState = new ApiSessionList(ctx)
|
||||
this.listState = new ApiSessionList(ctx, resolved.listWorkSliceMs)
|
||||
this.fileApplications = internals.fileApplications ?? nativeFileApplications
|
||||
this.openFileApplication = internals.openFileApplication ?? openNativeFileApplication
|
||||
this.openPath = internals.openPath ?? openNativeAssociatedPath
|
||||
@@ -245,7 +252,7 @@ export class SessionController extends TypertRemoteService {
|
||||
/**
|
||||
* Read all visible Session rows without resuming an Agent.
|
||||
* @param _request - reserved empty list request.
|
||||
* @param signal - cancellation for persistence reads.
|
||||
* @param signal - cancellation for persistence reads and summary generation.
|
||||
* @returns visible Session summaries ordered by activity.
|
||||
*/
|
||||
@Remote('list')
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
/** Cold-safe Session list and search projection. */
|
||||
|
||||
import { performance } from 'node:perf_hooks'
|
||||
import { scheduler } from 'node:timers/promises'
|
||||
import type { Context } from '@deepseek-ai/cordis'
|
||||
import type {} from '@deepseek-ai/dsh-agent-preset-registry'
|
||||
import type { ImageAttachmentLimits } from '@deepseek-ai/dsh-attachment'
|
||||
@@ -74,8 +76,11 @@ export function truncateUnicodeCodePoints(value: string, maximum: number): strin
|
||||
|
||||
/** Owns list projection registration, bounded cold summaries, and authorized search. */
|
||||
export class ApiSessionList {
|
||||
/** @param ctx - Host context carrying Session, query, persistence, and projection services. */
|
||||
constructor(private readonly ctx: Context) {
|
||||
/**
|
||||
* @param ctx - Host context carrying Session, query, persistence, and projection services.
|
||||
* @param workSliceMs - Resolved positive integral list-work budget in milliseconds.
|
||||
*/
|
||||
constructor(private readonly ctx: Context, private readonly workSliceMs: number) {
|
||||
ctx.sessionProjections.register<'sessionListMetadata', SessionListMetadata>({
|
||||
key: 'sessionListMetadata',
|
||||
stateSchema: sessionListMetadataSchema,
|
||||
@@ -120,7 +125,7 @@ export class ApiSessionList {
|
||||
|
||||
/**
|
||||
* Read every visible attached and persisted Session without activating an Agent.
|
||||
* @param signal - optional cancellation for persistence reads.
|
||||
* @param signal - optional cancellation for persistence reads and summary generation.
|
||||
* @returns visible Session summaries ordered by activity.
|
||||
*/
|
||||
async list(signal?: AbortSignal): Promise<SessionSummary[]> {
|
||||
@@ -129,16 +134,31 @@ export class ApiSessionList {
|
||||
signal?.throwIfAborted()
|
||||
const items: SessionSummary[] = []
|
||||
const cold: SessionHeader[] = []
|
||||
let yieldDeadline = performance.now() + this.workSliceMs
|
||||
for (const record of records) {
|
||||
signal?.throwIfAborted()
|
||||
const live = this.ctx.sessions.get(record.header.id)
|
||||
if (live !== undefined) {
|
||||
items.push(this.summaryFor(live))
|
||||
continue
|
||||
} else if (record.header.cwd !== undefined) {
|
||||
cold.push(record.header)
|
||||
}
|
||||
if (performance.now() >= yieldDeadline) {
|
||||
await scheduler.yield()
|
||||
signal?.throwIfAborted()
|
||||
yieldDeadline = performance.now() + this.workSliceMs
|
||||
}
|
||||
if (record.header.cwd === undefined) continue
|
||||
cold.push(record.header)
|
||||
}
|
||||
for (const header of cold) items.push(this.summarizeCold(header))
|
||||
for (const header of cold) {
|
||||
signal?.throwIfAborted()
|
||||
items.push(this.summarizeCold(header))
|
||||
if (performance.now() >= yieldDeadline) {
|
||||
await scheduler.yield()
|
||||
signal?.throwIfAborted()
|
||||
yieldDeadline = performance.now() + this.workSliceMs
|
||||
}
|
||||
}
|
||||
signal?.throwIfAborted()
|
||||
items.sort((left, right) => right.updatedAt - left.updatedAt)
|
||||
return items
|
||||
}
|
||||
|
||||
@@ -0,0 +1,338 @@
|
||||
/** Event-loop fairness and cancellation of synchronous Session-list summaries. */
|
||||
|
||||
import { Context } from '@deepseek-ai/cordis'
|
||||
import AgentRegistry, { type Agent } from '@deepseek-ai/dsh-agent'
|
||||
import SessionStore, { SessionId, type Session } from '@deepseek-ai/dsh-session'
|
||||
import SessionProjectionRegistry from '@deepseek-ai/dsh-session-projection'
|
||||
import type { SessionRecord } from '@deepseek-ai/dsh-session-query'
|
||||
import { resolve } from 'node:path'
|
||||
import { performance } from 'node:perf_hooks'
|
||||
import { scheduler } from 'node:timers/promises'
|
||||
import SessionController from '../src/index.ts'
|
||||
import { afterEach, describe, expect, it, vi } from 'vitest'
|
||||
import { ApiSessionList } from '../src/list.ts'
|
||||
|
||||
type Phase = 'live' | 'cold'
|
||||
interface Row {
|
||||
id: string
|
||||
phase: Phase
|
||||
cwd?: false
|
||||
}
|
||||
|
||||
const contexts = new Set<Context>()
|
||||
const immediates = new Set<ReturnType<typeof setImmediate>>()
|
||||
|
||||
afterEach(async () => {
|
||||
for (const handle of immediates) clearImmediate(handle)
|
||||
immediates.clear()
|
||||
vi.restoreAllMocks()
|
||||
try {
|
||||
await Promise.all([...contexts].map(ctx => ctx.fiber.dispose()))
|
||||
} finally {
|
||||
contexts.clear()
|
||||
}
|
||||
})
|
||||
|
||||
function betweenRows(action: () => void): void {
|
||||
const handle = setImmediate(() => {
|
||||
immediates.delete(handle)
|
||||
action()
|
||||
})
|
||||
immediates.add(handle)
|
||||
}
|
||||
|
||||
async function harness(rows: readonly Row[], summaryMs = 8, workSliceMs = 8) {
|
||||
const ctx = new Context()
|
||||
contexts.add(ctx)
|
||||
await ctx.plugin(SessionStore)
|
||||
await ctx.plugin(SessionProjectionRegistry)
|
||||
await ctx.plugin(AgentRegistry)
|
||||
const list = new ApiSessionList(ctx, workSliceMs)
|
||||
const clock = { now: 0 }
|
||||
const prepared = new Map<string, Session>()
|
||||
const agents = new Map<string, { status: Agent['status'] }>()
|
||||
const detach = new Map<string, () => void>()
|
||||
const trace: string[] = []
|
||||
const observer: { summary: (phase: Phase, id: string) => void } = { summary: () => {} }
|
||||
const recordSummary = (phase: Phase, id: string): void => {
|
||||
clock.now += summaryMs
|
||||
trace.push(`${phase}:${id}`)
|
||||
observer.summary(phase, id)
|
||||
}
|
||||
const attach = (id: string): void => {
|
||||
const session = prepared.get(id)
|
||||
if (session === undefined) throw new Error(`missing prepared Session ${id}`)
|
||||
const detachSession = ctx.sessions.enter(session)
|
||||
ctx.effect(() => detachSession, 'list-scheduling.session')
|
||||
ctx.sessions.announce(session)
|
||||
const agent = { id: session.id, session, ctx, status: 'idle' as Agent['status'] }
|
||||
// Only registry lookup and status are consumed; no Agent execution occurs in this fixture.
|
||||
const detachAgent = ctx.agents.enter(agent as Agent, undefined)
|
||||
ctx.effect(() => detachAgent, 'list-scheduling.agent')
|
||||
agents.set(id, agent)
|
||||
detach.set(id, () => { detachAgent(); detachSession() })
|
||||
}
|
||||
const records: SessionRecord[] = rows.map((row) => {
|
||||
const session = ctx.sessions.prepare(SessionId(row.id), {
|
||||
meta: { createdAt: 100, ...(row.cwd === false ? {} : { cwd: resolve('list-scheduling-fixture') }) },
|
||||
})
|
||||
prepared.set(row.id, session)
|
||||
if (row.phase === 'live') attach(row.id)
|
||||
return { header: session.header, live: row.phase === 'live', persisted: true }
|
||||
})
|
||||
const listSessions = vi.fn(async (_signal?: AbortSignal) => records)
|
||||
ctx.provide('sessionQuery', { listSessions } as never)
|
||||
ctx.provide('sessionProjectionCache', {
|
||||
cachedSnapshot: (header: SessionRecord['header']) => {
|
||||
recordSummary('cold', header.id)
|
||||
return { asOfSeq: -1, values: { sessionListMetadata: { blank: false, lastPromptAt: null } } }
|
||||
},
|
||||
cachedPredecessorTitle: () => undefined,
|
||||
} as never)
|
||||
const summaryFor = list.summaryFor.bind(list)
|
||||
vi.spyOn(list, 'summaryFor').mockImplementation((session) => {
|
||||
const summary = summaryFor(session)
|
||||
recordSummary('live', session.id)
|
||||
return summary
|
||||
})
|
||||
vi.spyOn(performance, 'now').mockImplementation(() => clock.now)
|
||||
return { ctx, list, listSessions, trace, observer, attach, agents, detach, clock }
|
||||
}
|
||||
|
||||
describe('Session-list work slices', () => {
|
||||
it.each<Phase>(['live', 'cold'])('keeps cheap %s summaries in one synchronous slice', async (phase) => {
|
||||
const h = await harness([{ id: 'first', phase }, { id: 'middle', phase }, { id: 'last', phase }], 2)
|
||||
const yieldWork = vi.spyOn(scheduler, 'yield')
|
||||
h.observer.summary = (_phase, id) => {
|
||||
if (id !== 'first') return
|
||||
queueMicrotask(() => { h.trace.push('microtask') })
|
||||
betweenRows(() => { h.trace.push('immediate') })
|
||||
}
|
||||
|
||||
await h.list.list()
|
||||
expect(yieldWork).not.toHaveBeenCalled()
|
||||
expect(h.trace).toEqual([`${phase}:first`, `${phase}:middle`, `${phase}:last`, 'microtask'])
|
||||
await new Promise<void>((resolve) => { betweenRows(resolve) })
|
||||
expect(h.trace.at(-1)).toBe('immediate')
|
||||
})
|
||||
|
||||
it('accumulates work across both phases and resets the deadline after waiting', async () => {
|
||||
const h = await harness([{ id: 'first', phase: 'live' }, { id: 'second', phase: 'cold' }, { id: 'last', phase: 'cold' }], 4)
|
||||
const yieldWork = vi.spyOn(scheduler, 'yield')
|
||||
h.observer.summary = (_phase, id) => {
|
||||
if (id !== 'first') return
|
||||
betweenRows(() => {
|
||||
h.clock.now += 1_000
|
||||
h.trace.push('other-work')
|
||||
})
|
||||
}
|
||||
|
||||
await h.list.list()
|
||||
expect(h.trace).toEqual(['live:first', 'cold:second', 'other-work', 'cold:last'])
|
||||
expect(yieldWork).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('honors an overridden budget without yielding early', async () => {
|
||||
const h = await harness([{ id: 'first', phase: 'cold' }, { id: 'last', phase: 'cold' }], 10, 16)
|
||||
const yieldWork = vi.spyOn(scheduler, 'yield')
|
||||
h.observer.summary = (_phase, id) => {
|
||||
if (id === 'first') betweenRows(() => { h.trace.push('other-work') })
|
||||
}
|
||||
await h.list.list()
|
||||
expect(h.trace).toEqual(['cold:first', 'cold:last', 'other-work'])
|
||||
expect(yieldWork).toHaveBeenCalledTimes(1)
|
||||
})
|
||||
|
||||
it('includes skipped-record classification in the work budget', async () => {
|
||||
const h = await harness([
|
||||
{ id: 'first', phase: 'cold', cwd: false },
|
||||
{ id: 'second', phase: 'cold', cwd: false },
|
||||
{ id: 'last', phase: 'cold', cwd: false },
|
||||
], 0)
|
||||
const controller = new AbortController()
|
||||
const reason = new Error('stop classification')
|
||||
const get = h.ctx.sessions.get.bind(h.ctx.sessions)
|
||||
const lookup = vi.spyOn(h.ctx.sessions, 'get').mockImplementation((id) => {
|
||||
h.clock.now += 4
|
||||
if (id === SessionId('first')) betweenRows(() => { controller.abort(reason) })
|
||||
return get(id)
|
||||
})
|
||||
await expect(h.list.list(controller.signal)).rejects.toBe(reason)
|
||||
expect(lookup).toHaveBeenCalledTimes(2)
|
||||
expect(h.trace).toEqual([])
|
||||
})
|
||||
|
||||
it.each(['first', 'last'])('honors synchronous cancellation in the %s cheap row without yielding', async (abortAt) => {
|
||||
const h = await harness([{ id: 'first', phase: 'cold' }, { id: 'last', phase: 'cold' }], 0)
|
||||
const yieldWork = vi.spyOn(scheduler, 'yield')
|
||||
const controller = new AbortController()
|
||||
const reason = new Error('stop without waiting')
|
||||
h.observer.summary = (_phase, id) => { if (id === abortAt) controller.abort(reason) }
|
||||
await expect(h.list.list(controller.signal)).rejects.toBe(reason)
|
||||
expect(yieldWork).not.toHaveBeenCalled()
|
||||
expect(h.trace).toEqual(abortAt === 'first' ? ['cold:first'] : ['cold:first', 'cold:last'])
|
||||
})
|
||||
|
||||
it('resolves default and overridden budgets without changing native-opening input', () => {
|
||||
expect(SessionController.Config({}).listWorkSliceMs).toBe(16)
|
||||
expect(SessionController.Config({ nativeOpen: false, listWorkSliceMs: 32 })).toMatchObject({ nativeOpen: false, listWorkSliceMs: 32 })
|
||||
})
|
||||
|
||||
it('routes an overridden budget through the public Session Controller', async () => {
|
||||
const h = await harness([{ id: 'first', phase: 'cold' }, { id: 'middle', phase: 'cold' }, { id: 'last', phase: 'cold' }], 4)
|
||||
const dispose = (): void => {}
|
||||
h.ctx.provide('typert', { lookups: { configure: () => dispose }, contexts: { configureHost: () => dispose } } as never)
|
||||
h.ctx.provide('fileUploads', { registerAgentResolver: () => dispose } as never)
|
||||
h.ctx.provide('agentDefaultModel', { currentSelection: () => ({ provider: 'test', model: 'test' }) } as never)
|
||||
h.ctx.provide('workspaceRegistry', { list: () => [], archivedSessionIds: [] } as never)
|
||||
const controller = new SessionController(h.ctx, { nativeOpen: false, listWorkSliceMs: 4 }, { canOpenPath: () => false })
|
||||
const yieldWork = vi.spyOn(scheduler, 'yield')
|
||||
const result = await controller.list({}, new AbortController().signal)
|
||||
expect(result.items).toHaveLength(3)
|
||||
expect(yieldWork).toHaveBeenCalledTimes(3)
|
||||
})
|
||||
|
||||
it.each([0, -1, 0.5, Number.NaN, Number.POSITIVE_INFINITY])('rejects invalid work-slice budget %s', (listWorkSliceMs) => {
|
||||
expect(() => SessionController.Config({ listWorkSliceMs })).toThrow()
|
||||
})
|
||||
|
||||
it.each<Phase>(['live', 'cold'])('yields between %s summaries that each exhaust the work slice', async (phase) => {
|
||||
const h = await harness([{ id: 'first', phase }, { id: 'middle', phase }, { id: 'last', phase }])
|
||||
h.observer.summary = (_phase, id) => { betweenRows(() => { h.trace.push(`immediate:${id}`) }) }
|
||||
|
||||
const items = await h.list.list()
|
||||
h.trace.push('resolved')
|
||||
|
||||
expect(h.trace).toEqual([
|
||||
`${phase}:first`, 'immediate:first', `${phase}:middle`, 'immediate:middle',
|
||||
`${phase}:last`, 'immediate:last', 'resolved',
|
||||
])
|
||||
expect(items.map(item => item.sessionId)).toEqual(['first', 'middle', 'last'])
|
||||
})
|
||||
|
||||
it.each([
|
||||
{ phase: 'live', abortAt: 'first' },
|
||||
{ phase: 'live', abortAt: 'last' },
|
||||
{ phase: 'cold', abortAt: 'first' },
|
||||
{ phase: 'cold', abortAt: 'last' },
|
||||
] as const)('preserves cancellation after the $abortAt $phase row', async ({ phase, abortAt }) => {
|
||||
const h = await harness([{ id: 'first', phase }, { id: 'middle', phase }, { id: 'last', phase }])
|
||||
const lookup = vi.spyOn(h.ctx.sessions, 'get')
|
||||
const controller = new AbortController()
|
||||
const reason = { cancellation: `${phase}-${abortAt}` }
|
||||
h.observer.summary = (_phase, id) => {
|
||||
if (id === abortAt) betweenRows(() => { controller.abort(reason) })
|
||||
}
|
||||
|
||||
await expect(h.list.list(controller.signal)).rejects.toBe(reason)
|
||||
expect(h.trace).toEqual(abortAt === 'first'
|
||||
? [`${phase}:first`]
|
||||
: [`${phase}:first`, `${phase}:middle`, `${phase}:last`])
|
||||
expect(h.listSessions).toHaveBeenCalledExactlyOnceWith(controller.signal)
|
||||
if (phase === 'live' && abortAt === 'first') {
|
||||
expect(lookup).toHaveBeenCalledExactlyOnceWith(SessionId('first'))
|
||||
}
|
||||
})
|
||||
|
||||
it('does not start cold summaries after cancellation during the last live row', async () => {
|
||||
const h = await harness([{ id: 'cold', phase: 'cold' }, { id: 'live', phase: 'live' }])
|
||||
const controller = new AbortController()
|
||||
const reason = new Error('cancel before cold phase')
|
||||
h.observer.summary = (phase) => {
|
||||
if (phase === 'live') betweenRows(() => { controller.abort(reason) })
|
||||
}
|
||||
|
||||
await expect(h.list.list(controller.signal)).rejects.toBe(reason)
|
||||
expect(h.trace).toEqual(['live:live'])
|
||||
})
|
||||
|
||||
it('rejects a pre-aborted request before querying headers', async () => {
|
||||
const h = await harness([{ id: 'live', phase: 'live' }])
|
||||
const controller = new AbortController()
|
||||
const reason = new Error('already cancelled')
|
||||
controller.abort(reason)
|
||||
|
||||
await expect(h.list.list(controller.signal)).rejects.toBe(reason)
|
||||
expect(h.listSessions).not.toHaveBeenCalled()
|
||||
expect(h.trace).toEqual([])
|
||||
})
|
||||
|
||||
it('checks cancellation after the header query before summarizing any row', async () => {
|
||||
const h = await harness([{ id: 'cold', phase: 'cold' }])
|
||||
const controller = new AbortController()
|
||||
const reason = new Error('query cancelled')
|
||||
h.listSessions.mockImplementation(async () => {
|
||||
controller.abort(reason)
|
||||
return []
|
||||
})
|
||||
|
||||
await expect(h.list.list(controller.signal)).rejects.toBe(reason)
|
||||
expect(h.trace).toEqual([])
|
||||
})
|
||||
|
||||
it('returns an empty list without invoking a summary', async () => {
|
||||
const h = await harness([])
|
||||
await expect(h.list.list()).resolves.toEqual([])
|
||||
expect(h.trace).toEqual([])
|
||||
})
|
||||
|
||||
it('omits cwd-less cold rows but retains cwd-less attached Sessions', async () => {
|
||||
const h = await harness([
|
||||
{ id: 'omitted-first', phase: 'cold', cwd: false },
|
||||
{ id: 'visible-cold', phase: 'cold' },
|
||||
{ id: 'visible-live', phase: 'live', cwd: false },
|
||||
{ id: 'omitted-last', phase: 'cold', cwd: false },
|
||||
])
|
||||
|
||||
const items = await h.list.list()
|
||||
|
||||
expect(h.trace).toEqual(['live:visible-live', 'cold:visible-cold'])
|
||||
expect(items.map(item => item.sessionId)).toEqual(['visible-live', 'visible-cold'])
|
||||
expect(items[0]).not.toHaveProperty('cwd')
|
||||
})
|
||||
|
||||
it('keeps live-before-cold stable ties despite interleaved query order', async () => {
|
||||
const h = await harness([
|
||||
{ id: 'cold-a', phase: 'cold' }, { id: 'live-a', phase: 'live' },
|
||||
{ id: 'cold-b', phase: 'cold' }, { id: 'live-b', phase: 'live' },
|
||||
])
|
||||
|
||||
const items = await h.list.list()
|
||||
|
||||
expect(h.trace).toEqual(['live:live-a', 'live:live-b', 'cold:cold-a', 'cold:cold-b'])
|
||||
expect(items.map(item => item.sessionId)).toEqual(['live-a', 'live-b', 'cold-a', 'cold-b'])
|
||||
expect(items.map(item => item.updatedAt)).toEqual([100, 100, 100, 100])
|
||||
})
|
||||
|
||||
it('observes later attachment and status changes without reclassifying queued cold rows', async () => {
|
||||
const h = await harness([
|
||||
{ id: 'queued', phase: 'cold' }, { id: 'first', phase: 'live' },
|
||||
{ id: 'removed', phase: 'live' }, { id: 'promoted', phase: 'cold' },
|
||||
{ id: 'status', phase: 'live' },
|
||||
])
|
||||
h.observer.summary = (_phase, id) => {
|
||||
if (id !== 'first') return
|
||||
betweenRows(() => {
|
||||
h.detach.get('removed')!()
|
||||
h.attach('queued')
|
||||
h.attach('promoted')
|
||||
h.agents.get('first')!.status = 'running'
|
||||
h.agents.get('promoted')!.status = 'running'
|
||||
h.agents.get('status')!.status = 'running'
|
||||
})
|
||||
}
|
||||
|
||||
const items = await h.list.list()
|
||||
|
||||
expect(h.trace).toEqual(['live:first', 'live:promoted', 'live:status', 'cold:queued', 'cold:removed'])
|
||||
expect(items.map(({ sessionId, agentAvailable, running }) => ({ sessionId, agentAvailable, running }))).toEqual([
|
||||
{ sessionId: 'first', agentAvailable: true, running: false },
|
||||
{ sessionId: 'promoted', agentAvailable: true, running: true },
|
||||
{ sessionId: 'status', agentAvailable: true, running: true },
|
||||
{ sessionId: 'queued', agentAvailable: false, running: false },
|
||||
{ sessionId: 'removed', agentAvailable: false, running: false },
|
||||
])
|
||||
expect(h.ctx.agents.get(SessionId('queued'))).toBeDefined()
|
||||
expect(h.ctx.sessions.get(SessionId('removed'))).toBeUndefined()
|
||||
})
|
||||
})
|
||||
@@ -96,7 +96,7 @@ function installSearchQuery(
|
||||
describe('session.search', () => {
|
||||
it('rejects search when the query service is absent', async () => {
|
||||
const ctx = await baseContext()
|
||||
const list = new ApiSessionList(ctx)
|
||||
const list = new ApiSessionList(ctx, 8)
|
||||
|
||||
await expect(list.search('query', new AbortController().signal)).rejects.toMatchObject({
|
||||
code: 'gateway/internal',
|
||||
|
||||
@@ -1928,7 +1928,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
|
||||
{
|
||||
signature: '@Remote(\'list\') async list(_request: SessionListRequest, signal: AbortSignal): Promise<SessionListValue>',
|
||||
description: 'Read all visible Session rows without resuming an Agent.',
|
||||
parameters: [{ name: '_request', description: 'reserved empty list request.' }, { name: 'signal', description: 'cancellation for persistence reads.' }],
|
||||
parameters: [{ name: '_request', description: 'reserved empty list request.' }, { name: 'signal', description: 'cancellation for persistence reads and summary generation.' }],
|
||||
returns: 'visible Session summaries ordered by activity.',
|
||||
},
|
||||
{
|
||||
|
||||
Generated
+3
@@ -985,6 +985,9 @@ importers:
|
||||
playwright:
|
||||
specifier: ^1.49.0
|
||||
version: 1.61.1
|
||||
zod:
|
||||
specifier: ^4.4.3
|
||||
version: 4.4.3
|
||||
|
||||
native/system:
|
||||
devDependencies:
|
||||
|
||||
Reference in New Issue
Block a user