diff --git a/CMakeLists.txt b/CMakeLists.txt index b6703557..bed67f32 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -105,6 +105,10 @@ termforge_warnings(search_test) add_executable(paste_test experiments/shell/paste_test.cpp) termforge_warnings(paste_test) +add_executable(sender_fence_test experiments/shell/sender_fence_test.cpp) +termforge_warnings(sender_fence_test) +find_package(Threads REQUIRED) +target_link_libraries(sender_fence_test PRIVATE Threads::Threads) add_executable(clipboard_policy_test experiments/shell/clipboard_policy_test.cpp) termforge_warnings(clipboard_policy_test) add_executable(keyboard_test experiments/shell/keyboard_test.cpp) @@ -293,6 +297,8 @@ if(BUILD_TESTING) add_test(NAME selection_core COMMAND selection_test) add_test(NAME search_core COMMAND search_test) add_test(NAME paste COMMAND paste_test) + add_test(NAME sender_fence COMMAND sender_fence_test) + set_tests_properties(sender_fence PROPERTIES TIMEOUT 10) add_test(NAME clipboard_policy COMMAND clipboard_policy_test) add_test(NAME keyboard COMMAND keyboard_test) add_test(NAME text_action_policy COMMAND text_action_policy_test) diff --git a/docs/EVIDENCE.md b/docs/EVIDENCE.md index 52600ade..0819d3cf 100644 --- a/docs/EVIDENCE.md +++ b/docs/EVIDENCE.md @@ -48,6 +48,7 @@ R1 固定总出口的源码为 `ce9b24d`，早期 `2dc8d6d` / `9b16acc` 仍分 | 纯 AST 夹具的生命周期记录 | [#40](https://github.com/JavaZeroo/termforge/issues/40)、[计时、进度与回收验证](../research/ime/results/2026-10-26-inventory-lifecycle/validation.json)、[专用 CI 40253727745](https://github.com/JavaZeroo/termforge/actions/runs/40253727745) | 旧 10 秒 AST 夹具及专用 CI 记录启动、AST 和 Job0；原超时根因未确定。诊断增强见 #40 阶段保留，不由一次 CI 成功结项。 | | 原生 margin 逐字段合同与修复候选（#56 仍开放） | [合同、源码与路径](research/conpty-native-margin-path.md)、[独立 oracle](../experiments/conpty/margin_contract.py)、[固定运行](../experiments/conpty/results/2026-10-26-native-margin-path/validation.json) | 四阶段文字/格属性/光标独立判据；系统与官方原生 final 仍红，官方80列两次原5秒生命周期失败保留。共享单 margin 源码支持小 MIT fork 候选；patch 仅可应用，未编译、未运行、不计修复。 | | ConPTY 系统/固定官方后端的同源 margin A/B（#41、#47 未关闭） | [合同与第一方来源](research/conpty-margin-backends.md)、[固定 175B 与 16 case 结果](../experiments/conpty/results/2026-10-26-margin-backends/validation.json) | 相同 175B、两网格/两 mode/两 fresh：系统末端 margin 错，固定官方后端的末端正确；CHAR_INFO 仍错。禁用 VT 由窄屏反例拒绝；仅诊断可行性，不证明 PR #50 或 main 联合整合通过。 | +| 实验输入 worker 持块与停止（#62） | [sender fence 合同](research/sender-fence.md)、[固定源码与实际字段](../experiments/shell/results/2026-10-26-sender-fence/validation.json) | `519a3f3` 的系统/官方各两进程、每次 stop/cancel/竞争，共 12 条实际 worker 记录，均 join/Job0；另有各一次原 focus 单测和各 2/2 纯 CTest。明确修正旧 never-read/output-first 不能保证 16KiB 的夹具假设，同 EXE 旧顺序仍红。实现仅在 PR #50 实验整合，main 未包含；probe 未链接候选 VT，纯 EOF 为模拟，#53 UI/3MB/768B/八会话及完整官方矩阵未验收。 | | 工作区几何与交互的固定记录 | [#38](https://github.com/JavaZeroo/termforge/issues/38)、[工作区几何与输出验证](../experiments/shell/results/2026-10-26-workspace-geometry/validation.json)、[resize 限定证据](../experiments/core/results/2026-10-26-resize-replay/validation.json) | 固定 27f376d 的客户区/布局/输出、默认/候选恢复及窄屏记录；其完整回归仍有 #36/#38 红项，不代表当前所有配置或 R2 通过。 | | 短窗口选区夹具与 hosted 四 shell 前置（#38） | [固定锚点回归](../experiments/shell/test_shell.py)、[准备入口](../tools/prepare_hosted_shells.py)、[夹具收尾合同](research/hosted-wsl-fixture-cleanup.md)、[云端结果与剩余阻塞](https://github.com/JavaZeroo/termforge/issues/38) | 短窗口保留相同选择断言；仅指定一次性 hosted VM 准备固定 Ubuntu。夹具强制收尾与产品自然退出分列，未知归属/回收仍失败。 | | 工作区恢复的运输完成端点（#38 夹具修正） | [限定合同](research/workspace-restore-transport-oracle.md)、[原失败及默认/候选验证](../experiments/shell/results/2026-10-26-workspace-transport-oracle/validation.json) | OSC 标题中的重复标记与 column-one CUP 误导旧端点；新判定保留完整 UTF-8、期限、负载及恢复断言。仅修夹具，不宣称修复产品恢复。 | diff --git a/docs/FEATURES.md b/docs/FEATURES.md index 5425f04a..3de64265 100644 --- a/docs/FEATURES.md +++ b/docs/FEATURES.md @@ -12,7 +12,7 @@ |---|---|---| | 默认 main：自研核心 + 系统 ConPTY | 默认构建与已发布 Alpha 的基础路线，不需要 Rust | 已有 R1 限定验收；VT/Unicode 能力仍为有限子集，不能引用候选核心的字素或 TUI 结果 | | 显式候选 main：`termforge.4` + 系统 ConPTY | `TERMFORGE_BUILD_CANDIDATE_ADAPTER=ON`；按 [ADR-0003](adr/0003-core-reference-and-reuse.md)推进 R2 新终端能力 | [固定来源](../experiments/candidates/alacritty/PROVENANCE.md)、[适配边界](../experiments/core/CANDIDATE.md)与 [#39 字素合同](research/streaming-graphemes.md)；候选正确性通过不关闭系统 ConPTY 或 R2 父项 | -| 实验分支：`termforge.4` + 固定官方 ConPTY | [PR #50（Draft）](https://github.com/JavaZeroo/termforge/pull/50)，联合源码 `3e5b96d` 已重建并[冻结真实基线](research/joint-conpty-baseline.md)，尚未合并或通过产品整体验收 | [#47](https://github.com/JavaZeroo/termforge/issues/47) 跟踪后端整合；终端 margin 与原生状态分列。[resize 残尾 #51](https://github.com/JavaZeroo/termforge/issues/51)、[Console-width Vim #52](https://github.com/JavaZeroo/termforge/issues/52)、[未完成粘贴的确定性取消夹具 #53](https://github.com/JavaZeroo/termforge/issues/53)仍开放，现有 completed 观察不能证明部分发送取消通过或产品 Cancel 故障；另有 worker gate [#62](https://github.com/JavaZeroo/termforge/issues/62) 和工作区 [#63](https://github.com/JavaZeroo/termforge/issues/63) 回归 | +| 实验分支：`termforge.4` + 固定官方 ConPTY | [PR #50（Draft）](https://github.com/JavaZeroo/termforge/pull/50)，联合源码 `3e5b96d` 已重建并[冻结真实基线](research/joint-conpty-baseline.md)，尚未合并或通过产品整体验收 | [#47](https://github.com/JavaZeroo/termforge/issues/47) 跟踪后端整合；终端 margin 与原生状态分列。[resize 残尾 #51](https://github.com/JavaZeroo/termforge/issues/51)、[Console-width Vim #52](https://github.com/JavaZeroo/termforge/issues/52)、[未完成粘贴的确定性取消夹具 #53](https://github.com/JavaZeroo/termforge/issues/53)仍开放，现有 completed 观察不能证明部分发送取消通过或产品 Cancel 故障；worker gate [#62](https://github.com/JavaZeroo/termforge/issues/62) 已有[独立夹具修正与双后端实际 worker 验证](research/sender-fence.md)，默认产品路径不启用 fence，也不替代 #53 的 UI 取消/关闭；工作区 [#63](https://github.com/JavaZeroo/termforge/issues/63) 回归仍开放 | 以下能力默认指已合并实现；某个配置的通过结果不能转用于另一个配置。R1 的真实系统 CF/拼音验收属于固定自研核心配置，不外推到候选 `.4` 或实验后端联合配置。 diff --git a/docs/milestones/R2-local-terminal.md b/docs/milestones/R2-local-terminal.md index 327004f7..33c3655f 100644 --- a/docs/milestones/R2-local-terminal.md +++ b/docs/milestones/R2-local-terminal.md @@ -42,7 +42,7 @@ flowchart LR ## 下一批的任务边界 -1. **优先完成联合基线。** 从主线与 Draft 的明确身份整合，不在旧配置上连续堆修补丁。#51–#53 可以用原冻结输入独立定位，但最终修复验收必须回到联合基线；不会因一次未复现自动获得“已修复”的因果结论。 +1. **在已冻结基线上逐项闭合后端回归。** #55 已交付含红的真实基线；接续 #51–#53 与 #63 的独立失败。#62 的[真实 writer 持块、取消与停止夹具](../research/sender-fence.md)不降低原 4MiB/16KiB/期限，下一步把默认关闭的 fence 接到 #53 原 UI/3MB/八会话入口；这不代表这些 UI 路径已通过。#51–#53 可以用原冻结输入独立定位，最终修复验收须在联合配置重建；一次未复现不自动构成因果修复。 2. **并行处理平台语义与性能。** [原生语义处理路径 #56](https://github.com/JavaZeroo/termforge/issues/56)先明确 #41 原合同需要什么、哪些路径可支持；保留必要的上游或独立实现工作，不用“平台不同”豁免缺口。#54 现在就冻结重复方法、完成端点和预算决定程序，再定位吞吐成本；不等到 #34 才补性能。 3. **在主线推进独立行为合同。** #29 核对极窄/极短、软硬行、历史/淘汰、选择/Find、连续 resize 和 child ack；#30 核对 IME 组合/提交/取消与输出/resize/切换、应用键盘/粘贴接收、实际复制及同次八会话。已验收的 #39 只提供首批字素基础，最终配置的交叉验收仍需执行。 4. **补终端模式和应用矩阵。** #27 明确 DECOM/DECAWM、pending-wrap、保存恢复、异常序列与 margin×resize；#28 先固定应用版本/入口/动作及协议字节预期，再逐应用完成进入、操作、resize、退出、主屏恢复。Vim 的一个修复不替代完整 TUI 验收。 diff --git a/docs/research/sender-fence.md b/docs/research/sender-fence.md new file mode 100644 index 00000000..f48f492b --- /dev/null +++ b/docs/research/sender-fence.md @@ -0,0 +1,63 @@ +# 实际输入 worker 的持块合同与 sender fence + +本合同对应 [#62](https://github.com/JavaZeroo/termforge/issues/62)，为 [#53](https://github.com/JavaZeroo/termforge/issues/53)提供可复用的协调机制。**实现属于 `fix/conpty-backend-r47` / [Draft #50](https://github.com/JavaZeroo/termforge/pull/50) 的实验整合，尚未进入 main。** 固定运行源码为 `519a3f37d52983062cd99b7efd805f183961fa4d`；完整 source/tree/blob、实际 EXE、官方后端文件对、原始日志 SHA 和逐字段结果见[验证记录](../../experiments/shell/results/2026-10-26-sender-fence/validation.json)。这是实际 `InputQueue` / `PasteTransfer` / ConPTY worker 的限定验证，不是 UI、VT 或操作系统背压验收。 + +## 修正的夹具假设 + +旧夹具先启动输出洪流并将 reader 持在 gate，再向完全不读输入的 child 发送 4MiB ASCII。它要求观察到 writer 已提交至少 16KiB，但固定官方后端的一次实际拒绝状态是 `sent=5120 / accepted=6144 / in_flight=1024`，未达到原门槛。不能假设所有宿主都能在 child 不读且输出被持住时预接收 16KiB；本批没有测定内核或宿主缓冲容量，也没有证明其是唯一影响因素。 + +原 `af563ba` 的 focus 单测真实失败及只有 EXE 身份的 `2ac023b…` 诊断保留；最终同一 EXE 的 `--ungated-writer-observation` 仍按旧顺序返回 2，并再次给出上述低 sent 状态。**没有把旧顺序改写为已通过。** 首次 Python 准备缺少 `termforge_shell.exe`、运行零项测试，单独归为构建前置失败，不计产品失败。缺少固定源码绑定的早期 WIP 绿色不计最终验证。 + +## 新的明确顺序 + +4MiB 请求、真实 1024B 分块、16KiB 已发送门槛、reader/output gate、resize 与回收检查保留。夹具明确改变交错顺序： + +1. child 用实际控制台读取核对恰好 16KiB `a`，之后不再读输入；只在核对完成后设置独立 primed event。 +2. 原 writer 经真实 `take_chunk()` 领取第 17 块，在 queue 发布 `in_flight=1024` 并解锁后，由 sender fence 持住。此时真实计数为 `sent=16384 / accepted=17408 / total=4194304`。 +3. parent 同时确认 held 状态和 child primed event，再开启原 flood phase、使 reader 进入原输出 gate。 +4. 执行真实 stop、cancel 或 cancel/stop 竞争；继续原 resize generation、child 实际尺寸 102×42、自然退出及自有 Session Job0 检查。 + +sender fence 只能固定达到门槛后的观察，不能使原来在门槛之前阻塞的 OS 写入自行前进。held 表示 writer 拥有一块真实 payload，不表示 `WriteFile` 已进入或被系统阻塞。输出 gate 的记录来自 reader callback 进入；本批没有独立 flood 字节计数或实际呈现端点。 + +## 生命周期与接口 + +`SenderFence` 是绑定单次 transfer 的测试控制面，普通产品 `InputQueue::begin_paste(transfer)` 默认不启用；队列探针显式传入 fence。构造时固定 transfer 身份、非零 generation 和绝对 deadline，`begin_paste(transfer, fence)` 在通知 writer 前完成单次 admission。绑定不同 transfer、重复 admission、错误 ticket、未 held 就 release、重复 release 和过期 release 均被拒绝。 + +- `snapshot(ticket)` 按值读取 held 状态及真实 progress；不读取或复制输入正文。 +- `release(ticket)` 只允许匹配的 held ticket 在 deadline 前继续发送，不生成写确认。 +- `InputQueue::cancel_paste()` 先调用真实 `PasteTransfer::cancel()`，再通知 fence；已领取块继续原 `Session::send → WriteFile → acknowledge_written → finish_chunk`，最终取消保持 `sent==accepted`。 +- `InputQueue::stop()` 先发布 stopping 并唤醒 fence，再沿原取消同步 I/O 与 join 路径退出。writer 醒来后重新检查 stopping；尚未提交的持块由原 writer failure 清理，不能伪造 ack 或直接当作已发送。 +- `abort(ticket, control_eof)` 将控制面中止表现为失败；本批 EOF 仅是纯协议中的方法调用模拟，没有执行实际 IPC EOF。 + +等待 fence 时不持 queue mutex；cancel/stop 在释放 queue mutex 后通知 fence，防止与 join 形成锁环。`CancelSynchronousIo` 本身不能唤醒 condition-variable wait。fence 强持有绑定 transfer，writer 和 queue 持有 fence 到等待/退出结束。 + +达到持块、child primed、reader gate 及取消完成的观察共同使用原 3 秒绝对期限；没有每阶段重新延长该观察期限。其他既有 ready、resize、child 尺寸及退出等待分别保留 3 秒，原 Python probe 进程外层仍为 10 秒。纯协议缺 GO 使用 1 秒绝对 deadline，单独证明有限等待；这些带同步的运行不是性能样本。 + +## 固定结果与计数 + +最终原始批次有**两种后端各两次 probe 进程运行，每次实际执行三个 worker 动作，共 12 条逐字段审计记录**。原始 launcher 标签 `held-stop-0/1` 不表示只运行 stop。另有两种配置各一次原 focus Python 单测通过，单独记录，不重复并入上述 12 条矩阵。两种配置的 `paste` / `sender_fence` CTest 各 2/2。 + +| 动作 | 动作前 sent / accepted / in_flight | 动作后 sent / accepted / in_flight | 实际状态与回收 | +|---|---|---|---| +| stop | 16384 / 17408 / 1024 | 16384 / 17408 / 0 | failed，retained capacity 15，Job0 | +| cancel | 16384 / 17408 / 1024 | 17408 / 17408 / 0 | cancelled，retained capacity 15，Job0 | +| cancel/stop 竞争 | 16384 / 17408 / 1024 | 本批为 16384 / 17408 / 0 | 本批均由 stop 获胜、failed、Job0；不能宣称两个交错都被实际观测 | + +所有正例保持 input join、resize working/pending/completed generation 1/3/3 和 child 102×42；这些 generation 字段属于 **resize worker**，不是 sender ticket 的独立运行日志。官方配置的构建定义及固定包选择 Console flags24，系统配置 flags0；worker JSON 没有独立输出 flags，证据来源在验证记录中分列。 + +`shell_queue_probe` 没有链接候选 VT 核心。构建目录叫 `candidate` / `candidate-system`，以及适配开关为 ON，均不能把该结果扩大为候选 VT、Unicode 或 TUI 验收。 + +## 复验与剩余边界 + +在已经明确配置并按贡献指南完成构建的目录运行： + +```powershell +cmake --build build/candidate --config Release --target shell_queue_probe paste_test sender_fence_test +ctest --test-dir build/candidate -C Release -R '^(paste|sender_fence)$' --no-tests=error --output-on-failure +$env:TERMFORGE_BIN_DIR = (Resolve-Path build/candidate/bin/Release).Path +python -m unittest experiments.shell.test_shell.ShellTests.test_queues_apply_bounded_backpressure_and_reject_input -v +``` + +系统配置单独使用对应构建目录，不能复用另一 EXE 的身份。旧顺序诊断使用 `shell_queue_probe --ungated-writer-observation`，由 10 秒 subprocess 监督，保留预期 return2 与具体进度；不能将它作为需要重跑取绿的产品测试。 + +#53 的真实 UI cancel/close、原 3,000,000B 汉字负载、768B receiver checkpoint、真实 flood、完整输入前缀及同次八会话尚未由本批验收。后续必须在发送前预绑定 fence，并让真实 UI 取消进入 queue 的取消通知入口；不能把本夹具额外读取 16KiB 的 priming 原样移到必须保持 768B checkpoint 的 #53。完整官方回归、长期资源、呈现、平台和 WindTerm 性能差距继续独立处理。 diff --git a/experiments/shell/README.md b/experiments/shell/README.md index 9df2016b..e4196662 100644 --- a/experiments/shell/README.md +++ b/experiments/shell/README.md @@ -64,6 +64,7 @@ R1 的 [#12–#17 限定合同](../../docs/milestones/R1-local-text.md)和 [#18 - ConPTY 输出线程将最多 64 KiB 的读取块放入 256 KiB 队列；满时等待窗口消费，正常运行不丢弃字节。 - UI 每个 16 ms 定时消息以 8 KiB 小块消费，最多 64 KiB；块间检查 4 ms 预算，随后发布一次快照。单次 feed、快照和绘制仍可能超过预算，它不是硬实时保证。解析和快照发布同属 UI 线程；dirty 范围包含内容与旧/新光标行，多次失效可合并为一次 WM_PAINT。 - 输入写线程独立服务至多 16 KiB 的队列，每条消息不超过 1 KiB。UI 不等待 WriteFile；容量满时拒绝该次输入并发出提示，不宣称输入已成功。关闭期间停止接收新输入。 +- PR #50 实验分支的原生 worker 测试可显式预绑定 [sender fence](../../docs/research/sender-fence.md)，main 尚未包含：仅在真实已提交至少 16KiB、领取下一块并发布 in-flight 后持住 writer，cancel/stop 使用独立唤醒路径。普通产品默认关闭。队列探针先让 child 核验恰好 16KiB 后停止读，再启动原 flood/output gate；这是对宿主预接收容量假设的明确夹具修正，旧顺序可用 `--ungated-writer-observation` 保留。两种后端实际 stop/cancel/竞争与纯协议结果见[固定验证](results/2026-10-26-sender-fence/validation.json)，不代表 OS 写阻塞、候选 VT 或 #53 的 UI/3MB/768B 工作流通过。 - resize 由独立 worker 执行，最多一个在途和一个合并请求；UI 先更新核心与快照并继续消费字节。关闭在回收 ConPTY 之前 join 此 worker。字节流没有 resize 世代边界，不保证所有中间帧及历史与 ConPTY 自身重排完全一致。 - 关闭工作在后台执行，UI 继续消费尾部输出；停止输入写入并反复取消该线程的同步 I/O，终止本会话 Job，关闭 ConPTY，等待读取线程退出。队列排空后执行 `finish()`、提交最终画面并销毁窗口。只终止自有 Job 内的进程，包括 shell 留下的后代。 - 正常 `exit 37` 保留实际退出码 37；关闭正在输出的会话通过 Job 取消。外部自动监督器另建拥有范围明确的 Job：宿主收到 `GO` 前不创建 shell，超时后回收整个自有树。 diff --git a/experiments/shell/queue_probe.cpp b/experiments/shell/queue_probe.cpp index 64e05dd9..62a3bc0c 100644 --- a/experiments/shell/queue_probe.cpp +++ b/experiments/shell/queue_probe.cpp @@ -20,7 +20,7 @@ static std::wstring executable_path() { if (!size || size == path.size()) tf_conpty::fail("GetModuleFileName"); path.resize(size); return path; } -static int child(const wchar_t* ready_name) { +static int child(const wchar_t* ready_name, bool prime_input) { tf_conpty::Handle ready(OpenEventW(EVENT_MODIFY_STATE, FALSE, ready_name)); if (!ready.get()) tf_conpty::fail("OpenEvent(queue child)"); const std::wstring query_name = std::wstring(ready_name) + L"-query"; @@ -31,8 +31,27 @@ static int child(const wchar_t* ready_name) { !SetConsoleMode(output, ENABLE_PROCESSED_OUTPUT | ENABLE_VIRTUAL_TERMINAL_PROCESSING)) tf_conpty::fail("SetConsoleMode(queue child)"); if (!SetEvent(ready.get())) tf_conpty::fail("SetEvent(queue child)"); - // Never reads input. Real console output fills while the parent reader is - // held at an explicit gate. No system clipboard or desktop input is used. + size_t prefix_read = 0; + if (prime_input) { + tf_conpty::Handle primed(OpenEventW(EVENT_MODIFY_STATE, FALSE, (std::wstring(ready_name) + L"-primed").c_str())); + tf_conpty::Handle flood(OpenEventW(SYNCHRONIZE, FALSE, (std::wstring(ready_name) + L"-flood").c_str())); + check(primed.get() && flood.get(), "open queue input phase gates"); + // A host's unread-input capacity is not a portable 16 KiB guarantee. + // Consume and verify exactly that prefix, then never read again. The + // sender still submits the original 4 MiB through the real InputQueue. + while (prefix_read < 16384) { + char bytes[256]; DWORD count = 0; + check(ReadFile(input, bytes, sizeof bytes, &count, nullptr) && count && count <= 16384 - prefix_read, + "read queue priming prefix"); + check(std::string_view(bytes, count).find_first_not_of('a') == std::string_view::npos, + "queue priming prefix differs from independent ASCII oracle"); + prefix_read += count; + } + check(SetEvent(primed.get()) != FALSE, "signal verified queue priming prefix"); + check(WaitForSingleObject(flood.get(), 3000) == WAIT_OBJECT_0, "queue flood phase release"); + } + // From this point input is never read. Output fills while the parent reader + // is held at its explicit gate; the old observation mode never primes input. const std::string bytes(65536, 'F'); for (;;) { if (WaitForSingleObject(query.get(), 0) == WAIT_OBJECT_0) { @@ -44,7 +63,8 @@ static int child(const wchar_t* ready_name) { std::this_thread::yield(); } const std::string reply = "\r\nTFQUEUE-SIZE:" + std::to_string(dimensions.dwSize.X) + "x" + - std::to_string(dimensions.dwSize.Y) + ":END\r\n"; + std::to_string(dimensions.dwSize.Y) + ":END\r\nTFQUEUE-PRIMED:" + + std::to_string(prefix_read) + ":END\r\n"; DWORD count = 0; if (!WriteFile(output, reply.data(), static_cast(reply.size()), &count, nullptr) || count != reply.size()) tf_conpty::fail("WriteFile(queue size)"); @@ -128,9 +148,11 @@ static void lifecycle_edges() { std::cout << "{\"case\":\"lifecycle-edges\",\"stopped_uncompleted_generation\":3,\"input_error_preserved\":true," "\"resize_error_preserved\":true}\n"; } -static void real_workers() { +enum class WriterAction { stop, cancel, cancel_stop_race }; +static void real_workers(bool fence_sender, WriterAction action) { std::mutex gate_mutex; std::condition_variable gate_changed; bool reader_entered = false, reader_released = false, resize_entered=false, resize_released=false; + bool reader_gate_enabled = !fence_sender; OutputQueue output; tf_conpty::Session session; InputQueue input(session); @@ -154,27 +176,96 @@ static void real_workers() { if (!ready.get()) tf_conpty::fail("CreateEvent(queue ready)"); tf_conpty::Handle query(CreateEventW(nullptr, FALSE, FALSE, (name + L"-query").c_str())); if (!query.get()) tf_conpty::fail("CreateEvent(queue query)"); + tf_conpty::Handle primed(CreateEventW(nullptr, TRUE, FALSE, (name + L"-primed").c_str())); + tf_conpty::Handle flood(CreateEventW(nullptr, TRUE, FALSE, (name + L"-flood").c_str())); + check(primed.get() && flood.get(), "create queue input phase gates"); try { - session.start(executable_path(), L"--queue-child \"" + name + L"\"", [&](std::string_view bytes) { - { std::unique_lock lock(gate_mutex); reader_entered = true; gate_changed.notify_all(); - gate_changed.wait(lock, [&] { return reader_released; }); } + session.start(executable_path(), (fence_sender ? L"--queue-child-primed \"" : L"--queue-child \"") + name + L"\"", [&](std::string_view bytes) { + { std::unique_lock lock(gate_mutex); + if (reader_gate_enabled) { + reader_entered = true; gate_changed.notify_all(); + gate_changed.wait(lock, [&] { return reader_released; }); + } } output.push(bytes); }); check(WaitForSingleObject(ready.get(), 3000) == WAIT_OBJECT_0, "real console child ready"); - { std::unique_lock lock(gate_mutex); + if (!fence_sender) { std::unique_lock lock(gate_mutex); check(gate_changed.wait_for(lock, std::chrono::seconds(3), [&] { return reader_entered; }), "real output reader reached gate"); } tf_shell::PreparedPaste prepared; prepared.bytes.assign(4 * 1024 * 1024, 'a'); auto transfer = std::make_shared(1, std::move(prepared)); - check(input.begin_paste(transfer), "real input transfer accepted"); input.start(); + const auto observation_deadline = std::chrono::steady_clock::now() + std::chrono::seconds(3); + auto fence = fence_sender ? std::make_shared(transfer, 1, observation_deadline) : nullptr; + check(input.begin_paste(transfer, fence), "real input transfer accepted"); input.start(); InputQueue::Snapshot held{}; - await([&] { held = input.snapshot(); return held.in_flight_bytes && transfer->progress().sent >= 16384; }, - "real input writer holds a chunk"); + try { + while (true) { + held = input.snapshot(); + const bool held_fence = !fence || fence->snapshot(fence->ticket()).state == tf_shell_test::SenderFence::State::held; + if (held_fence && held.in_flight_bytes && transfer->progress().sent >= 16384) break; + check(std::chrono::steady_clock::now() < observation_deadline, "real input writer holds a chunk"); + std::this_thread::yield(); + } + } catch (...) { + const auto observed = transfer->progress(); + bool observed_reader_entered = false, observed_reader_released = false; + { std::lock_guard lock(gate_mutex); + observed_reader_entered = reader_entered; observed_reader_released = reader_released; } + std::cerr << "{\"case\":\"writer-observation-rejected\",\"total\":" << observed.total + << ",\"accepted\":" << observed.accepted << ",\"sent\":" << observed.sent + << ",\"paste_in_flight\":" << observed.in_flight << ",\"queue_in_flight\":" << held.in_flight_bytes + << ",\"state\":" << static_cast(observed.state) + << ",\"reader_entered\":" << observed_reader_entered << ",\"reader_released\":" << observed_reader_released << "}\n"; + throw; + } + if (fence_sender) { + while (WaitForSingleObject(primed.get(), 0) != WAIT_OBJECT_0) { + check(std::chrono::steady_clock::now() < observation_deadline, "real child verified the 16 KiB prefix"); + std::this_thread::yield(); + } + { std::lock_guard lock(gate_mutex); reader_gate_enabled = true; } + check(SetEvent(flood.get()) != FALSE, "release real queue child output flood"); + std::unique_lock lock(gate_mutex); + check(gate_changed.wait_until(lock, observation_deadline, [&] { return reader_entered; }), "real output reader reached gate"); + } check(!held.queued_bytes && held.in_flight_bytes <= 1024 && held.started && !held.joined && !held.stopping && !held.failed, "dequeued input remains in flight"); const auto before_stop = transfer->progress(); + if (action == WriterAction::cancel) { + input.cancel_paste(); + while (!tf_shell::paste_terminal(transfer->progress().state)) { + check(std::chrono::steady_clock::now() < observation_deadline, "real queue cancellation finishes its claimed chunk"); + std::this_thread::yield(); + } + const auto cancelled = transfer->progress(); + check(cancelled.state == tf_shell::PasteState::cancelled && cancelled.sent == before_stop.accepted && + cancelled.accepted == before_stop.accepted && !cancelled.in_flight && cancelled.retained_capacity <= 32, + "real queue cancellation uses send acknowledgement for its claimed chunk"); + } else if (action == WriterAction::cancel_stop_race) { + std::promise start; + const auto go = start.get_future().share(); + std::thread cancel([&] { go.wait(); input.cancel_paste(); }); + start.set_value(); input.stop(); cancel.join(); + } input.stop(); const auto joined_input = input.snapshot(); check(joined_input.started && joined_input.joined && joined_input.stopping && !joined_input.queued_bytes && !joined_input.in_flight_bytes && !joined_input.failed, "input stop joins and releases held payload"); + if (fence_sender) { + const auto stopped_fence = fence->snapshot(fence->ticket()); + const auto stopped_paste = transfer->progress(); + check(stopped_fence.ever_held && stopped_fence.wait_finished && + stopped_fence.state == tf_shell_test::SenderFence::State::aborted && + stopped_fence.abort_reason == tf_shell_test::SenderFence::AbortReason::stop, + "stop wakes the held writer without a control GO"); + check(tf_shell::paste_terminal(stopped_paste.state) && !stopped_paste.in_flight && + stopped_paste.state != tf_shell::PasteState::completed && + stopped_paste.sent >= before_stop.sent && stopped_paste.sent <= before_stop.accepted && + stopped_paste.accepted == before_stop.accepted && + stopped_paste.retained_capacity <= 32, + "stopped pre-send chunk is released without a fabricated write acknowledgement"); + if (action == WriterAction::stop) + check(stopped_paste.state == tf_shell::PasteState::failed && stopped_paste.sent == before_stop.sent, + "stop never submits its pre-send held chunk"); + } resize.request(40, 100); resize.start(); {std::unique_lock lock(gate_mutex); check(gate_changed.wait_for(lock,std::chrono::seconds(3),[&] {return resize_entered;}),"resize worker reaches explicit pre-API gate");} @@ -196,22 +287,33 @@ static void real_workers() { check(joined_resize.started && joined_resize.joined && joined_resize.stopping && !joined_resize.working && !joined_resize.pending, "resize stop joins worker"); if (!SetEvent(query.get())) tf_conpty::fail("SetEvent(queue size query)"); - std::string tail; + std::string tail; bool size_seen = false, prefix_seen = false; await([&] { tail += output.take(65536); - const bool found = tail.find("TFQUEUE-SIZE:102x42:END") != std::string::npos; + size_seen = size_seen || tail.find("TFQUEUE-SIZE:102x42:END") != std::string::npos; + prefix_seen = prefix_seen || tail.find(fence_sender ? "TFQUEUE-PRIMED:16384:END" : "TFQUEUE-PRIMED:0:END") != std::string::npos; if (tail.size() > 128) tail.erase(0, tail.size() - 128); - return found; + return size_seen && prefix_seen; }, "real child observes latest requested dimensions"); check(session.wait_exit(3000) && session.exit_code == 0, "real queue child exits normally after size report"); cleanup(); check(session.cleanup_complete && !session.active_after && session.child_signaled && session.exit_code == 0, "real queue fixture Job cleanup"); - std::cout << "{\"case\":\"real-workers\",\"input_queued\":" << held.queued_bytes + const char* action_name = action == WriterAction::stop ? "stop" : + action == WriterAction::cancel ? "cancel" : "cancel-stop-race"; + const auto after = transfer->progress(); + std::cout << "{\"case\":\"real-workers\",\"action\":\"" << action_name << "\",\"input_queued\":" << held.queued_bytes << ",\"input_in_flight\":" << held.in_flight_bytes << ",\"input_joined\":true,\"working_generation\":" << coalesced.working_generation << ",\"pending_generation\":" << coalesced.pending_generation << ",\"completed_generation\":" << complete.completed_generation << ",\"child_columns\":102,\"child_rows\":42,\"input_submitted_before_stop\":" << before_stop.sent + << ",\"input_accepted_before_stop\":" << before_stop.accepted + << ",\"paste_in_flight_before_stop\":" << before_stop.in_flight + << ",\"total\":" << before_stop.total << ",\"child_prefix_read\":" << (fence_sender ? 16384 : 0) + << ",\"sender_fence_held\":" << (fence_sender ? "true" : "false") + << ",\"sent_after\":" << after.sent << ",\"accepted_after\":" << after.accepted + << ",\"in_flight_after\":" << after.in_flight << ",\"retained_capacity_after\":" << after.retained_capacity + << ",\"state_after\":" << static_cast(after.state) << ",\"job_active_after\":0}\n"; } catch (...) { const auto original = std::current_exception(); @@ -221,14 +323,21 @@ static void real_workers() { } int wmain(int argc, wchar_t** argv) { try { - if (argc == 3 && std::wstring(argv[1]) == L"--queue-child") return child(argv[2]); + if (argc == 3 && std::wstring(argv[1]) == L"--queue-child") return child(argv[2], false); + if (argc == 3 && std::wstring(argv[1]) == L"--queue-child-primed") return child(argv[2], true); + bool fence_sender = true; + if (argc == 2 && std::wstring(argv[1]) == L"--ungated-writer-observation") { fence_sender = false; argc = 1; } if (argc == 2 && std::wstring(argv[1]) == L"--supervised") { std::string command; check(static_cast(std::getline(std::cin, command)) && command == "GO", "owned supervisor gate"); argc = 1; } check(argc == 1, "unknown queue probe arguments"); - bounded_queues(); lifecycle_edges(); real_workers(); + bounded_queues(); lifecycle_edges(); real_workers(fence_sender, WriterAction::stop); + if (fence_sender) { + real_workers(true, WriterAction::cancel); + real_workers(true, WriterAction::cancel_stop_race); + } std::cout << "bounded queues, backpressure, ordering, in-flight resize and stop verified\n"; return 0; } catch (const std::exception& e) { std::cerr << e.what() << '\n'; return 2; } diff --git a/experiments/shell/queues.hpp b/experiments/shell/queues.hpp index 7d8e4480..0cc8379a 100644 --- a/experiments/shell/queues.hpp +++ b/experiments/shell/queues.hpp @@ -2,6 +2,7 @@ #pragma once #include "../conpty/session.hpp" #include "paste.hpp" +#include "sender_fence.hpp" #include #include @@ -58,6 +59,7 @@ class InputQueue { std::exception_ptr error_; std::thread writer_; std::shared_ptr paste_; + std::shared_ptr sender_fence_; public: struct Snapshot { size_t queued_bytes = 0, in_flight_bytes = 0, high_water_bytes = 0, rejected = 0; @@ -87,17 +89,27 @@ class InputQueue { (paste_ && !tf_shell::paste_terminal(paste_->progress().state)); }); if (stopping_) break; std::string bytes; + std::shared_ptr fence; writing_paste.reset(); if (!chunks_.empty()) { bytes = std::move(chunks_.front()); chunks_.pop_front(); bytes_ -= bytes.size(); } else { writing_paste = paste_; bytes = writing_paste->take_chunk(); + fence = sender_fence_; high_water = (std::max)(high_water, bytes.size()); } in_flight_bytes_ = bytes.size(); lock.unlock(); if (bytes.empty()) continue; // Cancelled between the predicate and take. if (writing_paste) { + if (fence) { + fence->before_send(writing_paste, bytes.size()); + // A CV wait is not cancelled by CancelSynchronousIo. + // stop() first publishes stopping and wakes it; do + // not start a new write after that observation. + std::lock_guard stopped(mutex_); + if (stopping_) throw std::runtime_error("input stopped at sender fence"); + } session_.send(bytes, [&](size_t written) { writing_paste->acknowledge_written(written); }); writing_paste->finish_chunk(); } else session_.send(bytes); @@ -132,15 +144,30 @@ class InputQueue { // handle and target identity; later tab switches never retarget this queue. // Previously accepted keystrokes drain first. New keystrokes are rejected // while pasting, so they cannot interleave with clipboard bytes. - bool begin_paste(const std::shared_ptr& paste) { + bool begin_paste(const std::shared_ptr& paste, + const std::shared_ptr& fence = {}) { std::lock_guard lock(mutex_); if (error_) std::rethrow_exception(error_); - if (stopping_ || !paste || (paste_ && !tf_shell::paste_terminal(paste_->progress().state)) || !paste->claim_writer()) return false; - paste_ = paste; ready_.notify_one(); return true; + if (stopping_ || !paste || (paste_ && !tf_shell::paste_terminal(paste_->progress().state))) return false; + if (fence && !fence->attach(paste)) return false; + if (!paste->claim_writer()) { + if (fence) fence->abort(fence->ticket()); + return false; + } + paste_ = paste; sender_fence_ = fence; ready_.notify_one(); return true; + } + void cancel_paste() { + std::shared_ptr fence; + { std::lock_guard lock(mutex_); + if (paste_) paste_->cancel(); + fence = sender_fence_; ready_.notify_all(); } + if (fence) fence->cancelled(fence->ticket()); } void stop() noexcept { + std::shared_ptr fence; { std::lock_guard lock(mutex_); stopping_ = true; chunks_.clear(); bytes_ = 0; - if (paste_) paste_->cancel(); ready_.notify_all(); } + if (paste_) paste_->cancel(); fence = sender_fence_; ready_.notify_all(); } + if (fence) fence->abort(fence->ticket(), tf_shell_test::SenderFence::AbortReason::stop); if (writer_.joinable()) { // Retry covers the race where cancellation runs just before WriteFile. while (WaitForSingleObject(writer_.native_handle(), 10) == WAIT_TIMEOUT) diff --git a/experiments/shell/results/2026-10-26-sender-fence/validation.json b/experiments/shell/results/2026-10-26-sender-fence/validation.json new file mode 100644 index 00000000..f8349890 --- /dev/null +++ b/experiments/shell/results/2026-10-26-sender-fence/validation.json @@ -0,0 +1,811 @@ +{ + "schema": "termforge-sender-fence-validation-v1", + "date": "2026-10-26", + "issues": [ + 62, + 53 + ], + "source_commit": "519a3f37d52983062cd99b7efd805f183961fa4d", + "source_tree_git_sha1": "add42d40b4e1a5bf62bb0fc51fc4396849619d88", + "source_files": { + "CMakeLists.txt": { + "git_blob_sha1": "bed67f32a924786aa497927328194a81acaf4641", + "git_blob_sha256": "e95d1cefe78af2b86c7e8d5619f42cb9f974fcb3783daa6a09d8ad824c5d4027", + "checkout_sha256": "e95d1cefe78af2b86c7e8d5619f42cb9f974fcb3783daa6a09d8ad824c5d4027", + "checkout_relation": "identical" + }, + "cmake/ConptyBackend.cmake": { + "git_blob_sha1": "6b0b9910fb36a31e486f9fbe2b33e906b68b43f0", + "git_blob_sha256": "6a9f9fb69f90a1db7ad5eb970e61d0913ce4ff75a121f267a369468344fffcde", + "checkout_sha256": "6a9f9fb69f90a1db7ad5eb970e61d0913ce4ff75a121f267a369468344fffcde", + "checkout_relation": "identical" + }, + "experiments/shell/queues.hpp": { + "git_blob_sha1": "0cc8379a1f52d061228b3d8c442295974e2c7c2d", + "git_blob_sha256": "eadbfe8b6034cc436327bfc2fc49cf2c4463a52af67a809eb586f42cf1baa389", + "checkout_sha256": "eadbfe8b6034cc436327bfc2fc49cf2c4463a52af67a809eb586f42cf1baa389", + "checkout_relation": "identical" + }, + "experiments/shell/sender_fence.hpp": { + "git_blob_sha1": "f17603ba662551664ee38e0dc0004f89913c3a77", + "git_blob_sha256": "04579e11de96237fafea2da48fd2895a4c97f0d407bc7bebf79848540dd1867c", + "checkout_sha256": "04579e11de96237fafea2da48fd2895a4c97f0d407bc7bebf79848540dd1867c", + "checkout_relation": "identical" + }, + "experiments/shell/sender_fence_test.cpp": { + "git_blob_sha1": "770a6e28809e03be7099eac09870b4763016b4cb", + "git_blob_sha256": "de84629f788768064969eea76baffff980f02090e68c33d36e2523b4e32d094a", + "checkout_sha256": "de84629f788768064969eea76baffff980f02090e68c33d36e2523b4e32d094a", + "checkout_relation": "identical" + }, + "experiments/shell/queue_probe.cpp": { + "git_blob_sha1": "62a3bc0c18285aa092dce9368133ad1e9148d5c0", + "git_blob_sha256": "d635b457f79ae0a1add9b630fc11c5ba5d5993070291fcac939252ea5863c4de", + "checkout_sha256": "d635b457f79ae0a1add9b630fc11c5ba5d5993070291fcac939252ea5863c4de", + "checkout_relation": "identical" + }, + "experiments/shell/paste.hpp": { + "git_blob_sha1": "cc72b93d1e4e3fe4537fe950afd51581bdc1373b", + "git_blob_sha256": "f905a693d840256ca7c6a8ffed2601bff41f74e549413f38822db7aceb1c7c82", + "checkout_sha256": "f905a693d840256ca7c6a8ffed2601bff41f74e549413f38822db7aceb1c7c82", + "checkout_relation": "identical" + }, + "experiments/shell/paste_test.cpp": { + "git_blob_sha1": "e3cd94fcf9f97a511144a899a34a5e7ff8889035", + "git_blob_sha256": "f4e056681ffd175928060191bc20cbfdc889c8d6ffa76843eda2365243dff8c4", + "checkout_sha256": "f4e056681ffd175928060191bc20cbfdc889c8d6ffa76843eda2365243dff8c4", + "checkout_relation": "identical" + }, + "experiments/shell/resize.hpp": { + "git_blob_sha1": "20843d154fffdd982e1a9221984245eb0d4cc654", + "git_blob_sha256": "37d22ba31fc3fee8004a3dd555f53fa55cb398057b6576f5a2d9b5403e2f5cd0", + "checkout_sha256": "37d22ba31fc3fee8004a3dd555f53fa55cb398057b6576f5a2d9b5403e2f5cd0", + "checkout_relation": "identical" + }, + "experiments/conpty/session.hpp": { + "git_blob_sha1": "d44f36661594a5b76b62d1ecb95cf166db22e90a", + "git_blob_sha256": "6945e6ffba1d61e66266f867d3cd124eb63c39da37e27662a5eaf544f60ed96b", + "checkout_sha256": "6945e6ffba1d61e66266f867d3cd124eb63c39da37e27662a5eaf544f60ed96b", + "checkout_relation": "identical" + }, + "experiments/conpty/backend.hpp": { + "git_blob_sha1": "3195dceaefa166b1251839108d29c5fd034aa2fb", + "git_blob_sha256": "5ee3ab95dc4b9aefce23ff7900cda2ecdd9cd09d1b1044074de2f5e572c31489", + "checkout_sha256": "5ee3ab95dc4b9aefce23ff7900cda2ecdd9cd09d1b1044074de2f5e572c31489", + "checkout_relation": "identical" + }, + "experiments/shell/test_shell.py": { + "git_blob_sha1": "e4d01513cf8d1169511c688b9819097c176b7360", + "git_blob_sha256": "ed785bc51b85b1bb47217d0430a8c8e80984140709507515a1132d051e603954", + "checkout_sha256": "ed785bc51b85b1bb47217d0430a8c8e80984140709507515a1132d051e603954", + "checkout_relation": "identical" + }, + "experiments/build_paths.py": { + "git_blob_sha1": "3a2b2503bcee700b3adea8e832a4a619b0e1c29a", + "git_blob_sha256": "1236b17dceacda610ec3d6eb1518742c0165af633c05a317f1d64f361791d7d9", + "checkout_sha256": "1236b17dceacda610ec3d6eb1518742c0165af633c05a317f1d64f361791d7d9", + "checkout_relation": "identical" + }, + "third_party/conpty/manifest.json": { + "git_blob_sha1": "a3f7fb23a92762d8decb196a07893628ac759c63", + "git_blob_sha256": "e75e522a3355c0a66c6d60cafa02413a83d8757262747f650b77c959c09ed439", + "checkout_sha256": "e75e522a3355c0a66c6d60cafa02413a83d8757262747f650b77c959c09ed439", + "checkout_relation": "identical" + }, + "third_party/conpty/LICENSE": { + "git_blob_sha1": "8cb179cdb69492ba3b8285e61a67fc1fbfe82cb2", + "git_blob_sha256": "ad0cf28f3381ca9bb0bf101d127402d44c17bfa0991e1a00bff7ae6679e9dada", + "checkout_sha256": "ad0cf28f3381ca9bb0bf101d127402d44c17bfa0991e1a00bff7ae6679e9dada", + "checkout_relation": "identical" + } + }, + "scope": "Windows actual InputQueue/PasteTransfer/ConPTY worker coordination and corrected fixture, not VT/UI/OS-backpressure acceptance", + "configurations": { + "official-console-width": { + "raw_build_directory_label": "candidate", + "configuration": "Release", + "architecture": "x64", + "probe_binaries_sha256": { + "shell_queue_probe.exe": "d12a1c20f9398c48047f0173294772f1d4c0364f17590990a45906fb53849377", + "sender_fence_test.exe": "74235824777afc99616147c24c9860aa737ca3c99f6e86df5c7e8a74e305edfc", + "paste_test.exe": "5138c2a5371f27735cd18002790cb36abebfe34ba948e235ca5bb0f383063806" + }, + "probe_links_candidate_vt_core": false, + "configured_creation_flags": 24, + "flags_evidence": "fixed backend.hpp/configured(), package manifest and per-target build definition; JSON worker records do not independently print backend flags", + "runtime_flags_field_in_worker_json": null, + "official_pair_sha256": { + "conpty.dll": "feeef341d891643c62d30b6b07800bc70f0bc148f44cb8c3bee84aa557ae805a", + "OpenConsole.exe": "3d66b23d0a71bb8eed2b77edc8b9df9bf54ce6c8fb74c863a30e760997f80586" + }, + "cmake_cache_sha256": "339b01918f5cc2864cca3ca6342d83d7add9e781fcdcd923a42ebaa7db77b16e", + "probe_project_sha256": "65ac79d7794f612f60828d2883a0bd83d3558b036fa7a7dd60d064e50729fd6c", + "candidate_adapter_build_option": "ON", + "candidate_adapter_option_is_not_probe_vt_coverage": true + }, + "system": { + "raw_build_directory_label": "candidate-system", + "configuration": "Release", + "architecture": "x64", + "probe_binaries_sha256": { + "shell_queue_probe.exe": "9df846bd32afdc5df45fdf6422cb998dd496f85f027f5d4dc76d3925a115c73c", + "sender_fence_test.exe": "e57b18ac11e83b36141436be02907d76276cf5431de22a2637d475c8612288ef", + "paste_test.exe": "56e40c363f2287b9a9e5b3b530037c5a5428e37a2b165e7857f74cb5f85dd910" + }, + "probe_links_candidate_vt_core": false, + "configured_creation_flags": 0, + "flags_evidence": "fixed backend.hpp/configured(), package manifest and per-target build definition; JSON worker records do not independently print backend flags", + "runtime_flags_field_in_worker_json": null, + "official_pair_sha256": null, + "cmake_cache_sha256": "9691f5adb31684c5fe1433e9187b58dacf4956f9baffcac61932078aa60ed354", + "probe_project_sha256": "a9d786a5f72e3b084f20eccabb98ea4875873320e5afe5dc9257097f6adfc43f", + "candidate_adapter_build_option": "ON", + "candidate_adapter_option_is_not_probe_vt_coverage": true + } + }, + "official_package": { + "schema": "termforge-conpty-package-v1", + "name": "Microsoft.Windows.Console.ConPTY", + "version": "1.25.260930003", + "architecture": "win-x64", + "license": "MIT", + "source_commit": "dc8ae365847d41f62d1ebca93815bb00959bc12f", + "source_tag": "v1.25.2733.0", + "license_source_url": "https://raw.githubusercontent.com/microsoft/terminal/dc8ae365847d41f62d1ebca93815bb00959bc12f/LICENSE", + "license_source_sha256": "5d177f23ecfeb0ea8e050b6a5a16355e1ae9a0b286436ca8f83ed08b3795be6b", + "bundled_license_sha256": "ad0cf28f3381ca9bb0bf101d127402d44c17bfa0991e1a00bff7ae6679e9dada", + "license_normalization": "CRLF to LF only; complete text verified equal", + "package_url": "https://github.com/microsoft/terminal/releases/download/v1.25.2733.0/Microsoft.Windows.Console.ConPTY.1.25.260930003.nupkg", + "package_bytes": 1750533, + "package_sha256": "6d6f8b008c655d814c498d8205d4012dcd4d0b6dfc1096461d0a185d1e3caf03", + "files": { + "conpty.dll": "feeef341d891643c62d30b6b07800bc70f0bc148f44cb8c3bee84aa557ae805a", + "OpenConsole.exe": "3d66b23d0a71bb8eed2b77edc8b9df9bf54ce6c8fb74c863a30e760997f80586" + }, + "package_members": { + "conpty.dll": "runtimes/win-x64/native/conpty.dll", + "OpenConsole.exe": "build/native/runtimes/x64/OpenConsole.exe" + }, + "experimental": true, + "configured_creation_flags": 24, + "configured_glyph_width": "PSEUDOCONSOLE_GLYPH_WIDTH_CONSOLE", + "default_backend": false, + "supported_client_platforms_verified": false + }, + "workload": { + "request_bytes": 4194304, + "request_content": "ASCII a repeated 4194304 times", + "request_sha256": "299285fc41a44cdb038b9fdaf494c76ca9d0c866672b2b266c1a0c17dda60a05", + "write_chunk_limit_bytes": 1024, + "unchanged_minimum_sent_bytes": 16384, + "verified_child_prefix_bytes": 16384, + "verified_child_prefix_sha256": "f3336bea752b5a28743033dd2c844a4a63fba08871aaee2586a2bf2d69be83a2", + "held_claimed_chunk_ordinal_one_based": 17, + "output_read_chunk_limit_bytes": 65536, + "output_queue_capacity_bytes": 262144, + "input_queue_capacity_bytes": 16384 + }, + "fixture_ordering": { + "original": "child never reads; output reader gate held before input transfer", + "corrected": "child really reads and checks exactly 16KiB ASCII then stops reading; actual writer holds the next claimed chunk after 16KiB sent; child prefix event is observed; original flood and reader output gate start; real stop/cancel/race follows", + "ordering_changed": true, + "original_os_backpressure_fixed_claim": false, + "fence_is_not_os_write_block": true, + "actual_transfer_or_ack_replaced": false, + "default_product_fence_enabled": false, + "queue_probe_default_uses_explicit_fence": true + }, + "deadlines_seconds": { + "absolute_input_observation": 3, + "child_ready_wait": 3, + "child_flood_phase_wait": 3, + "resize_gate_wait": 3, + "resize_settle_wait": 3, + "child_size_query_wait": 3, + "natural_exit_wait": 3, + "outer_probe_process": 10, + "pure_missing_go_fence": 1, + "pure_missing_go_future_check": 2, + "sender_fence_ctest": 10 + }, + "runs": [ + { + "configuration": "official-console-width", + "raw_configuration_label": "candidate", + "raw_mode_label": "ungated", + "process_exit": 2, + "outer_timeout_seconds": 10, + "observed_elapsed_seconds": 3.1399999998975545, + "probe_exe_sha256": "d12a1c20f9398c48047f0173294772f1d4c0364f17590990a45906fb53849377", + "stdout_sha256": "6c7b8d3e6e5c826e51f177bd11aaf91f636a86d3616470f9ec61ef0a546c642f", + "stderr_sha256": "10a9ae7cd1bcd8d7bfd7a78282b2afc62e8977127eda8f11c76b191b9afe2007", + "actual_json_records": [ + { + "case": "bounded", + "output_high_water": 262144, + "input_high_water": 16384, + "rejected": 2 + }, + { + "case": "lifecycle-edges", + "stopped_uncompleted_generation": 3, + "input_error_preserved": true, + "resize_error_preserved": true + } + ], + "scenario": "original never-read/output-first ordering", + "verdict": "original observation rejected at unchanged floor", + "actual_stderr_json_records": [ + { + "case": "writer-observation-rejected", + "total": 4194304, + "accepted": 6144, + "sent": 5120, + "paste_in_flight": 1024, + "queue_in_flight": 1024, + "state": 1, + "reader_entered": 1, + "reader_released": 0 + } + ], + "error": "real input writer holds a chunk", + "job_active_after": null, + "job_cleanup_reported_in_this_raw_record": false + }, + { + "configuration": "official-console-width", + "raw_configuration_label": "candidate", + "raw_mode_label": "held-stop-0", + "process_exit": 0, + "outer_timeout_seconds": 10, + "observed_elapsed_seconds": 2.4060000001918525, + "probe_exe_sha256": "d12a1c20f9398c48047f0173294772f1d4c0364f17590990a45906fb53849377", + "stdout_sha256": "7695b76d6631cd239a878a348e5c369e220fdc9a4a95e36f2386550c6f441687", + "stderr_sha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", + "actual_json_records": [ + { + "case": "bounded", + "output_high_water": 262144, + "input_high_water": 16384, + "rejected": 2 + }, + { + "case": "lifecycle-edges", + "stopped_uncompleted_generation": 3, + "input_error_preserved": true, + "resize_error_preserved": true + }, + { + "case": "real-workers", + "action": "stop", + "input_queued": 0, + "input_in_flight": 1024, + "input_joined": true, + "working_generation": 1, + "pending_generation": 3, + "completed_generation": 3, + "child_columns": 102, + "child_rows": 42, + "input_submitted_before_stop": 16384, + "input_accepted_before_stop": 17408, + "paste_in_flight_before_stop": 1024, + "total": 4194304, + "child_prefix_read": 16384, + "sender_fence_held": true, + "sent_after": 16384, + "accepted_after": 17408, + "in_flight_after": 0, + "retained_capacity_after": 15, + "state_after": 5, + "job_active_after": 0 + }, + { + "case": "real-workers", + "action": "cancel", + "input_queued": 0, + "input_in_flight": 1024, + "input_joined": true, + "working_generation": 1, + "pending_generation": 3, + "completed_generation": 3, + "child_columns": 102, + "child_rows": 42, + "input_submitted_before_stop": 16384, + "input_accepted_before_stop": 17408, + "paste_in_flight_before_stop": 1024, + "total": 4194304, + "child_prefix_read": 16384, + "sender_fence_held": true, + "sent_after": 17408, + "accepted_after": 17408, + "in_flight_after": 0, + "retained_capacity_after": 15, + "state_after": 4, + "job_active_after": 0 + }, + { + "case": "real-workers", + "action": "cancel-stop-race", + "input_queued": 0, + "input_in_flight": 1024, + "input_joined": true, + "working_generation": 1, + "pending_generation": 3, + "completed_generation": 3, + "child_columns": 102, + "child_rows": 42, + "input_submitted_before_stop": 16384, + "input_accepted_before_stop": 17408, + "paste_in_flight_before_stop": 1024, + "total": 4194304, + "child_prefix_read": 16384, + "sender_fence_held": true, + "sent_after": 16384, + "accepted_after": 17408, + "in_flight_after": 0, + "retained_capacity_after": 15, + "state_after": 5, + "job_active_after": 0 + } + ], + "scenario": "verified prefix then pre-send held block, flood/output gate, real action", + "verdict": "passed", + "audited_worker_records": 3 + }, + { + "configuration": "official-console-width", + "raw_configuration_label": "candidate", + "raw_mode_label": "held-stop-1", + "process_exit": 0, + "outer_timeout_seconds": 10, + "observed_elapsed_seconds": 2.4840000001713634, + "probe_exe_sha256": "d12a1c20f9398c48047f0173294772f1d4c0364f17590990a45906fb53849377", + "stdout_sha256": "7695b76d6631cd239a878a348e5c369e220fdc9a4a95e36f2386550c6f441687", + "stderr_sha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", + "actual_json_records": [ + { + "case": "bounded", + "output_high_water": 262144, + "input_high_water": 16384, + "rejected": 2 + }, + { + "case": "lifecycle-edges", + "stopped_uncompleted_generation": 3, + "input_error_preserved": true, + "resize_error_preserved": true + }, + { + "case": "real-workers", + "action": "stop", + "input_queued": 0, + "input_in_flight": 1024, + "input_joined": true, + "working_generation": 1, + "pending_generation": 3, + "completed_generation": 3, + "child_columns": 102, + "child_rows": 42, + "input_submitted_before_stop": 16384, + "input_accepted_before_stop": 17408, + "paste_in_flight_before_stop": 1024, + "total": 4194304, + "child_prefix_read": 16384, + "sender_fence_held": true, + "sent_after": 16384, + "accepted_after": 17408, + "in_flight_after": 0, + "retained_capacity_after": 15, + "state_after": 5, + "job_active_after": 0 + }, + { + "case": "real-workers", + "action": "cancel", + "input_queued": 0, + "input_in_flight": 1024, + "input_joined": true, + "working_generation": 1, + "pending_generation": 3, + "completed_generation": 3, + "child_columns": 102, + "child_rows": 42, + "input_submitted_before_stop": 16384, + "input_accepted_before_stop": 17408, + "paste_in_flight_before_stop": 1024, + "total": 4194304, + "child_prefix_read": 16384, + "sender_fence_held": true, + "sent_after": 17408, + "accepted_after": 17408, + "in_flight_after": 0, + "retained_capacity_after": 15, + "state_after": 4, + "job_active_after": 0 + }, + { + "case": "real-workers", + "action": "cancel-stop-race", + "input_queued": 0, + "input_in_flight": 1024, + "input_joined": true, + "working_generation": 1, + "pending_generation": 3, + "completed_generation": 3, + "child_columns": 102, + "child_rows": 42, + "input_submitted_before_stop": 16384, + "input_accepted_before_stop": 17408, + "paste_in_flight_before_stop": 1024, + "total": 4194304, + "child_prefix_read": 16384, + "sender_fence_held": true, + "sent_after": 16384, + "accepted_after": 17408, + "in_flight_after": 0, + "retained_capacity_after": 15, + "state_after": 5, + "job_active_after": 0 + } + ], + "scenario": "verified prefix then pre-send held block, flood/output gate, real action", + "verdict": "passed", + "audited_worker_records": 3 + }, + { + "configuration": "system", + "raw_configuration_label": "candidate-system", + "raw_mode_label": "held-stop-0", + "process_exit": 0, + "outer_timeout_seconds": 10, + "observed_elapsed_seconds": 0.9840000001713634, + "probe_exe_sha256": "9df846bd32afdc5df45fdf6422cb998dd496f85f027f5d4dc76d3925a115c73c", + "stdout_sha256": "7695b76d6631cd239a878a348e5c369e220fdc9a4a95e36f2386550c6f441687", + "stderr_sha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", + "actual_json_records": [ + { + "case": "bounded", + "output_high_water": 262144, + "input_high_water": 16384, + "rejected": 2 + }, + { + "case": "lifecycle-edges", + "stopped_uncompleted_generation": 3, + "input_error_preserved": true, + "resize_error_preserved": true + }, + { + "case": "real-workers", + "action": "stop", + "input_queued": 0, + "input_in_flight": 1024, + "input_joined": true, + "working_generation": 1, + "pending_generation": 3, + "completed_generation": 3, + "child_columns": 102, + "child_rows": 42, + "input_submitted_before_stop": 16384, + "input_accepted_before_stop": 17408, + "paste_in_flight_before_stop": 1024, + "total": 4194304, + "child_prefix_read": 16384, + "sender_fence_held": true, + "sent_after": 16384, + "accepted_after": 17408, + "in_flight_after": 0, + "retained_capacity_after": 15, + "state_after": 5, + "job_active_after": 0 + }, + { + "case": "real-workers", + "action": "cancel", + "input_queued": 0, + "input_in_flight": 1024, + "input_joined": true, + "working_generation": 1, + "pending_generation": 3, + "completed_generation": 3, + "child_columns": 102, + "child_rows": 42, + "input_submitted_before_stop": 16384, + "input_accepted_before_stop": 17408, + "paste_in_flight_before_stop": 1024, + "total": 4194304, + "child_prefix_read": 16384, + "sender_fence_held": true, + "sent_after": 17408, + "accepted_after": 17408, + "in_flight_after": 0, + "retained_capacity_after": 15, + "state_after": 4, + "job_active_after": 0 + }, + { + "case": "real-workers", + "action": "cancel-stop-race", + "input_queued": 0, + "input_in_flight": 1024, + "input_joined": true, + "working_generation": 1, + "pending_generation": 3, + "completed_generation": 3, + "child_columns": 102, + "child_rows": 42, + "input_submitted_before_stop": 16384, + "input_accepted_before_stop": 17408, + "paste_in_flight_before_stop": 1024, + "total": 4194304, + "child_prefix_read": 16384, + "sender_fence_held": true, + "sent_after": 16384, + "accepted_after": 17408, + "in_flight_after": 0, + "retained_capacity_after": 15, + "state_after": 5, + "job_active_after": 0 + } + ], + "scenario": "verified prefix then pre-send held block, flood/output gate, real action", + "verdict": "passed", + "audited_worker_records": 3 + }, + { + "configuration": "system", + "raw_configuration_label": "candidate-system", + "raw_mode_label": "held-stop-1", + "process_exit": 0, + "outer_timeout_seconds": 10, + "observed_elapsed_seconds": 1.031999999890104, + "probe_exe_sha256": "9df846bd32afdc5df45fdf6422cb998dd496f85f027f5d4dc76d3925a115c73c", + "stdout_sha256": "7695b76d6631cd239a878a348e5c369e220fdc9a4a95e36f2386550c6f441687", + "stderr_sha256": "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855", + "actual_json_records": [ + { + "case": "bounded", + "output_high_water": 262144, + "input_high_water": 16384, + "rejected": 2 + }, + { + "case": "lifecycle-edges", + "stopped_uncompleted_generation": 3, + "input_error_preserved": true, + "resize_error_preserved": true + }, + { + "case": "real-workers", + "action": "stop", + "input_queued": 0, + "input_in_flight": 1024, + "input_joined": true, + "working_generation": 1, + "pending_generation": 3, + "completed_generation": 3, + "child_columns": 102, + "child_rows": 42, + "input_submitted_before_stop": 16384, + "input_accepted_before_stop": 17408, + "paste_in_flight_before_stop": 1024, + "total": 4194304, + "child_prefix_read": 16384, + "sender_fence_held": true, + "sent_after": 16384, + "accepted_after": 17408, + "in_flight_after": 0, + "retained_capacity_after": 15, + "state_after": 5, + "job_active_after": 0 + }, + { + "case": "real-workers", + "action": "cancel", + "input_queued": 0, + "input_in_flight": 1024, + "input_joined": true, + "working_generation": 1, + "pending_generation": 3, + "completed_generation": 3, + "child_columns": 102, + "child_rows": 42, + "input_submitted_before_stop": 16384, + "input_accepted_before_stop": 17408, + "paste_in_flight_before_stop": 1024, + "total": 4194304, + "child_prefix_read": 16384, + "sender_fence_held": true, + "sent_after": 17408, + "accepted_after": 17408, + "in_flight_after": 0, + "retained_capacity_after": 15, + "state_after": 4, + "job_active_after": 0 + }, + { + "case": "real-workers", + "action": "cancel-stop-race", + "input_queued": 0, + "input_in_flight": 1024, + "input_joined": true, + "working_generation": 1, + "pending_generation": 3, + "completed_generation": 3, + "child_columns": 102, + "child_rows": 42, + "input_submitted_before_stop": 16384, + "input_accepted_before_stop": 17408, + "paste_in_flight_before_stop": 1024, + "total": 4194304, + "child_prefix_read": 16384, + "sender_fence_held": true, + "sent_after": 16384, + "accepted_after": 17408, + "in_flight_after": 0, + "retained_capacity_after": 15, + "state_after": 5, + "job_active_after": 0 + } + ], + "scenario": "verified prefix then pre-send held block, flood/output gate, real action", + "verdict": "passed", + "audited_worker_records": 3 + } + ], + "counts": { + "audited_raw_positive_probe_processes": 4, + "positive_probe_processes_per_configuration": 2, + "real_worker_records_per_positive_process": 3, + "audited_raw_positive_worker_records": 12, + "raw_official_original_order_negative_probe_processes": 1, + "focused_original_python_test_processes": 2, + "focused_python_passes_per_configuration": 1, + "focused_ctest_cases_per_configuration": 2, + "focused_ctest_passes_per_configuration": 2, + "focus_reexecutions_not_merged_into_raw_worker_matrix": true + }, + "raw_mode_label_note": "held-stop-0/held-stop-1 are launcher labels; each positive process actually executes stop, cancel and cancel-stop-race. working/pending/completed_generation fields are resize worker generations, not sender-fence ticket generations.", + "state_enum_from_fixed_source": { + "sending": 1, + "cancelled": 4, + "failed": 5 + }, + "historical_evidence": { + "original_focus_red": { + "declared_source_commit": "af563ba15aa5abf90a09ca55eb9c177ac873ecf0", + "artifact": "original-af563-official.log", + "tests": 1, + "failures": 1, + "probe_exit": 2, + "elapsed_seconds": 3.156, + "error": "real input writer holds a chunk", + "old_probe_exe_sha256": null, + "binary_identity_gap_preserved": true + }, + "diagnostic_red": { + "artifact": "diagnostic-ungated.json", + "exe_sha256": "2ac023b60cab1f9576deb023524dfc35d7a5382c742a0afdb1460557f4f314d3", + "source_commit": null, + "source_blob_binding_recorded": false, + "verdict": "diagnostic only; same low-sent state independently reproduced by the frozen final-source ungated mode" + }, + "preparation_error": { + "artifact": "12bce48-unittest-official.log", + "tests_run": 0, + "category": "missing build prerequisite termforge_shell during Python setUpClass", + "product_failure": false + }, + "excluded_wip_green_artifacts": [ + "fenced-official-first.json", + "fenced-official-second.json" + ], + "excluded_wip_reason": "No fixed tested source/blob manifest; retained hashes only and not counted as final validation." + }, + "pure_protocol_coverage": { + "named_ctest": "sender_fence", + "cases": [ + "transfer/generation identity", + "single admission", + "release before held rejected", + "wrong generation release rejected", + "duplicate release rejected", + "unchanged sent floor", + "cancel before wait and during wait", + "stop abort wakes waiter", + "control EOF simulated by abort(control_eof)", + "missing GO absolute deadline", + "accepted==total rejected" + ], + "transport_submission_claim": false, + "control_eof_test_is_actual_ipc_eof": false, + "synthetic_acknowledgements_used_only_for_unit_state_preparation": true + }, + "raw_artifacts": { + "519a3f3-actual.json": { + "bytes": 10117, + "sha256": "b2a97b94d2b7f23abb1fdb9ee68dfd7790c3c2c7922e9938cb5d38c10d31c692" + }, + "519a3f3-ctest-official.log": { + "bytes": 431, + "sha256": "68b9733ddd7de3dce29a28d5f84c43740b6d116b2b89bd52be2edc74dce962db" + }, + "519a3f3-ctest-system.log": { + "bytes": 445, + "sha256": "ff6f170436ad8b1af865dadd90232a20e79edfa1106697b08957268be6ca6707" + }, + "519a3f3-unittest-official.log": { + "bytes": 264, + "sha256": "5830aaf8d029050b7a70038e11edbaad47ed6822244f11ebd7b021701d904f3b" + }, + "519a3f3-unittest-system.log": { + "bytes": 264, + "sha256": "ddc73cb2e99d0a726194666dd0404b50589cd3f1b2041fd85ff988e4b3ce6ea2" + }, + "original-af563-official.log": { + "bytes": 901, + "sha256": "6bec80326dc29f217d2cd675587e4b9eedfb66b3fe290e1d1029f6237775895e" + }, + "diagnostic-ungated.json": { + "bytes": 641, + "sha256": "857b52ebb4fe6941bc05087066b5cf95c579ea0dcf181895b1ebd1856a733386" + }, + "12bce48-unittest-official.log": { + "bytes": 1009, + "sha256": "7fe1fd6a52955471838beb59a40b6b2b40d312aa508a26ab1c377e23cfe4944d" + }, + "fenced-official-first.json": { + "bytes": 446, + "sha256": "fd6c97429175e8fa1702b2a86c8a3e187945329f8b0de0b2d9663ebeb595dbb1" + }, + "fenced-official-second.json": { + "bytes": 905, + "sha256": "15004e8ec93a44edbcd86db9723f772dbd41323b2618ef5c091436052a27e21a" + }, + "final-focused-build.log": { + "bytes": 1002, + "sha256": "a0a67f9e7d8801adb5792d2657100cb56633e431532903f6609f139a79d4e727" + }, + "focused-system-build.log": { + "bytes": 945, + "sha256": "831b74495e8d074d1235df6c0bd14c74b2af6ac4775492ff8ddbfc4968d77371" + }, + "integrated-official-build.log": { + "bytes": 5688, + "sha256": "eb2dfd3ff41e0c46eb630d95cee43eea857ec4a174e049f8525a88fdac45cd53" + }, + "integrated-system-build.log": { + "bytes": 1128, + "sha256": "2a02506ab66b1c40f67be7d1d92ccbbad3a414eb1b38d0267bc93878aed4829c" + }, + "race-official-build.log": { + "bytes": 224, + "sha256": "4fca76e212d24f44c36174d9ad29ef8113d897a92f452b81f57bf1b4c00b2d9c" + }, + "race-system-build.log": { + "bytes": 895, + "sha256": "607ed7488a999b0b06e664b7543c9b16804dbb7543c3ddd37a940ed66b5ff3d7" + } + }, + "environment": { + "proof_host_platform_observed": "Windows-11-10.0.26200-SP0", + "per_run_cpu_power_or_idle_identity": null, + "known_shared_host": true, + "performance_claim": false, + "observed_elapsed_is_not_performance_sample": true + }, + "limitations": [ + "Queue probe does not link candidate VT core; candidate build directory labels do not constitute candidate terminal validation.", + "Fixture ordering is deliberately corrected; original never-read/output-first rejection remains reproducible with the same final EXE.", + "Original ungated rejection does not report post-cleanup Job count; null is retained rather than inferred from later successful runs.", + "Each positive worker JSON reports own Session Job zero; outer supervisor Job/desktop creation or recovery is not established by these direct subprocess runs.", + "No new UI cancel/close, 3000000-byte Han request, 768-byte checkpoint, system clipboard, eight-session workflow or full official regression matrix acceptance.", + "Pure control EOF is simulated through the in-process abort API; no actual IPC EOF regression is claimed.", + "The inspected sample set observes stop winning all cancel-stop races; assertions permit either bounded valid interleaving, but these runs do not prove both schedules occurred.", + "Ticket and flags behavior is supported by fixed source/build definitions and scoped protocol tests; worker JSON does not independently log sender ticket identity or flags.", + "No performance timing, long-run resource, portability, kernel buffer size or WindTerm equivalence conclusion." + ], + "privacy": "Published run objects are parsed synthetic JSON fields only; raw test paths, desktop names, process identities, unfiltered stdout/stderr and environment variables remain local.", + "r53_ui_accepted": false, + "r2_accepted": false +} diff --git a/experiments/shell/sender_fence.hpp b/experiments/shell/sender_fence.hpp new file mode 100644 index 00000000..f17603ba --- /dev/null +++ b/experiments/shell/sender_fence.hpp @@ -0,0 +1,147 @@ +// Opt-in, one-transfer test coordination. A held payload is not OS backpressure. +#pragma once +#include "paste.hpp" +#include +#include +#include +#include + +namespace tf_shell_test { +class SenderFence { +public: + using Clock = std::chrono::steady_clock; + enum class State { armed, held, released, cancelled, aborted, timed_out, failed }; + enum class AbortReason { none, stop, control_eof }; + struct Ticket { + const SenderFence* owner = nullptr; + const tf_shell::PasteTransfer* transfer = nullptr; + uint64_t generation = 0; + }; + struct Snapshot { + uint64_t generation = 0; + State state = State::armed; + AbortReason abort_reason = AbortReason::none; + bool attached = false, ever_held = false, wait_finished = false; + size_t owned_bytes = 0; + tf_shell::PasteProgress held_progress; + }; + static constexpr size_t minimum_sent = 16384; +private: + const std::shared_ptr transfer_; + const uint64_t generation_; + const Clock::time_point deadline_; + mutable std::mutex mutex_; + std::condition_variable changed_; + Snapshot snapshot_; + + bool matches(Ticket ticket) const noexcept { + return ticket.owner == this && ticket.transfer == transfer_.get() && ticket.generation == generation_; + } + [[noreturn]] static void rejected_wait(State state) { + if (state == State::timed_out) throw std::runtime_error("sender fence deadline expired"); + throw std::runtime_error("sender fence aborted or invalid"); + } +public: + SenderFence(std::shared_ptr transfer, uint64_t generation, + Clock::time_point deadline) + : transfer_(std::move(transfer)), generation_(generation), deadline_(deadline) { + if (!transfer_ || !generation_ || deadline_ <= Clock::now()) + throw std::invalid_argument("sender fence identity or deadline invalid"); + const auto progress = transfer_->progress(); + if (progress.state != tf_shell::PasteState::ready || progress.accepted || progress.sent || + progress.in_flight || progress.total <= minimum_sent) + throw std::invalid_argument("sender fence must precede a large paste admission"); + snapshot_.generation = generation_; + } + Ticket ticket() const noexcept { return {this, transfer_.get(), generation_}; } + Snapshot snapshot(Ticket ticket) const { + std::lock_guard lock(mutex_); + if (!matches(ticket)) throw std::invalid_argument("sender fence ticket differs"); + return snapshot_; + } + // Queue admission only. Failed admission leaves no worker waiting and must + // not replace an existing queue binding or claim a mismatched transfer. + bool attach(const std::shared_ptr& transfer) { + std::lock_guard lock(mutex_); + if (transfer.get() != transfer_.get() || snapshot_.attached || snapshot_.state != State::armed) + return false; + const auto progress = transfer_->progress(); + if (progress.state != tf_shell::PasteState::ready || progress.accepted || progress.sent || progress.in_flight) + return false; + if (Clock::now() >= deadline_) { snapshot_.state = State::timed_out; return false; } + snapshot_.attached = true; + return true; + } + bool release(Ticket ticket) { + std::lock_guard lock(mutex_); + if (!matches(ticket) || snapshot_.state != State::held) return false; + if (Clock::now() >= deadline_) { + snapshot_.state = State::timed_out; + changed_.notify_all(); + return false; + } + snapshot_.state = State::released; + changed_.notify_all(); + return true; + } + bool abort(Ticket ticket, AbortReason reason = AbortReason::control_eof) { + std::lock_guard lock(mutex_); + if (!matches(ticket) || reason == AbortReason::none || snapshot_.state == State::aborted || + snapshot_.state == State::timed_out || snapshot_.state == State::failed) return false; + snapshot_.state = State::aborted; + snapshot_.abort_reason = reason; + changed_.notify_all(); + return true; + } + // A real PasteTransfer::cancel must occur first. Wake permits only its + // already claimed chunk to use the unchanged send/ack/finish path. + bool cancelled(Ticket ticket) { + std::lock_guard lock(mutex_); + if (!matches(ticket) || (snapshot_.state != State::armed && snapshot_.state != State::held)) return false; + if (Clock::now() >= deadline_) { + snapshot_.state = State::timed_out; + changed_.notify_all(); + return false; + } + const auto state = transfer_->progress().state; + if (state != tf_shell::PasteState::cancelling && state != tf_shell::PasteState::cancelled) return false; + snapshot_.state = State::cancelled; + changed_.notify_all(); + return true; + } + // Exactly one real writer calls this after take_chunk/in-flight publication + // and after releasing the queue lock, immediately before Session::send. + void before_send(const std::shared_ptr& transfer, size_t owned_bytes) { + std::unique_lock lock(mutex_); + if (transfer.get() != transfer_.get() || !snapshot_.attached) + throw std::logic_error("sender fence writer identity differs"); + if (snapshot_.state == State::released || snapshot_.state == State::cancelled) return; + if (snapshot_.state != State::armed) rejected_wait(snapshot_.state); + if (Clock::now() >= deadline_) { + snapshot_.state = State::timed_out; + rejected_wait(snapshot_.state); + } + const auto progress = transfer_->progress(); + if (progress.state == tf_shell::PasteState::cancelling) return; + if (tf_shell::paste_terminal(progress.state)) { + snapshot_.state = State::failed; + rejected_wait(snapshot_.state); + } + if (progress.sent < minimum_sent) return; + if (!owned_bytes || owned_bytes > tf_shell::PasteTransfer::chunk_limit || + progress.in_flight != owned_bytes || progress.accepted != progress.sent + owned_bytes || + progress.accepted >= progress.total || progress.state != tf_shell::PasteState::sending) { + snapshot_.state = State::failed; + rejected_wait(snapshot_.state); + } + snapshot_.state = State::held; + snapshot_.ever_held = true; + snapshot_.held_progress = progress; + snapshot_.owned_bytes = owned_bytes; + const bool woken = changed_.wait_until(lock, deadline_, [&] { return snapshot_.state != State::held; }); + if (!woken) snapshot_.state = State::timed_out; + snapshot_.wait_finished = true; + if (snapshot_.state != State::released && snapshot_.state != State::cancelled) rejected_wait(snapshot_.state); + } +}; +} // namespace tf_shell_test diff --git a/experiments/shell/sender_fence_test.cpp b/experiments/shell/sender_fence_test.cpp new file mode 100644 index 00000000..770a6e28 --- /dev/null +++ b/experiments/shell/sender_fence_test.cpp @@ -0,0 +1,157 @@ +// Pure coordination tests. Synthetic acknowledgements prepare protocol state; +// these tests never establish real transport submission or OS backpressure. +#include "sender_fence.hpp" +#include +#include +#include + +namespace { +using Fence = tf_shell_test::SenderFence; +using Transfer = tf_shell::PasteTransfer; +using namespace std::chrono_literals; + +void check(bool condition, const char* detail) { + if (!condition) throw std::runtime_error(detail); +} +template void rejects(F function, const char* detail) { + bool rejected = false; + try { function(); } catch (const std::exception&) { rejected = true; } + check(rejected, detail); +} +std::shared_ptr transfer(size_t bytes = 4 * 1024 * 1024) { + tf_shell::PreparedPaste prepared; + prepared.bytes.assign(bytes, 'a'); + return std::make_shared(7, std::move(prepared)); +} +std::shared_ptr fence(const std::shared_ptr& paste, + Fence::Clock::duration duration = 2s) { + return std::make_shared(paste, 19, Fence::Clock::now() + duration); +} +// Unit setup only: prepare the exact real-PasteTransfer state reached after +// sixteen complete 1024B writes, then claim the seventeenth chunk. +std::string claimed_chunk(const std::shared_ptr& paste) { + for (size_t i = 0; i < 16; ++i) { + const auto bytes = paste->take_chunk(); + check(bytes.size() == 1024, "unit prefix chunk differs"); + paste->acknowledge_written(bytes.size()); + paste->finish_chunk(); + } + return paste->take_chunk(); +} +std::future wait(const std::shared_ptr& gate, const std::shared_ptr& paste, + size_t owned_bytes) { + return std::async(std::launch::async, [gate, paste, owned_bytes] { + try { gate->before_send(paste, owned_bytes); return true; } + catch (const std::exception&) { return false; } + }); +} +void held(const std::shared_ptr& gate) { + const auto deadline = Fence::Clock::now() + 1s; + while (gate->snapshot(gate->ticket()).state != Fence::State::held) { + check(Fence::Clock::now() < deadline, "unit writer did not reach held state"); + std::this_thread::yield(); + } +} +void identities_and_admission() { + auto paste = transfer(); + rejects([&] { Fence invalid(paste, 0, Fence::Clock::now() + 1s); }, "zero generation accepted"); + rejects([&] { Fence invalid(paste, 1, Fence::Clock::now()); }, "expired deadline accepted"); + rejects([&] { Fence invalid({}, 1, Fence::Clock::now() + 1s); }, "missing transfer accepted"); + auto gate = fence(paste); + auto ticket = gate->ticket(); + check(!gate->release(ticket), "release before held accepted"); + check(!gate->cancelled(ticket), "fake cancellation accepted"); + auto other = transfer(); + check(!gate->attach(other), "different transfer admitted"); + check(gate->attach(paste) && !gate->attach(paste), "attachment is not single use"); + ++ticket.generation; + check(!gate->release(ticket) && !gate->abort(ticket), "wrong generation changed fence"); + rejects([&] { gate->snapshot(ticket); }, "wrong generation observed fence"); + ticket = gate->ticket(); ticket.transfer = other.get(); + rejects([&] { gate->snapshot(ticket); }, "wrong transfer observed fence"); +} +void threshold_and_release() { + auto paste = transfer(); auto gate = fence(paste); + check(gate->attach(paste), "attach failed"); + const auto first = paste->take_chunk(); + gate->before_send(paste, first.size()); + check(!gate->snapshot(gate->ticket()).ever_held, "fence lowered sent floor"); + paste->acknowledge_written(first.size()); paste->finish_chunk(); + for (size_t i = 1; i < 16; ++i) { + const auto chunk = paste->take_chunk(); + paste->acknowledge_written(chunk.size()); paste->finish_chunk(); + } + const auto current = paste->take_chunk(); + auto worker = wait(gate, paste, current.size()); held(gate); + const auto before = gate->snapshot(gate->ticket()); + check(before.ever_held && before.owned_bytes == 1024 && before.held_progress.sent == 16384 && + before.held_progress.accepted == 17408 && before.held_progress.in_flight == 1024, + "held snapshot does not describe real transfer state"); + auto wrong = gate->ticket(); ++wrong.generation; + check(!gate->release(wrong) && worker.wait_for(0ms) == std::future_status::timeout, + "wrong generation released held writer"); + check(gate->release(gate->ticket()) && !gate->release(gate->ticket()), "release duplicate accepted"); + check(worker.get(), "valid release failed"); + check(paste->progress().sent == 16384, "fence fabricated write acknowledgement"); + paste->acknowledge_written(current.size()); paste->finish_chunk(); + check(gate->snapshot(gate->ticket()).wait_finished, "release did not finish wait"); +} +void cancellation_preserves_claimed_chunk(bool before_wait) { + auto paste = transfer(); auto gate = fence(paste); + check(gate->attach(paste), "cancel attach failed"); + const auto current = claimed_chunk(paste); + std::future worker; + if (!before_wait) { worker = wait(gate, paste, current.size()); held(gate); } + paste->cancel(); + check(gate->cancelled(gate->ticket()), "real cancellation did not wake fence"); + if (before_wait) gate->before_send(paste, current.size()); + else check(worker.get(), "cancellation aborted claimed chunk instead of allowing it"); + check(paste->progress().state == tf_shell::PasteState::cancelling && paste->progress().sent == 16384, + "cancellation prematurely completed an unsubmitted chunk"); + paste->acknowledge_written(current.size()); paste->finish_chunk(); + const auto after = paste->progress(); + check(after.state == tf_shell::PasteState::cancelled && after.sent == after.accepted && + after.sent == 17408 && !after.in_flight && after.retained_capacity <= 32, + "cancel accounting or capacity differs"); +} +void abort_wakes(Fence::AbortReason reason) { + auto paste = transfer(); auto gate = fence(paste); + check(gate->attach(paste), "abort attach failed"); + const auto current = claimed_chunk(paste); + auto worker = wait(gate, paste, current.size()); held(gate); + check(gate->abort(gate->ticket(), reason) && !gate->abort(gate->ticket(), reason), "abort duplicate accepted"); + check(worker.wait_for(1s) == std::future_status::ready && !worker.get(), "abort did not finish wait as failure"); + check(gate->snapshot(gate->ticket()).abort_reason == reason && paste->progress().sent == 16384, + "abort lost reason or fabricated acknowledgement"); + paste->fail("unit writer aborted before submission"); + check(!paste->progress().in_flight && paste->progress().retained_capacity <= 32, "writer failure did not release payload"); +} +void no_go_times_out() { + auto paste = transfer(); auto gate = fence(paste, 1s); + check(gate->attach(paste), "timeout attach failed"); + const auto current = claimed_chunk(paste); + auto worker = wait(gate, paste, current.size()); + check(worker.wait_for(2s) == std::future_status::ready && !worker.get(), "missing GO waited without bound"); + const auto after = gate->snapshot(gate->ticket()); + check(after.state == Fence::State::timed_out && after.ever_held && after.wait_finished && + !gate->release(gate->ticket()), "timeout was converted to release"); +} +void premature_completion_is_rejected() { + auto paste = transfer(17 * 1024); auto gate = fence(paste); + check(gate->attach(paste), "small large-paste attach failed"); + const auto current = claimed_chunk(paste); + rejects([&] { gate->before_send(paste, current.size()); }, "last accepted chunk was treated as a partial transfer"); + check(!gate->snapshot(gate->ticket()).ever_held, "completed admission manufactured held state"); +} +} // namespace + +int main() { + try { + identities_and_admission(); threshold_and_release(); + cancellation_preserves_claimed_chunk(false); cancellation_preserves_claimed_chunk(true); + abort_wakes(Fence::AbortReason::stop); abort_wakes(Fence::AbortReason::control_eof); + no_go_times_out(); premature_completion_is_rejected(); + std::cout << "sender fence pure protocol checks passed; no transport or OS backpressure claim\n"; + return 0; + } catch (const std::exception& error) { std::cerr << error.what() << '\n'; return 1; } +} diff --git a/experiments/shell/test_shell.py b/experiments/shell/test_shell.py index 09782b1c..e4d01513 100644 --- a/experiments/shell/test_shell.py +++ b/experiments/shell/test_shell.py @@ -463,6 +463,29 @@ def create_job(): def test_queues_apply_bounded_backpressure_and_reject_input(self): result = subprocess.run([str(binary("shell_queue_probe"))], capture_output=True, timeout=10) self.assertEqual(result.returncode, 0, result.stderr) + rows = [json.loads(line) for line in result.stdout.decode("utf-8").splitlines() if line.startswith("{")] + actual = [row for row in rows if row.get("case") == "real-workers"] + self.assertEqual([row.get("action") for row in actual], ["stop", "cancel", "cancel-stop-race"], result.stdout) + for row in actual: + for field, value in dict(total=4 * 1024 * 1024, input_submitted_before_stop=16384, + input_accepted_before_stop=17408, input_in_flight=1024, + paste_in_flight_before_stop=1024, child_prefix_read=16384, + sender_fence_held=True, input_joined=True, input_queued=0, + working_generation=1, pending_generation=3, completed_generation=3, + child_columns=102, child_rows=42, job_active_after=0).items(): + with self.subTest(action=row["action"], field=field): + self.assertEqual(row.get(field), value) + self.assertEqual(row["accepted_after"], 17408) + self.assertEqual(row["in_flight_after"], 0) + self.assertLessEqual(row["retained_capacity_after"], 32) + if row["action"] == "cancel": + self.assertEqual((row["sent_after"], row["state_after"]), (17408, 4)) + elif row["action"] == "stop": + self.assertEqual((row["sent_after"], row["state_after"]), (16384, 5)) + else: + self.assertIn(row["state_after"], (4, 5)) + self.assertGreaterEqual(row["sent_after"], 16384) + self.assertLessEqual(row["sent_after"], 17408) def test_invalid_arguments_create_no_session(self): result = subprocess.run([str(self.exe), "--test", "invalid", "unused.bmp"],