test(core): 带 resize 时间线的回放工具（解阻 #194）

CMakeLists.txt

@@ -24,6 +24,10 @@ add_executable(replay experiments/core/replay.cpp)
 target_link_libraries(replay PRIVATE termforge_core)
 termforge_warnings(replay)
 
+add_executable(resize_replay experiments/core/resize_replay.cpp)
+target_link_libraries(resize_replay PRIVATE termforge_core)
+termforge_warnings(resize_replay)
+
 add_executable(terminal_trace experiments/core/trace.cpp)
 target_link_libraries(terminal_trace PRIVATE termforge_core)
 termforge_warnings(terminal_trace)
@@ -350,6 +354,7 @@ add_test(NAME edit_csi_core COMMAND edit_csi_test)
     set_tests_properties(candidate_shell_tui PROPERTIES TIMEOUT 120)
   endif()
   termforge_add_python_test(core -m unittest experiments.core.test_core experiments.core.test_unicode -v)
+  termforge_add_python_test(resize_replay_contract -m unittest experiments.core.test_resize_replay -v)
   termforge_add_python_test(recovery -m unittest experiments.core.test_recovery -v)
   termforge_add_python_test(measure -m unittest experiments.measure.test_measure experiments.measure.test_r54_contract.CostPlanTests -v)
   termforge_add_python_test(binary_paths -m unittest experiments.test_build_paths -v)

experiments/core/resize_replay.cpp

@@ -0,0 +1,114 @@
+// Replay one captured byte stream across an explicit resize timeline.
+//
+// The plain `replay` tool feeds a whole capture at one fixed geometry, so a
+// capture that changed size mid-session (DPI switch, zoom, restored window)
+// cannot be replayed faithfully. This tool reads the same bytes but applies a
+// caller-supplied sequence of geometries, reporting the viewport and buffer
+// state after each step so a repaint hypothesis can be checked against real
+// bytes (#194).
+#include "terminal.hpp"
+#include "history_view.hpp"
+#include <chrono>
+#include <fstream>
+#include <iostream>
+#include <iterator>
+#include <sstream>
+#include <stdexcept>
+#include <string>
+#include <vector>
+
+namespace {
+
+using namespace termforge;
+
+// One step of the timeline: resize to rows x columns, then consume `bytes`
+// bytes of the capture.
+struct Step {
+    int rows = 0;
+    int columns = 0;
+    size_t bytes = 0;
+    std::string label;
+};
+
+void require(bool condition, const char* message) {
+    if (!condition) throw std::runtime_error(message);
+}
+
+std::vector<Step> read_steps(const char* path) {
+    std::ifstream input(path);
+    require(static_cast<bool>(input), "cannot open steps file");
+    std::vector<Step> steps;
+    std::string line;
+    while (std::getline(input, line)) {
+        if (line.empty() || line[0] == '#') continue;
+        std::istringstream fields(line);
+        Step step;
+        require(static_cast<bool>(fields >> step.rows >> step.columns >> step.bytes), "malformed step line");
+        require(step.rows > 0 && step.columns > 0, "step geometry must be positive");
+        std::getline(fields, step.label);
+        if (!step.label.empty() && step.label.front() == ' ') step.label.erase(0, 1);
+        steps.push_back(step);
+    }
+    require(!steps.empty(), "steps file has no steps");
+    return steps;
+}
+
+void quoted(std::ostream& out, const std::string& value) {
+    out << '"';
+    for (const char ch : value) {
+        if (ch == '"' || ch == '\\') out << '\\' << ch;
+        else if (ch == '\n') out << "\\n";
+        else out << ch;
+    }
+    out << '"';
+}
+
+} // namespace
+
+int main(int argc, char** argv) {
+    try {
+        if (argc != 4) throw std::runtime_error("usage: resize_replay input steps history");
+        std::ifstream input(argv[1], std::ios::binary | std::ios::ate);
+        require(static_cast<bool>(input), "cannot open input file");
+        require(input.tellg() <= 64 * 1024 * 1024, "input file too large");
+        input.seekg(0);
+        const std::string data((std::istreambuf_iterator<char>(input)), {});
+        const std::vector<Step> steps = read_steps(argv[2]);
+        const size_t history = std::stoull(argv[3]);
+
+        const Step& first = steps.front();
+        Parser parser(first.rows, first.columns, history);
+        HistoryViewport viewport;
+        const auto start = std::chrono::steady_clock::now();
+        size_t consumed = 0;
+        for (size_t index = 0; index < steps.size(); ++index) {
+            const Step& step = steps[index];
+            if (index > 0) parser.screen.resize(step.rows, step.columns);
+            const size_t take = (std::min)(step.bytes, data.size() - consumed);
+            parser.feed(std::string_view(data).substr(consumed, take));
+            parser.finish();
+            consumed += take;
+            const ViewportState state = viewport.reconcile(parser.screen);
+            std::cout << "{\"step\":" << index + 1 << ",\"label\":";
+            quoted(std::cout, step.label);
+            std::cout << ",\"rows\":" << step.rows << ",\"columns\":" << step.columns
+                      << ",\"consumed\":" << take << ",\"total_consumed\":" << consumed
+                      << ",\"history\":" << parser.screen.history_lines().size()
+                      << ",\"retained\":" << parser.screen.retained_size()
+                      << ",\"top\":" << state.top << ",\"maximum_top\":" << state.maximum_top
+                      << ",\"following\":" << (state.following ? "true" : "false")
+                      << ",\"cursor_row\":" << parser.screen.row
+                      << ",\"cursor_column\":" << parser.screen.column << "}\n";
+        }
+        const double ms =
+            std::chrono::duration<double, std::milli>(std::chrono::steady_clock::now() - start).count();
+        std::cout << "{\"bytes\":" << data.size() << ",\"consumed\":" << consumed << ",\"steps\":"
+                  << steps.size() << ",\"parse_ms\":" << ms << ",\"screen\":";
+        parser.screen.json(std::cout);
+        std::cout << "}\n";
+        return 0;
+    } catch (const std::exception& error) {
+        std::cerr << error.what() << '\n';
+        return 2;
+    }
+}

