test(history): align output fixtures with native path styles

tests/test_execution_output_database.py

@@ -17,9 +17,11 @@ except ImportError:
     import test_execution_history_database as history_tests
 
 
-def bundle(filename="first.png", *, root="a" * 64, node="9", ordinal=0):
+def bundle(
+    filename="first.png", *, root="a" * 64, node="9", ordinal=0, path_style="posix"
+):
     return OutputBundle(
-        (OutputReference(node, ordinal, "output", "", filename, "posix", root),),
+        (OutputReference(node, ordinal, "output", "", filename, path_style, root),),
         "complete" if root else "partial",
         "reported" if root else "root_unavailable",
     )

tests/test_execution_output_gallery.py

@@ -1,5 +1,7 @@
 """Mutable metadata galleries never own or retarget frozen output references."""
 
+from pathlib import PurePosixPath, PureWindowsPath
+
 from PIL import Image
 from PIL.PngImagePlugin import PngInfo
 
@@ -18,11 +20,11 @@ class OutputGalleryTests(AdminAPITestCase):
         db = self.api.db
         start = self.start(db, item)
         db.write_execution_start(start)
-        key = snapshot_output_roots(
-            "posix", str(root or self.output_dir), None
-        ).output_key
+        output_root = root or self.output_dir
+        style = "windows" if isinstance(output_root, PureWindowsPath) else "posix"
+        key = snapshot_output_roots(style, str(output_root), None).output_key
         db.write_execution_outputs_for_finish(
-            self.finish(start), bundle(filename, root=key)
+            self.finish(start), bundle(filename, root=key, path_style=style)
         )
         return start.run_id, db.get_execution_outputs(start.run_id)
 
@@ -48,6 +50,20 @@ class OutputGalleryTests(AdminAPITestCase):
             )
         }
 
+    def test_capture_matches_root_path_style_on_each_platform(self) -> None:
+        roots = (
+            (PurePosixPath("/output"), "posix"),
+            (PureWindowsPath(r"D:\output"), "windows"),
+        )
+        for item, (root, style) in enumerate(roots):
+            with self.subTest(style=style):
+                run, evidence = self.capture(item=item, root=root)
+                self.assertEqual(evidence["state"], "complete")
+                reference = evidence["references"][0]
+                self.assertEqual(reference["path_style"], style)
+                self.assertEqual(len(reference["root_key"]), 64)
+                self.assert_evidence(run, evidence)
+
     def test_metadata_disabled_or_malformed_still_captures_without_gallery_rows(self):
         for item, metadata in enumerate(({}, {"prompt": "{malformed"})):
             with self.subTest(metadata=metadata):

tests/test_execution_output_hooks.py

@@ -1,6 +1,8 @@
 """Exact successful host callbacks persist independently bounded output evidence."""
 
 from dataclasses import replace
+import os
+from pathlib import Path
 import threading
 import sys
 from types import SimpleNamespace
@@ -309,21 +311,53 @@ class OutputHooksTests(fixtures.unittest.TestCase):
         self.assertEqual(registration.writer.context_budget.charged_bytes, 0)
 
     def test_default_adapter_uses_configured_output_and_temp_namespaces(self):
-        folder_paths = SimpleNamespace(
-            get_output_directory=lambda: "/output", get_temp_directory=lambda: "/temp"
+        native_root = Path(self.directory.name)
+        cases = (
+            (
+                os.name,
+                str(native_root / "output"),
+                str(native_root / "temp"),
+                "windows" if os.name == "nt" else "posix",
+            ),
+            ("posix", "/configured", "/configured", "posix"),
+            ("nt", r"D:\configured", r"D:\configured", "windows"),
         )
-        with patch.dict(sys.modules, {"folder_paths": folder_paths}):
-            registration = HistoryRegistration(
-                self.queue, fixtures.JOB, lambda: self.db
+        for item, (platform, output_root, temp_root, style) in enumerate(cases):
+            folder_paths = SimpleNamespace(
+                get_output_directory=lambda: output_root,
+                get_temp_directory=lambda: temp_root,
             )
-            self.registrations.append(registration)
-            self.wait_for(lambda: registration.writer.status().accepting)
-            self.dequeue(0)
-            self.queue.task_done(0, history(), fixtures.status())
-            self.drain(registration)
-        self.assertEqual(self.outputs(0)["state"], "complete")
-        self.assertEqual(len(self.outputs(0)["references"][0]["root_key"]), 64)
-        self.assertEqual(registration._roots, {})
+            with (
+                self.subTest(platform=platform, root=output_root),
+                patch.dict(sys.modules, {"folder_paths": folder_paths}),
+                patch("utils.execution_history.os", SimpleNamespace(name=platform)),
+            ):
+                registration = HistoryRegistration(
+                    self.queue, fixtures.JOB, lambda: self.db
+                )
+                self.registrations.append(registration)
+                self.wait_for(lambda: registration.writer.status().accepting)
+                self.dequeue(item)
+                payload = history()
+                payload["outputs"]["save"]["images"].append(
+                    {"filename": "a.png", "subfolder": "", "type": "temp"}
+                )
+                self.queue.task_done(item, payload, fixtures.status())
+                self.drain(registration)
+                evidence = self.outputs(item)
+                self.assertEqual(evidence["state"], "complete")
+                references = evidence["references"]
+                self.assertEqual(len(references[0]["root_key"]), 64)
+                self.assertEqual(len(references[1]["root_key"]), 64)
+                self.assertEqual([r["path_style"] for r in references], [style] * 2)
+                self.assertEqual(
+                    [r["folder_type"] for r in references], ["output", "temp"]
+                )
+                self.assertNotEqual(
+                    references[0]["root_key"], references[1]["root_key"]
+                )
+                self.assertEqual(registration._roots, {})
+                self.assertTrue(registration.dispose())
 
     def test_sealed_context_is_discarded_before_output_reservation(self):
         pool = ContextBudget(1048576)