3.2.10. Invalidation agent
Understanding the Invalidation Process
To effectively invalidate temporal events, the agent performs checks in both directions:
- Incoming vs. Existing: Are incoming events invalidated by events already present?
- Existing vs. Incoming: Are current events invalidated by the new incoming events?
This bi-directional assessment results in a clear True/False decision.
Event Invalidation Prompt
The prompt has three key components:
<ol style="margin-left: 1em; line-height: 1.6; padding-left: 0.5em;"> <li style="margin-bottom: 1.2em;"><strong>Task Setup</strong><br /> Defines two roles—<code>primary</code> and <code>secondary</code>—for event comparison. The assessment checks if the <code>primary</code> event is invalidated by the <code>secondary</code> event.</li>Provides clear criteria on interpreting temporal metadata. Importantly, invalidation must rely solely on the relationships explicitly stated between events. External information cannot influence the decision.
python
event_invalidation_prompt = """
Task: Analyze the primary event against the secondary event and determine if the primary event is invalidated by the secondary event.
Only set dates if they explicitly relate to the validity of the relationship described in the text.
IMPORTANT: Only invalidate events if they are directly invalidated by the other event given in the context. Do NOT use any external knowledge to determine validity ranges.
Only use dates that are directly stated to invalidate the relationship. The invalid_at for the invalidated event should be the valid_at of the event that caused the invalidation.
Invalidation Guidelines:
1. Dates are given in ISO 8601 format (YYYY-MM-DDTHH:MM:SS.SSSSSSZ).
2. Where invalid_at is null, it means this event is still valid and considered to be ongoing.
3. Where invalid_at is defined, the event has previously been invalidated by something else and can be considered "finished".
4. An event can refine the invalid_at of a finished event to an earlier date only.
5. An event cannot invalidate an event that chronologically occurred after it.
6. An event cannot be invalidated by an event that chronologically occurred before it.
7. An event cannot invalidate itself.
---
Primary Event:
{% if primary_event -%}
Statement: {{primary_event}}
{%- endif %}
{% if primary_triplet -%}
Triplet: {{primary_triplet}}
{%- endif %}
Valid_at: {{primary_event.valid_at}}
Invalid_at: {{primary_event.invalid_at}}
---
Secondary Event:
{% if secondary_event -%}
Statement: {{secondary_event}}
{%- endif %}
{% if secondary_triplet -%}
Triplet: {{secondary_triplet}}
{%- endif %}
Valid_at: {{secondary_event.valid_at}}
Invalid_at: {{secondary_event.invalid_at}}
---
Return: "True" if the primary event is invalidated or its invalid_at is refined else "False"
"""Requirements to be compared for Invalidation
We can only invalidate dynamic facts that haven't been marked invalid yet. These facts serve as our primary events, while potential candidates for invalidation are our secondary events. To streamline the invalidation process, consider these guidelines when evaluating secondary events:
- Must be a FACT type and not Atemporal
- Share at least one canonical entity at the triplet level
- Belong to the same semantic predicate group at the triplet level (defined below)
- Temporally overlap and be currently ongoing
- Have a statement cosine similarity above the threshold (currently set to 0.5)
- The similarity threshold (0.5) helps us filter noise effectively by selecting only the
top_kmost relevant results. Low-level semantic similarities are acceptable since our goal is refining the data sent to the LLM for further assessment
When invalidation occurs, we annotate the affected events with expired_at and invalidated_by to clearly indicate cause-and-effect relationships.
python
PREDICATE_GROUPS: list[list[str]] = [
["IS_A", "HAS_A", "LOCATED_IN", "HOLDS_ROLE", "PART_OF"],
["PRODUCES", "SELLS", "SUPPLIES", "DISCONTINUED", "SECURED"],
["LAUNCHED", "DEVELOPED", "ADOPTED_BY", "INVESTS_IN", "COLLABORATES_WITH"],
["HAS_REVENUE", "INCREASED", "DECREASED", "RESULTED_IN", "TARGETS"],
]When we put all of this together, the workflow for our InvalidationAgent looks like this:
We use select_events_temporally() to filter events by:
- Checking if they're static or dynamic.
- Determining if their time ranges overlap with our incoming event.
- Handling dynamic events carefully, especially "ongoing" ones without an
invalid_at, or events with various overlaps.
Then, filter_by_embedding_similarity() compares events based on semantic similarity:
- It calculates cosine similarity between embeddings.
- Events below a similarity threshold (
_similarity_threshold = 0.5) are filtered out. - We keep only the top-K most similar events (
_top_k = 10).
With select_temporally_relevant_events_for_invalidation(), we:
- Apply temporal filters first.
- Then apply embedding similarity filters.
- This gives us a refined list of events most likely interacting or conflicting with the incoming one.
The LLM-based invalidation_step() (powered by GPT-4.1-mini) determines whether the incoming event invalidates another event:
- If it does, we update:
invalid_atto match the secondary event'svalid_at.expired_atwith the current timestamp.invalidated_bywith the ID of the secondary event.
We use bi_directional_event_invalidation() to check:
- If the incoming event invalidates existing events.
- If existing, later events invalidate the incoming event, especially if the incoming one is dynamic and currently valid.
The invalidation below represents this part of our pipeline:
python
import asyncio
import logging
import pickle
import sqlite3
from collections import Counter, defaultdict
from collections.abc import Coroutine
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime
from typing import Any
from jinja2 import DictLoader, Environment
from openai import AsyncOpenAI
from scipy.spatial.distance import cosine
from tenacity import retry, stop_after_attempt, wait_random_exponential
class InvalidationAgent:
"""Handles temporal-based operations for extracting and processing temporal events from text."""
def __init__(self, max_workers: int = 5) -> None:
"""Initialize the TemporalAgent with a client."""
self.max_workers = max_workers
self._executor = ThreadPoolExecutor(max_workers=max_workers)
self.logger = logging.getLogger(__name__)
self._client = AsyncOpenAI()
self._model = "gpt-4.1-mini"
self._similarity_threshold = 0.5
self._top_k = 10
self._env = Environment(loader=DictLoader({
"event_invalidation.jinja": event_invalidation_prompt,
}))
@staticmethod
def cosine_similarity(v1: list[float], v2: list[float]) -> float:
"""Calculate cosine similarity between two vectors."""
return float(1 - cosine(v1, v2))
@staticmethod
def get_incoming_temporal_bounds(
event: TemporalEvent,
) -> dict[str, datetime] | None:
"""Get temporal bounds of all temporal events associated with a statement."""
if (event.temporal_type == TemporalType.ATEMPORAL) or (event.valid_at is None):
return None
temporal_bounds = {"start": event.valid_at, "end": event.valid_at}
if event.temporal_type == TemporalType.DYNAMIC:
if event.invalid_at:
temporal_bounds["end"] = event.invalid_at
return temporal_bounds
def select_events_temporally(
self,
triplet_events: list[tuple[Triplet, TemporalEvent]],
temp_bounds: dict[str, datetime],
dynamic: bool = False,
) -> list[tuple[Triplet, TemporalEvent]]:
"""Select temporally relevant events (static or dynamic) based on temporal bounds.
Groups events into before, after, and overlapping categories based on their temporal bounds.
Args:
triplet_events: List of (Triplet, TemporalEvent) tuples to filter
temp_bounds: Dict with 'start' and 'end' datetime bounds
dynamic: If True, filter dynamic events; if False, filter static events
n_window: Number of events to include before and after bounds
Returns:
Dict with keys '{type}_before', '{type}_after', '{type}_overlap' where type is 'dynamic' or 'static'
"""
def _check_overlaps_dynamic(event: TemporalEvent, start: datetime, end: datetime) -> bool:
"""Check if the dynamic event overlaps with the temporal bounds of the incoming event."""
if event.temporal_type != TemporalType.DYNAMIC:
return False
event_start = event.valid_at or datetime.min
event_end = event.invalid_at
# 1. Event contains the start
if (event_end is not None) and (event_start <= start <= event_end):
return True
# 2. Ongoing event starts before the incoming start
if (event_end is None) and (event_start <= start):
return True
# 3. Event starts within the incoming interval
if start <= event_start <= end:
return True
return False
# Filter by temporal type
target_type = TemporalType.DYNAMIC if dynamic else TemporalType.STATIC
filtered_events = [(triplet, event) for triplet, event in triplet_events if event.temporal_type == target_type]
# Sort by valid_at timestamp
sorted_events = sorted(filtered_events, key=lambda te: te[1].valid_at or datetime.min)
start = temp_bounds["start"]
end = temp_bounds["end"]
if dynamic:
overlap: list[tuple[Triplet, TemporalEvent]] = [
(triplet, event) for triplet, event in sorted_events if _check_overlaps_dynamic(event, start, end)
]
else:
overlap = []
if start != end:
overlap = [(triplet, event) for triplet, event in sorted_events if event.valid_at and start <= event.valid_at <= end]
return overlap
def filter_by_embedding_similarity(
self,
reference_event: TemporalEvent,
candidate_pairs: list[tuple[Triplet, TemporalEvent]],
) -> list[tuple[Triplet, TemporalEvent]]:
"""Filter triplet-event pairs by embedding similarity."""
pairs_with_similarity = [
(triplet, event, self.cosine_similarity(reference_event.embedding, event.embedding)) for triplet, event in candidate_pairs
]
filtered_pairs = [
(triplet, event) for triplet, event, similarity in pairs_with_similarity if similarity >= self._similarity_threshold
]
sorted_pairs = sorted(filtered_pairs, key=lambda x: self.cosine_similarity(reference_event.embedding, x[1].embedding), reverse=True)
return sorted_pairs[: self._top_k]
def select_temporally_relevant_events_for_invalidation(
self,
incoming_event: TemporalEvent,
candidate_triplet_events: list[tuple[Triplet, TemporalEvent]],
) -> list[tuple[Triplet, TemporalEvent]] | None:
"""Select the temporally relevant events based on temporal range of incoming event."""
temporal_bounds = self.get_incoming_temporal_bounds(event=incoming_event)
if not temporal_bounds:
return None
# First apply temporal filtering - find overlapping events
selected_statics = self.select_events_temporally(
triplet_events=candidate_triplet_events,
temp_bounds=temporal_bounds,
)
selected_dynamics = self.select_events_temporally(
triplet_events=candidate_triplet_events,
temp_bounds=temporal_bounds,
dynamic=True,
)
# Then filter by semantic similarity
similar_static = self.filter_by_embedding_similarity(reference_event=incoming_event, candidate_pairs=selected_statics)
similar_dynamics = self.filter_by_embedding_similarity(reference_event=incoming_event, candidate_pairs=selected_dynamics)
return similar_static + similar_dynamics
@retry(wait=wait_random_exponential(multiplier=1, min=1, max=30), stop=stop_after_attempt(3))
async def invalidation_step(
self,
primary_event: TemporalEvent,
primary_triplet: Triplet,
secondary_event: TemporalEvent,
secondary_triplet: Triplet,
) -> TemporalEvent:
"""Check if primary event should be invalidated by secondary event.
Args:
primary_event: Event to potentially invalidate
primary_triplet: Triplet associated with primary event
secondary_event: Event that might cause invalidation
secondary_triplet: Triplet associated with secondary event
Returns:
TemporalEvent: Updated primary event (may have invalid_at and invalidated_by set)
"""
template = self._env.get_template("event_invalidation.jinja")
prompt = template.render(
primary_event=primary_event.statement,
primary_triplet=f"({primary_triplet.subject_name}, {primary_triplet.predicate}, {primary_triplet.object_name})",
primary_valid_at=primary_event.valid_at,
primary_invalid_at=primary_event.invalid_at,
secondary_event=secondary_event.statement,
secondary_triplet=f"({secondary_triplet.subject_name}, {secondary_triplet.predicate}, {secondary_triplet.object_name})",
secondary_valid_at=secondary_event.valid_at,
secondary_invalid_at=secondary_event.invalid_at,
)
response = await self._client.responses.parse(
model=self._model,
temperature=0,
input=prompt,
)
# Parse boolean response
response_bool = str(response).strip().lower() == "true" if response else False
if not response_bool:
return primary_event
# Create updated event with invalidation info
updated_event = primary_event.model_copy(
update={
"invalid_at": secondary_event.valid_at,
"expired_at": datetime.now(),
"invalidated_by": secondary_event.id,
}
)
return updated_event
async def bi_directional_event_invalidation(
self,
incoming_triplet: Triplet,
incoming_event: TemporalEvent,
existing_triplet_events: list[tuple[Triplet, TemporalEvent]],
) -> tuple[TemporalEvent, list[TemporalEvent]]:
"""Validate and update temporal information for triplet events with full bidirectional invalidation.
Args:
incoming_triplet: The new triplet
incoming_event: The new event associated with the triplet
existing_triplet_events: List of existing (triplet, event) pairs to validate against
Returns:
tuple[TemporalEvent, list[TemporalEvent]]: (updated_incoming_event, list_of_changed_existing_events)
"""
changed_existing_events: list[TemporalEvent] = []
updated_incoming_event = incoming_event
# Filter for dynamic events that can be invalidated
dynamic_events_to_check = [
(triplet, event) for triplet, event in existing_triplet_events if event.temporal_type == TemporalType.DYNAMIC
]
# 1. Check if incoming event invalidates existing dynamic events
if dynamic_events_to_check:
tasks = [
self.invalidation_step(
primary_event=existing_event,
primary_triplet=existing_triplet,
secondary_event=incoming_event,
secondary_triplet=incoming_triplet,
)
for existing_triplet, existing_event in dynamic_events_to_check
]
updated_events = await asyncio.gather(*tasks)
for original_pair, updated_event in zip(dynamic_events_to_check, updated_events, strict=True):
original_event = original_pair[1]
if (updated_event.invalid_at != original_event.invalid_at) or (
updated_event.invalidated_by != original_event.invalidated_by
):
changed_existing_events.append(updated_event)
# 2. Check if existing events invalidate the incoming dynamic event
if incoming_event.temporal_type == TemporalType.DYNAMIC and incoming_event.invalid_at is None:
# Only check events that occur after the incoming event
invalidating_events = [
(triplet, event)
for triplet, event in existing_triplet_events
if (incoming_event.valid_at and event.valid_at and incoming_event.valid_at < event.valid_at)
]
if invalidating_events:
tasks = [
self.invalidation_step(
primary_event=incoming_event,
primary_triplet=incoming_triplet,
secondary_event=existing_event,
secondary_triplet=existing_triplet,
)
for existing_triplet, existing_event in invalidating_events
]
updated_events = await asyncio.gather(*tasks)
# Find the earliest invalidation
valid_invalidations = [(e.invalid_at, e.invalidated_by) for e in updated_events if e.invalid_at is not None]
if valid_invalidations:
earliest_invalidation = min(valid_invalidations, key=lambda x: x[0])
updated_incoming_event = incoming_event.model_copy(
update={
"invalid_at": earliest_invalidation[0],
"invalidated_by": earliest_invalidation[1],
"expired_at": datetime.now(),
}
)
return updated_incoming_event, changed_existing_events
@staticmethod
def resolve_duplicate_invalidations(changed_events: list[TemporalEvent]) -> list[TemporalEvent]:
"""Resolve duplicate invalidations by selecting the most restrictive (earliest) invalidation.
When multiple incoming events invalidate the same existing event, we should apply
the invalidation that results in the shortest validity range (earliest invalid_at).
Args:
changed_events: List of events that may contain duplicates with different invalidations
Returns:
List of deduplicated events with the most restrictive invalidation applied
"""
if not changed_events:
return []
# Count occurrences of each event ID
id_counts = Counter(str(event.id) for event in changed_events)
resolved_events = []
# Group events by ID only for those with duplicates
events_by_id = defaultdict(list)
for event in changed_events:
event_id = str(event.id)
if id_counts[event_id] == 1:
resolved_events.append(event)
else:
events_by_id[event_id].append(event)
# Deduplicate only those with duplicates
for _id, event_versions in events_by_id.items():
invalidated_versions = [e for e in event_versions if e.invalid_at is not None]
if not invalidated_versions:
resolved_events.append(event_versions[0])
else:
most_restrictive = min(invalidated_versions, key=lambda e: (e.invalid_at if e.invalid_at is not None else datetime.max))
resolved_events.append(most_restrictive)
return resolved_events
async def _execute_task_pool(
self,
tasks: list[Coroutine[Any, Any, tuple[TemporalEvent, list[TemporalEvent]]]],
batch_size: int = 10
) -> list[Any]:
"""Execute tasks in batches using a pool to control concurrency.
Args:
tasks: List of coroutines to execute
batch_size: Number of tasks to process concurrently
Returns:
List of results from all tasks
"""
all_results = []
for i in range(0, len(tasks), batch_size):
batch = tasks[i:i + batch_size]
batch_results = await asyncio.gather(*batch, return_exceptions=True)
all_results.extend(batch_results)
# Small delay between batches to prevent overload
if i + batch_size < len(tasks):
await asyncio.sleep(0.1)
return all_results
async def process_invalidations_in_parallel(
self,
incoming_triplets: list[Triplet],
incoming_events: list[TemporalEvent],
existing_triplets: list[Triplet],
existing_events: list[TemporalEvent],
) -> tuple[list[TemporalEvent], list[TemporalEvent]]:
"""Process invalidations for multiple triplets in parallel.
Args:
incoming_triplets: List of new triplets to process
incoming_events: List of events associated with incoming triplets
existing_triplets: List of existing triplets from DB
existing_events: List of existing events from DB
Returns:
tuple[list[TemporalEvent], list[TemporalEvent]]:
- List of updated incoming events (potentially invalidated)
- List of existing events that were updated (deduplicated)
"""
# Create mappings for faster lookups
event_map = {str(e.id): e for e in existing_events}
incoming_event_map = {str(t.event_id): e for t, e in zip(incoming_triplets, incoming_events, strict=False)}
# Prepare tasks for parallel processing
tasks = []
for incoming_triplet in incoming_triplets:
incoming_event = incoming_event_map[str(incoming_triplet.event_id)]
# Get related triplet-event pairs
related_pairs = [
(t, event_map[str(t.event_id)])
for t in existing_triplets
if (str(t.subject_id) == str(incoming_triplet.subject_id) or str(t.object_id) == str(incoming_triplet.object_id))
and str(t.event_id) in event_map
]
# Filter for temporal relevance
all_relevant_events = self.select_temporally_relevant_events_for_invalidation(
incoming_event=incoming_event,
candidate_triplet_events=related_pairs,
)
if not all_relevant_events:
continue
# Add task for parallel processing
task = self.bi_directional_event_invalidation(
incoming_triplet=incoming_triplet,
incoming_event=incoming_event,
existing_triplet_events=all_relevant_events,
)
tasks.append(task)
# Process all invalidations in parallel with pooling
if not tasks:
return [], []
# Use pool size based on number of workers, but cap it
pool_size = min(self.max_workers * 2, 10) # Adjust these numbers based on your needs
results = await self._execute_task_pool(tasks, batch_size=pool_size)
# Collect all results (may contain duplicates)
updated_incoming_events = []
all_changed_existing_events = []
for result in results:
if isinstance(result, Exception):
self.logger.error(f"Task failed with error: {str(result)}")
continue
updated_event, changed_events = result
updated_incoming_events.append(updated_event)
all_changed_existing_events.extend(changed_events)
# Resolve duplicate invalidations for existing events
deduplicated_existing_events = self.resolve_duplicate_invalidations(all_changed_existing_events)
# Resolve duplicate invalidations for incoming events (in case multiple triplets from same event)
deduplicated_incoming_events = self.resolve_duplicate_invalidations(updated_incoming_events)
return deduplicated_incoming_events, deduplicated_existing_events
@staticmethod
def batch_fetch_related_triplet_events(
conn: sqlite3.Connection,
incoming_triplets: list[Triplet],
) -> tuple[list[Triplet], list[TemporalEvent]]:
"""
Batch fetch all existing triplets and their events from the DB that are related to any of the incoming triplets.
Related means:
- Share a subject or object entity
- Predicate is in the same group
- Associated event is a FACT
Returns two lists: triplets and events (with mapping via event_id).
"""
# 1. Build sets of all relevant entity IDs and predicate groups
entity_ids = set()
predicate_to_group = {}
for group in PREDICATE_GROUPS:
group_list = list(group)
for pred in group_list:
predicate_to_group[pred] = group_list
relevant_predicates = set()
for triplet in incoming_triplets:
entity_ids.add(str(triplet.subject_id))
entity_ids.add(str(triplet.object_id))
group = predicate_to_group.get(str(triplet.predicate), [])
if group:
relevant_predicates.update(group)
# 2. Prepare SQL query
entity_placeholders = ",".join(["?"] * len(entity_ids))
predicate_placeholders = ",".join(["?"] * len(relevant_predicates))
query = f"""
SELECT
t.id,
t.subject_name,
t.subject_id,
t.predicate,
t.object_name,
t.object_id,
t.value,
t.event_id,
e.chunk_id,
e.statement,
e.triplets,
e.statement_type,
e.temporal_type,
e.valid_at,
e.invalid_at,
e.created_at,
e.expired_at,
e.invalidated_by,
e.embedding
FROM triplets t
JOIN events e ON t.event_id = e.id
WHERE
(t.subject_id IN ({entity_placeholders}) OR t.object_id IN ({entity_placeholders}))
AND t.predicate IN ({predicate_placeholders})
AND e.statement_type = ?
"""
params = list(entity_ids) + list(entity_ids) + list(relevant_predicates) + [StatementType.FACT]
cursor = conn.cursor()
cursor.execute(query, params)
rows = cursor.fetchall()
triplets = []
events = []
events_by_id = {}
for row in rows:
triplet = Triplet(
id=row[0],
subject_name=row[1],
subject_id=row[2],
predicate=Predicate(row[3]),
object_name=row[4],
object_id=row[5],
value=row[6],
event_id=row[7],
)
event_id = row[7]
triplets.append(triplet)
if event_id not in events_by_id:
events_by_id[event_id] = TemporalEvent(
id=row[7],
chunk_id=row[8],
statement=row[9],
triplets=TemporalEvent.parse_triplets_json(row[10]),
statement_type=row[11],
temporal_type=row[12],
valid_at=row[13],
invalid_at=row[14],
created_at=row[15],
expired_at=row[16],
invalidated_by=row[17],
embedding=pickle.loads(row[18]) if row[18] else [0] * 1536,
)
events = list(events_by_id.values())
return triplets, eventsWe can create a batch processing function for invalidation for a set of Temporal Events. This is where we filter our Statements to type FACT before passing into the invalidation agent to process.
python
async def batch_process_invalidation(
conn: sqlite3.Connection, all_events: list[TemporalEvent], all_triplets: list[Triplet], invalidation_agent: InvalidationAgent
) -> tuple[list[TemporalEvent], list[TemporalEvent]]:
"""Process invalidation for all FACT events that are temporal.
Args:
conn: SQLite database connection
all_events: List of all extracted events
all_triplets: List of all extracted triplets
invalidation_agent: The invalidation agent instance
Returns:
tuple[list[TemporalEvent], list[TemporalEvent]]:
- final_events: All events (updated incoming events)
- events_to_update: Existing events that need DB updates
"""
def _get_fact_triplets(
all_events: list[TemporalEvent],
all_triplets: list[Triplet],
) -> list[Triplet]:
"""
Return only those triplets whose associated event is of statement_type FACT.
"""
fact_event_ids = {
event.id for event in all_events if (event.statement_type == StatementType.FACT) and (event.temporal_type != TemporalType.ATEMPORAL)
}
return [triplet for triplet in all_triplets if triplet.event_id in fact_event_ids]
# Prepare a list of triplets whose associated event is a FACT and not ATEMPORAL
fact_triplets = _get_fact_triplets(all_events, all_triplets)
if not fact_triplets:
return all_events, []
# Create event map for quick lookup
all_events_map = {event.id: event for event in all_events}
# Build aligned lists of valid triplets and their corresponding events
fact_events: list[TemporalEvent] = []
valid_fact_triplets: list[Triplet] = []
for triplet in fact_triplets:
# Handle potential None event_id and ensure type safety
if triplet.event_id is not None:
event = all_events_map.get(triplet.event_id)
if event:
fact_events.append(event)
valid_fact_triplets.append(triplet)
else:
print(f"Warning: Could not find event for fact_triplet with event_id {triplet.event_id}")
else:
print(f"Warning: Fact triplet {triplet.id} has no event_id, skipping invalidation")
if not valid_fact_triplets:
return all_events, []
# Batch fetch all related existing triplets and events
existing_triplets, existing_events = invalidation_agent.batch_fetch_related_triplet_events(conn, valid_fact_triplets)
# Process all invalidations in parallel
updated_incoming_fact_events, changed_existing_events = await invalidation_agent.process_invalidations_in_parallel(
incoming_triplets=valid_fact_triplets,
incoming_events=fact_events,
existing_triplets=existing_triplets,
existing_events=existing_events,
)
# Create mapping for efficient updates
updated_incoming_event_map = {event.id: event for event in updated_incoming_fact_events}
# Reconstruct final events list with updates applied
final_events = []
for original_event in all_events:
if original_event.id in updated_incoming_event_map:
final_events.append(updated_incoming_event_map[original_event.id])
else:
final_events.append(original_event)
return final_events, changed_existing_events