[Core] Fix itemsToParse not applied during webhook live event processing (#2868)

# Description

What - Fixed `itemsToParse` expansion being silently skipped during
webhook/live event processing, causing entity mapping failures for any
integration using `itemsToParse` (e.g. GitLab-v2 `file` kind with YAML
arrays).


Why - The live events path in `_parse_raw_event_results_to_entities`
passed raw items directly to `entity_processor.parse_items()` without
expansion, and `handle_items_to_parse` had an implicit dependency on the
`event` context which is only populated during resync — making it
impossible to safely reuse in the live events path.


How - Refactored `handle_items_to_parse` to accept
`items_to_parse_top_level_transform` as an explicit parameter (removing
the event context dependency), then added a `_expand_raw_item` async
generator to `LiveEventsMixin` that applies the expansion when
`itemsToParse` is configured before passing items to
`entity_processor.parse_items()`.


## Type of change

Please leave one option from the following and delete the rest:

- [ ] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] New Integration (non-breaking change which adds a new integration)
- [ ] Breaking change (fix or feature that would cause existing
functionality to not work as expected)
- [ ] Non-breaking change (fix of existing functionality that will not
change current behavior)
- [ ] Documentation (added/updated documentation)

<h4> All tests should be run against the port production
environment(using a testing org). </h4>

### Core testing checklist

- [ ] Integration able to create all default resources from scratch
- [ ] Resync finishes successfully
- [ ] Resync able to create entities
- [ ] Resync able to update entities
- [ ] Resync able to detect and delete entities
- [ ] Scheduled resync able to abort existing resync and start a new one
- [ ] Tested with at least 2 integrations from scratch
- [ ] Tested with Kafka and Polling event listeners
- [ ] Tested deletion of entities that don't pass the selector


### Integration testing checklist

