Feature/event handler retries

[Astrasu]

Background

The original startup path called await eventing.on_startup() inside the FastAPI lifespan
and re-raised any connection error, killing the process if the NATS broker or Dapr sidecar
was not yet ready. The first implementation on this branch added a retry loop and a
subscriptions_ready flag directly inside EventHandlingBase to fix that crash.

A subsequent code review identified two blockers:

1. Health check gap. NATSClient.health_check() checked only whether the transport
   socket was open (is_connected). It was unaware of whether topic subscriptions had
   actually been established. A pod could pass the readiness probe and start receiving
   traffic before any handler was subscribed.

2. Missing issue link. The PR body lacked a Closes #… / Fixes #… reference.

Attempting to fix the health check exposed a deeper problem: the retry and subscription
lifecycle had been placed in the wrong layer entirely.

Separation-of-concerns issue uncovered

EventHandlingBase is a REST API dispatch class. Its role is to receive an incoming
HTTP/CloudEvent request and route it through the handler chain. It had no business knowing
whether a broker connection existed, how many times to retry it, or when subscriptions
became active.

The health check (ClientHealthChecker) already operated on the transport clients
(NATSClient, DaprClient), not on the eventing classes. Adding subscription readiness
tracking to EventHandlingBase would have required either:

• Reaching across the layer boundary to expose internal client state from the eventing
  class, or
• Duplicating the readiness flag in two places.

Both options made the health check unreliable and the design harder to follow.

The correct fix was to move the entire connection lifecycle — connection, retry loop,
subscription management, disconnect/reconnect handling — into the clients, where the
health checker already lived.

Changes

NATSClient and DaprClient

Both clients now expose an identical resilience interface so they can be used
interchangeably by the eventing layer:

async def subscribe(self, topic_callbacks: dict[str, Callable[[CloudEvent], Awaitable[None]]]) -> None:
    """Register all topic→callback mappings and start the background retry task (non-blocking)."""

@property
def subscriptions_ready(self) -> bool:
    """True once connect + all subscriptions completed without error."""

async def health_check(self) -> ComponentHealth:
    """Unhealthy until subscriptions_ready is True, or if the broker becomes unreachable."""

async def close(self) -> None:
    """Cancel the retry task then close the transport."""

Key behaviours:

• subscribe() returns immediately. Connection and subscription happen in a background
  asyncio.Task so on_startup() never blocks or raises.
• The retry loop reads event_client_max_retries (default -1 = indefinite) and
  event_client_retry_delay (default 5.0 s) from config.
• health_check() returns unhealthy until both transport connectivity and topic
  subscriptions are confirmed. The readiness probe therefore reflects true handler
  readiness, not just socket state.
• NATS: disconnected_cb clears subscriptions_ready; reconnected_cb re-subscribes
  JetStream topics manually (Core NATS re-subscribes automatically via the library) and
  restores the flag. This keeps the readiness probe accurate across runtime reconnects.
• Dapr: no persistent socket, so _connect_and_subscribe() pings
  {dapr_url}/v1.0/healthz. If the sidecar disappears after startup, the live ping in
  health_check() catches it without extra polling.
• close() cancels an in-progress retry task cleanly before tearing down the transport.

ClientBase

The abstract subscribe() signature was updated from (topic: str, callback: Callable)
to (topic_callbacks: dict[str, Callable]) so the full mapping can be handed to the
client atomically. AI clients (OpenAIClient, VLLMClient) implement it as a no-op with
the updated signature.

NatsEventing and DaprEventing

on_startup() now:

1. Fetches the transport client from the registry.
2. Builds a {topic: callback} dict from all registered event handlers (and, for NATS,
   from config-declared topics).
3. Calls await client.subscribe(topic_callbacks) — which returns immediately.

on_shutdown() is a no-op; the lifespan already calls client.close() during teardown.

The POST /nats/subscribe/{topic} REST endpoint was removed. Runtime subscription
mutation via REST conflicted with the managed subscription model (_topic_callbacks is
fixed at startup) and the existing POST /events/{topic} endpoint already serves the
testing/debugging use case by routing a payload directly through the handler chain.

The GET /dapr/subscribe endpoint was kept. The Dapr sidecar calls it at startup for
passive topic discovery; it is not a mutation and does not interact with
_subscriptions_ready.

