From 67a6dcfe69aafc4e5bd40578480a44b99f048c8b Mon Sep 17 00:00:00 2001
From: dr-hoseyn <95236112+dr-hoseyn@users.noreply.github.com>
Date: Sat, 31 Oct 2026 11:06:15 +0330
Subject: [PATCH] fix(tools): reject thread filters on collapsed samples

---
 tools/symbolize_samples.py            |  2 ++
 tools/tests/test_symbolize_samples.py | 16 ++++++++++++++++
 2 files changed, 18 insertions(+)

diff --git a/tools/symbolize_samples.py b/tools/symbolize_samples.py
index 73534b8c81..5f5eee70ad 100644
--- a/tools/symbolize_samples.py
+++ b/tools/symbolize_samples.py
@@ -272,6 +272,8 @@ def load(path, thread=None):
         frames, _, count = line.rpartition(" ")
         if not frames or not count.isdigit():
             raise ValueError(f"{path}:{number}: expected 'frame;...;frame count', 'module offset self inclusive' or 'thread <id> samples <n>'")
+        if thread is not None:
+            raise ValueError(f"{path}:{number}: thread filtering requires an address capture; collapsed stacks have no thread IDs")
         samples.stacks[frames] += int(count)
     return samples
 
diff --git a/tools/tests/test_symbolize_samples.py b/tools/tests/test_symbolize_samples.py
index 07af2d78f9..90e6db15bf 100644
--- a/tools/tests/test_symbolize_samples.py
+++ b/tools/tests/test_symbolize_samples.py
@@ -67,6 +67,22 @@ def build_pe(path, with_symbols=True, with_exports=False, second_code=False, tim
 
 
 class SymbolizeSamplesTests(unittest.TestCase):
+    def test_collapsed_stacks_reject_thread_filter_without_writing_output(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            capture = root / "samples.txt"
+            capture.write_text("example.prx+0x1010 5\n", encoding="utf-8")
+            output = root / "output.txt"
+            output.write_bytes(b"existing output\n")
+            result = subprocess.run([sys.executable, symbolize_samples.__file__, str(capture),
+                                     "--libs", str(root), "--thread", "123", "--out", str(output)],
+                                    capture_output=True, text=True, timeout=5)
+            self.assertEqual(result.returncode, 2, result.stdout + result.stderr)
+            self.assertIn("thread filtering requires an address capture", result.stderr)
+            self.assertEqual(result.stdout, "")
+            self.assertNotIn("Traceback", result.stderr)
+            self.assertEqual(output.read_bytes(), b"existing output\n")
+
     def setUp(self):
         self.directory = tempfile.TemporaryDirectory()
         self.root = Path(self.directory.name)