- [ ] Integration able to create all default resources from scratch
- [ ] Completed a full resync from a freshly installed integration and
it completed successfully
- [ ] Resync able to create entities
- [ ] Resync able to update entities
- [ ] Resync able to detect and delete entities
- [ ] Resync finishes successfully
- [ ] If new resource kind is added or updated in the integration, add
example raw data, mapping and expected result to the `examples` folder
in the integration directory.
- [ ] If resource kind is updated, run the integration with the example
data and check if the expected result is achieved
- [ ] If new resource kind is added or updated, validate that
live-events for that resource are working as expected
- [ ] Docs PR link [here](#)

### Preflight checklist

- [ ] Handled rate limiting
- [ ] Handled pagination
- [ ] Implemented the code in async
- [ ] Support Multi account

## Screenshots

Include screenshots from your environment showing how the resources of
the integration will look.

## API Documentation

Provide links to the API documentation used for this integration.

Author: oiadebayo

CHANGELOG.md

@@ -7,6 +7,13 @@ this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 <!-- towncrier release notes start -->
 
+## 0.38.11 (2026-03-11)
+
+### Improvements
+
+- Added support for items to parse in live events context
+
+
 ## 0.38.10 (2026-03-10)
 
 ### Improvements

port_ocean/core/integrations/mixins/live_events.py

@@ -1,9 +1,13 @@
+from typing import AsyncGenerator
+
 from loguru import logger
 from port_ocean.clients.port.types import UserAgentType
+from port_ocean.core.handlers.port_app_config.models import ResourceConfig
 from port_ocean.core.handlers.webhook.webhook_event import WebhookEventRawResults
 from port_ocean.core.integrations.mixins.handler import HandlerMixin
-from port_ocean.core.integrations.mixins.utils import is_lakehouse_data_enabled
+from port_ocean.core.integrations.mixins.utils import handle_items_to_parse, is_lakehouse_data_enabled
 from port_ocean.core.models import Entity, LakehouseOperation
+from port_ocean.core.ocean_types import RAW_ITEM
 from port_ocean.context.ocean import ocean
 
 
@@ -28,6 +32,20 @@ async def sync_raw_results(self, webhook_events_raw_result: list[WebhookEventRaw
             await self._delete_entities(entities_to_delete)
 
 
+    async def _expand_raw_item(
+        self, raw_item: RAW_ITEM, resource: ResourceConfig
+    ) -> AsyncGenerator[list[RAW_ITEM], None]:
+        if resource.port.items_to_parse:
+            async for batch in handle_items_to_parse(
+                [raw_item],
+                resource.port.items_to_parse_name,
+                resource.port.items_to_parse,
+                resource.port.items_to_parse_top_level_transform,
+            ):
+                yield batch
+        else:
+            yield [raw_item]
+
     async def _parse_raw_event_results_to_entities(self, webhook_events_raw_result: list[WebhookEventRawResults]) -> tuple[list[Entity], list[Entity]]:
         """Parse the webhook event raw results and return a list of entities.
 
@@ -38,18 +56,21 @@ async def _parse_raw_event_results_to_entities(self, webhook_events_raw_result:
         entities_not_passed: list[Entity] = []
         entities_to_delete: list[Entity] = []
         for webhook_event_raw_result in webhook_events_raw_result:
+            resource = webhook_event_raw_result.resource
             for raw_item in webhook_event_raw_result.updated_raw_results:
-                calaculation_results = await self.entity_processor.parse_items(
-                    webhook_event_raw_result.resource, [raw_item], parse_all=True, send_raw_data_examples_amount=0
-                )
-                entities.extend(calaculation_results.entity_selector_diff.passed)
-                entities_not_passed.extend(calaculation_results.entity_selector_diff.failed)
+                async for batch in self._expand_raw_item(raw_item, resource):
+                    calculation_results = await self.entity_processor.parse_items(
+                        resource, batch, parse_all=True, send_raw_data_examples_amount=0
+                    )
+                    entities.extend(calculation_results.entity_selector_diff.passed)
+                    entities_not_passed.extend(calculation_results.entity_selector_diff.failed)
 
             for raw_item in webhook_event_raw_result.deleted_raw_results:
-                deletion_results = await self.entity_processor.parse_items(
-                    webhook_event_raw_result.resource, [raw_item], parse_all=True, send_raw_data_examples_amount=0
-                )
-                entities_to_delete.extend(deletion_results.entity_selector_diff.passed)
+                async for batch in self._expand_raw_item(raw_item, resource):
+                    deletion_results = await self.entity_processor.parse_items(
+                        resource, batch, parse_all=True, send_raw_data_examples_amount=0
+                    )
+                    entities_to_delete.extend(deletion_results.entity_selector_diff.passed)
 
         entities_to_remove = []
         for entity in entities_to_delete + entities_not_passed:

port_ocean/core/integrations/mixins/sync_raw.py

@@ -127,6 +127,7 @@ async def _execute_resync_tasks(
                         resource_config.kind,
                         resource_config.port.items_to_parse_name,
                         resource_config.port.items_to_parse,
+                        resource_config.port.items_to_parse_top_level_transform,
                     )
                 )
             else:

port_ocean/core/integrations/mixins/utils.py

@@ -7,7 +7,6 @@
 from loguru import logger
 
 from port_ocean.clients.port.utils import _http_client as _port_http_client
-from port_ocean.context.event import event
 from port_ocean.context.ocean import ocean
 from port_ocean.core.handlers import JQEntityProcessor
 from port_ocean.core.ocean_types import (
@@ -133,7 +132,7 @@ async def resync_function_wrapper(
         results = await fn(kind)
         return validate_result(results)
 
-async def handle_items_to_parse(result: RAW_RESULT, items_to_parse_name: str, items_to_parse: str | None = None) -> AsyncGenerator[list[dict[str, Any]], None]:
+async def handle_items_to_parse(result: RAW_RESULT, items_to_parse_name: str, items_to_parse: str | None = None, items_to_parse_top_level_transform: bool = True) -> AsyncGenerator[list[dict[str, Any]], None]:
     delete_target = extract_jq_deletion_path_revised(items_to_parse) or '.'
     jq_expression = f". | del({delete_target})"
     batch_size = ocean.config.yield_items_to_parse_batch_size
@@ -142,8 +141,14 @@ async def handle_items_to_parse(result: RAW_RESULT, items_to_parse_name: str, it
 
     for item in result:
         items_to_parse_data = await entity_processor._search(item, items_to_parse)
-        if event.resource_config.port.items_to_parse_top_level_transform:
-            item = await entity_processor._search(item, jq_expression)
+        if items_to_parse_top_level_transform:
+            transformed = await entity_processor._search(item, jq_expression)
+            if transformed is None:
+                logger.warning(
+                    f"Top-level transform '{jq_expression}' returned None for item, skipping..."
+                )
+                continue
+            item = transformed
         if not isinstance(items_to_parse_data, list):
             logger.warning(
                 f"Failed to parse items for JQ expression {items_to_parse}, Expected list but got {type(items_to_parse_data)}."
@@ -162,7 +167,7 @@ async def handle_items_to_parse(result: RAW_RESULT, items_to_parse_name: str, it
             yield batch
 
 async def resync_generator_wrapper(
-    fn: Callable[[str], ASYNC_GENERATOR_RESYNC_TYPE], kind: str, items_to_parse_name: str, items_to_parse: str | None = None
+    fn: Callable[[str], ASYNC_GENERATOR_RESYNC_TYPE], kind: str, items_to_parse_name: str, items_to_parse: str | None = None, items_to_parse_top_level_transform: bool = True
 ) -> ASYNC_GENERATOR_RESYNC_TYPE:
     generator = fn(kind)
     errors = []
@@ -173,7 +178,8 @@ async def resync_generator_wrapper(
                     result = validate_result(await anext(generator))
 
                     if items_to_parse:
-                        items_to_parse_generator = handle_items_to_parse(result, items_to_parse_name, items_to_parse)
+                        items_to_parse_generator = handle_items_to_parse(result, items_to_parse_name, items_to_parse, items_to_parse_top_level_transform)
+                        del result
                         async for batch in items_to_parse_generator:
                             yield batch
                     else:

port_ocean/tests/core/handlers/mixins/test_live_events.py

@@ -424,6 +424,160 @@ async def test_sync_raw_results_one_raw_result_entity_upserted(
     mock_live_events_mixin.entities_state_applier.delete.assert_not_called()
 
 
+file_resource_config_with_items_to_parse = ResourceConfig(
+    kind="file",
+    selector=Selector(query="true"),
+    port=PortResourceConfig(
+        entity=MappingsConfig(
+            mappings=EntityMapping(
+                identifier=".item.identifier",
+                title=".item.title",
+                blueprint='"fileBlueprint"',
+                properties={"repoName": ".repo.name"},
+                relations={},
+            )
+        ),
+        itemsToParse=".file.content",
+    ),
+)
+
+file_raw_item_with_array = {
+    "file": {
+        "path": "port.yml",
+        "content": [
+            {"identifier": "svc-one", "title": "Service One"},
+            {"identifier": "svc-two", "title": "Service Two"},
+        ],
+    },
+    "repo": {"name": "my-repo"},
+}
+
+
+@pytest.mark.asyncio
+async def test_expand_raw_item_without_items_to_parse(
+    mock_live_events_mixin: LiveEventsMixin,
+) -> None:
+    """When itemsToParse is not configured, _expand_raw_item yields the original item unchanged"""
+    resource = ResourceConfig(
+        kind="repository",
+        selector=Selector(query="true"),
+        port=PortResourceConfig(
+            entity=MappingsConfig(
+                mappings=EntityMapping(
+                    identifier=".name",
+                    title=".name",
+                    blueprint='"service"',
+                    properties={},
+                    relations={},
+                )
+            )
+        ),
+    )
+    raw_item = {"name": "my-repo", "url": "https://example.com/my-repo"}
+
+    batches: list[Any] = []
+    async for batch in mock_live_events_mixin._expand_raw_item(raw_item, resource):
+        batches.extend(batch)
+
+    assert batches == [raw_item]
+
+
+@pytest.mark.asyncio
+async def test_expand_raw_item_with_items_to_parse(
+    mock_live_events_mixin: LiveEventsMixin,
+    mock_context: PortOceanContext,
+) -> None:
+    """When itemsToParse is configured, _expand_raw_item fans out array elements into separate items"""
+    mock_ocean_utils = MagicMock()
+    mock_ocean_utils.config.yield_items_to_parse_batch_size = 100
+    mock_ocean_utils.app.integration.entity_processor = JQEntityProcessor(mock_context)
+
+    with patch("port_ocean.core.integrations.mixins.utils.ocean", mock_ocean_utils):
+        batches: list[Any] = []
+        async for batch in mock_live_events_mixin._expand_raw_item(
+            file_raw_item_with_array, file_resource_config_with_items_to_parse
+        ):
+            batches.extend(batch)
+
+    assert len(batches) == 2
+    assert batches[0]["item"] == {"identifier": "svc-one", "title": "Service One"}
+    assert batches[1]["item"] == {"identifier": "svc-two", "title": "Service Two"}
+    assert "content" not in batches[0].get("file", {})
+    assert batches[0]["repo"] == {"name": "my-repo"}
+
+
+@pytest.mark.asyncio
+async def test_parse_raw_event_results_items_to_parse_expansion(
+    mock_live_events_mixin: LiveEventsMixin,
+) -> None:
+    """When itemsToParse is configured, _parse_raw_event_results_to_entities produces one entity per array element"""
+    expanded_item_one = {
+        "file": {"path": "port.yml"},
+        "repo": {"name": "my-repo"},
+        "item": {"identifier": "svc-one", "title": "Service One"},
+    }
+    expanded_item_two = {
+        "file": {"path": "port.yml"},
+        "repo": {"name": "my-repo"},
+        "item": {"identifier": "svc-two", "title": "Service Two"},
+    }
+
+    entity_one = Entity(
+        identifier="svc-one",
+        blueprint="fileBlueprint",
+        title="Service One",
+        team=[],
+        properties={"repoName": "my-repo"},
+        relations={},
+    )
+    entity_two = Entity(
+        identifier="svc-two",
+        blueprint="fileBlueprint",
+        title="Service Two",
+        team=[],
+        properties={"repoName": "my-repo"},
+        relations={},
+    )
+
+    async def mock_expand(raw_item: Any, resource: Any) -> Any:
+        yield [expanded_item_one]
+        yield [expanded_item_two]
+
+    mock_live_events_mixin._expand_raw_item = mock_expand  # type: ignore
+
+    def make_calculation_result(passed: list[Entity]) -> CalculationResult:
+        return CalculationResult(
+            entity_selector_diff=EntitySelectorDiff(passed=passed, failed=[]),
+            errors=[],
+            misconfigured_entity_keys={},
+        )
+
+    mock_live_events_mixin.entity_processor.parse_items = AsyncMock(  # type: ignore
+        side_effect=[
+            make_calculation_result([entity_one]),
+            make_calculation_result([entity_two]),
+        ]
+    )
+
+    webhook_result = WebhookEventRawResults(
+        updated_raw_results=[file_raw_item_with_array],
+        deleted_raw_results=[],
+    )
+    webhook_result.resource = file_resource_config_with_items_to_parse
+
+    entities_to_create, entities_to_delete = (
+        await mock_live_events_mixin._parse_raw_event_results_to_entities(
+            [webhook_result]
+        )
+    )
+
+    assert len(entities_to_create) == 2
+    assert entities_to_create[0].identifier == "svc-one"
+    assert entities_to_create[1].identifier == "svc-two"
+    assert entities_to_delete == []
+    assert mock_live_events_mixin.entity_processor.parse_items.call_count == 2
+
+
 @pytest.mark.asyncio
 async def test_sync_raw_results_entity_deleted(
     mock_live_events_mixin: LiveEventsMixin,