EventHandlingBase

All retry and connection lifecycle code was removed. The class reverts to its original
role: a pure event-dispatch base combining REST routing (RestApiBase) with CloudEvent
processing (CloudEventProcessorMixin). It exposes only:

• handle_event() — structured dispatch with HTTP result encoding.
• publish() — abstract; implemented by concrete eventing subclasses.
• _process_cloud_event() — error-classified dispatch with logging.

Tests

File	Change
test_nats_client.py	Full rewrite: init fields, retry loop (max_retries 0/2/−1), disconnect/reconnect callbacks, health check readiness, managed subscribe, task cancellation
test_dapr_client.py	Same structure mirroring NATS; sidecar-ping readiness path
test_nats.py	Replaced old _connect_and_subscribe / _subscribe_to_topic tests with TestNatsEventingOnStartup covering topic collection and client.subscribe() delegation
test_dapr.py	Added TestDaprEventingOnStartup mirroring NATS startup tests
test_event_handling_base.py	Removed TestStartWithRetry (retry logic no longer lives here); retained TestUnwrapNestedCloudEvent and TestHandleEvent
test_openai_client.py, test_vllm_client.py	Updated subscribe call to dict API

Config keys

Key	Type	Default	Meaning
event_client_max_retries	int	-1	Retries after the first failure. -1 = indefinite. 0 = single attempt, raise on failure.
event_client_retry_delay	float	5.0	Seconds between attempts.

Closes #28

[pajoma]

Verdict: DENY

Target Summary:
PR #27, "Feature/event handler retries", is open against develop from feature/event_handler_retries: #27

The PR adds asynchronous retry handling for event broker startup, updates NATS/Dapr eventing classes, adds unit tests, documents broker startup resilience, and bumps the package version from 0.6.1 to 0.6.2.

Evidence Reviewed:

• Live PR metadata, body, files, commits, comments, reviews, checks, and closing issue references were fetched on 2026-06-17.
• PR comments/reviews: none.
• Prior reviewer verdicts: none.
• Checks: Lint, Type Check & Test is passing.
• Closing issue references: none returned by GitHub GraphQL; the PR body does not contain Closes #..., Fixes #..., or Resolves #....
• Diff reviewed: docs, pyproject.toml, src/blueprint/agents/io/api/eventing/{event_handling_base,dapr,nats}.py, src/blueprint/agents/io/api/utilities/root.py, and eventing unit tests.
• Local context inspected: ClientHealthChecker, HealthCheckCache, NATSClient.health_check, DaprClient.health_check, current eventing startup/subscription flow, and recent develop history.

Technical Assessment:
Brooks-Lint PR Review result: Health Score 95/100. Scope: PR #27 diff, with local context for health/readiness and client behavior. No generated files were reviewed.

Blocking finding:

Change Propagation / Correctness Contract Drift - readiness does not prove subscriptions are active
Symptom: The PR documentation and body claim /health/ready remains 503 until the broker is reachable and "all topic subscriptions are active", but the changed implementation only retries _connect_and_subscribe() in a background task. Readiness still delegates to ClientHealthChecker, which calls each client's connect() and then health_check(). NATSClient.health_check() reports healthy when the client is connected; it does not know whether auto-subscription completed. If NATS connection succeeds but one configured subscription fails, _start_with_retry() may keep retrying or eventually fail, while readiness can still return healthy because the client connection exists.
Source: A Philosophy of Software Design - Information Leakage; Software Engineering at Google - Hyrum's Law / observable contract.
Consequence: Kubernetes can route traffic to a pod that is connected to NATS but has not actually subscribed to required topics, contradicting the documented startup contract and hiding message-consumption failure behind a green readiness probe.
Remedy: Track eventing startup/subscription readiness explicitly and wire that state into the health/readiness path, or adjust the contract and docs if readiness is only meant to reflect client connectivity. Add a test where client connection succeeds but subscription fails, and assert /health/ready remains unavailable until subscriptions are established.

Test assessment:
The new retry-loop unit tests cover retry counts, cancellation, and exhaustion. They do not cover the readiness contract that the PR explicitly documents. Because that contract is operationally important for Kubernetes behavior, the missing test is part of the blocker above, not a non-blocking suggestion.

Workflow Assessment:

• Base/head branches are plausible: develop <- feature/event_handler_retries.
• PR is open, not draft, and merge state is clean.
• Required checks visible to GitHub are passing.
• Blocking workflow issue: the PR has no linked story/issue and no GitHub closing syntax. The required PR gate says a PR body must contain Closes #123, Fixes #123, or Resolves #123; a plain description is not enough.

Blocking Concerns:

1. Readiness can report healthy based on client connectivity even when event subscriptions are not active, contradicting the PR's documented operational contract.
2. The PR does not link or close a user story/issue with GitHub closing syntax.

Recommended Next Steps:

1. Add explicit eventing/subscription readiness state and expose it through the existing health checker path, or narrow the docs/body to say readiness only reflects client connectivity.
2. Add a regression test for "broker connects, subscription fails" proving readiness stays down until subscriptions succeed.
3. Update the PR body with Closes #..., Fixes #..., or Resolves #... for the relevant user story/issue.

Posted: submitted as a formal PR review requesting changes.

DENY

[pajoma]

Verdict: DENY

Target Summary:
PR #27, "Feature/event handler retries", is open against develop from feature/event_handler_retries: #27

The latest actionable update is commit e93484d98dece0daf3682286444be7c526fefa4b from 2026-06-18 plus the rewritten PR body. There are no issue comments and no inline review comments; the latest discussion artifact remains the prior CHANGES_REQUESTED review from 2026-06-17.

Evidence Reviewed:

• Live PR metadata, body, files, commits, comments, reviews, check rollup, and current head SHA were fetched on 2026-06-22.
• Prior reviewer verdict: CHANGES_REQUESTED on 2026-06-17.
• Latest commit reviewed: e93484d98dece0daf3682286444be7c526fefa4b.
• Diff reviewed against refs/remotes/origin/develop: 21 files, 845 additions, 275 deletions.
• Local/GitHub context inspected: NATSClient, DaprClient, ClientBase, NatsEventing, DaprEventing, readiness-related tests, PR body, and visible CI check rollup.
• Checks: GitHub currently reports one Lint, Type Check & Test failure for the pull_request run and one passing push run; PR merge state is UNSTABLE.

Technical Assessment:
Brooks-Lint PR Review result: Health Score 100/100 for the updated eventing/retry scope. No generated files were reviewed.

The previous technical blocker is resolved. NATSClient.health_check() now returns unhealthy when managed subscriptions are not ready, _start_with_retry() only flips readiness true after _connect_and_subscribe() succeeds, and _connect_and_subscribe() subscribes all registered topics before readiness is set. The Dapr path follows the same readiness pattern by gating readiness on sidecar reachability. The tests now cover the key regression shape, including connected-but-subscriptions-not-ready and connect-succeeds/subscription-fails behavior.

I did not find a replacement code-level blocker in the updated eventing implementation.

Workflow Assessment:

• Base/head branches are plausible: develop <- feature/event_handler_retries.
• PR is open and not draft.
• Prior blocker resolved: the readiness/subscription contract is now represented in client health checks and tests.
• Prior blocker still open: the PR body still does not contain real GitHub closing syntax such as Closes #123, Fixes #123, or Resolves #123. It describes that the link was missing, but it does not actually link or close a user story/issue.
• New/remaining workflow blocker: GitHub's current check rollup is not clean. The pull_request-triggered Lint, Type Check & Test run failed at the mypy step with exit code 127, while a separate push-triggered run passed. Until the PR-visible required check state is clean, this is not merge-ready.

Blocking Concerns:

1. The PR still has no linked/closing user story or issue via real GitHub closing syntax.
2. The PR currently has an unstable check state with a visible failed Lint, Type Check & Test pull_request run.

Recommended Next Steps:

1. Add the actual closing reference to the PR body, for example Closes #123, Fixes #123, or Resolves #123 for the relevant story.
2. Re-run or fix the failing pull_request CI path so the PR check rollup is clean on the current head commit.

Posted: submitted as a formal PR review requesting changes.

DENY

[Mathes76]

Verdict: DENY

Target Summary:
PR #27, "Feature/event handler retries", is open against develop from feature/event_handler_retries: #27

