Update journey last_modified on node, edge, and link mutations

Adding, updating, or deleting nodes, edges, or links changes the
journey's structure. These mutations now update last_modified on the
parent journey document via _touch_journey_last_modified helper.
Also covers set/unset journey metadata and set_node_properties.

Signed-off-by: Dor Zohar <dor@emcie.co>

Author: mc-dorzo

src/parlant/core/journeys.py

@@ -1069,6 +1069,13 @@ def _deserialize_link(self, doc: JourneyLinkAssociationDocument) -> JourneyLink:
             merge_node_id=JourneyNodeId(doc["merge_node_id"]),
         )
 
+    async def _touch_journey_last_modified(self, journey_id: JourneyId) -> None:
+        """Update the journey's last_modified timestamp."""
+        await self._collection.update_one(
+            filters={"id": {"$eq": journey_id}},
+            params={"last_modified": datetime.now(timezone.utc).isoformat()},
+        )
+
     @staticmethod
     def assemble_content(
         title: str,
@@ -1519,6 +1526,8 @@ async def create_node(
                 document=self._serialize_node(node, journey_id)
             )
 
+            await self._touch_journey_last_modified(journey_id)
+
         return node
 
     @override
@@ -1566,6 +1575,8 @@ async def update_node(
                 params=cast(JourneyNodeAssociationDocument, to_json_dict(updated)),
             )
 
+            await self._touch_journey_last_modified(doc["journey_id"])
+
         assert result.updated_document
 
         return self._deserialize_node(result.updated_document)
@@ -1583,7 +1594,9 @@ async def delete_node(
             if not node_doc:
                 raise ItemNotFoundError(item_id=UniqueId(node_id))
 
-            edges = await self.list_edges(journey_id=node_doc["journey_id"], node_id=node_id)
+            journey_id = node_doc["journey_id"]
+
+            edges = await self.list_edges(journey_id=journey_id, node_id=node_id)
 
             for edge in edges:
                 await self.delete_edge(edge.id)
@@ -1592,6 +1605,8 @@ async def delete_node(
                 filters={"node_id": {"$eq": node_id}}
             )
 
+            await self._touch_journey_last_modified(journey_id)
+
         if result.deleted_count == 0:
             raise ItemNotFoundError(item_id=UniqueId(node_id))
 
@@ -1690,7 +1705,10 @@ async def set_journey_metadata(
 
             result = await self._collection.update_one(
                 filters={"id": {"$eq": journey_id}},
-                params={"metadata": updated_metadata},
+                params={
+                    "metadata": updated_metadata,
+                    "last_modified": datetime.now(timezone.utc).isoformat(),
+                },
             )
 
         assert result.updated_document
@@ -1713,7 +1731,10 @@ async def unset_journey_metadata(
 
             result = await self._collection.update_one(
                 filters={"id": {"$eq": journey_id}},
-                params={"metadata": updated_metadata},
+                params={
+                    "metadata": updated_metadata,
+                    "last_modified": datetime.now(timezone.utc).isoformat(),
+                },
             )
 
         assert result.updated_document
@@ -1734,7 +1755,10 @@ async def set_node_properties(
 
             result = await self._collection.update_one(
                 filters={"id": {"$eq": journey_id}},
-                params={"node_properties": node_properties},
+                params={
+                    "node_properties": node_properties,
+                    "last_modified": datetime.now(timezone.utc).isoformat(),
+                },
             )
 
         assert result.updated_document
@@ -1765,6 +1789,8 @@ async def create_edge(
                 document=self._serialize_edge(edge, journey_id)
             )
 
+            await self._touch_journey_last_modified(journey_id)
+
         return edge
 
     @override
@@ -1799,6 +1825,8 @@ async def update_edge(
                 params=cast(JourneyEdgeAssociationDocument, to_json_dict(updated)),
             )
 
+            await self._touch_journey_last_modified(doc["journey_id"])
+
         assert result.updated_document
 
         return self._deserialize_edge(result.updated_document)
@@ -1836,10 +1864,15 @@ async def delete_edge(
         edge_id: JourneyEdgeId,
     ) -> None:
         async with self._lock.writer_lock:
+            edge_doc = await self._edge_association_collection.find_one({"id": {"$eq": edge_id}})
+
             result = await self._edge_association_collection.delete_one(
                 filters={"id": {"$eq": edge_id}}
             )
 
+            if result.deleted_count > 0 and edge_doc:
+                await self._touch_journey_last_modified(edge_doc["journey_id"])
+
         if result.deleted_count == 0:
             raise ItemNotFoundError(item_id=UniqueId(edge_id))
 
@@ -2033,6 +2066,7 @@ async def create_link(
 
         async with self._lock.writer_lock:
             await self._link_association_collection.insert_one(document=self._serialize_link(link))
+            await self._touch_journey_last_modified(journey_id)
 
         return link
 
@@ -2074,7 +2108,7 @@ async def delete_link(
 
         link = self._deserialize_link(link_doc)
 
-        # Delete the merge fork node
+        # Delete the merge fork node (also touches last_modified via delete_node)
         await self.delete_node(link.merge_node_id)
 
         async with self._lock.writer_lock: