Add Stage 5 published artifact index

Author: anth-volk

changelog.d/1044.added

@@ -0,0 +1 @@
+Added the Stage 5 published artifact index JSONL artifact.

docs/engineering/stages/release_promotion.md

@@ -122,3 +122,18 @@ Runtime step manifests for `5_validate_and_promote_release` should include the
 contract as a JSON `contract` output. They may still record legacy validated
 input artifacts for compatibility, but the contract is the preferred semantic
 entry point for Stage 5 status and lineage.
+
+## Published Artifact Index
+
+Stage 5 also writes `published_artifact_index.jsonl` under the run-local
+`diagnostics/` directory. Each JSONL row describes one promoted artifact or
+release metadata artifact with its canonical `run_id`, candidate version,
+release version, source-stage metadata, final Hugging Face URI, and GCS URI
+when the artifact is mirrored to GCS.
+
+Build index rows from typed release candidate and promotion-result objects, not
+from console logs. Release manifest entries may supply final checksum, size,
+revision, and kind fields for promoted data artifacts; the index should leave
+the release manifest schema unchanged. The release promotion contract must
+reference the index as a `published_artifact_index` output so dashboards and AI
+systems can discover the per-artifact rows from the Stage 5 contract.

modal_app/pipeline.py

@@ -586,19 +586,26 @@ def _write_release_promotion_contract_for_run(
     run_context: RunContext,
     rel_paths: list[str],
     promotion_result,
-) -> ArtifactReference:
-    """Write Stage 5's run-local contract and return its manifest reference."""
+) -> tuple[ArtifactReference, ...]:
+    """Write Stage 5's run-local index/contract and return manifest references."""
 
     from policyengine_us_data.release_promotion import (
         build_legacy_release_candidate_bundle,
+        build_published_artifact_index,
+        published_artifact_index_artifact_ref,
+        published_artifact_index_path,
+        release_promotion_contract_repo_path,
         release_promotion_contract_path,
+        write_published_artifact_index,
         write_release_promotion_contract,
     )
+    from policyengine_us_data.stage_contracts import ArtifactRef
 
     run_dir = _run_dir(run_context.run_id)
