diag(shell): 失败转储的结构化解析 (#73)

experiments/shell/summarize_failure.py

@@ -0,0 +1,56 @@
+"""Summarise the workspace failure dump from a saved shell log (#73).
+
+The dump is three shapes of line:
+
+    workspace-failure mode=<mode> stage=<n> what=<text>
+      tab <group>:<index> input_high_water=<n> ... exit_code=<n>
+    workspace-failure mode=<mode> what=<text>
+
+This module turns them into rows so a failure can be read without scrolling a
+raw CI log. It parses text only; it never runs the shell.
+"""
+import re
+
+MODE = re.compile(r"workspace-failure mode=(?P<mode>\S+)(?: stage=(?P<stage>\d+))? what=(?P<what>.*)")
+TAB = re.compile(r"\btab (?P<group>main|bottom):(?P<index>\d+) (?P<fields>.*)")
+
+
+def parse(text):
+    """Return one dict per workspace failure, each with its tab counters."""
+    failures = []
+    current = None
+    for line in text.splitlines():
+        found = MODE.search(line)
+        if found:
+            current = dict(mode=found.group("mode"),
+                           stage=int(found.group("stage")) if found.group("stage") else None,
+                           what=found.group("what").strip(),
+                           tabs=[])
+            failures.append(current)
+            continue
+        found = TAB.search(line)
+        if found and current is not None:
+            counters = {}
+            for item in found.group("fields").split():
+                if "=" in item:
+                    key, value = item.split("=", 1)
+                    counters[key] = value
+            current["tabs"].append(dict(group=found.group("group"),
+                                        index=int(found.group("index")),
+                                        counters=counters))
+    return failures
+
+
+def describe(failure):
+    """Return a one-line verdict for the counters that decide the shape."""
+    verdicts = []
+    for tab in failure["tabs"]:
+        counters = tab["counters"]
+        read = counters.get("output_bytes")
+        consumed = counters.get("consumed")
+        drained = read is not None and read == consumed
+        verdicts.append(f"{tab['group']}:{tab['index']} "
+                        f"rejected={counters.get('input_rejected')} "
+                        f"drained={'yes' if drained else 'no'} "
+                        f"child={counters.get('exit_code')}")
+    return verdicts

experiments/shell/test_summarize_failure.py

@@ -0,0 +1,42 @@
+"""Contracts for the workspace failure dump summary (#73)."""
+import unittest
+
+from experiments.shell.summarize_failure import describe, parse
+
+DUMP = """ValueError: shell probe failed: workspace-failure mode=ui-copy-system stage=7 what=selection scenario exceeded its bounded deadline
+  tab bottom:0 input_high_water=26 input_rejected=0 output_bytes=688 read_error=0 consumed=688 stage=0 closing=false child_signaled=false exit_code=259
+  tab main:0 input_high_water=2 input_rejected=0 output_bytes=577 read_error=0 consumed=577 stage=0 closing=false child_signaled=false exit_code=259
+"""
+
+OLD_STYLE = """workspace-failure mode=ui-selection what=selection scenario exceeded its bounded deadline
+  tab bottom:0 input_high_water=3 input_rejected=0 output_bytes=637 read_error=0 consumed=637 stage=0 closing=false child_signaled=false exit_code=259
+"""
+
+
+class WorkspaceFailureDump(unittest.TestCase):
+    def test_stage_and_both_tabs_are_read(self):
+        failures = parse(DUMP)
+        self.assertEqual(len(failures), 1)
+        self.assertEqual(failures[0]["mode"], "ui-copy-system")
+        self.assertEqual(failures[0]["stage"], 7)
+        self.assertEqual([(tab["group"], tab["index"]) for tab in failures[0]["tabs"]],
+                         [("bottom", 0), ("main", 0)])
+
+    def test_missing_stage_still_parses(self):
+        failures = parse(OLD_STYLE)
+        self.assertEqual(len(failures), 1)
+        self.assertIsNone(failures[0]["stage"])
+
+    def test_drained_and_child_state_are_reported(self):
+        verdicts = describe(parse(DUMP)[0])
+        self.assertEqual(verdicts[0], "bottom:0 rejected=0 drained=yes child=259")
+        self.assertEqual(verdicts[1], "main:0 rejected=0 drained=yes child=259")
+
+    def test_partial_read_is_not_called_drained(self):
+        failure = parse(DUMP)[0]
+        failure["tabs"][0]["counters"]["consumed"] = "1"
+        self.assertIn("drained=no", describe(failure)[0])
+
+
+if __name__ == "__main__":
+    unittest.main()