[ai] Clear a stale crash report before a test run

Task: 2026/07/29/clear-stale-crash-report-before-test-run
This commit is contained in:
John Preston
2026-07-30 14:39:55 +04:00
parent f238e46ff9
commit 08f466f719
3 changed files with 371 additions and 70 deletions

View File

@@ -115,7 +115,19 @@ The workspace helper's `test-run` command performs exactly these steps before ev
contract those commands implement.
1. Require `test_TelegramForcePortable`. Its absence is the only portable-account setup blocker.
2. If `TelegramForcePortable/testing` exists, the live folder is already the reusable test copy:
touch none of the three folders and proceed straight to testing.
never copy, move, or delete any of the three folders. Clear only what an earlier run left
inside the live copy — move a non-empty `TelegramForcePortable/tdata/working` into
`<EVIDENCE_DIR>/stale-crash/` and every `TelegramForcePortable/tdata/dumps/*.dmp` into
`<EVIDENCE_DIR>/stale-crash/dumps/`, then proceed straight to testing. `<EVIDENCE_DIR>` must be
a run-specific directory the repository ignores, never a tracked one: a preserved minidump
is routinely tens of megabytes, and a tracked destination sweeps it into the wrapper's
publishing commit. Never delete either: the leftover `tdata/working` is what blinds the next
run (the app shows its "previous launch was not finished properly" window instead of starting,
so the run writes no `test_log.txt` and reads as a hang), while a leftover `.dmp` never blocks
a launch and is moved only to keep a later run's `dumps` report free of old minidumps.
`test-run` names every moved file and its destination in `stale_crash_cleared`, refuses to
launch when the report itself cannot be moved, and leaves a minidump it cannot move in place,
reported with a null destination.
3. If `TelegramForcePortable` exists without the marker, it is the user's real data: move it to
`real_TelegramForcePortable` when that is absent. If `real_...` already exists, the unmarked
live folder is the user's manual restore of that same preserved data — recursively delete the

View File

