diff --git a/README.md b/README.md index d8095002..038fec1c 100644 --- a/README.md +++ b/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 diff --git a/roar/application/publish/lineage.py b/roar/application/publish/lineage.py index 30fdec6b..c34564ac 100644 --- a/roar/application/publish/lineage.py +++ b/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] diff --git a/roar/db/lineage_order.py b/roar/db/lineage_order.py new file mode 100644 index 00000000..2090e967 --- /dev/null +++ b/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) diff --git a/roar/db/services/lineage.py b/roar/db/services/lineage.py index 57a02099..7aec2a2b 100644 --- a/roar/db/services/lineage.py +++ b/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"]) diff --git a/tests/application/publish/test_lineage.py b/tests/application/publish/test_lineage.py index ff247561..969f6298 100644 --- a/tests/application/publish/test_lineage.py +++ b/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."""