The latest reviewed head is 79f252c22d5485987d52ca190d0e466c665fba60 (updated 2026-06-24). The PR moves the broker connection/retry/subscription lifecycle out of EventHandlingBase and into the transport clients (NATSClient, DaprClient), gates health_check() on subscription readiness, removes the POST /nats/subscribe/{topic} endpoint, and rewrites the eventing unit tests.

Evidence Reviewed:

• Live PR metadata, body, files, commits, comments, reviews, and check runs were fetched on 2026-06-25.
• Prior reviewer verdicts: two CHANGES_REQUESTED reviews on 2026-06-17 and 2026-06-22 (pajoma).
• Latest commit reviewed: 79f252c22d5485987d52ca190d0e466c665fba60.
• Checks: both Lint, Type Check & Test runs on the head commit report success.
• Closing issue reference: the PR body now contains Closes #28 — the prior workflow blocker is resolved.
• Prior readiness/subscription-contract blocker: resolved. health_check() on both clients now returns unhealthy until subscriptions_ready is set.
• Diff reviewed: src/blueprint/agents/clients/io/{nats_client,dapr_client}.py, client_base.py, src/blueprint/agents/io/api/eventing/{event_handling_base,nats,dapr}.py, the AI clients, root.py, .github/actions/setup-python-env/action.yml, docs, and the eventing/client unit tests.

Technical Assessment:
The core refactor is sound. Moving connection, retry, reconnect, and subscription-readiness tracking into the clients restores separation of concerns, and the readiness probe now reflects true handler readiness rather than socket state. Retry semantics (-1 = indefinite, 0 = single attempt) are consistent across NATS and Dapr and match the documented config keys. Task teardown is clean (close() cancels the retry task and suppresses CancelledError; _on_retry_done retrieves the exception). The NATS JetStream-vs-Core reconnect distinction is correct.

One blocking finding and one item that needs clarification before merge.

1. (Blocking) Dapr: subscription callbacks are built and stored but never invoked; docstring states the opposite (information leakage / dead contract).
Symptom: DaprEventing.on_startup() builds a {topic: callback} mapping via _make_event_callback(topic) and passes it to DaprClient.subscribe(), which stores it in self._topic_callbacks. Nothing in DaprClient ever reads _topic_callbacks. Incoming Dapr events arrive over POST /events/{topic} and are dispatched directly by DaprEventing.publish() -> _process_cloud_event(), independent of these callbacks. The DaprClient.subscribe docstring asserts the mapping is "for use by DaprEventing.publish()", which is not true.
Consequence: Dead state plus misleading documentation. A maintainer reading the Dapr path will assume the stored callbacks participate in dispatch and may wire future behavior onto state that is never consulted. For Dapr, subscribe() functionally needs only to set _subscriptions_managed=True and start the sidecar-ping retry task.
Remedy: Either drop _make_event_callback/_topic_callbacks from the Dapr path and have subscribe() take just what it needs to manage readiness, or, if the mapping is retained intentionally, correct the docstring to describe its actual (non-dispatch) role.

2. (Needs clarification) CI: virtualenv caching removed in .github/actions/setup-python-env/action.yml, outside this PR's stated scope.
Symptom: The composite action drops the actions/cache@v4 step that cached .venv keyed on pyproject.toml (-8 lines). This is unrelated to event-handler retries and removes caching for every CI run using this action.
Consequence: If the removal was deliberate (e.g. cache key collisions or stale-venv issues), the rationale is not recorded in the PR; if it was incidental, it is a silent CI-time regression bundled into a feature PR.
Remedy: Confirm the intent — keep it with a one-line rationale in the PR body, or restore the cache step / split it into its own change.

Workflow Assessment:

• Base/head branches are plausible: develop <- feature/event_handler_retries.
• PR is open, not draft, merge state clean.
• Required checks on the head commit are green.
• Closing syntax (Closes #28) is present — the prior workflow blocker is resolved.

Blocking Concerns:

1. Dapr subscription callbacks are built and stored but never invoked, and the DaprClient.subscribe docstring misdescribes their purpose.

Recommended Next Steps:

1. Remove the unused Dapr callback machinery (or correct the docstring if the mapping is kept intentionally) so the Dapr readiness path contains no dead dispatch state.
2. Clarify the CI virtualenv-cache removal — document the reason in the PR body or restore/split it out.

Posted: submitted as a formal PR review requesting changes.

DENY