fix(lineage): preserve producers through identical-content copies

README.md

@@ -394,6 +394,9 @@ Labels are stored locally by default. You can explicitly reconcile current local
 
 Register session, job, step, or artifact lineage with GLaaS.
 
+Publishing an unchanged copy of an artifact preserves its earlier producer steps,
+including when a packaging command reads back the files it just copied.
+
 ```bash
 roar register model.pt              # Register model lineage
 roar register --dry-run model.pt    # Preview without registering

roar/application/publish/lineage.py

@@ -11,6 +11,7 @@
 from ...core.digests import extract_primary_digest
 from ...core.interfaces.lineage import LineageData
 from ...db.context import create_database_context
+from ...db.lineage_order import ProducerOrder, preceding_producer, producer_order
 from ...db.query_context import create_query_database_context
 from ...execution.framework.registry import is_execution_task_job
 
@@ -396,17 +397,20 @@ def _get_lineage_jobs_read_only(
                 resolved_ids.append(artifact["id"])
 
         visited_jobs: set[int] = set()
-        visited_artifacts: set[str] = set()
+        visited_artifacts: set[tuple[str, ProducerOrder | None]] = set()
         jobs: list[dict] = []
 
-        def trace_upstream(artifact_id: str, current_depth: int) -> None:
-            if current_depth > max_depth or artifact_id in visited_artifacts:
+        def trace_upstream(
+            artifact_id: str, current_depth: int, before: ProducerOrder | None = None
+        ) -> None:
+            visit = (artifact_id, before)
+            if current_depth > max_depth or visit in visited_artifacts:
                 return
-            visited_artifacts.add(artifact_id)
+            visited_artifacts.add(visit)
 
             artifact_jobs = ctx_db.artifacts.get_jobs(artifact_id)
             produced_by = artifact_jobs.get("produced_by", [])
-            producer = produced_by[0] if produced_by else None
+            producer = preceding_producer(produced_by, before)
 
             if producer and producer["id"] not in visited_jobs:
                 visited_jobs.add(producer["id"])
@@ -428,7 +432,7 @@ def trace_upstream(artifact_id: str, current_depth: int) -> None:
                 ]
 
                 for inp in inputs:
-                    trace_upstream(inp["artifact_id"], current_depth + 1)
+                    trace_upstream(inp["artifact_id"], current_depth + 1, producer_order(producer))
 
                 outputs = ctx_db.jobs.get_outputs(producer["id"])
                 job_dict["_output_artifact_ids"] = [out["artifact_id"] for out in outputs]

roar/db/lineage_order.py

@@ -0,0 +1,18 @@
+"""Chronological producer selection for content-addressed lineage traversal."""
+
+from typing import Any
+
+ProducerOrder = tuple[float, int]
+
+
+def producer_order(job: dict[str, Any]) -> ProducerOrder:
+    """Use local insertion order to break equal start-time ties deterministically."""
+    return float(job["timestamp"]), int(job["id"])
+
+
+def preceding_producer(
+    producers: list[dict[str, Any]], before: ProducerOrder | None
+) -> dict[str, Any] | None:
+    """Choose the latest producer preceding the consuming job, when supplied."""
+    eligible = (job for job in producers if before is None or producer_order(job) < before)
+    return max(eligible, key=producer_order, default=None)

roar/db/services/lineage.py

@@ -9,6 +9,7 @@
 from ...core.digests import extract_primary_digest
 from ...core.interfaces.repositories import ArtifactRepository, JobRepository
 from ...core.interfaces.services import LineageService
+from ..lineage_order import ProducerOrder, preceding_producer, producer_order
 
 
 class DefaultLineageService(LineageService):
@@ -121,18 +122,21 @@ def get_lineage_jobs(
                     resolved_ids.append(artifact["id"])
 
         visited_jobs: set[int] = set()
-        visited_artifacts: set[str] = set()
+        visited_artifacts: set[tuple[str, ProducerOrder | None]] = set()
         jobs: list[dict[str, Any]] = []
 
-        def trace_upstream(artifact_id: str, current_depth: int):
-            if current_depth > max_depth or artifact_id in visited_artifacts:
+        def trace_upstream(
+            artifact_id: str, current_depth: int, before: ProducerOrder | None = None
+        ):
+            visit = (artifact_id, before)
+            if current_depth > max_depth or visit in visited_artifacts:
                 return
-            visited_artifacts.add(artifact_id)
+            visited_artifacts.add(visit)
 
             # Find the job that produced this artifact
             artifact_jobs = self._artifact_repo.get_jobs(artifact_id)
             produced_by = artifact_jobs.get("produced_by", [])
-            producer = produced_by[0] if produced_by else None
+            producer = preceding_producer(produced_by, before)
 
             if producer and producer["id"] not in visited_jobs:
                 visited_jobs.add(producer["id"])
@@ -156,7 +160,9 @@ def trace_upstream(artifact_id: str, current_depth: int):
                 ]
 
                 for inp in inputs:
-                    trace_upstream(inp["artifact_id"], current_depth + 1)
+                    # Copying and then hashing an output reads the same content
+                    # identity. Continue to its earlier producer, not this job again.
+                    trace_upstream(inp["artifact_id"], current_depth + 1, producer_order(producer))
 
                 # Get outputs
                 outputs = self._job_repo.get_outputs(producer["id"])

tests/application/publish/test_lineage.py

@@ -3,6 +3,8 @@
 import sqlite3
 from unittest.mock import Mock
 
+import pytest
+
 from roar.application.publish.lineage import (
     LineageCollector,
     _extract_primary_digest,
@@ -12,6 +14,56 @@
 from roar.db.schema import SCHEMA, run_migrations
 
 
+@pytest.mark.parametrize("method", ["collect", "collect_step", "collect_step_read_only"])
+@pytest.mark.parametrize("later_input_producer", [False, True])
+@pytest.mark.parametrize("same_timestamp", [False, True])
+def test_packaged_identical_content_preserves_training_ancestry(
+    tmp_path, method, later_input_producer, same_timestamp
+):
+    """Copying and hashing a checkpoint/report must preserve their earlier producers."""
+    roar_dir = tmp_path / ".roar"
+    with create_database_context(roar_dir) as db:
+        session_id = db.sessions.create(git_repo="/repo", git_commit="abc", make_active=True)
+        artifacts = {}
+        for name, digit in (("base", "1"), ("checkpoint", "2"), ("report", "3")):
+            artifacts[name], _ = db.artifacts.register(
+                {"blake3": digit * 64}, size=8, path=f"/work/{name}"
+            )
+        uids = {}
+        specs = [
+            ("fetch", ["base"], ["base"]),
+            ("train", ["base"], ["checkpoint"]),
+            ("evaluate", ["checkpoint"], ["report"]),
+            # Packaging copies both files, then hashes the copied outputs.
+            ("package", ["checkpoint", "report"], ["checkpoint", "report"]),
+        ]
+        if later_input_producer:
+            specs.append(("unrelated_later_fetch", [], ["base"]))
+        for step, (name, inputs, outputs) in enumerate(specs, 1):
+            job_id, uids[name] = db.jobs.create(
+                f"python {name}.py",
+                1.0 if same_timestamp else float(step),
+                session_id=session_id,
+                step_number=step,
+                duration_seconds=0.5,
+                exit_code=0,
+            )
+            for artifact in inputs:
+                db.jobs.add_input(job_id, artifacts[artifact], f"/work/{artifact}")
+            for artifact in outputs:
+                path = f"/release/{artifact}" if name == "package" else f"/work/{artifact}"
+                db.jobs.add_output(job_id, artifacts[artifact], path)
+
+    collector = LineageCollector()
+    if method == "collect":
+        lineage = collector.collect(["2" * 64, "3" * 64], roar_dir)
+    else:
+        lineage = getattr(collector, method)(session_id, 4, roar_dir)
+    assert [job["job_uid"] for job in lineage.jobs] == [
+        uids[name] for name in ("fetch", "train", "evaluate", "package")
+    ]
+
+
 class TestComputeIoSignature:
     """Tests for compute_io_signature function."""