1510 add reconstructed ephemeris to kernels that trigger only new coverage (#1512)

* Handle reconstructed SPICE coverage triggers

* Clarify predicted SPICE trigger filtering

* Explain growing SPICE kernels

* Add predicts back to event bridge kernels

* Update test with correct spice event list

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>

Author: tmplummer

sds_data_manager/lambda_code/SDSCode/pipeline_lambdas/spice_indexer.py

@@ -519,6 +519,7 @@ def send_spice_event(spice_obj: SPICEFilePath, s3_key: str):
     spice_events = [
         "attitude_history",
         "attitude_predict",
+        "pointing_attitude",
         "ephemeris_reconstructed",
         "ephemeris_nominal",
         "ephemeris_predict",

sds_data_manager/orchestration/imap_job.py

@@ -634,7 +634,11 @@ def trigger_from_new_non_science_inputs(
             if new_files:
                 latest_ingestion_date = max(f.ingestion_date for f in new_files)
 
-            if cursor_str == config.MISSION_START_TIME:
+            if dependency.source in spice.NON_TRIGGERING_KERNEL_TYPES:
+                context.log.info(
+                    f"Skipping trigger evaluation for {dependency.source}."
+                )
+            elif cursor_str == config.MISSION_START_TIME:
                 # Just process everything, don't bother looping through all new files.
                 partitions = dagster_utilities.get_affected_partitions(
                     context,

sds_data_manager/orchestration/spice.py

@@ -14,6 +14,25 @@
 logger.setLevel(logging.INFO)
 
 
+# These kernels are delivered as append-only time series: newer files either
+# extend an existing coverage window or replace it with a higher version of the
+# same window, so downstream triggering should be narrowed to genuinely new
+# coverage where possible.
+GROWING_KERNEL_TYPES = (
+    "attitude_history",
+    "pointing_attitude",
+    "ephemeris_reconstructed",
+)
+# Kernel types that we don't want to trigger processing.
+# Predicted ephemeris appears as `ephemeris_predicted` in dependency YAML and
+# as `ephemeris_predict` in indexed SPICE metadata, so guard against both.
+NON_TRIGGERING_KERNEL_TYPES = (
+    "attitude_predict",
+    "ephemeris_predict",
+    "ephemeris_predicted",
+)
+
+
 def check_requested_kernels(combined_kernel_sources, metakernel_files):
     """Check if all requested kernels are present in the metakernel files.
 
@@ -197,7 +216,14 @@ def _parse_interval_list(
     if not raw_intervals:
         return []
     return [
-        [datetime.datetime.fromisoformat(start), datetime.datetime.fromisoformat(end)]
+        [
+            start
+            if isinstance(start, datetime.datetime)
+            else datetime.datetime.fromisoformat(start),
+            end
+            if isinstance(end, datetime.datetime)
+            else datetime.datetime.fromisoformat(end),
+        ]
         for start, end in raw_intervals
     ]
 
@@ -269,15 +295,6 @@ def subtract_intervals(
     return leftover
 
 
-# Kernel types whose coverage grows by appending segments to the same file
-# series over time (attitude_history: MOC-delivered AH kernels; pointing_attitude:
-# the DPS kernel produced by the spacecraft l1a pointing-attitude job). Both use
-# the same start-date_end-date_version filename convention and the same
-# ingestion-time SPICE segment decomposition, so the same trigger-range rules
-# apply to both.
-GROWING_KERNEL_TYPES = ("attitude_history", "pointing_attitude")
-
-
 def get_growing_kernel_trigger_ranges(
     session, kernel_type: str, new_files: list["models.SPICEFiles"]
 ) -> list[tuple[datetime.datetime, datetime.datetime]]:

tests/lambda_endpoints/test_spice_dependency_triggers.py

@@ -0,0 +1,141 @@
+"""Tests for SPICE dependency trigger narrowing."""
+
+from datetime import datetime
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+from dagster import DagsterInstance, build_sensor_context
+
+from sds_data_manager.lambda_code.SDSCode.database import models
+from sds_data_manager.orchestration.dependency import DependencyConfigReader
+from sds_data_manager.orchestration.imap_job import IMAPJobHandler
+
+
+def _mag_job_handler():
+    """Return a job handler with ephemeris dependencies."""
+    reader = DependencyConfigReader()
+    return IMAPJobHandler(reader.config[("mag", "l1d", "norm-srf")])
+
+
+def _session_returning(new_files, predecessor):
+    """Return a mock session for trigger range testing."""
+    session = MagicMock()
+    session.query.return_value.filter.return_value.all.return_value = new_files
+    first = session.query.return_value.filter.return_value.order_by.return_value.first
+    first.return_value = predecessor
+    return session
+
+
+def _spice_file(file_name, min_dt, max_dt, intervals):
+    """Build a lightweight SPICE file object for tests."""
+    return SimpleNamespace(
+        file_name=file_name,
+        min_date_datetime=min_dt,
+        max_date_datetime=max_dt,
+        file_intervals_datetime=intervals,
+        ingestion_date=datetime(2025, 1, 5),
+    )
+
+
+def test_ephemeris_reconstructed_only_triggers_new_daily_partitions():
+    """Ephemeris reconstructed kernels should only trigger for new coverage."""
+    instance = DagsterInstance.ephemeral()
+    instance.add_dynamic_partitions(
+        "daily_partitions",
+        [
+            "daily_2025-01-01T00:00:00_to_2025-01-02T00:00:00",
+            "daily_2025-01-02T00:00:00_to_2025-01-03T00:00:00",
+            "daily_2025-01-03T00:00:00_to_2025-01-04T00:00:00",
+        ],
+    )
+    context = build_sensor_context(instance=instance)
+
+    job_handler = _mag_job_handler()
+    dependency = next(
+        dep
+        for dep in job_handler.job_config.spice_inputs
+        if dep.source == "ephemeris_reconstructed"
+    )
+
+    min_dt = datetime(2025, 1, 1)
+    old_max = datetime(2025, 1, 2)
+    new_max = datetime(2025, 1, 4)
+    predecessor = _spice_file(
+        "imap_2025_001_2025_002_01.bsp",
+        min_dt,
+        old_max,
+        [[min_dt, old_max]],
+    )
+    new_file = _spice_file(
+        "imap_2025_001_2025_004_01.bsp",
+        min_dt,
+        new_max,
+        [[min_dt, old_max], [old_max, new_max]],
+    )
+    session = _session_returning([new_file], predecessor)
+    cursors = {dependency.to_dagster_name(): "2024-01-01T00:00:00"}
+
+    with patch(
+        "sds_data_manager.orchestration.imap_job.db.Session"
+    ) as mock_session_cls:
+        mock_session_cls.return_value.__enter__.return_value = session
+        mock_session_cls.return_value.__exit__.return_value = False
+        partitions = job_handler.trigger_from_new_non_science_inputs(
+            context,
+            dependency,
+            cursors,
+            models.SPICEFiles,
+            models.SPICEFiles.kernel_type,
+            None,
+            "min_date_datetime",
+            "max_date_datetime",
+        )
+
+    assert set(partitions) == {
+        "daily_2025-01-02T00:00:00_to_2025-01-03T00:00:00",
+        "daily_2025-01-03T00:00:00_to_2025-01-04T00:00:00",
+    }
+
+
+def test_predicted_ephemeris_does_not_trigger_partitions():
+    """Predicted ephemeris kernels should not trigger reprocessing."""
+    instance = DagsterInstance.ephemeral()
+    instance.add_dynamic_partitions(
+        "daily_partitions",
+        ["daily_2025-01-01T00:00:00_to_2025-01-02T00:00:00"],
+    )
+    context = build_sensor_context(instance=instance)
+
+    job_handler = _mag_job_handler()
+    dependency = next(
+        dep
+        for dep in job_handler.job_config.spice_inputs
+        if dep.source == "ephemeris_predicted"
+    )
+
+    new_file = _spice_file(
+        "imap_2025_001_2025_002_01_pred.bsp",
+        datetime(2025, 1, 1),
+        datetime(2025, 1, 2),
+        [[datetime(2025, 1, 1), datetime(2025, 1, 2)]],
+    )
+    session = _session_returning([new_file], None)
+    cursors = {dependency.to_dagster_name(): "2024-01-01T00:00:00"}
+
+    with patch(
+        "sds_data_manager.orchestration.imap_job.db.Session"
+    ) as mock_session_cls:
+        mock_session_cls.return_value.__enter__.return_value = session
+        mock_session_cls.return_value.__exit__.return_value = False
+        partitions = job_handler.trigger_from_new_non_science_inputs(
+            context,
+            dependency,
+            cursors,
+            models.SPICEFiles,
+            models.SPICEFiles.kernel_type,
+            None,
+            "min_date_datetime",
+            "max_date_datetime",
+        )
+
+    assert partitions == []

tests/lambda_endpoints/test_spice_indexer_lambda.py

@@ -4,7 +4,7 @@
 import os
 from datetime import datetime, timedelta
 from pathlib import Path
-from unittest.mock import patch
+from unittest.mock import Mock, patch
 
 import numpy as np
 import pytest
@@ -456,6 +456,26 @@ def test_send_spice_event(session, events_client, s3_client):
         spice_indexer.lambda_handler(event, None)
 
 
+def test_send_spice_event_filters_kernel_types(events_client):
+    """Only reconstructed/current coverage kernels should emit events."""
+    with patch(
+        "sds_data_manager.lambda_code.SDSCode.pipeline_lambdas.spice_indexer.boto3.client",
+        return_value=events_client,
+    ):
+        for kernel_type in (
+            "attitude_history",
+            "pointing_attitude",
+            "ephemeris_reconstructed",
+            "attitude_predict",
+            "ephemeris_predict",
+        ):
+            result = spice_indexer.send_spice_event(
+                Mock(spice_metadata={"type": kernel_type}),
+                "imap/spice/test/file",
+            )
+            assert result["ResponseMetadata"]["HTTPStatusCode"] == 200
+
+
 @patch(
     "sds_data_manager.lambda_code.SDSCode.pipeline_lambdas.spice_indexer.download_from_s3"
 )