marin: rename .artifact -> artifact.json with legacy fallback

Addresses #5732 (comment).
`Artifact.save` now writes `artifact.json`; `from_path` reads it first and
falls back to the legacy `.artifact` dotfile so historical GCS outputs remain
loadable. The fallback can be deleted once those prefixes are gone.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: ravwojdyla

lib/marin/src/marin/execution/artifact.py

@@ -15,7 +15,10 @@
 
 
 class Artifact:
-    __artifact_file_name = ".artifact"
+    __artifact_file_name = "artifact.json"
+    # Legacy filename written before the rename; read-only fallback so historical
+    # GCS outputs remain loadable. Safe to remove once those prefixes are gone.
+    __legacy_artifact_file_name = ".artifact"
 
     @overload
     @classmethod
@@ -34,29 +37,32 @@ def from_path(
         If ``base_path`` is a relative path (no URL scheme, doesn't start with ``/``),
         it is resolved against ``marin_prefix()``.
 
-        If ``base_path`` has no ``.artifact`` file but its ``.executor_status`` file
-        contains ``SUCCESS``, returns a :class:`PathMetadata` pointing at ``base_path``
-        — provided the caller asked for no specific type or for ``PathMetadata``.
+        If ``base_path`` has no ``artifact.json`` (or legacy ``.artifact``) file but
+        its ``.executor_status`` file contains ``SUCCESS``, returns a
+        :class:`PathMetadata` pointing at ``base_path`` — provided the caller asked
+        for no specific type or for ``PathMetadata``.
         """
 
         if isinstance(base_path, StepSpec):
             base_path = base_path.output_path
         elif _is_relative_path(base_path):
             base_path = f"{marin_prefix()}/{base_path}"
 
-        try:
-            with open_url(f"{base_path}/{cls.__artifact_file_name}", "rb") as fd:
-                if artifact_type is None:
-                    return json.load(fd)
-                if not issubclass(artifact_type, BaseModel):
-                    raise TypeError(f"artifact_type must be a pydantic BaseModel subclass, got {artifact_type!r}")
-                return artifact_type.model_validate_json(fd.read())
-        except FileNotFoundError:
-            return cls._from_executor_status(base_path, artifact_type)
+        for file_name in (cls.__artifact_file_name, cls.__legacy_artifact_file_name):
+            try:
+                with open_url(f"{base_path}/{file_name}", "rb") as fd:
+                    if artifact_type is None:
+                        return json.load(fd)
+                    if not issubclass(artifact_type, BaseModel):
+                        raise TypeError(f"artifact_type must be a pydantic BaseModel subclass, got {artifact_type!r}")
+                    return artifact_type.model_validate_json(fd.read())
+            except FileNotFoundError:
+                continue
+        return cls._from_executor_status(base_path, artifact_type)
 
     @classmethod
     def _from_executor_status(cls, base_path: str, artifact_type: type[T] | None) -> "T | PathMetadata":
-        """Fallback when ``.artifact`` is absent: synthesize a :class:`PathMetadata`
+        """Fallback when no artifact file is present: synthesize a :class:`PathMetadata`
         if the step published ``.executor_status = SUCCESS``.
 
         Only valid when the caller wants no type or ``PathMetadata`` — other types

tests/execution/test_step_runner.py

@@ -125,7 +125,7 @@ def test_artifact_load_relative_path_resolves_against_marin_prefix(tmp_path: Pat
 
 
 def test_artifact_from_executor_status_success_untyped(tmp_path: Path):
-    """No .artifact, but .executor_status=SUCCESS: synthesize PathMetadata."""
+    """No artifact file, but .executor_status=SUCCESS: synthesize PathMetadata."""
     (tmp_path / ".executor_status").write_text("SUCCESS")
 
     loaded = Artifact.from_path(tmp_path.as_posix())
@@ -134,29 +134,37 @@ def test_artifact_from_executor_status_success_untyped(tmp_path: Path):
 
 
 def test_artifact_from_executor_status_success_typed_pathmetadata(tmp_path: Path):
-    """No .artifact, but .executor_status=SUCCESS and caller asks for PathMetadata."""
+    """No artifact file, but .executor_status=SUCCESS and caller asks for PathMetadata."""
     (tmp_path / ".executor_status").write_text("SUCCESS")
 
     loaded = Artifact.from_path(tmp_path.as_posix(), PathMetadata)
     assert loaded == PathMetadata(path=tmp_path.as_posix())
 
 
 def test_artifact_from_executor_status_success_typed_other_raises(tmp_path: Path):
-    """No .artifact, .executor_status=SUCCESS, but caller asks for a different type."""
+    """No artifact file, .executor_status=SUCCESS, but caller asks for a different type."""
     (tmp_path / ".executor_status").write_text("SUCCESS")
 
     with pytest.raises(FileNotFoundError, match="cannot synthesize"):
         Artifact.from_path(tmp_path.as_posix(), TokenizeMetadata)
 
 
 def test_artifact_from_executor_status_non_success_raises(tmp_path: Path):
-    """No .artifact, .executor_status present but not SUCCESS."""
+    """No artifact file, .executor_status present but not SUCCESS."""
     (tmp_path / ".executor_status").write_text("RUNNING")
 
     with pytest.raises(FileNotFoundError, match="not 'SUCCESS'"):
         Artifact.from_path(tmp_path.as_posix())
 
 
+def test_artifact_load_legacy_dotfile(tmp_path: Path):
+    """Historical outputs wrote `.artifact`; from_path should still load them."""
+    (tmp_path / ".artifact").write_text(json.dumps({"path": "/legacy"}))
+
+    loaded = Artifact.from_path(tmp_path.as_posix(), PathMetadata)
+    assert loaded == PathMetadata(path="/legacy")
+
+
 def test_artifact_save_and_load_untyped(tmp_path: Path):
     artifact = TokenizeMetadata(path="/tokenized", num_tokens=42)
     Artifact.save(artifact, tmp_path.as_posix())
@@ -440,7 +448,7 @@ def test_runner_skips_completed_steps(tmp_path: Path):
     runner1.run(steps)
 
     # Record modification times
-    tokenize_artifact_path = os.path.join(steps[1].output_path, ".artifact")
+    tokenize_artifact_path = os.path.join(steps[1].output_path, "artifact.json")
     mtime_before = os.path.getmtime(tokenize_artifact_path)
 
     # Re-run — all steps should be skipped