Add scoped full recalculation to Impulse reporting

docs/impulse/docs/config/configuration.md

@@ -354,6 +354,42 @@ mode-resolution rules and what counts as a definition change.
 
 ---
 
+## full_recalculation (optional)
+
+Forces a full recalculation of specific entities on an **incremental** run: the listed
+aggregations, events, and/or calculated channels recompute over **all** matching containers and
+have their gold rows fully replaced, regardless of whether their `definition_hash` changed. This is
+the same treatment a definition change already receives, applied on demand — useful to backfill
+after fixing a persistence bug or correcting upstream data, without rerunning everything in full
+mode. In full mode it is a no-op, since every entity already recomputes over all containers.
+
+Entities are identified by their human-readable `name`. A name that matches no registered entity
+fails the run fast with a `ValueError`, so typos surface immediately rather than silently doing
+nothing.
+
+| Field                 | Type        | Default | Description                                      |
+|-----------------------|-------------|---------|--------------------------------------------------|
+| `aggregations`        | `list[str]` | `[]`    | Names of aggregations to fully recompute.        |
+| `events`              | `list[str]` | `[]`    | Names of events to fully recompute.              |
+| `calculated_channels` | `list[str]` | `[]`    | Names of calculated channels to fully recompute. |
+
+```json
+{
+  "full_recalculation": {
+    "aggregations": ["rpm_hist_p1"],
+    "events": ["overspeed"],
+    "calculated_channels": ["power_kw"]
+  }
+}
+```
+
+:::note Incremental only
+`full_recalculation` only has an effect when the run is incremental. A first run, or any full run,
+already recomputes every entity over all containers, so the scope is ignored.
+:::
+
+---
+
 ## calculated_channels (optional)
 
 Controls the optional `calculated_channel_metrics` output. By default a report

docs/impulse/docs/references/api/impulse_reporting/config/config_parser.md

@@ -265,6 +265,33 @@ attribute key of the same name.
 registered KPI (see ``calculated_channel_kpis.KPI_BUILDERS``); an unknown
 name is rejected at validation. Duplicates are removed (order preserved).
 
