feat: Webhook operator for posting intermediate results back to webserver layer

[jdye64]

As the system is processing in "online" mode the system needs a way to report the current status of the processing back to the client. This requires getting those details first to the web server layer and then it sends it back to the client. This operator assists in getting those results back to the webserver before the entire job is completed.

Checklist

• I am familiar with the Contributing Guidelines.
• New or existing tests cover these changes.
• The documentation is up to date with these changes.
• If adjusting docker-compose.yaml environment variables have you ensured those are mimicked in the Helm values.yaml file.

[greptile-apps]

Greptile Summary

This PR adds WebhookNotifyOperator, a side-effect-only graph stage that HTTP-POSTs processed batch results to a configurable URL so the web server layer can stream intermediate progress back to the client. The operator is always appended last in the pipeline, uses a lazily-created requests.Session (with retry) that is reused across batches, and passes data through unmodified regardless of HTTP outcome. The API key leakage and per-batch session issues from the prior review have both been addressed.

Confidence Score: 5/5

Safe to merge; all remaining findings are P2 style/hardening suggestions that do not block correct runtime behaviour.

The two major issues from the previous review (API key leakage and per-batch session creation) are both resolved. The new code is well-structured, properly documented, and ships with a solid test suite. The three remaining comments are all P2: a missing DataFrame type guard in process(), mocking depth in tests, and lack of URL scheme validation — none cause incorrect behaviour in the intended usage.

No files require special attention; the URL validation suggestion in params/models.py is the highest-value hardening opportunity.

Security Review

• No URL scheme validation on endpoint_url (params/models.py): Any string is accepted, including file://, ftp://, or internal IP ranges. A misconfigured or adversarially supplied URL fails silently (exception is swallowed) so there is no early-exit warning. Using pydantic.AnyHttpUrl would restrict accepted schemes to http/https and surface misconfigurations at construction time.
• The previously flagged NVIDIA API key auto-injection into the Authorization header has been resolved: WebhookParams carries no api_key field, so _ParamsModel._resolve_api_keys is a no-op for this class.

Important Files Changed

Filename	Overview
nemo_retriever/src/nemo_retriever/graph/webhook_operator.py	New side-effect-only operator that POSTs batch results to a webhook URL; session is lazily created and reused; exception handling correctly logs network failures and passes data through
nemo_retriever/src/nemo_retriever/params/models.py	Adds WebhookParams with no api_key field (addressing previous security concern); fields lack Field(description=...) but that pattern is inconsistent across existing models
nemo_retriever/src/nemo_retriever/graph/ingestor_runtime.py	Threads webhook_params through resolve/append helpers as keyword-only args (safe after *); webhook stage is always appended last after all pending_stages, consistent with documented behaviour
nemo_retriever/tests/test_webhook_operator.py	Good coverage of serialize helpers, column-subset logic, no-op paths, happy path, error handling, session reuse, and retry adapter; one issue: mocks _get_session (internal) rather than requests.Session at the boundary

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart TD
    A[GraphIngestor.webhook\nrecords stage + params] --> B[build_graph]
    B --> C[_resolve_execution_inputs\nextracts webhook_params from plan]
    C --> D[_append_ordered_transform_stages]
    D --> E{webhook_params set\nAND endpoint_url truthy?}
    E -- No --> F[Skip: no-op]
    E -- Yes --> G[Append WebhookNotifyOperator\nalways AFTER all other stages]
    G --> H[WebhookNotifyOperator.process]
    H --> I{data is empty?}
    I -- Yes --> J[Debug log, return data]
    I -- No --> K[_dataframe_to_records\nserialise to JSON-safe dicts]
    K --> L[_get_session\nlazy-init, reuse across batches]
    L --> M[session.post with Retry adapter]
    M --> N{HTTP success?}
    N -- Yes --> O[Log INFO, return data unchanged]
    N -- No --> P[logger.exception, return data unchanged]

Loading

Prompt To Fix All With AI

This is a comment left during a code review.
Path: nemo_retriever/src/nemo_retriever/graph/webhook_operator.py
Line: 108-116

Comment:
**`_dataframe_to_records` called outside the try/except block**

`_dataframe_to_records(data, ...)` calls `df.to_dict()` and accesses `df.columns`. If an upstream operator ever passes something other than a `pd.DataFrame` (e.g., a `ControlMessage`, a list, or `None`), the resulting `AttributeError` propagates uncaught and halts the pipeline, whereas a network failure on the same path is silently logged and swallowed. Since this is a side-effect-only operator that is meant to be non-fatal, consider wrapping the entire notification block (including serialisation) in the try/except, or add an explicit `isinstance(data, pd.DataFrame)` guard before the serialisation call.

How can I resolve this? If you propose a fix, please make it concise.

---

This is a comment left during a code review.
Path: nemo_retriever/tests/test_webhook_operator.py
Line: 160-175

Comment:
**Mocking at internal boundary rather than the `requests` adapter**

`@patch("...WebhookNotifyOperator._get_session")` mocks a private implementation method rather than the external I/O boundary (`requests.Session.post`). Per the project's mocking discipline rules, mocks should be placed at the boundary so that internal refactors (e.g., renaming `_get_session`) don't silently stop exercising the real path. Patching `requests.Session` (or its `post` method) directly, and letting `_get_session` run, would give stronger assurance that the adapter/retry configuration is actually applied during the call.

How can I resolve this? If you propose a fix, please make it concise.

---

This is a comment left during a code review.
Path: nemo_retriever/src/nemo_retriever/params/models.py
Line: 420-424

Comment:
**No validation that `endpoint_url` is an HTTP/HTTPS URL**

Any arbitrary string (e.g., `"file:///etc/passwd"`, `"ftp://..."`, or a bare hostname) is accepted and only rejected at HTTP-request time. Because the exception is swallowed and logged, a misconfigured or malicious URL fails silently. A simple Pydantic `@field_validator` (or `AnyHttpUrl` type) on `endpoint_url` would surface mistakes early and prevent the operator from being inadvertently pointed at non-HTTP schemes or internal services.

```python
from pydantic import AnyHttpUrl

endpoint_url: Optional[AnyHttpUrl] = None
```

How can I resolve this? If you propose a fix, please make it concise.

_{Reviews (2): Last reviewed commit: "Merge branch 'main' into feat/post-per-f..." | Re-trigger Greptile}