@@ -50,6 +50,7 @@ OVERLAY_PATHS_FILE = "test-overlay.paths"
OVERLAY_PATCH_FILE = "test-overlay.patch"
TEST_LOG_FILE = "test_log.txt"
TEST_COMPLETE_MARKER = "TEST_COMPLETE"
STALE_CRASH_DIR = "stale-crash"
class WorkspaceError(RuntimeError):
@@ -1092,25 +1093,59 @@ def portable_root_for(exe, override):
return exe.parent
def clear_crash_marker(live):
"""Drop a stale crash dump so the next launch is not held for a human.
def unique_destination(directory, name):
candidate = directory / name
index = 2
while candidate.exists():
candidate = directory / (
f"{Path(name).stem}-{index:02d}{Path(name).suffix}"
)
index += 1
return candidate
`Sandbox::singleInstanceChecked()` builds a `LastCrashedWindow` and waits
for the user whenever `tdata/working` is non-empty, and `-testagent` does
not bypass it — the gate sits above `launchApplication()`. A previous run
that died after `TEST_COMPLETE` therefore blocks every later launch until
the file is removed by hand. The live folder is a disposable copy and the
crash check after a run only counts a dump written during that run, so
clearing it here loses nothing.
"""
working = live / "tdata" / "working"
def move_stale_leftover(path, target):
for attempt in reversed(range(5)):
moved = unique_destination(target, path.name)
try:
if working.is_file() and working.stat().st_size > 0:
working.unlink()
return True
shutil.move(path, moved)
return moved
except OSError:
try:
moved.unlink(missing_ok=True)
except OSError:
pass
return False
if not attempt:
raise
time.sleep(0.2)
def clear_stale_crash_state(live, destination):
report = live / "tdata" / "working"
dumps_dir = live / "tdata" / "dumps"
leftovers = []
if report.is_file() and report.stat().st_size > 0:
leftovers.append(("report", report))
if dumps_dir.is_dir():
leftovers.extend(
("dump", path) for path in sorted(dumps_dir.glob("*.dmp"))
if path.is_file()
)
cleared = []
for kind, path in leftovers:
target = destination / "dumps" if kind == "dump" else destination
target.mkdir(parents=True, exist_ok=True)
try:
moved = move_stale_leftover(path, target)
except OSError as error:
if kind == "report":
raise WorkspaceError(
f"Cannot clear the stale crash report {path}: {error}"
) from error
cleared.append({"from": str(path), "kind": kind, "to": None})
continue
cleared.append({"from": str(path), "kind": kind, "to": str(moved)})
return cleared
def setup_test_account(root):
@@ -1120,8 +1155,6 @@ def setup_test_account(root):
if not golden.is_dir():
raise WorkspaceError(f"Missing golden test account: {golden}")
if (live / PORTABLE_MARKER).exists():
if clear_crash_marker(live):
return "reused-marked-live-crash-cleared"
return "reused-marked-live"
if live.exists():
if real.exists():
@@ -1262,15 +1295,24 @@ def command_test_run(args):
log_path = run_dir / TEST_LOG_FILE
if log_path.exists():
log_path.unlink()
stdout_path = run_dir / "app_stdout.txt"
stderr_path = run_dir / "app_stderr.txt"
working = portable / PORTABLE_LIVE / "tdata" / "working"
dumps_dir = portable / PORTABLE_LIVE / "tdata" / "dumps"
environment = os.environ.copy()
environment["TDESKTOP_TEST_EVIDENCE_DIR"] = str(run_dir)
environment.update(parse_env_values(args.env))
cleared = (
clear_stale_crash_state(
portable / PORTABLE_LIVE, run_dir / STALE_CRASH_DIR
)
if account == "reused-marked-live"
else []
)
stdout_path = run_dir / "app_stdout.txt"
stderr_path = run_dir / "app_stderr.txt"
working = portable / PORTABLE_LIVE / "tdata" / "working"
dumps_dir = portable / PORTABLE_LIVE / "tdata" / "dumps"
launched_at = time.time()
with stdout_path.open("wb") as out, stderr_path.open("wb") as err:
process = subprocess.Popen(
@@ -1364,6 +1406,7 @@ def command_test_run(args):
"outcome": outcome,
"portable_root": str(portable),
"run_dir": str(run_dir),
"stale_crash_cleared": cleared,
"stderr_tail": tail_of_file(stderr_path),
"stragglers_killed": stragglers,
"test_complete": test_complete,

View File

@@ -876,13 +876,35 @@ inbox_receipt: receipts/2026/07/19/test.md
self.assertIn("projects/old-project/tasks.md", staged)
def write_fake_exe(path, script):
def write_fake_exe(path, script, windows_script):
path.parent.mkdir(parents=True, exist_ok=True)
if os.name == "nt":
path = path.with_suffix(".cmd")
path.write_text("@echo off\n" + windows_script, encoding="utf-8")
return path
path.write_text("#!/bin/sh\n" + script, encoding="utf-8")
path.chmod(0o755)
return path
def write_complete_markers_exe(path):
return write_fake_exe(path, (
'LOG="$TDESKTOP_TEST_EVIDENCE_DIR/test_log.txt"\n'
'echo "TEST_STEP: open settings" >> "$LOG"\n'
'echo "TEST_RESULT: PASS: row painted" >> "$LOG"\n'
'echo "SCREENSHOT: /tmp/fake.png" >> "$LOG"\n'
'echo "TEST_COMPLETE" >> "$LOG"\n'
"exit 0\n"
), (
'set "LOG=%TDESKTOP_TEST_EVIDENCE_DIR%\\test_log.txt"\n'
'echo TEST_STEP: open settings>>"%LOG%"\n'
'echo TEST_RESULT: PASS: row painted>>"%LOG%"\n'
'echo SCREENSHOT: /tmp/fake.png>>"%LOG%"\n'
'echo TEST_COMPLETE>>"%LOG%"\n'
"exit /b 0\n"
))
def make_portable_root(root):
debug = root / "out" / "Debug"
golden = debug / workspace.PORTABLE_GOLDEN
@@ -891,6 +913,16 @@ def make_portable_root(root):
return debug
def plant_leftover_crash_state(live):
dumps_dir = live / "tdata" / "dumps"
dumps_dir.mkdir(parents=True, exist_ok=True)
report = live / "tdata" / "working"
report.write_bytes(b"Assertion: previous run\n" * 20)
dump = dumps_dir / "stale.dmp"
dump.write_bytes(b"MDMP stale minidump\n")
return report, dump
def source_repo_with_task(root, kind="implement"):
source = root / "source"
git_repo(source)
@@ -932,6 +964,20 @@ def run_command(handler, **kwargs):
return json.loads(out.getvalue())
def run_test_run(exe, run_dir, **overrides):
arguments = {
"exe": str(exe),
"run_dir": str(run_dir),
"portable_root": None,
"deadline": 20.0,
"quiet": 10.0,
"grace": 5.0,
"env": None,
}
arguments.update(overrides)
return run_command(workspace.command_test_run, **arguments)
class MechanicsTest(unittest.TestCase):
def test_parse_env_values_requires_name_value_pairs(self):
self.assertEqual(
@@ -986,29 +1032,12 @@ class MechanicsTest(unittest.TestCase):
with self.assertRaisesRegex(workspace.WorkspaceError, "unmarked"):
workspace.reset_broken_test_account(root)
@unittest.skipUnless(os.name == "posix", "posix launch mechanics")
def test_test_run_reports_complete_markers(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
debug = make_portable_root(root)
exe = write_fake_exe(debug / "Telegram", (
'LOG="$TDESKTOP_TEST_EVIDENCE_DIR/test_log.txt"\n'
'echo "TEST_STEP: open settings" >> "$LOG"\n'
'echo "TEST_RESULT: PASS: row painted" >> "$LOG"\n'
'echo "SCREENSHOT: /tmp/fake.png" >> "$LOG"\n'
'echo "TEST_COMPLETE" >> "$LOG"\n'
"exit 0\n"
))
result = run_command(
workspace.command_test_run,
exe=str(exe),
run_dir=str(root / "run1"),
portable_root=None,
deadline=20.0,
quiet=10.0,
grace=5.0,
env=["EXTRA_FLAG=1"],
)
exe = write_complete_markers_exe(debug / "Telegram")
result = run_test_run(exe, root / "run1", env=["EXTRA_FLAG=1"])
self.assertEqual(result["outcome"], "exited")
self.assertEqual(result["verdict_hint"], "complete")
self.assertTrue(result["test_complete"])
@@ -1018,54 +1047,253 @@ class MechanicsTest(unittest.TestCase):
self.assertEqual(result["markers"]["screenshots"], ["/tmp/fake.png"])
self.assertFalse(result["crash_report_fresh"])
@unittest.skipUnless(os.name == "posix", "posix launch mechanics")
def test_test_run_reports_crash_diagnostics(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
debug = make_portable_root(root)
live_tdata = debug / workspace.PORTABLE_LIVE / "tdata"
live_working = live_tdata / "working"
exe = write_fake_exe(debug / "Telegram", (
'LOG="$TDESKTOP_TEST_EVIDENCE_DIR/test_log.txt"\n'
'echo "TEST_STEP: about to crash" >> "$LOG"\n'
f'mkdir -p "{debug}/{workspace.PORTABLE_LIVE}/tdata"\n'
f'echo "Assertion: boom" > "{debug}/{workspace.PORTABLE_LIVE}/tdata/working"\n'
f'mkdir -p "{live_tdata}"\n'
f'echo "Assertion: boom" > "{live_working}"\n'
"exit 0\n"
), (
'set "LOG=%TDESKTOP_TEST_EVIDENCE_DIR%\\test_log.txt"\n'
'echo TEST_STEP: about to crash>>"%LOG%"\n'
f'if not exist "{live_tdata}" mkdir "{live_tdata}"\n'
f'echo Assertion: boom>"{live_working}"\n'
"exit /b 0\n"
))
result = run_command(
workspace.command_test_run,
exe=str(exe),
run_dir=str(root / "run1"),
portable_root=None,
deadline=20.0,
quiet=10.0,
grace=5.0,
env=None,
)
result = run_test_run(exe, root / "run1")
self.assertEqual(result["outcome"], "exited")
self.assertEqual(result["verdict_hint"], "crash")
self.assertFalse(result["test_complete"])
self.assertTrue(result["crash_report_fresh"])
self.assertIn("Assertion: boom", result["crash_report_excerpt"])
@unittest.skipUnless(os.name == "posix", "posix launch mechanics")
def test_test_run_kills_on_deadline(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
debug = make_portable_root(root)
exe = write_fake_exe(debug / "Telegram", "sleep 30\n")
result = run_command(
workspace.command_test_run,
exe=str(exe),
run_dir=str(root / "run1"),
portable_root=None,
deadline=2.0,
quiet=30.0,
grace=5.0,
env=None,
exe = write_fake_exe(
debug / "Telegram", "sleep 30\n", ":loop\ngoto loop\n",
)
result = run_test_run(
exe, root / "run1", deadline=2.0, quiet=30.0,
)
self.assertEqual(result["outcome"], "deadline-killed")
self.assertEqual(result["verdict_hint"], "hang")
self.assertFalse(result["test_complete"])
def test_test_run_clears_and_preserves_stale_crash_state(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary).resolve()
debug = make_portable_root(root)
golden = debug / workspace.PORTABLE_GOLDEN
real = debug / workspace.PORTABLE_REAL
(real / "tdata" / "dumps").mkdir(parents=True)
(real / "tdata" / "working").write_bytes(b"real crash\n")
(real / "tdata" / "dumps" / "real.dmp").write_bytes(b"real dump\n")
self.assertEqual(workspace.setup_test_account(debug), "fresh-copy")
live = debug / workspace.PORTABLE_LIVE
report, dump = plant_leftover_crash_state(live)
report_payload = report.read_bytes()
dump_payload = dump.read_bytes()
exe = write_complete_markers_exe(debug / "Telegram")
run_dir = root / "run1"
result = run_test_run(exe, run_dir)
stale = run_dir / workspace.STALE_CRASH_DIR
self.assertEqual(result["account"], "reused-marked-live")
self.assertTrue(result["test_complete"])
self.assertEqual(result["verdict_hint"], "complete")
self.assertEqual(result["markers"]["pass"], ["row painted"])
self.assertEqual(result["stale_crash_cleared"], [
{
"from": str(report),
"kind": "report",
"to": str(stale / "working"),
},
{
"from": str(dump),
"kind": "dump",
"to": str(stale / "dumps" / "stale.dmp"),
},
])
self.assertFalse(report.exists())
self.assertFalse(dump.exists())
self.assertEqual((stale / "working").read_bytes(), report_payload)
self.assertEqual(
(stale / "dumps" / "stale.dmp").read_bytes(),
dump_payload,
)
self.assertIsNone(result["crash_report"])
self.assertFalse(result["crash_report_fresh"])
self.assertEqual(result["dumps"], [])
self.assertEqual(
(golden / "tdata" / "key_data").read_text(encoding="utf-8"),
"golden\n",
)
self.assertFalse((golden / "tdata" / "working").exists())
self.assertFalse((golden / workspace.PORTABLE_MARKER).exists())
self.assertEqual(
(real / "tdata" / "working").read_bytes(),
b"real crash\n",
)
self.assertEqual(
(real / "tdata" / "dumps" / "real.dmp").read_bytes(),
b"real dump\n",
)
def test_test_run_without_leftovers_reports_nothing_cleared(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
debug = make_portable_root(root)
self.assertEqual(workspace.setup_test_account(debug), "fresh-copy")
exe = write_complete_markers_exe(debug / "Telegram")
run_dir = root / "run1"
result = run_test_run(exe, run_dir)
self.assertEqual(result["account"], "reused-marked-live")
self.assertEqual(result["stale_crash_cleared"], [])
self.assertFalse((run_dir / workspace.STALE_CRASH_DIR).exists())
self.assertTrue(result["test_complete"])
def test_test_run_leaves_an_unmarked_live_folder_alone(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
debug = make_portable_root(root)
golden = debug / workspace.PORTABLE_GOLDEN
(golden / "tdata" / "dumps").mkdir(parents=True)
(golden / "tdata" / "working").write_bytes(b"golden crash\n")
golden_dump = golden / "tdata" / "dumps" / "golden.dmp"
golden_dump.write_bytes(b"golden dump\n")
live = debug / workspace.PORTABLE_LIVE
(live / "tdata" / "dumps").mkdir(parents=True)
(live / "tdata" / "working").write_bytes(b"user crash\n")
(live / "tdata" / "dumps" / "user.dmp").write_bytes(b"user dump\n")
(live / "tdata" / "user_file").write_text("mine\n", encoding="utf-8")
exe = write_complete_markers_exe(debug / "Telegram")
run_dir = root / "run1"
result = run_test_run(exe, run_dir)
real = debug / workspace.PORTABLE_REAL
self.assertEqual(result["account"], "preserved-real")
self.assertEqual(result["stale_crash_cleared"], [])
self.assertFalse((run_dir / workspace.STALE_CRASH_DIR).exists())
self.assertEqual(
(real / "tdata" / "working").read_bytes(),
b"user crash\n",
)
self.assertEqual(
(real / "tdata" / "dumps" / "user.dmp").read_bytes(),
b"user dump\n",
)
self.assertTrue((real / "tdata" / "user_file").is_file())
self.assertEqual(
(golden / "tdata" / "working").read_bytes(),
b"golden crash\n",
)
self.assertEqual(golden_dump.read_bytes(), b"golden dump\n")
self.assertTrue((live / workspace.PORTABLE_MARKER).is_file())
self.assertEqual(
(live / "tdata" / "working").read_bytes(),
b"golden crash\n",
)
self.assertEqual(
(live / "tdata" / "dumps" / "golden.dmp").read_bytes(),
b"golden dump\n",
)
def test_test_run_refuses_to_launch_when_the_stale_report_cannot_move(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
debug = make_portable_root(root)
self.assertEqual(workspace.setup_test_account(debug), "fresh-copy")
live = debug / workspace.PORTABLE_LIVE
report, dump = plant_leftover_crash_state(live)
report_payload = report.read_bytes()
exe = write_complete_markers_exe(debug / "Telegram")
run_dir = root / "run1"
with mock.patch.object(
workspace.shutil, "move", side_effect=OSError("locked"),
):
with self.assertRaisesRegex(workspace.WorkspaceError, "working"):
run_test_run(exe, run_dir)
self.assertEqual(report.read_bytes(), report_payload)
self.assertTrue(dump.is_file())
self.assertFalse((run_dir / "app_stdout.txt").exists())
def test_test_run_reports_a_dump_that_could_not_be_moved(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary).resolve()
debug = make_portable_root(root)
self.assertEqual(workspace.setup_test_account(debug), "fresh-copy")
live = debug / workspace.PORTABLE_LIVE
report, dump = plant_leftover_crash_state(live)
dump_payload = dump.read_bytes()
exe = write_complete_markers_exe(debug / "Telegram")
run_dir = root / "run1"
real_move = workspace.shutil.move
def move_unless_dump(source, target):
if str(source).endswith(".dmp"):
Path(target).write_bytes(Path(source).read_bytes())
raise OSError("locked")
return real_move(source, target)
with mock.patch.object(
workspace.shutil, "move", side_effect=move_unless_dump,
):
result = run_test_run(exe, run_dir)
stale = run_dir / workspace.STALE_CRASH_DIR
self.assertTrue(result["test_complete"])
self.assertEqual(result["verdict_hint"], "complete")
self.assertEqual(result["stale_crash_cleared"], [
{
"from": str(report),
"kind": "report",
"to": str(stale / "working"),
},
{
"from": str(dump),
"kind": "dump",
"to": None,
},
])
self.assertEqual(dump.read_bytes(), dump_payload)
self.assertEqual(result["dumps"], [])
self.assertEqual(list((stale / "dumps").iterdir()), [])
self.assertEqual(
sorted(path.name for path in stale.iterdir()),
["dumps", "working"],
)
def test_test_run_discards_a_partial_report_copy_from_a_failed_move(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary).resolve()
debug = make_portable_root(root)
self.assertEqual(workspace.setup_test_account(debug), "fresh-copy")
live = debug / workspace.PORTABLE_LIVE
report, dump = plant_leftover_crash_state(live)
report_payload = report.read_bytes()
exe = write_complete_markers_exe(debug / "Telegram")
run_dir = root / "run1"
def write_then_fail(source, target):
Path(target).write_bytes(Path(source).read_bytes())
raise OSError("locked")
with mock.patch.object(
workspace.shutil, "move", side_effect=write_then_fail,
):
with self.assertRaisesRegex(workspace.WorkspaceError, "working"):
run_test_run(exe, run_dir)
stale = run_dir / workspace.STALE_CRASH_DIR
self.assertEqual(report.read_bytes(), report_payload)
self.assertTrue(dump.is_file())
self.assertEqual(list(stale.iterdir()), [])
self.assertFalse((run_dir / "app_stdout.txt").exists())
def test_portable_root_for_prefers_app_bundle_parent(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
@@ -1086,6 +1314,24 @@ class MechanicsTest(unittest.TestCase):
root / "out" / "Debug",
)
def test_unique_destination_avoids_an_existing_name(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
self.assertEqual(
workspace.unique_destination(root, "working"),
root / "working",
)
(root / "working").write_text("stale\n", encoding="utf-8")
self.assertEqual(
workspace.unique_destination(root, "working"),
root / "working-02",
)
(root / "stale.dmp").write_text("dump\n", encoding="utf-8")
self.assertEqual(
workspace.unique_destination(root, "stale.dmp"),
root / "stale-02.dmp",
)
def test_overlay_save_and_apply_roundtrip(self):
with tempfile.TemporaryDirectory() as temporary:
root = Path(temporary)
@@ -1376,7 +1622,7 @@ class MechanicsTest(unittest.TestCase):
(source / "tracked.txt").write_text("dirty\n", encoding="utf-8")
(source / "unrelated.txt").write_text("stray\n", encoding="utf-8")
debug = make_portable_root(root)
exe = write_fake_exe(debug / "Telegram", "exit 0\n")
exe = write_fake_exe(debug / "Telegram", "exit 0\n", "exit /b 0\n")
with mock.patch.object(
workspace, "task_action_config", return_value=(config, slot),
):