+## FullRecalculation
+
+```python
+class FullRecalculation(BaseModel)
+```
+
+Scoped full-recalculation request, by entity name.
+
+Lists aggregations / events / calculated channels — identified by their
+human-readable ``name`` — that should be fully recalculated on the next run,
+regardless of whether their definition hash changed.
+
+Only meaningful in incremental mode: listed entities recompute over **all**
+containers and fully replace their gold rows (the same treatment a
+definition-hash change already receives), while every other entity stays
+incremental. In full mode this is a no-op, since everything recomputes over all
+containers anyway.
+
+Names that do not match any registered entity are rejected at the start of
+``Report.determine_report`` (fail fast on typos / stale names).
+
+**Arguments**:
+
+- `aggregations` (`list of str, default=[]`): Names of aggregations to fully recalculate.
+- `events` (`list of str, default=[]`): Names of events to fully recalculate.
+- `calculated_channels` (`list of str, default=[]`): Names of calculated channels to fully recalculate.
+
 ## ImpulseConfig
 
 ```python
@@ -288,6 +315,10 @@ Attributes
  calculated_channels : CalculatedChannels, optional
      Optional calculated-channel output configuration (e.g. opting in to the
      ``calculated_channel_metrics`` table). Defaults to CalculatedChannels().
+ full_recalculation : FullRecalculation, optional
+     Optional scoped full-recalculation request naming aggregations / events /
+     calculated channels to fully recompute on the next incremental run
+     regardless of definition-hash changes. Defaults to None (no override).
  measurement_dimensions : list of str, optional
      Column names to surface from ``container_metrics`` into the
      gold-layer ``measurement_dimension`` table. Names are matched

docs/impulse/docs/references/report/index.md

@@ -119,6 +119,10 @@ Only the hashed attributes matter. Anything else is cosmetic and won't trigger r
 
 Renaming an aggregation, tweaking the description, or changing units keeps the hash stable. No reprocessing. `channel_names` (and a cross-channel statistic's `channel_name`) **do** affect the hash for `StatsAggregator` / `PointValueAggregator`, because they are the fact table's `channel_name` merge key — renaming forces a recompute so old-name rows are pruned rather than left stale.
 
+#### Forcing a full recalculation
+
+To recompute specific entities over all containers even when their `definition_hash` is unchanged, list their names under [`full_recalculation`](../../config/configuration.md#full_recalculation-optional) in config. The listed aggregations, events, and calculated channels are treated as **changed** for that run — they recompute over all matching containers and their gold rows are fully replaced — while every other entity stays incremental. This is meant for backfills after a persistence fix or corrected upstream data. In full mode it is a no-op, and a name that matches no registered entity fails the run fast.
+
 #### Container-update detection
 
 `ContainerUpsertDetector.detect_upserted_containers` finds two things and unions them:

skills/impulse-config/SKILL.md

@@ -6,7 +6,7 @@ description: >
   "configure an Impulse report", set the source/sink tables, filter which containers are processed,
   choose RLE vs RAW, turn on incremental processing, run without writing (sinkless), remap column
   names, or scope by project. Covers source, unity_sink, container_filters, query_engine, solver_config,
-  incremental, measurement_dimensions, and calculated_channels, all validated by Pydantic.
+  incremental, full_recalculation, measurement_dimensions, and calculated_channels, all validated by Pydantic.
 ---
 
 # Impulse — configuration
@@ -40,6 +40,7 @@ config = {
         ],
     },
     "incremental": {"enabled": True},
+    "full_recalculation": {"aggregations": ["rpm_hist_p1"]},  # optional
     "measurement_dimensions": ["container_id", "vehicle_key", "start_ts", "stop_ts"],
     "calculated_channels": {"emit_channel_metrics": True, "attribute_columns": ["unit"]},  # optional
 }
@@ -168,6 +169,22 @@ Reuses prior results for unchanged definitions and reprocesses only new/updated
 | `silver_last_modified_column` | `"timestamp"`   | Silver column used to detect container updates. |
 | `gold_last_modified_column`   | `"_created_at"` | Gold column used to detect prior-run freshness. |
 
+## full_recalculation (optional)
+
+Forces a full recalculation of specific entities on an **incremental** run: the listed aggregations,
+events, and/or calculated channels recompute over **all** matching containers and have their gold rows
+fully replaced, even when their `definition_hash` is unchanged (the same treatment a definition change
+already receives, applied on demand). Use it to backfill after fixing a persistence bug or correcting
+upstream data, without a full-mode rerun of everything. In full mode it is a no-op — everything already
+recomputes. Entities are named by their human-readable `name`; a name matching no registered entity
+fails the run fast with a `ValueError`, so typos surface immediately.
+
+| Field                 | Default | Description                                    |
+|-----------------------|---------|------------------------------------------------|
+| `aggregations`        | `[]`    | Names of aggregations to fully recompute.      |
+| `events`              | `[]`    | Names of events to fully recompute.            |
+| `calculated_channels` | `[]`    | Names of calculated channels to fully recompute. |
+
 ## measurement_dimensions (optional)
 
 List of `container_metrics` columns (post-mapping **internal** names) to surface into the gold

skills/impulse-reporting/SKILL.md

@@ -122,6 +122,12 @@ definitions reprocess only new/updated containers (persisted via Delta `MERGE` o
 changed or brand-new definitions reprocess all matching containers (replaced atomically via
 `replaceWhere` on `visual_id`/`event_id`/`channel_id`). A single run can mix both per entity.
 
+**Forcing a full recalculation.** To recompute specific *unchanged* entities over all containers — e.g.
+to backfill after a persistence fix or corrected upstream data — list their names under
+`full_recalculation` in config (see `impulse-config`). Those aggregations/events/calculated channels are
+treated as changed for that run regardless of their hash; everything else stays incremental. A no-op in
+full mode, and a name matching no registered entity fails fast.
+
 **What counts as a definition change** — only the hashed attributes; renames, descriptions, and units
 are cosmetic and do not trigger reprocessing:
 

src/impulse_reporting/config/config_parser.py

@@ -500,6 +500,38 @@ def _normalize_kpis(cls, value: list[str]) -> list[str]:
         return normalized
 
 
+class FullRecalculation(BaseModel):
+    """
+    Scoped full-recalculation request, by entity name.
+
+    Lists aggregations / events / calculated channels — identified by their
+    human-readable ``name`` — that should be fully recalculated on the next run,
+    regardless of whether their definition hash changed.
+
+    Only meaningful in incremental mode: listed entities recompute over **all**
+    containers and fully replace their gold rows (the same treatment a
+    definition-hash change already receives), while every other entity stays
+    incremental. In full mode this is a no-op, since everything recomputes over all
+    containers anyway.
+
+    Names that do not match any registered entity are rejected at the start of
+    ``Report.determine_report`` (fail fast on typos / stale names).
+
+    Attributes
+    ----------
+    aggregations : list of str, default=[]
+        Names of aggregations to fully recalculate.
+    events : list of str, default=[]
+        Names of events to fully recalculate.
+    calculated_channels : list of str, default=[]
+        Names of calculated channels to fully recalculate.
+    """
+
+    aggregations: list[str] = []
+    events: list[str] = []
+    calculated_channels: list[str] = []
+
+
 class ImpulseConfig(BaseModel):
     """
      Main configuration model.
@@ -519,6 +551,10 @@ class ImpulseConfig(BaseModel):
      calculated_channels : CalculatedChannels, optional
          Optional calculated-channel output configuration (e.g. opting in to the
          ``calculated_channel_metrics`` table). Defaults to CalculatedChannels().
+     full_recalculation : FullRecalculation, optional
+         Optional scoped full-recalculation request naming aggregations / events /
+         calculated channels to fully recompute on the next incremental run
+         regardless of definition-hash changes. Defaults to None (no override).
      measurement_dimensions : list of str, optional
          Column names to surface from ``container_metrics`` into the
          gold-layer ``measurement_dimension`` table. Names are matched
@@ -594,6 +630,7 @@ class ImpulseConfig(BaseModel):
     query_engine: QueryEngine = QueryEngine(solver=Solvers.DEFAULT_SOLVER)
     incremental: IncrementalConfig | None = None
     calculated_channels: CalculatedChannels = CalculatedChannels()
+    full_recalculation: FullRecalculation | None = None
 
     measurement_dimensions: list[str] = list(DEFAULT_MEASUREMENT_DIMENSIONS)
 

src/impulse_reporting/core/report.py

@@ -28,6 +28,7 @@
     dispatch_calculated_channel_metrics,
     dispatch_calculated_channels,
     dispatch_events,
+    full_recalc_names,
     group_selectables_by_type,
     merge_changed_unchanged,
     persist_channel_metrics,
@@ -38,6 +39,7 @@
     solve_calculated_channels_batched,
     solve_expressions_batched,
     split_by_hash_change,
+    validate_full_recalculation_scope,
 )
 from impulse_reporting.events.container_event import ContainerEvent
 from impulse_reporting.events.event import Event
@@ -549,6 +551,28 @@ def _validate_aggregation_events(self) -> None:
             )
             raise ValueError(error_message)
 
+    def _validate_full_recalculation_scope(self) -> None:
+        """Reject full-recalculation names that match no registered entity.
+
+        Thin wrapper delegating to :func:`report_utils.validate_full_recalculation_scope`;
+        supplies the report's registered entity names. Fails fast (mirroring
+        :meth:`_validate_aggregation_events`) so typos or stale names surface
+        immediately rather than silently recomputing nothing.
+
+        Raises
+        ------
+        ValueError
+            If any configured name does not match a registered entity of its kind.
+        """
+        validate_full_recalculation_scope(
+            self.config,
+            {
+                "events": self.get_events(),
+                "aggregations": [agg for page in self.pages for agg in page.aggregations],
+                "calculated_channels": self.get_calculated_channels(),
+            },
+        )
+
     @telemetry_logger("report", "persist_results")
     def persist_results(self, cleanup_temp_tables: bool | None = None):
         """
@@ -937,6 +961,9 @@ def determine_report(self, is_incremental: bool = None):
         # Validate that every aggregation references a registered event
         self._validate_aggregation_events()
 
+        # Validate that any scoped full-recalculation names match registered entities
+        self._validate_full_recalculation_scope()
+
         # Pin one consistent Delta snapshot of every configured silver input for
         # the whole run so mid-run table changes cannot leak across lazy stages
         # (issue #87).
@@ -982,7 +1009,13 @@ def determine_report(self, is_incremental: bool = None):
         # Split changed/unchanged definitions
         changed_events_by_type, unchanged_events_by_type, self._changed_event_ids = (
             split_by_hash_change(
-                events_by_type, EventType, self.sink, self.spark, hash_comparator, kind="event"
+                events_by_type,
+                EventType,
+                self.sink,
+                self.spark,
+                hash_comparator,
+                kind="event",
+                force_recalc_names=full_recalc_names(self.config, "events"),
             )
         )
         changed_aggs_by_type, unchanged_aggs_by_type, self._changed_aggregation_ids = (
@@ -993,6 +1026,7 @@ def determine_report(self, is_incremental: bool = None):
                 self.spark,
                 hash_comparator,
                 kind="aggregation",
+                force_recalc_names=full_recalc_names(self.config, "aggregations"),
             )
         )
 
@@ -1073,6 +1107,7 @@ def determine_report(self, is_incremental: bool = None):
                 self.spark,
                 hash_comparator,
                 kind="channel",
+                force_recalc_names=full_recalc_names(self.config, "calculated_channels"),
             )
         )
         # Collect the query-engine channel expressions across types for the batched