experiments/core/test_resize_replay.py

@@ -0,0 +1,68 @@
+"""Resize-aware replay: one capture read across an explicit geometry timeline."""
+import json
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+from experiments.build_paths import binary
+
+
+class ResizeReplayTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        folder = Path(self.temp.name)
+        self.input = folder / "input.bin"
+        self.steps = folder / "steps.txt"
+
+    def run_steps(self, data, steps, history=100):
+        self.input.write_bytes(data)
+        self.steps.write_text("".join(f"{rows} {columns} {count} {label}\n"
+                                      for rows, columns, count, label in steps), encoding="utf-8")
+        completed = subprocess.run(
+            [str(binary("resize_replay")), str(self.input), str(self.steps), str(history)],
+            capture_output=True, text=True, encoding="utf-8", timeout=10)
+        self.assertEqual(completed.returncode, 0, completed.stderr)
+        return [json.loads(line) for line in completed.stdout.splitlines()]
+
+    def test_resize_between_chunks_moves_the_viewport_and_grows_history(self):
+        data = "".join(f"L{index:02d}\r\n" for index in range(1, 21)).encode()
+        split = len("".join(f"L{index:02d}\r\n" for index in range(1, 13)).encode())
+        rows = self.run_steps(data, [(3, 40, split, "small"), (5, 40, len(data) - split, "grown")])
+        first, second = rows[0], rows[1]
+        self.assertEqual((first["rows"], second["rows"]), (3, 5))
+        self.assertTrue(first["following"] and second["following"])
+        self.assertEqual(second["total_consumed"], len(data))
+        self.assertGreater(second["history"], first["history"])
+        # The viewport follows the bottom, so retained rows are history plus one page.
+        self.assertEqual(second["retained"], second["history"] + 5)
+        self.assertEqual(first["retained"], first["history"] + 3)
+
+    def test_identical_bytes_scale_their_scrollback_with_page_height(self):
+        # #194 measured this on a real capture: the same repaint bytes pushed 30
+        # rows into scrollback behind a 7-row page and none behind a 24-row one.
+        data = "".join(f"R{index:02d}\r\n" for index in range(1, 21)).encode()
+        short, medium, tall = (self.run_steps(data, [(rows, 20, len(data), f"rows-{rows}")])[0]
+                               for rows in (4, 10, 24))
+        self.assertEqual((short["history"], medium["history"], tall["history"]), (17, 11, 0))
+        for row in (short, medium, tall):
+            self.assertEqual(row["retained"], row["history"] + row["rows"])
+
+    def test_a_step_longer_than_the_capture_consumes_only_what_is_left(self):
+        rows = self.run_steps(b"only\r\n", [(2, 20, 100, "overreach")])
+        self.assertEqual(rows[0]["consumed"], len(b"only\r\n"))
+        self.assertEqual(rows[0]["total_consumed"], len(b"only\r\n"))
+
+    def test_malformed_steps_are_rejected_before_any_parsing(self):
+        self.input.write_bytes(b"x")
+        self.steps.write_text("not a step\n", encoding="utf-8")
+        completed = subprocess.run(
+            [str(binary("resize_replay")), str(self.input), str(self.steps), "10"],
+            capture_output=True, text=True, encoding="utf-8", timeout=10)
+        self.assertNotEqual(completed.returncode, 0)
+        self.assertIn("malformed step", completed.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()