diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index d7b97a0..40eb72b 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -75,6 +75,8 @@ jobs:
           python -B -I -X utf8 scripts/repo_checks.py
       - name: script tests
         run: python -B -I -X utf8 scripts/test_repo_checks.py
+      - name: trim measurement tests
+        run: python -B -I -X utf8 scripts/test_measure_trim.py
 
   portable:
     strategy:
diff --git a/README.en.md b/README.en.md
index 7a58882..7671551 100644
--- a/README.en.md
+++ b/README.en.md
@@ -163,6 +163,10 @@ modavd mcp --root D:\ModAVD
 
 It supports both MCP 2026-07-28 (stateless, `server/discover`) and 2025-11-25 / 2025-06-18 (`initialize` handshake). The root folder is fixed at startup and is not a tool parameter. Tools: `list_instances`, `get_instance`, `check_host`, `create_instance`, `update_instance`, `delete_instance`, `start_instance` (with progress), `stop_instance`, `set_window_visible`, `shell`, `install_apk`, `disk_info` (data disk size), `trim_disk` (reclaim space while running; the emulator must support discard), `compact_disk` (shrink the data disk; the instance must be stopped), `get_app_proxy`, `set_app_proxy` (per-app proxy).
 
+`disk <i> trim`, the GUI trim job, and `trim_disk` return after requesting asynchronous guest maintenance.
+Success does not confirm that guest FITRIM finished or succeeded. The automatic trim log measures only the
+adb/su command round trip. Stable discard counters alone cannot establish completion.
+
 To register it in Claude Code (in PowerShell, `--` needs quotes, or it gets swallowed):
 
 ```powershell
diff --git a/README.md b/README.md
index 9669331..9255b0a 100644
--- a/README.md
+++ b/README.md
@@ -69,7 +69,7 @@ modavd [--root <目录>] [--json] [--quiet] [--verbose] <命令> [参数]
 | `start <i> [--window \| --headless] [--timeout 300] [--no-bridge]` | 开机（默认无头，`--window` 带窗口），等开完并做完开机后的设置，返回前确认 `/sdcard` 能写进文件（一般这时早已就绪；最多再等 60 秒，到点只警告）；实例正在关机时（比如刚点了模拟器窗口的 ×）先等它关完（最多 30 秒）再开机；实例还没有执行桥时接着装上，`--no-bridge` 这次不装。运行中不换方式：实例已经在运行时，给了 `--window` / `--headless` 而它不是这种方式，就报错（退出码 5），不接管也不重启，要换先 `stop` 再 `start`；两个都不给就照原样接管 |
 | `stop <i> [--force] [--cancel-create]`、`stop --all [--force]` | 关机：前台开着应用时先按 HOME，等 2 秒让它保存；再 sync，让安卓正常关机（前台应用先收到暂停、停止），不成再 sync + kill；`--force` 是 sync + 强杀。新建中默认拒绝，`--cancel-create` 显式取消；`--all` 也关闭新建中的实例，结果标记 `interrupted_create: true` |
 | `bridge <i> install\|uninstall\|verify` | 装、卸、检查 arm64 执行桥；开机时会自动装，一般不用手动跑（见下） |
-| `disk <i> trim` | 磁盘回收：在开着的实例里对数据盘触发一次 FITRIM，让模拟器释放已删文件占着的空间；模拟器太旧、不支持回收时返回不支持（退出码 4） |
+| `disk <i> trim` | 磁盘回收：在开着的实例里对数据盘触发一次 FITRIM；成功只表示触发请求已返回，不等待或确认客体回收完成；模拟器太旧、不支持回收时返回不支持（退出码 4） |
 | `disk <i> info` | 看数据盘的大小：逻辑大小、覆盖层在宿主上实际占的空间；关机且取得启动锁后才运行只读的 qcow2 元数据检查，给出已分配和估计可回收大小。整理中或锁不可用时，仅返回可读取的文件头、宿主占用和原因；不可读的数值省略，不修复镜像 |
 | `disk <i> compact` | 整理已关机实例的数据盘；预计收益不足时跳过重写；转换后先校验再原子替换，原盘在替换成功前不动；报告前后占用和增减；大盘预算可达数十小时，正常整理基本不会超时，停止请取消；仅支持 NTFS/ReFS，拒绝快照或重写空间不足，详见[预算与结果契约](internal/emu/COMPACT.md) |
 | `check [--memory M]` | 开机前的电脑检查 |
@@ -84,6 +84,10 @@ modavd [--root <目录>] [--json] [--quiet] [--verbose] <命令> [参数]
 - 根目录：`--root`，没给就用环境变量 `MODAVD_ROOT`，再没有就用 `modavd.exe` 所在目录或它的上一级。
 - 0.3.0 起没有 `restart`：先 `stop` 再 `start`。旧的调用报用法错误（退出码 2），提示改用这两步。
 
+`trim` 的 GUI 任务完成、MCP 成功返回和开机后自动回收日志都只表示触发请求已返回。
+自动回收日志里的秒数是 adb/su 命令往返耗时。客体是否完成或失败，需要另看客体信号，
+不能用暂时稳定的 discard 计数判定。
+
 **arm64 执行桥**：
 
 - 只对内置名单里的游戏生效，其余应用照旧走原来的转译层。桥装在实例自己的 system / vendor 覆盖层里，要 0.36 起的模拟器。
diff --git a/internal/cli/commands.go b/internal/cli/commands.go
index 6c87c9f..6bd3e4e 100644
--- a/internal/cli/commands.go
+++ b/internal/cli/commands.go
@@ -245,6 +245,7 @@ func init() {
 			name: "disk", args: "<i> trim|info|compact", summary: "磁盘回收：触发回收或整理把客体已删文件占的空间还给宿主；看数据盘大小",
 			details: "trim：在客体的数据盘（/data）上触发一次回收（FITRIM，经 vold 空闲维护），" +
 				"让模拟器释放已删除文件在数据盘镜像里占着的空间。实例要开着并已开机完成；" +
+				"命令成功只表示已触发，不等待客体回收完成，也不确认回收成功。" +
 				"模拟器版本太旧、数据盘不支持回收时返回不支持（退出码 4），不会白跑。" +
 				"info：报数据盘的逻辑大小、在宿主上实际占的空间，以及（实例关机时）已分配的大小和估计能回收多少；" +
 				"实例开着时已分配和可回收的数要关机后才有，只读，不改任何东西。" +
diff --git a/internal/emu/TRIM.md b/internal/emu/TRIM.md
new file mode 100644
index 0000000..86acd23
--- /dev/null
+++ b/internal/emu/TRIM.md
@@ -0,0 +1,101 @@
+# trim 时机测量
+
+本文记录 trim 请求的异步契约及复跑方法。具体版本的测量结果保存在 [ModAVD/ModAVD#95 的评论](https://github.com/ModAVD/ModAVD/issues/95#issuecomment-6654569636)中。
+
+## 返回和日志分别表示什么
+
+| 入口或信号 | 含义 |
+| --- | --- |
+| `modavd disk <i> trim` 成功及 `DiskTrimResult` | 管理器确认实例已开机、`vdc` 宣告支持 discard，随后 adb/su 中的 `sm idle-maint run` 成功返回。未等待或确认客体 FITRIM 成功。CLI 保留“已触发磁盘回收” |
+| `diskTrimTimeout` 的 2 分钟、`autoTrimTimeout` 的 30 秒 | 请求往返的超时，不限制后台维护的执行时间 |
+| 开机后自动 trim 日志里的秒数 | `runIdleMaint` 调用前后的宿主时间差，即 adb/su 请求往返，不是客体回收耗时。`diskTrimAt` 仍是上次自动尝试的时间 |
+| GUI `disk-trim` 的 `done`、100%、`finished_at` | 宿主请求任务返回。时间差还包含管理器检查与任务开销。进行中显示“已进行”，请求结束后显示“已触发”和“触发请求用时” |
+| MCP `trim_disk` 成功 | 同一个 `DiskTrim` 返回，没有额外客体完成检查。`discard_max_bytes` 是能力声明，不是本轮回收字节数 |
+| vold `Starting trim of /data` | vold 开始这一轮 `/data` trim 尝试，并不证明已经成功提交 discard |
+| vold `Trimmed … bytes on /data in …ms` | 对应 `/data` FITRIM 的明确成功返回信号。脚本要求窗口内只有一次开始，且开始、返回的进程和线程相同，返回日志必须排在开始日志之后 |
+| vold `Trim failed on /data: …` | 该次 `/data` trim 失败。不能因为 `sm` 退出 0 就忽略 |
+| vold `idle maintenance completed` | 维护线程结束，可能包含其它挂载点及失败的 `/data` trim。不能单独用来判定 `/data` 成功 |
+| `/proc/diskstats` discard 计数 | 内核设备累计的 discard 完成请求数、合并数、512 字节扇区数及毫秒数。计数变化是旁证，暂时稳定不能证明本轮结束 |
+
+调用链是 `internal/cli/commands.go`、`internal/ui/disk.go` 或 `internal/mcp/tools.go`
+→ `internal/emu/disk.go:DiskTrim` → `runIdleMaint` → `Manager.su` → 私有 adb server。
+自动触发来自 `internal/emu/autotrim.go`。管理器的这些路径都没有读取客体完成信号。
+脚本用正式包的客体日志观测 Android 行为，不需要拉取 AOSP 源码。
+
+FITRIM 返回也不等于宿主 qcow2 文件长度已缩小，日志中的字节数不能当作宿主释放字节数。
+没有匹配的明确返回、日志缺失或出现重叠轮次时，脚本输出 `unknown`，不根据计数或等待时长补判成功。
+
+## 复跑
+
+需要 Windows、Python 3.10+、已补全组件的现有正式包，以及从待测源码构建的管理器。
+不需要下载或同步 AOSP。先把正式包及 SDK **物理复制**到一个新的专用根，
+不要复制已有 `avd`、实例配置或使用共享数据盘。为新根写入独立配置，例如：
+
+```json
+{"portBase":27400,"adbServerPort":27498,"scriptAdbServerPort":27499,"instances":{}}
+```
+
+先确认整段端口空闲，而且没有其它验收正在使用该环境。只有一套可用环境时排队。
+以下端口是示例，复跑时需重新确认。脚本要求三个端口配置都是 JSON 整数，两种 adb server 端口必须不同，
+且不占实例端口、生产段或 adb 扫描端口；整个 100 端口段也必须避开生产及 adb 扫描范围。
+脚本在发出 adb 命令前检查整段与两种 server 端口的 TCP/UDP 占用，
+只允许当前实例的专用 qemu 和本根 adb 在其指定 TCP 端口监听；未能核实进程路径时拒绝继续。
+这只是开始时的快照，不能替代测试环境独占。
+
+```powershell
+$root = 'C:\Temp\modavd-trim-test'
+$manager = (Resolve-Path '.\out\tools\modavd.exe').Path
+& $manager --root $root create --index 0 --memory 4096 --cpu 4 --data-gb 16 --no-bridge
+```
+
+新建完成会正常关机。为排除开机后的自动请求，在**该专用根**的 `mas-avd.json` 中
+将 `instances.0.diskTrimAt` 设为当前 Unix 秒，再开机，不改其它实例：
+
+```powershell
+$metaPath = Join-Path $root 'mas-avd.json'
+$meta = Get-Content -LiteralPath $metaPath -Raw | ConvertFrom-Json -AsHashtable
+$meta.instances['0'].diskTrimAt = [DateTimeOffset]::UtcNow.ToUnixTimeSeconds()
+$meta | ConvertTo-Json -Depth 20 | Set-Content -LiteralPath $metaPath -Encoding utf8NoBOM
+& $manager --root $root start 0 --headless --no-bridge
+
+python -B -X utf8 scripts/measure_trim.py --root $root --manager $manager --command cli --write-mib 128 --exclusive-test-instance --output cli-work.json
+python -B -X utf8 scripts/measure_trim.py --root $root --manager $manager --command cli --exclusive-test-instance --output cli-idle.json
+python -B -X utf8 scripts/measure_trim.py --root $root --manager $manager --command sm --write-mib 128 --exclusive-test-instance --output sm-work.json
+python -B -X utf8 scripts/measure_trim.py --root $root --manager $manager --command sm --exclusive-test-instance --output sm-idle-1.json
+python -B -X utf8 scripts/measure_trim.py --root $root --manager $manager --command sm --exclusive-test-instance --output sm-idle-2.json
+```
+
+每次输出使用新文件名。`--write-mib` 默认为 0，最多 256 MiB；非零时，脚本只在
+`/data/local/tmp/modavd-trim-measure-<随机值>` 新建目录，以 `/dev/urandom` 实际写入文件，
+fsync、sync、记录 `du` 后删掉文件并再次 sync。最后只清理这个文件和空目录。
+脚本不装应用、不整理磁盘、不换包、不关机。
+
+`--exclusive-test-instance` 明确声明本实例由本次测试独占。根内的排他文件只防止两份测量脚本重叠，
+不能阻止其它管理器或 adb 使用者。采样期间不要让其它会话对本实例 trim、compact、安装、换包或关机。
+每轮完成后核对结果；若缺少明确结束信号，先查明是否仍有维护在执行，再触发下一轮。
+
+脚本从 mountinfo 的 `/data` 根挂载读取 major:minor，递归读取 sysfs `slaves`，
+保存每层设备名称、discard 上限并按设备号筛选 diskstats，不假定 `/data` 一定直接挂在 `vdc`。
+logcat 经 `su` 仅读取 `main,system`，避免非 root logcat 的内核读取权限错误；dmesg 另经 su 采集。
+每轮保存 boot ID、内核和二进制 SHA-256、vold 单调时间日志、dmesg 前后快照，
+以及默认约 10 秒的 diskstats 采样。采样循环的 sleep 为 0.1 秒，实际间隔还包含读数开销，
+以记录的客体 uptime 为准。采样有固定上限，主机断连也不会留下无限循环。
+
+`command.sent_at` / `returned_at` 使用宿主 UTC，`round_trip_seconds` 使用宿主单调时钟。
+CLI 模式测量整个 CLI 的检查和请求，不假装知道其内部 `sm` 精确发出时刻。
+`sm` 模式的 stdout 另存客体 `SM_SENT`、`SM_RETURNED` 与 uptime，包住实际 `sm` 调用。
+vold 与 diskstats 使用同一客体 uptime 时基，不将宿主墙钟直接减去客体时间。
+
+根目录字段只记录目录名，不保存绝对路径。命令输出和完整 dmesg 仍可能包含路径、用户名或其它本机信息，
+公开原始 JSON 前检查并脱敏。测试结束、确认采样退出后，再正常关闭专用实例及私有 adb server。
+
+## 自动验证
+
+```powershell
+python -B -I -X utf8 scripts/test_measure_trim.py
+go test -count=1 -run 'TestDiskJobs|TestDiskTrim|TestToolDefinitions|TestToolCalls' ./internal/ui ./internal/cli ./internal/mcp
+```
+
+解析测试覆盖明确返回、零字节返回、真实失败后维护仍结束、缺日志、旧日志、其它挂载点、
+设备号、缺少 discard 字段、进程或线程不匹配、重叠轮次、窗口外日志及稳定计数。
+GUI HTTP 测试检查完成消息只说“已触发”，MCP 测试检查工具公开说明，CLI 现有测试保留准确原文。
diff --git a/internal/emu/autotrim.go b/internal/emu/autotrim.go
index 4014e96..076c005 100644
--- a/internal/emu/autotrim.go
+++ b/internal/emu/autotrim.go
@@ -1,7 +1,7 @@
 package emu
 
 // The automatic trim of the data disk after boot (ModAVD/ModAVD#2, design §3.3 / §3.6): once the boot setup is done
-// and the emulator advertises discard on /data, the clusters the guest freed are handed back (the same guest trim as
+// and the emulator advertises discard on /data, request reclamation of freed clusters (the same guest request as
 // `modavd disk <i> trim`), at most once per 24 hours per instance, so nobody has to run it by hand.
 //
 // Strictly best effort and never in the boot's way: every problem (no discard, su failed, mas-avd.json unreadable or
@@ -21,9 +21,8 @@ const (
 	autoTrimInterval = 24 * time.Hour
 	// autoTrimProbeTimeout bounds the read of the discard limit.
 	autoTrimProbeTimeout = 10 * time.Second
-	// autoTrimTimeout bounds the trim itself, which the start waits for. Far shorter than the manual diskTrimTimeout:
-	// a boot must not wait minutes for it. A trim that did not finish in time is not tried again before the interval
-	// is over (see autoTrim).
+	// autoTrimTimeout bounds the sm request, which the start waits for. It does not
+	// bound or await vold's asynchronous FITRIM. See autoTrim for retry scheduling.
 	autoTrimTimeout = 30 * time.Second
 )
 
@@ -90,9 +89,9 @@ func (m *Manager) autoTrim(ctx context.Context, i int, mode bootSetup) {
 		return
 	}
 	if res.Code == 0 {
-		m.Log.Infof("实例 %d 开机后自动回收磁盘空间（%.1f 秒）", i, time.Since(started).Seconds())
+		m.Log.Infof("实例 %d 开机后已触发磁盘回收（命令往返 %.1f 秒，未等待客体回收完成）", i, time.Since(started).Seconds())
 	} else {
-		m.Log.Warnf("实例 %d 开机后自动回收磁盘空间没成功，%.0f 小时后再试: %s", i, autoTrimInterval.Hours(),
+		m.Log.Warnf("实例 %d 开机后触发磁盘回收失败，%.0f 小时后再试: %s", i, autoTrimInterval.Hours(),
 			orText(tail(res.Out, 200), fmt.Sprintf("退出码 %d", res.Code)))
 	}
 	if err := m.writeDiskTrimAt(i, time.Now().Unix()); err != nil {
diff --git a/internal/emu/disk.go b/internal/emu/disk.go
index e7a9b8a..36fa9bf 100644
--- a/internal/emu/disk.go
+++ b/internal/emu/disk.go
@@ -35,9 +35,8 @@ const (
 	userdataBacking = "userdata-qemu.img"
 )
 
-// diskTrimTimeout bounds the guest idle-maintenance run. A first full trim of a
-// large /data walks all the free space and can take a few seconds; later trims
-// are quick. Generous so a big first pass does not time out.
+// diskTrimTimeout bounds the sm request's adb/su round trip, not the asynchronous
+// vold maintenance or FITRIM. A successful return does not confirm guest completion.
 const diskTrimTimeout = 2 * time.Minute
 
 // vdcDiscardMaxPath is the data disk's advertised discard limit in the guest. 0
@@ -46,12 +45,12 @@ const diskTrimTimeout = 2 * time.Minute
 // backport, where a trim would silently do nothing.
 const vdcDiscardMaxPath = "/sys/block/vdc/queue/discard_max_bytes"
 
-// DiskTrimResult reports what `modavd disk <id> trim` did.
+// DiskTrimResult reports a successful trim request, not guest FITRIM completion.
 type DiskTrimResult struct {
 	Index int `json:"index"`
 	// DiscardMaxBytes is the data disk's advertised discard limit (guest
 	// /sys/block/vdc/queue/discard_max_bytes); >0 confirms the emulator's
-	// discard path is live.
+	// discard capability is advertised. It does not confirm a completed discard.
 	DiscardMaxBytes int64 `json:"discard_max_bytes"`
 }
 
@@ -74,15 +73,16 @@ func (m *Manager) probeDiscardMax(ctx context.Context, i int, timeout time.Durat
 	return parseDiscardMaxBytes(probe.Out)
 }
 
-// runIdleMaint triggers the guest's vold idle maintenance (it runs fstrim on the mounted file systems). Shared by
-// DiskTrim and the best-effort trim after boot (autotrim.go).
+// runIdleMaint asks vold to run idle maintenance asynchronously. The result and timeout
+// describe only the sm request. Shared by DiskTrim and the best-effort boot request.
 func (m *Manager) runIdleMaint(ctx context.Context, i int, timeout time.Duration) adb.Result {
 	return m.su(ctx, i, "sm idle-maint run", timeout)
 }
 
 // DiskTrim triggers a guest FITRIM on /data (via vold idle maintenance) so the
 // virtio-blk discard path frees the qcow2 clusters behind deleted files. The
-// instance must be running and booted. errs.Unsupported when the instance is
+// result confirms only that sm returned successfully, not that FITRIM finished or
+// succeeded. The instance must be running and booted. errs.Unsupported when the instance is
 // not up, or when the emulator does not advertise discard on the data disk (an
 // old build); errs.Internal when the guest command fails.
 func (m *Manager) DiskTrim(ctx context.Context, i int) (*DiskTrimResult, error) {
diff --git a/internal/emu/qemu_img_lifetime_windows_test.go b/internal/emu/qemu_img_lifetime_windows_test.go
index 38fcc3c..0574f40 100644
--- a/internal/emu/qemu_img_lifetime_windows_test.go
+++ b/internal/emu/qemu_img_lifetime_windows_test.go
@@ -61,7 +61,11 @@ func lifetimeChild() int {
 			return 93
 		}
 		defer syscall.CloseHandle(h)
-		if err := os.WriteFile(path+".ready", []byte(strconv.Itoa(os.Getpid())), 0o600); err != nil {
+		ready := path + ".ready"
+		if err := os.WriteFile(ready+".tmp", []byte(strconv.Itoa(os.Getpid())), 0o600); err != nil {
+			return 94
+		}
+		if err := os.Rename(ready+".tmp", ready); err != nil {
 			return 94
 		}
 		deadline := time.Now().Add(time.Minute)
diff --git a/internal/mcp/mcp.go b/internal/mcp/mcp.go
index 7e8e94e..8eb03a8 100644
--- a/internal/mcp/mcp.go
+++ b/internal/mcp/mcp.go
@@ -63,8 +63,8 @@ var instructions = fmt.Sprintf("Manages the ModAVD (patched Android Emulator) in
 	"is fixed by the server's --root; instances are numbered 0-%d except %d. Always shut instances down with "+
 	"stop_instance, which syncs the guest disk first: never kill emulator or qemu processes, that loses recently "+
 	"written game data. start_instance can take minutes and reports progress; delete_instance permanently removes "+
-	"an instance with all its game data. disk_info shows how big an instance's data disk is; trim_disk reclaims "+
-	"unused space of a running instance, compact_disk shrinks the file of a stopped one.", root.MaxIndex, root.ReservedIndex)
+	"an instance with all its game data. disk_info shows how big an instance's data disk is; trim_disk requests "+
+	"reclamation in a running instance without waiting for guest completion; compact_disk shrinks the file of a stopped one.", root.MaxIndex, root.ReservedIndex)
 
 // progressEvery throttles progress notifications that do not complete a step.
 const progressEvery = time.Second
diff --git a/internal/mcp/mcp_test.go b/internal/mcp/mcp_test.go
index d7d4c8e..b34c401 100644
--- a/internal/mcp/mcp_test.go
+++ b/internal/mcp/mcp_test.go
@@ -1567,6 +1567,11 @@ func TestToolDefinitions(t *testing.T) {
 		}
 	}
 	del := byName["delete_instance"]["inputSchema"].(map[string]any)
+	trimDescription := byName["trim_disk"]["description"].(string)
+	if !strings.Contains(trimDescription, "not that guest FITRIM finished or succeeded") ||
+		!strings.Contains(trimDescription, "timeout bounds only the request") {
+		t.Fatalf("trim contract must distinguish the request from guest completion: %s", trimDescription)
+	}
 	if !reflect.DeepEqual(del["required"], []any{"index", "confirm"}) {
 		t.Fatalf("delete required %v", del["required"])
 	}
diff --git a/internal/mcp/tools.go b/internal/mcp/tools.go
index cd089e8..61f4040 100644
--- a/internal/mcp/tools.go
+++ b/internal/mcp/tools.go
@@ -479,7 +479,8 @@ func toolDefs() []*toolDef {
 			"Trigger one reclaim of the unused space in a running, booted instance's data disk (guest fstrim via "+
 				"idle maintenance), so the host file can shrink as deleted game data is discarded. The instance "+
 				"must be online and the emulator must support discard (an old emulator build without it fails "+
-				"with unsupported). Safe to repeat; takes up to about 2 minutes. To shrink the file fully, stop "+
+				"with unsupported). Success means the request returned, not that guest FITRIM finished or succeeded. "+
+				"The 2-minute timeout bounds only the request. Safe to repeat. To shrink the file fully, stop "+
 				"the instance and use compact_disk.",
 			mutating(false, true), reflect.TypeFor[emu.DiskTrimResult](), []prop{indexProp(true)},
 			func(ctx context.Context, b backend, a args, _ func(emu.Progress)) (any, error) {
diff --git a/internal/ui/API.md b/internal/ui/API.md
index 9c6b04e..fb28cbb 100644
--- a/internal/ui/API.md
+++ b/internal/ui/API.md
@@ -42,7 +42,7 @@ HTTP 4xx / 5xx，体为 `{"error":{"code":"…","message":"…","advice":"…","
 | POST | `/api/instances/{i}/stop` | `{force?: bool, cancel_create?: bool}` | 202 `Job`；新建中默认 busy，显式 `cancel_create: true` 才取消；`force` 只管是否跳过 Android 正常关机。被取消的新建在结果里标记 `interrupted_create: true` |
 | POST | `/api/instances/{i}/install` | `multipart/form-data`，文件字段 `apk` | 202 `Job`（服务端存到临时文件，任务结束删掉） |
 | POST | `/api/instances/{i}/disk-info` | `{}` | 202 `Job`（`kind` 是 `disk-info`）：数据盘的占用，随时可以查（开机时 `allocated_bytes`、`reclaimable_bytes` 不返回，`needs_shutdown` 为 `true`）；结果是 emu 的 `DiskInfoResult`。整理中或锁不可用时返回可读取的文件信息及 `warnings`，不运行 qemu-img；读不到宿主占用时省略 `on_disk_bytes` |
-| POST | `/api/instances/{i}/disk-trim` | `{}` | 202 `Job`（`kind` 是 `disk-trim`）：让开着的安卓把删掉的文件占的空间还给模拟器（电脑上的文件要整理后才变小）；实例没开机完成、或模拟器版本不支持时任务失败，code 是 `unsupported`；结果是 `DiskTrimResult` |
+| POST | `/api/instances/{i}/disk-trim` | `{}` | 202 `Job`（`kind` 是 `disk-trim`）：请求开着的安卓回收空间；`done` 和 100% 仅表示触发请求已返回，不等待或确认客体 FITRIM 完成，任务耗时也是请求耗时；实例没开机完成、或模拟器版本不支持时任务失败，code 是 `unsupported`；结果是 `DiskTrimResult` |
 | POST | `/api/instances/{i}/disk-compact` | `{}` | 202 `Job`（`kind` 是 `disk-compact`）：整理已关机实例的数据盘文件；预估收益不足时跳过重写；实例没关机时任务失败，code 是 `busy`；结果是 `DiskCompactResult` |
 | POST | `/api/instances/{i}/resume` | `{}` | 202 `Job`（`kind` 是 `create`）：重做新建失败的那台的初始化 |
 | POST | `/api/instances/{i}/input` | `{type:"tap", x, y}`（坐标是这台实例显示的像素（1280×720 或 1920×1080，竖屏时宽高对调））或 `{type:"key", key:"back"\|"home"\|"recents"}` | `{}` |
diff --git a/internal/ui/disk.go b/internal/ui/disk.go
index 3410ef5..2e51444 100644
--- a/internal/ui/disk.go
+++ b/internal/ui/disk.go
@@ -36,12 +36,12 @@ func (s *server) handleDiskTrim(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	s.respondJob(w)(s.startJob(kindDiskTrim, i, func(ctx context.Context, j *job, b backend) (any, error) {
-		s.setProgress(j, jobProgress{Step: "disk_trim", Message: fmt.Sprintf("实例 %d 正在回收磁盘空间", i)})
+		s.setProgress(j, jobProgress{Step: "disk_trim", Message: fmt.Sprintf("实例 %d 正在触发磁盘回收", i)})
 		res, err := b.DiskTrim(ctx, i)
 		if err != nil {
 			return nil, err
 		}
-		s.setProgress(j, jobProgress{Step: "done", Message: fmt.Sprintf("实例 %d 已回收磁盘空间", i), Done: progressTotal, Total: progressTotal})
+		s.setProgress(j, jobProgress{Step: "done", Message: fmt.Sprintf("实例 %d 已触发磁盘回收", i), Done: progressTotal, Total: progressTotal})
 		return res, nil
 	}))
 }
diff --git a/internal/ui/disk_test.go b/internal/ui/disk_test.go
index 8e8821f..de53f51 100644
--- a/internal/ui/disk_test.go
+++ b/internal/ui/disk_test.go
@@ -56,6 +56,9 @@ func TestDiskJobs(t *testing.T) {
 	if v.State != stateDone || trim.Index != 4 || trim.DiscardMaxBytes != 2<<30 || v.Progress == nil || v.Progress.Step != "done" {
 		t.Fatalf("trim job %+v (%+v)", v, trim)
 	}
+	if v.Progress.Message != "实例 4 已触发磁盘回收" {
+		t.Fatalf("trim request must not claim guest completion: %+v", v.Progress)
+	}
 
 	var compact emu.DiskCompactResult
 	v = diskJob(t, h, 3, kindDiskCompact, &compact)
diff --git a/internal/ui/web/dev/mock.js b/internal/ui/web/dev/mock.js
index 10f89b6..52cd393 100644
--- a/internal/ui/web/dev/mock.js
+++ b/internal/ui/web/dev/mock.js
@@ -789,7 +789,7 @@ async function handle(url, init) {
           ...(running ? {} : { allocated_bytes: Math.round(d.alloc), reclaimable_bytes: Math.max(0, Math.round(d.onDisk - d.alloc)) }) });
       }, 500);
     } else if (act === 'disk-trim') {
-      step(j, 'disk_trim', `实例 ${i} 正在回收磁盘空间`, 0);
+      step(j, 'disk_trim', `实例 ${i} 正在触发磁盘回收`, 0);
       setTimeout(() => {
         if (inst.status !== 'online') return failJob(j, 'unsupported', `魔改 AVD 实例 ${i} 没有在运行，请先开机再回收磁盘空间`, '');
         d.alloc = Math.max(3 * GB, d.alloc - 1.5 * GB);
diff --git a/internal/ui/web/js/actions.js b/internal/ui/web/js/actions.js
index 08b82fb..fa3048d 100644
--- a/internal/ui/web/js/actions.js
+++ b/internal/ui/web/js/actions.js
@@ -41,7 +41,7 @@ function notify(job) {
   else if (job.kind === 'install') toast({ title: '安装完成', text: `${who}装好了 ${r.package || 'APK'}` });
   // the sizes are the answer, so they stay long enough to read
   else if (job.kind === 'disk-info') toast({ tone: 'info', title: `${who}的磁盘占用`, text: diskInfoText(r), timeout: 9000 });
-  else if (job.kind === 'disk-trim') toast({ title: `${who}已回收磁盘空间`, text: diskTrimText(), timeout: 6000 });
+  else if (job.kind === 'disk-trim') toast({ title: `${who}已触发磁盘回收`, text: diskTrimText(), timeout: 6000 });
   else if (job.kind === 'disk-compact') toast({ title: r.skipped ? `${who}已跳过磁盘重写` : `${who}磁盘整理好了`, text: diskCompactText(r), timeout: 6000 });
 }
 
diff --git a/internal/ui/web/js/cards.js b/internal/ui/web/js/cards.js
index 251d163..45f804b 100644
--- a/internal/ui/web/js/cards.js
+++ b/internal/ui/web/js/cards.js
@@ -221,7 +221,7 @@ function updateCard(c, inst) {
   if (busy) {
     const p = v.job && v.job.progress;
     const msg = (p && stripInst(p.message)) || (v.key === 'stopping' ? '正在关机…' : v.key === 'creating' ? '正在新建：后台开机做初始化…' : v.kind === 'install' ? '正在安装 APK…'
-      : v.kind === 'disk-trim' ? '正在回收磁盘空间…' : v.kind === 'disk-compact' ? '正在整理磁盘…' : '正在开机…');
+      : v.kind === 'disk-trim' ? '正在触发磁盘回收…' : v.kind === 'disk-compact' ? '正在整理磁盘…' : '正在开机…');
     setText(c.progMsg, msg);
     setText(c.lmsg, msg);
     const n = cp ?? pct(p);
diff --git a/internal/ui/web/js/format.js b/internal/ui/web/js/format.js
index ea95009..4962633 100644
--- a/internal/ui/web/js/format.js
+++ b/internal/ui/web/js/format.js
@@ -69,7 +69,7 @@ export function diskInfoText(r) {
 }
 
 export function diskTrimText() {
-  return '已让安卓把删掉的文件占的空间还回来；电脑上的磁盘文件要关机后用「整理磁盘」才会变小。';
+  return '不等待安卓回收完成；电脑上的磁盘文件要关机后用「整理磁盘」才会变小。';
 }
 
 export function diskCompactText(r) {
diff --git a/internal/ui/web/js/panels.js b/internal/ui/web/js/panels.js
index 43bf7a8..8391de9 100644
--- a/internal/ui/web/js/panels.js
+++ b/internal/ui/web/js/panels.js
@@ -254,8 +254,8 @@ function updateJob(r, j) {
   const running = j.state === 'running';
   const t0 = parseTime(j.created_at);
   const took = t0 && ((parseTime(j.finished_at) || Date.now()) - t0) / 1000;
-  setText(r.meta, clock(j.created_at) + (took ? ` · ${running ? '已进行' : '用时'} ${dur(took)}` : ''));
-  setText(r.pill, STATE[j.state] || j.state);
+  setText(r.meta, clock(j.created_at) + (took ? ` · ${running ? '已进行' : j.kind === 'disk-trim' ? '触发请求用时' : '用时'} ${dur(took)}` : ''));
+  setText(r.pill, j.kind === 'disk-trim' && j.state === 'done' ? '已触发' : STATE[j.state] || j.state);
   attr(r.pill, 'class', 'pill tone-' + (STATE_TONE[j.state] || 'idle'));
 
   r.prog.hidden = !running;
@@ -263,7 +263,7 @@ function updateJob(r, j) {
     const p = j.progress;
     const n = pct(p);
     setText(r.msg, (p && stripInst(p.message)) || (j.kind === 'stop' ? '正在关机…' : j.kind === 'install' ? '正在安装…'
-      : j.kind === 'disk-info' ? '正在查看磁盘占用…' : j.kind === 'disk-trim' ? '正在回收磁盘空间…' : j.kind === 'disk-compact' ? '正在整理磁盘…' : '进行中…'));
+      : j.kind === 'disk-info' ? '正在查看磁盘占用…' : j.kind === 'disk-trim' ? '正在触发磁盘回收…' : j.kind === 'disk-compact' ? '正在整理磁盘…' : '进行中…'));
     setText(r.pctEl, n == null ? '' : Math.round(n) + '%');
     r.bar.classList.toggle('indet', n == null);
     r.fill.style.width = n == null ? '' : n.toFixed(1) + '%';
diff --git a/scripts/measure_trim.py b/scripts/measure_trim.py
new file mode 100644
index 0000000..c7db2c6
--- /dev/null
+++ b/scripts/measure_trim.py
@@ -0,0 +1,354 @@
+"""Measure trim requests on an already booted, dedicated test instance. Stdlib only."""
+from __future__ import annotations
+
+import argparse
+from contextlib import contextmanager
+import datetime as dt
+import hashlib
+import json
+import os
+import platform
+from pathlib import Path
+import re
+import shlex
+import subprocess
+import threading
+import time
+import uuid
+
+
+def port_plan(meta: dict, index: int) -> tuple[int, int, int]:
+    """Reject metadata that could fall back to production ports in the manager."""
+    base, port, script_port = (meta.get(key) for key in ("portBase", "adbServerPort", "scriptAdbServerPort"))
+    if type(index) is not int or index not in set(range(10)) - {5}:
+        raise ValueError("invalid instance index")
+    if type(base) is not int or not 1100 <= base <= 65400 or base % 100:
+        raise ValueError("portBase must be a multiple of 100 in 1100..65400")
+    reserved = {5037, *range(5555, 5586), *range(20000, 20100)}
+    if reserved.intersection(range(base, base + 100)):
+        raise ValueError("the whole portBase segment must avoid production and adb scan ports")
+    instance_ports = {base + 10 * i + offset for i in range(10) if i != 5 for offset in range(3)}
+    for name, value in (("adbServerPort", port), ("scriptAdbServerPort", script_port)):
+        if type(value) is not int or not 1024 <= value <= 65535 or value in reserved | instance_ports:
+            raise ValueError(f"{name} must be explicit, private and outside instance ports")
+    if port == script_port:
+        raise ValueError("the two adb server ports must differ")
+    return base, port, script_port
+
+
+def listening_ports(ports: set[int]) -> list[dict]:
+    # Only validated integers enter the PowerShell code. Root paths never do.
+    command = "$ErrorActionPreference='Stop'; $ports=@(" + ",".join(map(str, sorted(ports))) + "); " + r"""
+$listeners = @(
+    Get-NetTCPConnection -State Listen | Where-Object { $_.LocalPort -in $ports } | ForEach-Object {
+        [pscustomobject]@{ Port=$_.LocalPort; Pid=$_.OwningProcess; Protocol='tcp' }
+    }
+    Get-NetUDPEndpoint | Where-Object { $_.LocalPort -in $ports } | ForEach-Object {
+        [pscustomobject]@{ Port=$_.LocalPort; Pid=$_.OwningProcess; Protocol='udp' }
+    }
+)
+@($listeners | ForEach-Object {
+    $owner = Get-CimInstance Win32_Process -Filter ('ProcessId = ' + $_.Pid)
+    [pscustomobject]@{ port=$_.Port; pid=$_.Pid; protocol=$_.Protocol; executable=$owner.ExecutablePath }
+}) | ConvertTo-Json -Compress
+"""
+    result = run(["powershell.exe", "-NoProfile", "-NonInteractive", "-Command", command])
+    if result["code"] != 0:
+        raise RuntimeError("cannot inspect the test port segment: " + result["stderr"])
+    rows = json.loads(result["stdout"] or "[]")
+    return rows if isinstance(rows, list) else [rows]
+
+
+def check_listeners(root: Path, base: int, port: int, script_port: int, index: int, listeners: list[dict]) -> list[dict]:
+    """Allow only this root's qemu and adb on their assigned TCP ports."""
+    qemu_ports = {base + 10 * index + offset for offset in range(3)}
+    qemu_dir = root / "sdk/emulator/qemu/windows-x86_64"
+    qemus = {(qemu_dir / name).resolve() for name in ("qemu-system-x86_64.exe", "qemu-system-x86_64-headless.exe")}
+    adb = (root / "sdk/platform-tools/adb.exe").resolve()
+    checked = []
+    for listener in listeners:
+        number = listener["port"]
+        role = "qemu" if number in qemu_ports else "adb" if number in (port, script_port) else None
+        path = listener.get("executable")
+        if (listener["protocol"] != "tcp" or role is None or not path
+                or Path(path).resolve() not in (qemus if role == "qemu" else {adb})):
+            raise ValueError(f"test port {number}/{listener['protocol']} is occupied by an unverified process (PID {listener['pid']})")
+        checked.append(dict(port=number, pid=listener["pid"], role=role, executable_name=Path(path).name))
+    return checked
+
+
+@contextmanager
+def measurement_lock(root: Path):
+    lock = root / ".trim-measure.lock"
+    # A failed exclusive open never removes another run's lock.
+    fd = os.open(lock, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    try:
+        yield fd
+    finally:
+        try:
+            os.close(fd)
+        finally:
+            lock.unlink()
+
+
+def cleanup_streams(streams):
+    errors = []
+    for process, thread in streams:
+        try:
+            if process.poll() is None:
+                process.terminate()
+            try:
+                process.wait(timeout=5)
+            except subprocess.TimeoutExpired:
+                process.kill()
+                process.wait(timeout=5)
+        except Exception as error:
+            errors.append(error)
+        finally:
+            thread.join(timeout=2)
+    if errors:
+        raise RuntimeError("failed to stop a measurement stream") from errors[0]
+
+
+def data_device(mountinfo: str) -> str:
+    mounts = [line.split() for line in mountinfo.splitlines()]
+    devices = {m[2] for m in mounts if len(m) > 6 and m[3:5] == ["/", "/data"]}
+    if len(devices) != 1 or not re.fullmatch(r"\d+:\d+", next(iter(devices), "")):
+        raise ValueError("cannot identify the /data mount device")
+    return devices.pop()
+
+
+def discard_stats(text: str, devices: set[str]) -> dict:
+    result = {}
+    for line in text.splitlines():
+        f = line.split()
+        if len(f) < 3 or f"{f[0]}:{f[1]}" not in devices:
+            continue
+        # Linux diskstats fields 12..15 follow the original eleven I/O fields.
+        # Old kernels have no discard fields. Do not turn missing counters into zero.
+        if len(f) < 18:
+            result[f"{f[0]}:{f[1]}"] = {"name": f[2], "discard": None}
+        else:
+            values = [int(x) for x in f[14:18]]
+            if any(x < 0 for x in values):
+                raise ValueError("negative diskstats counter")
+            result[f"{f[0]}:{f[1]}"] = dict(name=f[2], discard=dict(zip(
+                ("completed", "merged", "sectors", "milliseconds"), values)))
+    return result
+
+
+LOG = re.compile(r"^\s*(\d+\.\d+)\s+(\d+)\s+(\d+)\s+[VDIWEF]\s+(\S+)\s*:\s*(.*)$")
+
+
+def trim_evidence(log: str, token: str) -> dict:
+    """Require a /data start and explicit success on the same vold thread after our marker.
+
+    An idle-maintenance completion or stable counters alone never establishes /data success.
+    Exclusive access to this instance is a measurement precondition, not inferred from logs.
+    """
+    active = False
+    starts, ends, failures, maintenance = [], [], [], []
+    for line_number, line in enumerate(log.splitlines(), 1):
+        match = LOG.fullmatch(line)
+        if not match:
+            continue
+        stamp, pid, tid, tag, message = match.groups()
+        event = {"uptime_seconds": float(stamp), "pid": pid, "tid": tid, "log_line": line_number, "line": line}
+        if tag == "ModAVDTrimMeasure" and message == "BEGIN " + token:
+            active = True
+            starts, ends, failures, maintenance = [], [], [], []
+        if tag == "ModAVDTrimMeasure" and message == "END " + token:
+            active = False
+        if not active or tag != "vold":
+            continue
+        if message == "Starting trim of /data":
+            starts.append(event)
+        if re.fullmatch(r"Trimmed \d+ bytes on /data in \d+(?:\.\d+)?ms", message):
+            ends.append(event)
+        if message.startswith("Trim failed on /data:"):
+            failures.append(event)
+        if message == "idle maintenance completed":
+            maintenance.append(event)
+    status = "unknown"
+    if len(starts) == 1:
+        start = starts[0]
+        same = lambda e: ((e["pid"], e["tid"]) == (start["pid"], start["tid"])
+                          and e["uptime_seconds"] >= start["uptime_seconds"] and e["log_line"] > start["log_line"])
+        if len(failures) == 1 and same(failures[0]) and not ends:
+            status = "failed"
+        elif len(ends) == 1 and same(ends[0]) and not failures:
+            status = "observed_fitrim_return"
+    return dict(data_trim_status=status, data_starts=starts, data_returns=ends,
+                data_failures=failures, maintenance_completions=maintenance)
+
+
+def utc_now() -> str:
+    return dt.datetime.now(dt.timezone.utc).isoformat()
+
+
+def run(argv: list[str], timeout: float = 30) -> dict:
+    started = utc_now()
+    before = time.perf_counter()
+    try:
+        p = subprocess.run(argv, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout)
+        return dict(sent_at=started, returned_at=utc_now(), round_trip_seconds=time.perf_counter() - before,
+                    code=p.returncode, stdout=p.stdout, stderr=p.stderr)
+    except subprocess.TimeoutExpired as e:
+        return dict(sent_at=started, returned_at=utc_now(), round_trip_seconds=time.perf_counter() - before,
+                    code=None, timeout=True, stdout=(e.stdout or b"").decode("utf-8", "replace"),
+                    stderr=(e.stderr or b"").decode("utf-8", "replace"))
+
+
+def measure(args) -> dict:
+    root = args.root.resolve(strict=True)
+    meta = json.loads((root / "mas-avd.json").read_text(encoding="utf-8-sig"))
+    base, port, script_port = port_plan(meta, args.index)
+    listeners = check_listeners(root, base, port, script_port, args.index,
+                                listening_ports(set(range(base, base + 100)) | {port, script_port}))
+    qemu_processes = {(row["pid"], row["executable_name"]) for row in listeners if row["role"] == "qemu"}
+    if len(qemu_processes) != 1:
+        raise ValueError("cannot identify one running qemu on this instance's ports")
+    _, qemu_name = qemu_processes.pop()
+    adb = [str(root / "sdk/platform-tools/adb.exe"), "-P", str(port), "-s", f"emulator-{base + 10 * args.index}"]
+    shell = lambda script: adb + ["shell", "su 0 sh -c " + shlex.quote(script)]
+
+    def checked(script):
+        r = run(shell(script))
+        if r["code"] != 0:
+            raise RuntimeError(json.dumps(r))
+        return r["stdout"]
+
+    streams = []
+    with measurement_lock(root) as fd:
+        try:
+            os.write(fd, str(os.getpid()).encode())
+            mountinfo = checked("cat /proc/self/mountinfo")
+            device = data_device(mountinfo)
+            pending, graph = [device], {}
+            while pending:
+                dev = pending.pop()
+                if dev in graph:
+                    continue
+                raw = checked(f"cat /sys/dev/block/{dev}/uevent; echo SLAVES; "
+                              f"for s in /sys/dev/block/{dev}/slaves/*; do [ ! -e \"$s\" ] || cat \"$s/dev\"; done; "
+                              f"echo LIMIT; cat /sys/dev/block/{dev}/queue/discard_max_bytes")
+                props, rest = raw.split("SLAVES\n", 1)
+                slaves, limit = rest.split("LIMIT\n", 1)
+                children = slaves.split()
+                if any(not re.fullmatch(r"\d+:\d+", s) for s in children):
+                    raise ValueError("invalid sysfs slave device")
+                graph[dev] = dict(uevent=props, slaves=children, discard_max_bytes=int(limit.strip()))
+                pending.extend(children)
+            before = checked("cat /proc/diskstats")
+            report = dict(root_name=root.name, index=args.index, port_base=base, adb_server_port=port,
+                          script_adb_server_port=script_port, port_listeners=listeners,
+                          command_kind=args.command, write_mib=args.write_mib, started_at=utc_now(),
+                          mountinfo=mountinfo, data_device=device, device_graph=graph,
+                          boot_id=checked("cat /proc/sys/kernel/random/boot_id").strip(),
+                          kernel=checked("uname -a"), diskstats_before=discard_stats(before, set(graph)))
+            report["host"] = dict(platform=platform.platform(), python=platform.python_version())
+            report["binaries"] = {}
+            manager = args.manager or root / "tools/modavd.exe"
+            for name, path in {"manager": manager, "adb": Path(adb[0]),
+                               "qemu": root / "sdk/emulator/qemu/windows-x86_64" / qemu_name}.items():
+                report["binaries"][name] = hashlib.sha256(path.read_bytes()).hexdigest()
+            report["package_version"] = json.loads((root / "modavd-manifest.json").read_text(encoding="utf-8-sig"))["version"]
+            report["dmesg_before"] = run(shell("dmesg"))
+            token = uuid.uuid4().hex
+            workdir = "/data/local/tmp/modavd-trim-measure-" + token
+            # mkdir must succeed. Never reuse or recursively delete a preexisting directory.
+            checked(f"mkdir {workdir}")
+            try:
+                if args.write_mib:
+                    report["workload"] = run(shell(
+                        f"dd if=/dev/urandom of={workdir}/payload bs=1048576 count={args.write_mib} conv=fsync && "
+                        f"sync && du -k {workdir}/payload && rm {workdir}/payload && sync"), timeout=120)
+                    if report["workload"]["code"] != 0:
+                        raise RuntimeError("workload did not finish; see output")
+                lines, samples = [], []
+
+                def collect(argv, output):
+                    p = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
+                                         text=True, encoding="utf-8", errors="replace")
+                    def read():
+                        with p.stdout:
+                            for line in p.stdout:
+                                output.append(line)
+                    t = threading.Thread(target=read, daemon=True)
+                    t.start()
+                    streams.append((p, t))
+                    return p, t
+
+                collect(shell("exec logcat -b main,system -v monotonic -T 1 -s vold:* ModAVDTrimMeasure:I"), lines)
+                checked("log -t ModAVDTrimMeasure " + shlex.quote("BEGIN " + token))
+                deadline = time.monotonic() + 5
+                while not any("BEGIN " + token in x for x in lines):
+                    if time.monotonic() >= deadline:
+                        raise RuntimeError("logcat marker not received; no command issued")
+                    time.sleep(0.02)
+                # Finite guest sampler: a disconnected host cannot leave an unbounded guest loop.
+                count = int(args.observe_seconds * 10)
+                pattern = " || ".join(f'$1 ":" $2 == "{dev}"' for dev in graph)
+                sampler, sample_thread = collect(shell(
+                    f"for n in $(seq 1 {count}); do read up rest < /proc/uptime; echo SAMPLE $up; "
+                    f"awk '{pattern}' /proc/diskstats; sleep 0.1; done"), samples)
+                sample_deadline = time.monotonic() + 5
+                while not any(x.startswith("SAMPLE ") for x in samples):
+                    if time.monotonic() > sample_deadline:
+                        raise RuntimeError("diskstats sampler not ready; no command issued")
+                    time.sleep(0.02)
+                if args.command == "cli":
+                    command = [str(manager), "--root", str(root), "disk", str(args.index), "trim"]
+                else:
+                    # These uptime stamps bound the guest sm invocation, separate from host adb RTT.
+                    command = shell("echo SM_SENT; cat /proc/uptime; sm idle-maint run; rc=$?; "
+                                    "echo SM_RETURNED $rc; cat /proc/uptime; exit $rc")
+                report["command"] = run(command, timeout=150)
+                sampler.wait(timeout=args.observe_seconds + 15)
+                sample_thread.join(timeout=2)
+                report["sampler_code"] = sampler.returncode
+                report["diskstats_samples"] = "".join(samples)
+                report["diskstats_after"] = discard_stats(checked("cat /proc/diskstats"), set(graph))
+                checked("log -t ModAVDTrimMeasure " + shlex.quote("END " + token))
+                deadline = time.monotonic() + 5
+                while not any("END " + token in x for x in lines) and time.monotonic() < deadline:
+                    time.sleep(0.02)
+                report["vold_log"] = "".join(lines)
+                report["evidence"] = trim_evidence(report["vold_log"], token)
+                report["dmesg_after"] = run(shell("dmesg"))
+                report["same_boot"] = checked("cat /proc/sys/kernel/random/boot_id").strip() == report["boot_id"]
+                if not report["same_boot"]:
+                    report["evidence"]["data_trim_status"] = "unknown"
+                report["finished_at"] = utc_now()
+                return report
+            finally:
+                run(shell(f"rm -f {workdir}/payload; rmdir {workdir}"))
+        finally:
+            cleanup_streams(streams)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--root", type=Path, required=True)
+    parser.add_argument("--index", type=int, default=0)
+    parser.add_argument("--manager", type=Path, help="CLI built from the source under test")
+    parser.add_argument("--command", choices=("cli", "sm"), default="cli")
+    parser.add_argument("--write-mib", type=int, choices=range(0, 257), default=0, metavar="0..256")
+    parser.add_argument("--observe-seconds", type=int, choices=range(2, 61), default=10, metavar="2..60")
+    parser.add_argument("--exclusive-test-instance", action="store_true", required=True,
+                        help="acknowledge this is a dedicated instance with no other maintenance, installs, updates or shutdown")
+    parser.add_argument("--output", type=Path, required=True)
+    args = parser.parse_args()
+    if args.output.exists():
+        parser.error("output already exists")
+    report = measure(args)
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    with args.output.open("x", encoding="utf-8") as output:
+        output.write(json.dumps(report, ensure_ascii=False, indent=2) + "\n")
+    print(json.dumps(dict(command=report["command_kind"], write_mib=report["write_mib"],
+                         round_trip_seconds=report["command"]["round_trip_seconds"],
+                         command_code=report["command"]["code"], **report["evidence"]), ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/scripts/test_measure_trim.py b/scripts/test_measure_trim.py
new file mode 100644
index 0000000..6ee334c
--- /dev/null
+++ b/scripts/test_measure_trim.py
@@ -0,0 +1,175 @@
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import Mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+from measure_trim import (check_listeners, cleanup_streams, data_device, discard_stats,
+                          measurement_lock, port_plan, trim_evidence)
+
+
+def line(message, at=2, tid=8936, tag="vold"):
+    return f" {at:.3f} 171 {tid} D {tag} : {message}\n"
+
+
+class TrimEvidenceTest(unittest.TestCase):
+    begin = line("BEGIN test", at=1, tag="ModAVDTrimMeasure")
+    start = line("Starting trim of /data")
+    success = line("Trimmed 134217728 bytes on /data in 120ms", at=3)
+    failure = line("Trim failed on /data: Operation not supported on transport endpoint", at=3)
+
+    def status(self, text):
+        return trim_evidence(text, "test")["data_trim_status"]
+
+    def test_explicit_return(self):
+        self.assertEqual(self.status(self.begin + self.start + self.success), "observed_fitrim_return")
+
+    def test_zero_bytes_still_explicit_return(self):
+        self.assertEqual(self.status(self.begin + self.start + self.success.replace("134217728", "0")),
+                         "observed_fitrim_return")
+
+    def test_actual_failure_then_maintenance_completion(self):
+        # Observed on the unmodified 0.37.1 package: sm exits 0 despite /data failure.
+        text = self.begin + self.start + self.failure + line("idle maintenance completed", at=4)
+        evidence = trim_evidence(text, "test")
+        self.assertEqual(evidence["data_trim_status"], "failed")
+        self.assertEqual(len(evidence["maintenance_completions"]), 1)
+
+    def test_missing_or_ambiguous_evidence_is_unknown(self):
+        cases = {
+            "no log": "",
+            "stable counters are not a signal": "SAMPLE 2.0\n253 32 vdc 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0\n" * 10,
+            "old cycle": self.start + self.success + self.begin,
+            "missing marker": self.start + self.success,
+            "different marker": self.begin.replace("test", "other") + self.start + self.success,
+            "missing start": self.begin + self.success,
+            "maintenance only": self.begin + line("idle maintenance completed"),
+            "start only": self.begin + self.start,
+            "metadata only": self.begin + self.start.replace("/data", "/metadata") + self.success.replace("/data", "/metadata"),
+            "data_mirror": self.begin + self.start + self.success.replace("/data", "/data_mirror"),
+            "other thread": self.begin + self.start + self.success.replace("8936", "8937"),
+            "other process": self.begin + self.start + self.success.replace("171", "172"),
+            "other tag": self.begin + self.start + self.success.replace("vold", "app"),
+            "return before start": self.begin + self.start + self.success.replace("3.000", "1.500"),
+            "return earlier in log at same timestamp": self.begin + self.success.replace("3.000", "2.000") + self.start,
+            "two starts": self.begin + self.start * 2 + self.success,
+            "two returns": self.begin + self.start + self.success * 2,
+            "contradictory result": self.begin + self.start + self.success + self.failure,
+            "outside window": self.begin + line("END test", tag="ModAVDTrimMeasure") + self.start + self.success,
+        }
+        for name, text in cases.items():
+            with self.subTest(name=name):
+                self.assertEqual(self.status(text), "unknown")
+
+    def test_data_mount_uses_major_minor_not_name(self):
+        text = ("90 33 254:40 / /data rw - ext4 /dev/block/dm-40 rw\n"
+                "94 92 254:40 /user /data_mirror/data_ce/null rw - ext4 /dev/block/dm-40 rw\n"
+                "42 33 253:49 / /metadata rw - ext4 /dev/block/vdd rw\n")
+        self.assertEqual(data_device(text), "254:40")
+        for bad in ("", text.replace("/ /data ", "/ /data_mirror "),
+                    text + "91 33 253:32 / /data rw - ext4 /dev/block/vdc rw\n"):
+            with self.assertRaises(ValueError):
+                data_device(bad)
+
+    def test_diskstats_discard_columns(self):
+        raw = "253 32 vdc " + " ".join(str(n) for n in range(1, 18)) + "\n"
+        raw += "253 49 vdd " + "0 " * 17
+        result = discard_stats(raw, {"253:32"})
+        self.assertEqual(set(result), {"253:32"})
+        self.assertEqual(result["253:32"]["discard"],
+                         dict(completed=12, merged=13, sectors=14, milliseconds=15))
+        self.assertIsNone(discard_stats("253 32 vdc " + "0 " * 11, {"253:32"})["253:32"]["discard"])
+        self.assertEqual(discard_stats(raw, {"254:40"}), {})
+        with self.assertRaises(ValueError):
+            discard_stats(raw.replace("12", "-1"), {"253:32"})
+
+
+class MeasurementResourcesTest(unittest.TestCase):
+    def test_wait_failure_releases_lock_and_cleans_remaining_stream(self):
+        stuck = Mock()
+        stuck.poll.return_value = None
+        stuck.wait.side_effect = subprocess.TimeoutExpired("adb", 5)
+        stopped, stuck_thread, stopped_thread = Mock(), Mock(), Mock()
+        stopped.poll.return_value = None
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            with self.assertRaisesRegex(RuntimeError, "failed to stop"):
+                with measurement_lock(root) as fd:
+                    cleanup_streams([(stuck, stuck_thread), (stopped, stopped_thread)])
+            stuck.terminate.assert_called_once()
+            stuck.kill.assert_called_once()
+            stopped.terminate.assert_called_once()
+            stopped.wait.assert_called_once_with(timeout=5)
+            stuck_thread.join.assert_called_once_with(timeout=2)
+            stopped_thread.join.assert_called_once_with(timeout=2)
+            self.assertFalse((root / ".trim-measure.lock").exists())
+            with self.assertRaises(OSError):
+                os.fstat(fd)
+            with measurement_lock(root):
+                self.assertTrue((root / ".trim-measure.lock").exists())
+
+    def test_wait_timeout_kills_and_reaps(self):
+        process, thread = Mock(), Mock()
+        process.poll.return_value = None
+        process.wait.side_effect = [subprocess.TimeoutExpired("adb", 5), 0]
+        cleanup_streams([(process, thread)])
+        process.kill.assert_called_once()
+        self.assertEqual(process.wait.call_count, 2)
+
+    def test_existing_lock_is_preserved(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            path = root / ".trim-measure.lock"
+            path.write_text("another run", encoding="utf-8")
+            with self.assertRaises(FileExistsError):
+                with measurement_lock(root):
+                    self.fail("entered another run's lock")
+            self.assertEqual(path.read_text(encoding="utf-8"), "another run")
+
+
+class PortIsolationTest(unittest.TestCase):
+    meta = dict(portBase=27400, adbServerPort=27498, scriptAdbServerPort=27499)
+
+    def test_explicit_private_plan(self):
+        self.assertEqual(port_plan(self.meta, 0), (27400, 27498, 27499))
+        self.assertEqual(port_plan(dict(self.meta, portBase=65400), 9), (65400, 27498, 27499))
+
+    def test_invalid_or_shared_ports(self):
+        cases = [dict(self.meta, **{key: value})
+                 for key in self.meta for value in (None, True, "27400", 27400.0, 0, -1, 65536)]
+        cases += [dict(self.meta, portBase=base) for base in (1000, 20000, 20001, 5000, 5500, 65401, 65500)]
+        cases += [dict(self.meta, **{key: value})
+                  for key in ("adbServerPort", "scriptAdbServerPort")
+                  for value in (5037, 5555, 5585, 20048, 20049, 20050, 20091, 27400, 27401, 27492)]
+        cases.append(dict(self.meta, scriptAdbServerPort=27498))
+        for meta in cases:
+            with self.subTest(meta=meta), self.assertRaises(ValueError):
+                port_plan(meta, 0)
+        for index in (-1, 5, 10, True):
+            with self.subTest(index=index), self.assertRaises(ValueError):
+                port_plan(self.meta, index)
+
+    def test_entire_segment_and_process_ownership(self):
+        root = Path(tempfile.gettempdir()) / "trim-test-root"
+        qemu = root / "sdk/emulator/qemu/windows-x86_64/qemu-system-x86_64.exe"
+        adb = root / "sdk/platform-tools/adb.exe"
+        listener = lambda port, path, protocol="tcp": dict(port=port, pid=1234, protocol=protocol, executable=str(path))
+        good = [listener(p, qemu) for p in (27400, 27401, 27402)] + [listener(p, adb) for p in (27498, 27499)]
+        checked = check_listeners(root, 27400, 27498, 27499, 0, good)
+        self.assertEqual(len(checked), 5)
+        self.assertTrue(all("executable" not in row for row in checked))
+        headless = listener(27400, qemu.with_name("qemu-system-x86_64-headless.exe"))
+        self.assertEqual(check_listeners(root, 27400, 27498, 27499, 0, [headless])[0]["role"], "qemu")
+        bad = [listener(27497, adb), listener(27410, qemu), listener(27498, qemu),
+               listener(27499, adb, "udp"), listener(27400, root.parent / "other-root/qemu.exe"),
+               dict(port=27400, pid=1234, protocol="tcp", executable=None)]
+        for row in bad:
+            with self.subTest(listener=row), self.assertRaisesRegex(ValueError, "unverified process"):
+                check_listeners(root, 27400, 27498, 27499, 0, [row])
+
+
+if __name__ == "__main__":
+    unittest.main()