+    release_context = _release_promotion_context_from_run_context(run_context)
     contract_path = release_promotion_contract_path(run_dir)
     candidate_bundle = build_legacy_release_candidate_bundle(
-        context=_release_promotion_context_from_run_context(run_context),
+        context=release_context,
         rel_paths=rel_paths,
         artifact_metadata_by_path=_release_artifact_metadata_by_path(
             run_context.run_id,
@@ -608,23 +615,61 @@ def _write_release_promotion_contract_for_run(
             run_context.run_id
         ),
     )
+    contract_artifact = ArtifactRef(
+        logical_name="release_promotion_contract",
+        uri=(
+            f"hf://{release_context.hf_repo_name}/"
+            f"{release_promotion_contract_repo_path(release_context.run_id)}"
+        ),
+        media_type="application/json",
+        metadata={
+            "artifact_family": "stage_contract",
+            "source_stage_id": "5_validate_and_promote_release",
+            "relative_path": release_promotion_contract_repo_path(
+                release_context.run_id
+            ),
+        },
+    )
+    published_index_path = published_artifact_index_path(run_dir)
+    published_index_rows = build_published_artifact_index(
+        candidate_bundle=candidate_bundle,
+        promotion_result=promotion_result,
+        diagnostic_artifacts=(contract_artifact,),
+    )
+    write_published_artifact_index(published_index_rows, published_index_path)
+    published_index_manifest_ref = ArtifactReference.from_path(
+        published_index_path,
+        role="index",
+        base_dir=run_dir,
+        media_type="application/jsonl",
+    )
+    published_index_artifact = published_artifact_index_artifact_ref(
+        release_context,
+        row_count=len(published_index_rows),
+        sha256=f"sha256:{published_index_manifest_ref.sha256}",
+        size_bytes=published_index_manifest_ref.size_bytes,
+    )
     write_release_promotion_contract(
         contract_path=contract_path,
         candidate_bundle=candidate_bundle,
         promotion_result=promotion_result,
         created_at=datetime.now(timezone.utc).isoformat(),
         code_sha=meta.sha,
         package_version=meta.version,
+        published_artifact_index=published_index_artifact,
         metadata={
             "writer": "modal_app.pipeline.promote_run",
             "branch": meta.branch,
         },
     )
-    return ArtifactReference.from_path(
-        contract_path,
-        role="contract",
-        base_dir=run_dir,
-        media_type="application/json",
+    return (
+        ArtifactReference.from_path(
+            contract_path,
+            role="contract",
+            base_dir=run_dir,
+            media_type="application/json",
+        ),
+        published_index_manifest_ref,
     )
 
 
@@ -2154,7 +2199,7 @@ def promote_run(
         )
         print(f"  {promotion_stdout}")
         promotion_result = _promotion_result_from_stdout(promotion_stdout)
-        release_promotion_contract_ref = _write_release_promotion_contract_for_run(
+        release_promotion_refs = _write_release_promotion_contract_for_run(
             meta=meta,
             run_context=promotion_context,
             rel_paths=rel_paths,
@@ -2172,7 +2217,7 @@ def promote_run(
                     ArtifactReference.from_dict(artifact)
                     for artifact in promote_inputs["validated_step_outputs"]
                 ],
-                release_promotion_contract_ref,
+                *release_promotion_refs,
             ],
             reuse_decision="computed",
             vol=pipeline_volume,

policyengine_us_data/release_promotion/__init__.py

@@ -31,6 +31,19 @@
     release_promotion_contract_repo_path,
     write_release_promotion_contract,
 )
+from .published_index import (
+    PUBLISHED_ARTIFACT_INDEX_FILENAME,
+    PUBLISHED_ARTIFACT_INDEX_MEDIA_TYPE,
+    PublishedArtifactIndexRow,
+    build_published_artifact_index,
+    published_artifact_index_artifact_ref,
+    published_artifact_index_from_jsonl,
+    published_artifact_index_path,
+    published_artifact_index_repo_path,
+    published_artifact_index_to_jsonl,
+    read_published_artifact_index,
+    write_published_artifact_index,
+)
 from .results import (
     CleanupPromotionResult,
     CompletionMarkerPromotionResult,
@@ -55,11 +68,14 @@
     "RELEASE_VALIDATION_SUBSTAGE_ID",
     "RELEASE_PROMOTION_CONTRACT_FILENAME",
     "RELEASE_PROMOTION_CONTRACT_TYPE",
+    "PUBLISHED_ARTIFACT_INDEX_FILENAME",
+    "PUBLISHED_ARTIFACT_INDEX_MEDIA_TYPE",
     "CleanupPromotionResult",
     "CompletionMarkerPromotionResult",
     "FullPromotionResult",
     "GcsPromotionResult",
     "HuggingFacePromotionResult",
+    "PublishedArtifactIndexRow",
     "ReleaseArtifactSpec",
     "ReleaseCandidateInputBundle",
     "ReleasePromotionContractBuilder",
@@ -69,6 +85,7 @@
     "ReleaseManifestPromotionResult",
     "VersionManifestPromotionResult",
     "build_legacy_release_candidate_bundle",
+    "build_published_artifact_index",
     "build_release_promotion_contract",
     "build_release_candidate_bundle_from_stage4_contract",
     "build_release_candidate_shape_report",
@@ -78,9 +95,16 @@
     "infer_release_artifact_spec",
     "logical_name_for_release_path",
     "normalize_release_path",
+    "published_artifact_index_artifact_ref",
+    "published_artifact_index_from_jsonl",
+    "published_artifact_index_path",
+    "published_artifact_index_repo_path",
+    "published_artifact_index_to_jsonl",
     "release_promotion_contract_path",
     "release_promotion_contract_repo_path",
+    "read_published_artifact_index",
     "read_stage4_release_candidate_bundle",
     "strip_staging_prefix",
+    "write_published_artifact_index",
     "write_release_promotion_contract",
 ]

policyengine_us_data/release_promotion/contract.py

@@ -78,13 +78,18 @@ class ReleasePromotionContractBuilder:
     package_version: str | None = None
     validation: ValidationReport | None = None
     diagnostics: Sequence[DiagnosticRef] = ()
+    published_artifact_index: ArtifactRef | None = None
     metadata: Mapping[str, Any] = field(default_factory=dict)
 
     def __post_init__(self) -> None:
         if not isinstance(self.candidate_bundle, ReleaseCandidateInputBundle):
             raise ValueError("candidate_bundle must be ReleaseCandidateInputBundle")
         if not isinstance(self.promotion_result, FullPromotionResult):
             raise ValueError("promotion_result must be FullPromotionResult")
+        if self.published_artifact_index is not None and not isinstance(
+            self.published_artifact_index, ArtifactRef
+        ):
+            raise ValueError("published_artifact_index must be ArtifactRef")
         object.__setattr__(
             self,
             "diagnostics",
@@ -100,10 +105,15 @@ def build(self) -> StageContract:
 
         context = self.candidate_bundle.context
         inputs = _contract_inputs(self.candidate_bundle)
-        outputs = _contract_outputs(context, self.promotion_result)
+        outputs = _contract_outputs(
+            context,
+            self.promotion_result,
+            published_artifact_index=self.published_artifact_index,
+        )
         parameters = _contract_parameters(
             self.candidate_bundle,
             self.promotion_result,
+            published_artifact_index=self.published_artifact_index,
         )
         return StageContract(
             contract_type=RELEASE_PROMOTION_CONTRACT_TYPE,
@@ -122,6 +132,11 @@ def build(self) -> StageContract:
                     "context": context.to_dict(),
                     "candidate_bundle": self.candidate_bundle.to_dict(),
                     "promotion_result": self.promotion_result.to_dict(),
+                    "published_artifact_index": (
+                        self.published_artifact_index.to_dict()
+                        if self.published_artifact_index is not None
+                        else None
+                    ),
                     "outputs": [output.to_dict() for output in outputs],
                 }
             ),
@@ -152,6 +167,7 @@ def build_release_promotion_contract(
     package_version: str | None = None,
     validation: ValidationReport | None = None,
     diagnostics: Sequence[DiagnosticRef] = (),
+    published_artifact_index: ArtifactRef | None = None,
     metadata: Mapping[str, Any] | None = None,
 ) -> StageContract:
     """Build the Stage 5 release promotion contract."""
@@ -164,6 +180,7 @@ def build_release_promotion_contract(
         package_version=package_version,
         validation=validation,
         diagnostics=diagnostics,
+        published_artifact_index=published_artifact_index,
         metadata=metadata or {},
     ).build()
 
@@ -178,6 +195,7 @@ def write_release_promotion_contract(
     package_version: str | None = None,
     validation: ValidationReport | None = None,
     diagnostics: Sequence[DiagnosticRef] = (),
+    published_artifact_index: ArtifactRef | None = None,
     metadata: Mapping[str, Any] | None = None,
 ) -> StageContract:
     """Build, write, and return the Stage 5 release promotion contract."""
@@ -190,6 +208,7 @@ def write_release_promotion_contract(
         package_version=package_version,
         validation=validation,
         diagnostics=diagnostics,
+        published_artifact_index=published_artifact_index,
         metadata=metadata,
     )
     write_contract(contract, contract_path)
@@ -266,13 +285,15 @@ def _contract_inputs(
 def _contract_outputs(
     context: ReleasePromotionContext,
     result: FullPromotionResult,
+    *,
+    published_artifact_index: ArtifactRef | None = None,
 ) -> tuple[ArtifactRef, ...]:
     hf_base = f"hf://{context.hf_repo_name}"
     completion_marker_path = (
         result.completion_marker.marker_path
         or f"releases/{context.release_version}/release-complete.json"
     )
-    return (
+    outputs = (
         ArtifactRef(
             logical_name="huggingface_release_artifacts",
             uri=f"{hf_base}/",
@@ -339,11 +360,16 @@ def _contract_outputs(
             metadata={"artifact_family": "release_completion_marker"},
         ),
     )
+    if published_artifact_index is not None:
+        outputs = (*outputs, published_artifact_index)
+    return outputs
 
 
 def _contract_parameters(
     candidate_bundle: ReleaseCandidateInputBundle,
     result: FullPromotionResult,
+    *,
+    published_artifact_index: ArtifactRef | None = None,
 ) -> dict[str, Any]:
     context = candidate_bundle.context
     return {
@@ -363,6 +389,9 @@ def _contract_parameters(
         "source_output_contract_path": candidate_bundle.source_output_contract_path,
         "validation_report_paths": list(candidate_bundle.validation_report_paths),
         "diagnostics_manifest_path": candidate_bundle.diagnostics_manifest_path,
+        "published_artifact_index_path": _artifact_relative_path(
+            published_artifact_index
+        ),
     }
 
 
@@ -374,6 +403,7 @@ def _contract_metadata(
     outputs: Sequence[ArtifactRef],
     extra: Mapping[str, Any],
 ) -> dict[str, Any]:
+    outputs_by_name = {output.logical_name: output for output in outputs}
     return {
         **dict(extra),
         "contract_file": RELEASE_PROMOTION_CONTRACT_FILENAME,
@@ -383,10 +413,22 @@ def _contract_metadata(
         "cleanup": promotion_result.cleanup.to_dict(),
         "already_finalized": promotion_result.already_finalized,
         "promotion_result": promotion_result.to_dict(),
+        "published_artifact_index": (
+            outputs_by_name["published_artifact_index"].to_dict()
+            if "published_artifact_index" in outputs_by_name
+            else None
+        ),
         "public_refs": {output.logical_name: output.uri for output in outputs},
     }
 
 
+def _artifact_relative_path(artifact: ArtifactRef | None) -> str | None:
+    if artifact is None:
+        return None
+    relative_path = artifact.metadata.get("relative_path")
+    return relative_path if isinstance(relative_path, str) and relative_path else None
+
+
 def _execution_record(result: FullPromotionResult) -> ExecutionRecord:
     return ExecutionRecord(
         status="completed",
@@ -411,6 +453,16 @@ def _substage_records(
     promotion_result: FullPromotionResult,
 ) -> tuple[SubstageRecord, ...]:
     outputs_by_name = {artifact.logical_name: artifact for artifact in public_outputs}
+    finalization_outputs = [
+        outputs_by_name["release_manifest"],
+        outputs_by_name["versioned_release_manifest"],
+        outputs_by_name["trace_tro"],
+        outputs_by_name["versioned_trace_tro"],
+        outputs_by_name["version_manifest"],
+        outputs_by_name["release_completion_marker"],
+    ]
+    if "published_artifact_index" in outputs_by_name:
+        finalization_outputs.append(outputs_by_name["published_artifact_index"])
     return (
         SubstageRecord(
             substage_id="5a_validate_outputs",
@@ -442,14 +494,7 @@ def _substage_records(
         SubstageRecord(
             substage_id="5d_write_version_manifest",
             status="completed",
-            outputs=(
-                outputs_by_name["release_manifest"],
-                outputs_by_name["versioned_release_manifest"],
-                outputs_by_name["trace_tro"],
-                outputs_by_name["versioned_trace_tro"],
-                outputs_by_name["version_manifest"],
-                outputs_by_name["release_completion_marker"],
-            ),
+            outputs=tuple(finalization_outputs),
             reuse_mode="handoff",
             metadata={
                 "version_manifest_updated": promotion_result.version_manifest.updated,