feat: implement messaging, routing, and retry/cancel mechanisms (topics 7-9)

[gabriel-farache]

Summary

This PR consolidates topics 7, 8, and 9. #15 (topic 7) and #18 (topic 8) were closed in favor of merging that work directly through this PR.

Topic 7 — Messaging integration

• CloudEvent builder (NewCloudEvent, FormatSource) for spec-compliant CloudEvents
• NATS JetStream messaging client: connection lifecycle, topic subscription/dispatch, response CE publishing, cancel deny-list management
• Topic name validation and main/retry/cancel topic derivation (DeriveTopicNames)

Topic 8 — Resource operation routing

• Route CloudEvent create/delete/cancel requests to the correct service provider by service type, with health-aware dispatch, retry backoff + jitter, and deny-list cancel semantics
• Bug fixes and security hardening (ack-on-error, poison-pill ack-drop, TOCTOU race via atomic AddIfAbsent, bounded deny-list, payload validation, sanitized CE error details)

Topic 9 — Retry processor and cancel mechanisms

• Retry processor: re-forwards pending requests from the retry topic once an SP recovers, with per-message handler deadline and stable Idempotency-Key forwarding
• Health CE publisher: publishes dcm.agent.health.service-type-degraded / -unavailable CloudEvents on SP state transitions
• SP forwarder extracted into a dedicated component; router internals refactored (DenyList → ResourceSet, RouterDeps, retry-consumer lifecycle wiring)
• Transient in-flight lock (KeyLock) serializing concurrent forward attempts per resource_id (main-topic vs. retry-topic races)

[gabriel-farache]

/agentic_review

[qodo-code-review]

Code Review by Qodo

🐞 Bugs (0) 📘 Rule violations (0) 📎 Requirement gaps (0) 🎨 UX issues (0) 🔗 Cross-repo conflicts (0) 📜 Skill insights (0)

1. Retry fetch errors masked ✓ Resolved 🐞 Bug ☼ Reliability ⭐ New

Description

retry.Processor.fetchAllFromConsumer breaks on any Consumer.Fetch error but always returns
(collected, nil), so non-timeout JetStream failures are silently treated as success. Callers (e.g.,
ProcessOnTransition) therefore proceed with partial/empty results without error propagation.

Code

internal/routing/retry/processor.go[R621-624]

+		batch, fetchErr := cons.Fetch(fetchBatchSize, jetstream.FetchMaxWait(fetchMaxWait))
+		if fetchErr != nil {
+			break
+		}

Evidence

ProcessOnTransition relies on fetchAllFromConsumer returning an error to abort/log;
fetchAllFromConsumer currently breaks on fetchErr but returns nil error unconditionally, masking
those failures.

internal/routing/retry/processor.go[131-152]
internal/routing/retry/processor.go[598-635]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
`fetchAllFromConsumer` treats *all* `cons.Fetch` errors the same (break) and then returns `nil` error, masking real failures (auth/connection/consumer deleted/etc.).

## Issue Context
`ProcessOnTransition` only logs/returns errors when `fetchAllFromConsumer` returns a non-nil error; with the current implementation, many failures won’t be reported.

## Fix Focus Areas
- internal/routing/retry/processor.go[147-156]
- internal/routing/retry/processor.go[598-635]

## Proposed fix
- Distinguish expected “no more messages / timeout” from genuine errors:
 - If the error is the expected timeout (or context cancellation), stop fetching and return `collected, nil`.
 - Otherwise return `collected, fetchErr` (or wrap it) so callers can log and retry.
- Optionally log the fetch error (at least at debug/warn) when returning it to improve observability.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

2. Cancel lacks handler deadline ✓ Resolved 🐞 Bug ☼ Reliability ⭐ New

Description

handleCancelMessage calls the cancel handler with context.Background(), so cancel processing has no
timeout/cancellation. Because Router.HandleCancel may synchronously purge the retry topic using the
provided ctx, a hung JetStream/store operation can stall the cancel consumer (and startup
cancel-drain) indefinitely.

Code

internal/messaging/handlers.go[R34-37]

+	c.logMessageReceived(msg, resourceID, ceID, ceType)
+
+	if err := c.cancelHandler(context.Background(), msg.Data()); err != nil {
+		if nakErr := msg.NakWithDelay(c.nakDelay()); nakErr != nil {

Evidence

The cancel handler is invoked with context.Background(), and Router cancel handling uses the passed
context to drive synchronous retry-topic purge work; without a deadline, cancel processing can hang
the consumer loop.

internal/messaging/handlers.go[15-43]
internal/routing/router.go[373-420]
internal/routing/router.go[430-437]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
Cancel messages are processed with `context.Background()` in `handleCancelMessage`, so there is no per-message deadline/cancellation. Downstream cancel handling performs synchronous work (including retry-topic purge) and can block forever.

## Issue Context
- `Router.HandleCancel(ctx, ...)` calls `purgeFromRetryTopic(ctx, ...)`, which calls `FetchRetryMessages(ctx)`.
- If the cancel handler hangs, `drainCancelTopic` (which calls `handleCancelMessage` inline) can also hang, delaying/locking startup.

## Fix Focus Areas
- internal/messaging/handlers.go[15-49]
- internal/routing/router.go[373-455]

## Proposed fix
- In `handleCancelMessage`, create a per-message context with timeout (similar to `handleMainMessage`).
 - Prefer a timeout that is guaranteed `< CancelAckWait` to avoid redelivery during cancel handling (either a new `CancelHandlerTimeout` config, or `min(HandlerTimeout, CancelAckWait - safetyMargin)`).
- Pass that context into `c.cancelHandler(ctx, msg.Data())`.
- Optionally add a config invariant check analogous to `HandlerTimeout < AckWait`, but for cancel timeout vs `CancelAckWait`.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

3. Retry forwards can hang ✓ Resolved 🐞 Bug ☼ Reliability

Description

Retry processing on health transitions/restarts forwards to SPs without enforcing
ProcessorConfig.HandlerTimeout, so a hung SP call can block retry processing indefinitely when
invoked with the long-lived top-level context.

Code

internal/routing/retry/processor.go[R613-616]

+	err := routing.ForwardToSP(ctx, p.deps.Forwarder, sp, routing.ForwardParams{
+		ResourceID: res.resourceID, ServiceType: res.serviceType,
+		Spec: res.spec, EventID: res.eventID, IsCreate: res.ceType != cloudevent.TypeRequestDelete,
+	})

Evidence

The health-transition path uses the top-level context from main, and the retry processor forwarding
helper uses that context directly without applying HandlerTimeout, unlike the (unused here)
forwardAndAck path which does wrap with HandlerTimeout.

cmd/environment-agent/main.go[190-197]
internal/routing/retry/processor.go[148-193]
internal/routing/retry/processor.go[371-386]
internal/routing/retry/processor.go[589-616]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`ProcessorConfig.HandlerTimeout` is not applied to forwarding during transition/restart retry processing (`forwardRequest`), so retry processing can hang indefinitely if an external SP call hangs (common when HTTP client has no timeout).

### Issue Context
- `main.go` wires `retryProcessor.RunTransition(ctx, ...)` using the top-level `run` context (cancellation-only, no deadline).
- `ProcessOnTransition` → `processTransitionItems` → `forwardRequest` forwards using the provided `ctx` directly.
- While `forwardAndAck` wraps ctx with `HandlerTimeout`, it’s only used by `Processor.Start` (not used in main wiring).

### Fix Focus Areas
- cmd/environment-agent/main.go[190-197]
- internal/routing/retry/processor.go[148-193]
- internal/routing/retry/processor.go[371-386]
- internal/routing/retry/processor.go[589-626]

### Suggested direction
- In `forwardRequest`, create a per-message context:
 - If `p.deps.Config.HandlerTimeout > 0`, wrap with `context.WithTimeout(ctx, ...)` (or compute min(existing_deadline, handler_timeout)).
 - Ensure `defer cancel()`.
- Use that per-message context for `routing.ForwardToSP(...)` so hung SP calls are canceled reliably during transition/restart processing.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

View high (2)

4. Retry JS captured nil ✓ Resolved 🐞 Bug ≡ Correctness

Description

retry.ProcessorDeps stores a one-time snapshot of msgClient.JS(); because messaging.Client.Start is
explicitly non-blocking, this snapshot can be nil during startup and the retry processor will
permanently no-op all retry fetch/purge logic.

Code

cmd/environment-agent/main.go[R144-147]

+		Forwarder:           forwarder,
+		Publisher:           msgClient,
+		JS:                  msgClient.JS(),
+		DenyList:            denyList,

Evidence

main wires retry.NewProcessor with a one-time msgClient.JS() value; messaging initializes c.js
later during setup, and the retry processor explicitly treats deps.JS == nil as a no-op when
fetching from consumers.

cmd/environment-agent/main.go[108-159]
internal/messaging/client.go[81-129]
internal/messaging/client.go[132-148]
internal/messaging/client.go[299-307]
internal/routing/retry/processor.go[33-42]
internal/routing/retry/processor.go[659-662]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`retry.NewProcessor` is constructed with `JS: msgClient.JS()` once at startup. If NATS/JetStream isn’t ready yet (Start is non-blocking), `msgClient.JS()` returns nil and the retry processor will keep `deps.JS == nil` forever, disabling retry-topic draining and cancel purging.

### Issue Context
- `messaging.Client.Start` returns immediately and initializes JetStream later during connection setup.
- `retry.Processor.fetchAllFromConsumer` returns `(nil, nil)` when `deps.JS` is nil, silently disabling processing.

### Fix Focus Areas
- cmd/environment-agent/main.go[108-159]
- internal/messaging/client.go[81-129]
- internal/messaging/client.go[299-307]
- internal/routing/retry/processor.go[33-42]
- internal/routing/retry/processor.go[659-662]

### Suggested direction
- Change `ProcessorDeps.JS` to a provider (e.g., `func() jetstream.JetStream` or an interface with `JS() jetstream.JetStream`) and resolve JetStream at call time inside `fetchAllFromConsumer`.
- Alternatively, delay creating/binding the processor (and `router.SetRetryConsumer`) until messaging setup has initialized JetStream, or add a `processor.SetJS(...)` that is invoked after successful messaging setup.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

5. SetupOnce blocks recovery ✓ Resolved 🐞 Bug ☼ Reliability

Description

messaging.Client.doSetup uses sync.Once around setupStreamsAndConsume; if setupStreamsAndConsume
exits early due to a transient JetStream/stream/consumer error, setup will never be retried on later
reconnects and consumers may never start.

Code

internal/messaging/client.go[R127-129]

+	c.setupOnce.Do(func() {
+		c.setupStreamsAndConsume(ctx, conn)
+	})

Evidence

doSetup permanently gates setup behind setupOnce.Do, while setupStreamsAndConsume returns on
errors without any mechanism to reset/retry setup, so subsequent connect callbacks cannot recover.

internal/messaging/client.go[93-106]
internal/messaging/client.go[123-154]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`sync.Once` is used to guard a fallible initialization path (`setupStreamsAndConsume`). If any step fails and returns, `setupOnce` still becomes “done”, so later successful connects can’t retry setup and the client may remain permanently uninitialized.

### Issue Context
- `setupStreamsAndConsume` logs and returns on multiple errors.
- `doSetup` is invoked from the NATS `ConnectHandler`, but `sync.Once` prevents subsequent calls.

### Fix Focus Areas
- internal/messaging/client.go[93-106]
- internal/messaging/client.go[123-154]

### Suggested direction
- Replace `setupOnce` with a stateful retry mechanism:
 - e.g., a `setupMu` + `setupDone bool` set only after full successful setup.
 - On failure, keep `setupDone=false` so the next connect can retry.
- Consider adding backoff/logging for repeated setup failures to improve operability.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

6. Cancel panic terminates message ✓ Resolved 🐞 Bug ☼ Reliability ⭐ New

Description

On panic in cancel processing, handleCancelMessage calls msg.Term(), permanently discarding the
cancel message instead of retrying. This extends Term beyond the stated “parse failure” case and can
drop valid cancels when a transient bug/panic occurs.

Code

internal/messaging/handlers.go[R20-23]

+	defer func() {
+		if r := recover(); r != nil {
+			termErr := msg.Term()
+			attrs := []any{

Evidence

Cancel panic recovery explicitly calls Term(), while the messaging client’s consumer configuration
comment states cancels should only be Term’d on parse failure.

internal/messaging/handlers.go[15-32]
internal/messaging/client.go[415-417]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
`handleCancelMessage` recovers panics by calling `msg.Term()`, which permanently removes the cancel message. This contradicts the nearby documented intent that cancel messages are only Term’d on parse failure.

## Issue Context
Cancel consumer is configured without `MaxDeliver`, so Term is the primary way cancels get dropped.

## Fix Focus Areas
- internal/messaging/handlers.go[15-32]
- internal/messaging/client.go[415-417]

## Proposed fix
- Change the cancel panic recovery behavior to mirror the main handler behavior:
 - Prefer `msg.NakWithDelay(c.nakDelay())` (or at least `msg.Nak()`) on panic, so a transient panic doesn’t silently drop the cancel.
- If dropping on panic is intentional, update the comment in `initConsumers` to reflect that cancels may be Term’d on panic as well, and consider publishing a terminal error CloudEvent for observability.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

7. Start blocks during setup ✓ Resolved 🐞 Bug ☼ Reliability ⭐ New

Description

messaging.Client.Start is documented as non-blocking but calls doSetup synchronously when the
initial connection succeeds. doSetup can block up to requestStreamRetryTimeout waiting for the
request stream, so Start may block agent startup for a long time.

Code

internal/messaging/client.go[R201-204]

+	if conn.IsConnected() {
+		c.connected.Store(true)
+		c.doSetup(setupCtx, conn)
+	}

Evidence

Start explicitly documents non-blocking behavior, but invokes doSetup inline on an already-connected
connection. doSetup can block while creating request consumers, which retries until
requestStreamRetryTimeout expires.

internal/messaging/client.go[166-215]
internal/messaging/client.go[441-458]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
`messaging.Client.Start` claims to be non-blocking, but it calls `c.doSetup(...)` inline when `conn.IsConnected()`. `doSetup`/consumer creation can block for up to `requestStreamRetryTimeout`, so `Start` can block despite the documented contract.

## Issue Context
This is especially likely when NATS is reachable but the control-plane request stream (`RequestStreamName`) is not created yet.

## Fix Focus Areas
- internal/messaging/client.go[166-215]

## Proposed fix
- In `Start`, change the `if conn.IsConnected()` path to `go c.doSetup(setupCtx, conn)` (matching the ConnectHandler/ReconnectHandler behavior).
- Consider whether `Start(ctx)` should use the passed `ctx` (instead of `context.Background()`) for setup cancellation, or ensure `Stop()` always cancels/halts setup promptly.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

8. Cancel purge resets retries ✓ Resolved 🐞 Bug ☼ Reliability

Description

Both routing.Router.purgeFromRetryTopic and retry.Processor handle retry-topic messages by
republishing the payload back onto <topic>.retry and then ACKing the original, effectively
rewriting the retry backlog. This creates new JetStream messages with fresh delivery history, making
cancel operations O(backlog) and undermining NumDelivered/MaxDeliver-based poison termination so
persistently failing messages can evade cutoff and retry indefinitely.

Code

internal/routing/router.go[R385-388]

+		if err := rc.RepublishToRetry(ctx, m.Data); err != nil {
+			return fmt.Errorf("failed to republish non-matching message: %w", err)
+		}
+		if err := m.AckFunc(); err != nil {

Evidence

The cited cancellation path republished every non-matching retry message and then ACKed it,
replacing messages one-for-one with newly published instances and thereby resetting JetStream
delivery metadata while doing work proportional to the entire fetched retry backlog. Separately, the
retry processor similarly republishes messages to the retry subject and then ACKs the originals
(including in route/transition failure paths), and poison handling uses
msg.Metadata().NumDelivered to decide when to emit a MaxDeliver terminal error and Term() the
message; because republishing creates a new message instance with NumDelivered restarted, the
MaxDeliver cutoff no longer reflects true attempt count and can be bypassed.

internal/routing/router.go[370-391]
internal/routing/types.go[37-49]
internal/routing/retry/processor.go[353-372]
internal/routing/retry/processor.go[644-671]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
The retry pipeline currently “moves” retry messages by republishing them back onto `<topic>.retry` and then ACKing the original message (both in `Router.purgeFromRetryTopic` during cancel and within `retry.Processor` during retry/error transitions). This rewrites the retry backlog, resets JetStream delivery metadata (`msg.Metadata().NumDelivered`) that MaxDeliver poison handling relies on, makes cancel operations scale with the entire backlog, and can allow persistently failing messages to evade MaxDeliver termination.

## Issue Context
- In the router cancellation flow, non-matching (non-cancelled) retry messages are republished and then ACKed, effectively rewriting all unrelated messages just to cancel a subset.
- In the retry processor, multiple paths republish to the retry subject and then ACK the original, creating a new message instance with a fresh delivery history.
- Poison termination uses `msg.Metadata().NumDelivered` in `terminateIfMaxDeliver` to decide when to publish a MaxDeliver terminal error and `Term()` the message; republishing resets this accounting.
- The current `RetryTopicConsumer` abstraction appears to expose only `AckFunc` plus `RepublishToRetry`, which pushes callers toward republish+ack instead of preserving the original message.

## Fix Focus Areas
- internal/routing/router.go[370-391]
- internal/routing/types.go[37-49]
- internal/routing/retry/processor.go[644-672]
- internal/routing/retry/processor.go[353-374]

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

View medium (1)

9. Delete URL not escaped ✓ Resolved 🐞 Bug ≡ Correctness

Description

routing.Forwarder.deleteExternal concatenates endpoint + "/" + req.ResourceID; since resourceId is
only checked for non-empty, reserved URL characters in the ID can change the request path/query and
delete the wrong target or fail.

Code

internal/routing/forwarder.go[R103-105]

+func (f *Forwarder) deleteExternal(ctx context.Context, endpoint string, req DeleteResourceRequest) error {
+	url := endpoint + "/" + req.ResourceID
+	httpReq, err := http.NewRequestWithContext(ctx, http.MethodDelete, url, nil)

Evidence

Router only enforces resourceId is non-empty, and forwarder’s DELETE path uses raw string
concatenation, so special characters can alter the resulting request URL.

internal/routing/router.go[121-136]
internal/routing/forwarder.go[103-111]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
DELETE forwarding builds the URL via string concatenation with an unescaped `resourceId`. If `resourceId` contains `/`, `?`, or `#`, the final HTTP request path can be malformed or refer to a different path than intended.

### Issue Context
- Router accepts `resourceId` from inbound CloudEvent data and only validates it is non-empty.
- Forwarder uses the raw value when constructing the DELETE URL.

### Fix Focus Areas
- internal/routing/forwarder.go[103-111]
- internal/routing/router.go[121-136]

### Suggested direction
- Use structured URL construction:
 - `url.JoinPath(endpoint, resourceID)` (Go will escape path segments), or
 - `u, _ := url.Parse(endpoint); u.Path = path.Join(u.Path, url.PathEscape(resourceID));`
- Optionally validate `resourceId` format if the system expects a constrained character set.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

Tip of the day

💡 Did you know, you can show, collapse, or hide each part of a finding: code, evidence, and all

More tips ↗ | Customize Qodo ↗ | Qodo docs ↗

Previous reviews

Review updated until commit 045aeca

Results up to commit a7086d9 ⚖️ Balanced

🐞 Bugs (0) 📘 Rule violations (0) 📎 Requirement gaps (0) 🎨 UX issues (0) 🔗 Cross-repo conflicts (0) 📜 Skill insights (0)

1. Retry forwards can hang ✓ Resolved 🐞 Bug ☼ Reliability

Description

Retry processing on health transitions/restarts forwards to SPs without enforcing
ProcessorConfig.HandlerTimeout, so a hung SP call can block retry processing indefinitely when
invoked with the long-lived top-level context.

Code

internal/routing/retry/processor.go[R613-616]

+	err := routing.ForwardToSP(ctx, p.deps.Forwarder, sp, routing.ForwardParams{
+		ResourceID: res.resourceID, ServiceType: res.serviceType,
+		Spec: res.spec, EventID: res.eventID, IsCreate: res.ceType != cloudevent.TypeRequestDelete,
+	})

Evidence

The health-transition path uses the top-level context from main, and the retry processor forwarding
helper uses that context directly without applying HandlerTimeout, unlike the (unused here)
forwardAndAck path which does wrap with HandlerTimeout.

cmd/environment-agent/main.go[190-197]
internal/routing/retry/processor.go[148-193]
internal/routing/retry/processor.go[371-386]
internal/routing/retry/processor.go[589-616]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`ProcessorConfig.HandlerTimeout` is not applied to forwarding during transition/restart retry processing (`forwardRequest`), so retry processing can hang indefinitely if an external SP call hangs (common when HTTP client has no timeout).

### Issue Context
- `main.go` wires `retryProcessor.RunTransition(ctx, ...)` using the top-level `run` context (cancellation-only, no deadline).
- `ProcessOnTransition` → `processTransitionItems` → `forwardRequest` forwards using the provided `ctx` directly.
- While `forwardAndAck` wraps ctx with `HandlerTimeout`, it’s only used by `Processor.Start` (not used in main wiring).

### Fix Focus Areas
- cmd/environment-agent/main.go[190-197]
- internal/routing/retry/processor.go[148-193]
- internal/routing/retry/processor.go[371-386]
- internal/routing/retry/processor.go[589-626]

### Suggested direction
- In `forwardRequest`, create a per-message context:
 - If `p.deps.Config.HandlerTimeout > 0`, wrap with `context.WithTimeout(ctx, ...)` (or compute min(existing_deadline, handler_timeout)).
 - Ensure `defer cancel()`.
- Use that per-message context for `routing.ForwardToSP(...)` so hung SP calls are canceled reliably during transition/restart processing.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

2. Retry JS captured nil ✓ Resolved 🐞 Bug ≡ Correctness

Description

retry.ProcessorDeps stores a one-time snapshot of msgClient.JS(); because messaging.Client.Start is
explicitly non-blocking, this snapshot can be nil during startup and the retry processor will
permanently no-op all retry fetch/purge logic.

Code

cmd/environment-agent/main.go[R144-147]

+		Forwarder:           forwarder,
+		Publisher:           msgClient,
+		JS:                  msgClient.JS(),
+		DenyList:            denyList,

Evidence

main wires retry.NewProcessor with a one-time msgClient.JS() value; messaging initializes c.js
later during setup, and the retry processor explicitly treats deps.JS == nil as a no-op when
fetching from consumers.

cmd/environment-agent/main.go[108-159]
internal/messaging/client.go[81-129]
internal/messaging/client.go[132-148]
internal/messaging/client.go[299-307]
internal/routing/retry/processor.go[33-42]
internal/routing/retry/processor.go[659-662]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`retry.NewProcessor` is constructed with `JS: msgClient.JS()` once at startup. If NATS/JetStream isn’t ready yet (Start is non-blocking), `msgClient.JS()` returns nil and the retry processor will keep `deps.JS == nil` forever, disabling retry-topic draining and cancel purging.

### Issue Context
- `messaging.Client.Start` returns immediately and initializes JetStream later during connection setup.
- `retry.Processor.fetchAllFromConsumer` returns `(nil, nil)` when `deps.JS` is nil, silently disabling processing.

### Fix Focus Areas
- cmd/environment-agent/main.go[108-159]
- internal/messaging/client.go[81-129]
- internal/messaging/client.go[299-307]
- internal/routing/retry/processor.go[33-42]
- internal/routing/retry/processor.go[659-662]

### Suggested direction
- Change `ProcessorDeps.JS` to a provider (e.g., `func() jetstream.JetStream` or an interface with `JS() jetstream.JetStream`) and resolve JetStream at call time inside `fetchAllFromConsumer`.
- Alternatively, delay creating/binding the processor (and `router.SetRetryConsumer`) until messaging setup has initialized JetStream, or add a `processor.SetJS(...)` that is invoked after successful messaging setup.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

3. SetupOnce blocks recovery ✓ Resolved 🐞 Bug ☼ Reliability

Description

messaging.Client.doSetup uses sync.Once around setupStreamsAndConsume; if setupStreamsAndConsume
exits early due to a transient JetStream/stream/consumer error, setup will never be retried on later
reconnects and consumers may never start.

Code

internal/messaging/client.go[R127-129]

+	c.setupOnce.Do(func() {
+		c.setupStreamsAndConsume(ctx, conn)
+	})

Evidence

doSetup permanently gates setup behind setupOnce.Do, while setupStreamsAndConsume returns on
errors without any mechanism to reset/retry setup, so subsequent connect callbacks cannot recover.

internal/messaging/client.go[93-106]
internal/messaging/client.go[123-154]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
`sync.Once` is used to guard a fallible initialization path (`setupStreamsAndConsume`). If any step fails and returns, `setupOnce` still becomes “done”, so later successful connects can’t retry setup and the client may remain permanently uninitialized.

### Issue Context
- `setupStreamsAndConsume` logs and returns on multiple errors.
- `doSetup` is invoked from the NATS `ConnectHandler`, but `sync.Once` prevents subsequent calls.

### Fix Focus Areas
- internal/messaging/client.go[93-106]
- internal/messaging/client.go[123-154]

### Suggested direction
- Replace `setupOnce` with a stateful retry mechanism:
 - e.g., a `setupMu` + `setupDone bool` set only after full successful setup.
 - On failure, keep `setupDone=false` so the next connect can retry.
- Consider adding backoff/logging for repeated setup failures to improve operability.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

4. Delete URL not escaped ✓ Resolved 🐞 Bug ≡ Correctness

Description

routing.Forwarder.deleteExternal concatenates endpoint + "/" + req.ResourceID; since resourceId is
only checked for non-empty, reserved URL characters in the ID can change the request path/query and
delete the wrong target or fail.

Code

internal/routing/forwarder.go[R103-105]

+func (f *Forwarder) deleteExternal(ctx context.Context, endpoint string, req DeleteResourceRequest) error {
+	url := endpoint + "/" + req.ResourceID
+	httpReq, err := http.NewRequestWithContext(ctx, http.MethodDelete, url, nil)

Evidence

Router only enforces resourceId is non-empty, and forwarder’s DELETE path uses raw string
concatenation, so special characters can alter the resulting request URL.

internal/routing/router.go[121-136]
internal/routing/forwarder.go[103-111]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

### Issue description
DELETE forwarding builds the URL via string concatenation with an unescaped `resourceId`. If `resourceId` contains `/`, `?`, or `#`, the final HTTP request path can be malformed or refer to a different path than intended.

### Issue Context
- Router accepts `resourceId` from inbound CloudEvent data and only validates it is non-empty.
- Forwarder uses the raw value when constructing the DELETE URL.

### Fix Focus Areas
- internal/routing/forwarder.go[103-111]
- internal/routing/router.go[121-136]

### Suggested direction
- Use structured URL construction:
 - `url.JoinPath(endpoint, resourceID)` (Go will escape path segments), or
 - `u, _ := url.Parse(endpoint); u.Path = path.Join(u.Path, url.PathEscape(resourceID));`
- Optionally validate `resourceId` format if the system expects a constrained character set.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

Results up to commit 12f4e92 ⚖️ Balanced

🐞 Bugs (0) 📘 Rule violations (0) 📎 Requirement gaps (0) 🎨 UX issues (0) 🔗 Cross-repo conflicts (0) 📜 Skill insights (0)

1. Cancel purge resets retries ✓ Resolved 🐞 Bug ☼ Reliability

Description

Both routing.Router.purgeFromRetryTopic and retry.Processor handle retry-topic messages by
republishing the payload back onto <topic>.retry and then ACKing the original, effectively
rewriting the retry backlog. This creates new JetStream messages with fresh delivery history, making
cancel operations O(backlog) and undermining NumDelivered/MaxDeliver-based poison termination so
persistently failing messages can evade cutoff and retry indefinitely.

Code

internal/routing/router.go[R385-388]

+		if err := rc.RepublishToRetry(ctx, m.Data); err != nil {
+			return fmt.Errorf("failed to republish non-matching message: %w", err)
+		}
+		if err := m.AckFunc(); err != nil {

Evidence

The cited cancellation path republished every non-matching retry message and then ACKed it,
replacing messages one-for-one with newly published instances and thereby resetting JetStream
delivery metadata while doing work proportional to the entire fetched retry backlog. Separately, the
retry processor similarly republishes messages to the retry subject and then ACKs the originals
(including in route/transition failure paths), and poison handling uses
msg.Metadata().NumDelivered to decide when to emit a MaxDeliver terminal error and Term() the
message; because republishing creates a new message instance with NumDelivered restarted, the
MaxDeliver cutoff no longer reflects true attempt count and can be bypassed.

internal/routing/router.go[370-391]
internal/routing/types.go[37-49]
internal/routing/retry/processor.go[353-372]
internal/routing/retry/processor.go[644-671]

Agent prompt

The issue below was found during a code review. Follow the provided context and guidance below and implement a solution

## Issue description
The retry pipeline currently “moves” retry messages by republishing them back onto `<topic>.retry` and then ACKing the original message (both in `Router.purgeFromRetryTopic` during cancel and within `retry.Processor` during retry/error transitions). This rewrites the retry backlog, resets JetStream delivery metadata (`msg.Metadata().NumDelivered`) that MaxDeliver poison handling relies on, makes cancel operations scale with the entire backlog, and can allow persistently failing messages to evade MaxDeliver termination.

## Issue Context
- In the router cancellation flow, non-matching (non-cancelled) retry messages are republished and then ACKed, effectively rewriting all unrelated messages just to cancel a subset.
- In the retry processor, multiple paths republish to the retry subject and then ACK the original, creating a new message instance with a fresh delivery history.
- Poison termination uses `msg.Metadata().NumDelivered` in `terminateIfMaxDeliver` to decide when to publish a MaxDeliver terminal error and `Term()` the message; republishing resets this accounting.
- The current `RetryTopicConsumer` abstraction appears to expose only `AckFunc` plus `RepublishToRetry`, which pushes callers toward republish+ack instead of preserving the original message.

## Fix Focus Areas
- internal/routing/router.go[370-391]
- internal/routing/types.go[37-49]
- internal/routing/retry/processor.go[644-672]
- internal/routing/retry/processor.go[353-374]

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

---

[gabriel-farache]

/agentic_review

[qodo-code-review]

Code review by qodo was updated up to the latest commit 12f4e92

[gciavarrini]

Title/summary are Topic 9 (retry/cancel, AC-RCM-070 through AC-RCM-130), but later commits also add CP wire format and stream ownership, logging REQs/ACs (240 through 270), hardening DDs (250/260/270 and related), health IT rewiring, and .cursor/rules. Worth updating the PR body to list what is in scope now, or splitting / marking the non topic 9 work. “Section 4.9 all satisfied” also looks stale after logging/hardening and DD-250.

[qodo-code-review]

Code review by qodo was updated up to the latest commit 8ee6fd0

[qodo-code-review]

PR Summary by Qodo

feat(routing): retry processor, cancel purge, SP forwarder, health CE publisher (Topic 9)

✨ Enhancement 🧪 Tests 📝 Documentation ⚙️ Configuration changes 🕐 40+ Minutes

AI Description

• Add retry-topic processor that re-forwards queued requests on SP recovery with per-message
 deadlines.
• Add cancel handling that purges matching retry-topic messages and blocks late creates via
 deny-list.
• Extract SP forwarding into a dedicated forwarder supporting stable Idempotency-Key propagation.
• Publish health degraded/unavailable CloudEvents on provider health transitions; wire lifecycle in
 main.

Diagram

graph TD
  CP["Control Plane"] -->|"CE requests"| Router["Router"] -->|"healthy SP"| Forwarder["SP Forwarder"] -->|"HTTP/in-proc"| SP["Service Provider"]
  Router -->|"queue to retry"| RetryStream[("Retry Stream")]
  HealthMonitor(["Health Monitor"]) -->|"on transition"| RetryProcessor["Retry Processor"] -->|"drain & forward"| RetryStream
  RetryProcessor -->|"forward"| Forwarder
  HealthMonitor -->|"on transition"| CEPublisher(["Health CE Publisher"]) -->|"health CEs"| CP
  Router -. "share KeyLock/ResourceSet" .- RetryProcessor
  subgraph Legend
    direction LR
    _svc(["Service/Monitor"]) ~~~ _db[("Stream") ] ~~~ _mod["Module"]
  end

Loading

High-Level Assessment

The following are alternative approaches to this PR:

1. Distributed coordination (NATS KV / external lock) for in-flight guard

• ➕ Would prevent double-dispatch across multiple agent replicas
• ➕ Makes retry/main coordination robust under horizontal scaling
• ➖ Adds operational complexity and latency to every dispatch attempt
• ➖ Unnecessary if the agent remains single-instance per environment

2. Ack+republish instead of Nak-in-place for retry-topic messages

• ➕ Simplifies message movement model (always republish to reschedule)
• ➖ Resets delivery count and complicates any delivery-based observability/limits
• ➖ Adds extra publishes and increases stream churn compared to NakWithDelay

Recommendation: Given a single-instance agent, the PR’s in-process KeyLock/ResourceSet sharing between Router and Retry Processor is the right complexity/performance trade-off and directly targets main-vs-retry races. If/when the agent is scaled horizontally, revisit this design and move coordination to a distributed primitive.

Files changed (93) +15169 / -563

Enhancement (23) +3213 / -141

main.goWire messaging + routing + retry processor + health CE publisher +140/-24

Wire messaging + routing + retry processor + health CE publisher

• Builds and wires the messaging client, Router, Forwarder, Retry Processor, and Health CE publisher; adds startup ordering to run restart draining before live consumption.

cmd/environment-agent/main.go

builder.goAdd CloudEvent builder helpers +33/-0

Add CloudEvent builder helpers

• Adds helper constructors/utilities for producing consistent CloudEvents used by router/retry/health publishing.

internal/cloudevent/builder.go

publish.goAdd CloudEvent publish helper +33/-0

Add CloudEvent publish helper

• Centralizes CloudEvent publishing behavior and error propagation to callers.

internal/cloudevent/publish.go

types.goAdd CloudEvent type/subject constants +29/-0

Add CloudEvent type/subject constants

• Defines shared CE type and subject constants used across routing/messaging/health.

internal/cloudevent/types.go

client.goAdjust DCM client for messaging integration points +18/-14

Adjust DCM client for messaging integration points

• Updates DCM client wiring to work with the new messaging/lag-provider integration.

internal/dcm/client.go

registrar.goWire registrar to real consumer lag provider +74/-18

Wire registrar to real consumer lag provider

• Uses the messaging client as ConsumerLagProvider and updates registrar behavior accordingly.

internal/dcm/registrar.go

cepublisher.goPublish degraded/unavailable health CloudEvents on transitions +88/-0

Publish degraded/unavailable health CloudEvents on transitions

• Implements CEPublisher that resolves provider metadata and publishes health CloudEvents for Unhealthy/Unavailable transitions.

internal/health/cepublisher.go

monitor.goTrack serviceType and improve transition logging +42/-18

Track serviceType and improve transition logging

• Extends Monitor.RegisterProvider to include serviceType and adds structured logging around registration, checks, and transitions.

internal/health/monitor/monitor.go

panic.goAdd panic-to-error helper +0/-16

Add panic-to-error helper

• Introduces a helper for representing panic conditions in HTTP error handling.

internal/httperror/panic.go

problem.goAdd RFC7807-style problem helper +0/-22

Add RFC7807-style problem helper

• Introduces a helper type for structured problem responses.

internal/httperror/problem.go

client.goImplement JetStream messaging client (topics, consumers, publish) +637/-0

Implement JetStream messaging client (topics, consumers, publish)

• Adds a NATS/JetStream client with durable consumer setup, publish helpers (including Msg-Id), reconnect backoff, and controlled shutdown draining.

internal/messaging/client.go

handlers.goAdd main/cancel handler dispatch with deadlines and MaxDeliver guard +213/-0

Add main/cancel handler dispatch with deadlines and MaxDeliver guard

• Implements message dispatch with per-message handler context timeout, panic recovery, MaxDeliver terminal handling (main consumer), and consistent ack/nak behavior.

internal/messaging/handlers.go

topics.goDerive/validate main/cancel/retry subjects and names +146/-0

Derive/validate main/cancel/retry subjects and names

• Adds TopicNames derivation plus validation for NATS subjects and JetStream-safe stream/consumer names.

internal/messaging/topics.go

health_state.goHealth state tracking adjustments +3/-0

Health state tracking adjustments

• Small updates to provider health state representations used by routing/health subsystems.

internal/provider/health_state.go

provider.goProvider registry updates supporting routing resolution +8/-1

Provider registry updates supporting routing resolution

• Adjusts provider registry behavior used by routing resolution and retry processing.

internal/provider/provider.go

service.goPass serviceType into health monitor and adjust registration ordering +53/-21

Pass serviceType into health monitor and adjust registration ordering

• Updates provider service registration/monitor wiring to include serviceType and match new startup ordering constraints.

internal/provider/service/service.go

file.goEnhance file-backed provider store APIs and logging +56/-7

Enhance file-backed provider store APIs and logging

• Adjusts the file store constructor/signature and adds behavior needed by health CE publishing and routing lookups.

internal/provider/store/file.go

forwarder.goIntroduce SP forwarder with Idempotency-Key support +201/-0

Introduce SP forwarder with Idempotency-Key support

• Implements a forwarder abstraction handling external HTTP and embedded SPs, forwarding CE id as Idempotency-Key and logging safely.

internal/routing/forwarder.go

keylock.goAdd transient per-resource in-flight lock +47/-0

Add transient per-resource in-flight lock

• Introduces KeyLock to block concurrent forward attempts for the same resource ID without long-lived blocking.

internal/routing/keylock.go

resourceset.goIntroduce bounded LRU ResourceSet (deny-list/ledger) +110/-0

Introduce bounded LRU ResourceSet (deny-list/ledger)

• Adds a concurrent, bounded, LRU-evicting set used for deny-list and claimed-resources tracking.

internal/routing/resourceset.go

processor.goImplement retry processor (transition + restart flows) +651/-0

Implement retry processor (transition + restart flows)

• Adds Processor that fetches retry messages on provider transitions/restarts, dedups create/delete pairs, forwards with handler deadlines, and acks/naks appropriately.

internal/routing/retry/processor.go

router.goImplement Router with queueing, cancel purge, and retry/backoff forwarding +455/-0

Implement Router with queueing, cancel purge, and retry/backoff forwarding

• Introduces Router with decomposed request parsing, provider resolution, unhealthy queueing to retry topic, cancel deny-list + retry purge, and forward retry/backoff with in-flight locking.

internal/routing/router.go

types.goAdd routing interfaces and CE payload types +176/-0

Add routing interfaces and CE payload types

• Defines SPForwarder/Publisher/RetryTopicConsumer interfaces plus shared payload and error helpers used across router/retry/health.

internal/routing/types.go

Bug fix (1) +16 / -8

backoff.goMake backoff calculation overflow-safe +16/-8

Make backoff calculation overflow-safe

• Replaces float-based exponential backoff with integer doubling capped before overflow.

internal/backoff/backoff.go

Refactor (6) +83 / -7

server.goAdjust server implementation for new subsystem wiring +5/-2

Adjust server implementation for new subsystem wiring

• Small updates to align server behavior with newly introduced components/config.

internal/apiserver/server.go

backoff.goAlign DCM backoff usage with shared backoff changes +4/-1

Align DCM backoff usage with shared backoff changes

• Minor updates related to shared backoff behavior/usage.

internal/dcm/backoff.go

handler.goUpdate handler layer for new routing/messaging flow +11/-4

Update handler layer for new routing/messaging flow

• Small handler adjustments to match new routing/messaging data flow and error handling expectations.

internal/handler/handler.go

checker.goMinor checker-related update +2/-0

Minor checker-related update

• Small change to health checker wiring/definitions to support monitor changes.

internal/health/monitor/checker.go

statemachine.goState machine support update +13/-0

State machine support update

• Small additions supporting the updated monitor behavior and observability.

internal/health/monitor/statemachine.go

resolve.goExtract provider resolution helper +48/-0

Extract provider resolution helper

• Adds ResolveProvider helper used by Router and Retry Processor to determine provider and current health status.

internal/routing/resolve.go

Documentation (10) +2827 / -217

environment-agent.decisions.mdUpdate architecture decisions for routing/retry/cancel semantics +607/-9

Update architecture decisions for routing/retry/cancel semantics

• Documents new and amended design decisions related to retry processing, cancel/deny-list behavior, and concurrency controls.

.ai/decisions/environment-agent.decisions.md

environment-agent.spec.mdUpdate spec for Topic 9 routing/retry/health requirements +705/-143

Update spec for Topic 9 routing/retry/health requirements

• Extends specification content to cover retry processor behavior, deadlines/idempotency expectations, and health-event semantics (with deferred items noted).

.ai/specs/environment-agent.spec.md

2026-07-17-16-28-integration-tests.mdAdd integration test plan cases for messaging/routing/retry +903/-28

Add integration test plan cases for messaging/routing/retry

• Adds new integration-test plan entries and traceability updates for topic 7–9 behaviors.

.ai/test-plans/2026-07-17-16-28-integration-tests.md

2026-07-17-16-28-unit-tests.mdAdd unit test plan cases for routing/messaging primitives +429/-37

Add unit test plan cases for routing/messaging primitives

• Adds new unit-test plan entries and traceability updates for primitives and edge cases (topic validation, locks, etc.).

.ai/test-plans/2026-07-17-16-28-unit-tests.md

api.mdcAdd repository API development rules +25/-0

Add repository API development rules

• Adds Cursor rules to standardize API-related work in this repository.

.cursor/rules/api.mdc

commits.mdcAdd commit message rules +35/-0

Add commit message rules

• Adds Cursor rules for commit hygiene and formatting.

.cursor/rules/commits.mdc

go-patterns.mdcAdd Go implementation guidelines +49/-0

Add Go implementation guidelines

• Adds Cursor rules describing preferred Go patterns used in this codebase.

.cursor/rules/go-patterns.mdc

project.mdcAdd project-wide Cursor rules +33/-0

Add project-wide Cursor rules

• Adds Cursor project rules to align generated and authored changes with repo conventions.

.cursor/rules/project.mdc

testing.mdcAdd testing guidelines +39/-0

Add testing guidelines

• Adds Cursor rules for unit/integration testing conventions.

.cursor/rules/testing.mdc

doc.goAdd retry package documentation +2/-0

Add retry package documentation

• Adds package docs for retry processor subsystem.

internal/routing/retry/doc.go

Other (53) +9030 / -190

.gitignoreUpdate ignored artifacts +11/-3

Update ignored artifacts

• Adds/adjusts ignored files for local tooling and build/test artifacts.

.gitignore

MakefileMinor build/test target tweaks +2/-2

Minor build/test target tweaks

• Small Makefile adjustments to support updated build/test workflows.

Makefile

main_test.goUpdate main package tests for new wiring +87/-2

Update main package tests for new wiring

• Adjusts tests to reflect new startup behavior, dependencies, and invariants.

cmd/environment-agent/main_test.go

main_wiring_test.goAdd wiring test coverage for startup composition +120/-0

Add wiring test coverage for startup composition

• Adds targeted tests ensuring the runtime wiring between messaging/routing/retry/health is correct.

cmd/environment-agent/main_wiring_test.go

go.modAdd NATS + CloudEvents dependencies +17/-0

Add NATS + CloudEvents dependencies

• Introduces JetStream/NATS client/server libraries and CloudEvents SDK required for messaging and CE publishing.

go.mod

go.sumAdd checksums for new dependencies +41/-0

Add checksums for new dependencies

• Updates module checksums for newly introduced dependencies.

go.sum

server_integration_test.goUpdate apiserver integration tests +23/-5

Update apiserver integration tests

• Updates integration tests to match server wiring/behavior changes.

internal/apiserver/server_integration_test.go

backoff_test.goAdd backoff edge-case test coverage +23/-0

Add backoff edge-case test coverage

• Adds tests covering large attempts and boundary conditions for backoff calculation.

internal/backoff/backoff_test.go

builder_test.goTest CloudEvent builder helpers +47/-0

Test CloudEvent builder helpers

• Adds unit tests for CloudEvent construction helpers.

internal/cloudevent/builder_test.go

suite_test.goAdd cloudevent test suite bootstrap +13/-0

Add cloudevent test suite bootstrap

• Adds suite wiring for cloudevent package tests.

internal/cloudevent/suite_test.go

config.goAdd routing/messaging config + optional config-file merge +135/-7

Add routing/messaging config + optional config-file merge

• Introduces RoutingConfig, expands MessagingConfig, supports AGENT_CONFIG_FILE merge, and adds cross-field invariant validation for handler timeout vs ack wait.

internal/config/config.go

config_test.goTest config defaults, validation, and config-file loading +258/-19

Test config defaults, validation, and config-file loading

• Adds tests for new config fields, validation ranges, and file-vs-env precedence behavior.

internal/config/config_test.go

backoff_unit_test.goAdd DCM backoff unit tests +17/-0

Add DCM backoff unit tests

• Adds test coverage for DCM backoff behavior.

internal/dcm/backoff_unit_test.go

registrar_integration_test.goUpdate registrar integration tests +321/-27

Update registrar integration tests

• Adjusts integration tests for registrar changes and messaging-backed lag reporting.

internal/dcm/registrar_integration_test.go

cepublisher_test.goTest health CE publisher +237/-0

Test health CE publisher

• Adds unit tests covering type selection, store lookup failures, and publish error handling.

internal/health/cepublisher_test.go

health_integration_test.goUpdate health integration tests for CE publishing +33/-69

Update health integration tests for CE publishing

• Updates integration tests to account for health CE publication on transitions.

internal/health/health_integration_test.go

monitor_test.goUpdate monitor tests +234/-9

Update monitor tests

• Adds/updates tests for serviceType-aware registration and transition behavior.

internal/health/monitor/monitor_test.go

problem_unit_test.goUnit tests for problem helper +0/-28

Unit tests for problem helper

• Adds tests for the new problem response helper behavior.

internal/httperror/problem_unit_test.go

client_backoff_test.goTest messaging reconnect backoff behavior +78/-0

Test messaging reconnect backoff behavior

• Adds unit tests validating reconnect delay/backoff calculations.

internal/messaging/client_backoff_test.go

client_drain_test.goTest messaging Stop() drain behavior +65/-0

Test messaging Stop() drain behavior

• Adds tests ensuring in-flight handlers are drained before shutdown completes.

internal/messaging/client_drain_test.go

client_integration_test.goEnd-to-end tests for messaging client against NATS +1326/-0

End-to-end tests for messaging client against NATS

• Adds integration tests covering consumer setup, publish/consume flows, and shutdown behavior.

internal/messaging/client_integration_test.go

client_msgid_wire_test.goTest Msg-Id header wiring for dedup +104/-0

Test Msg-Id header wiring for dedup

• Verifies PublishWithMsgID sets JetStream deduplication headers correctly.

internal/messaging/client_msgid_wire_test.go

client_onsetupready_test.goTest OnSetupReady ordering semantics +190/-0

Test OnSetupReady ordering semantics

• Verifies deferred consumption starts only after setup callbacks run (used for restart draining).

internal/messaging/client_onsetupready_test.go

client_setup_test.goTest consumer/stream setup logic +108/-0

Test consumer/stream setup logic

• Covers durable consumer creation and startup-race tolerance for CP-owned stream existence.

internal/messaging/client_setup_test.go

handlers_test.goTest messaging handler ack/nak/panic behavior +769/-0

Test messaging handler ack/nak/panic behavior

• Adds extensive tests for handler execution, deadline handling, MaxDeliver, and error paths.

internal/messaging/handlers_test.go

router_integration_test.goIntegration tests for messaging ↔ router wiring +449/-0

Integration tests for messaging ↔ router wiring

• Validates that consumed messages invoke router handlers and produce expected outcomes.

internal/messaging/router_integration_test.go

suite_test.goAdd messaging test suite bootstrap +67/-0

Add messaging test suite bootstrap

• Adds suite wiring for messaging package tests.

internal/messaging/suite_test.go

topics_test.goTest topic derivation and validation +96/-0

Test topic derivation and validation

• Adds unit tests for topic naming rules and edge cases.

internal/messaging/topics_test.go

health_monitor_integration_test.goUpdate provider health monitor integration test coverage +3/-12

Update provider health monitor integration test coverage

• Updates integration tests to align with serviceType-aware health monitor registration.

internal/provider/health_monitor_integration_test.go

health_state_test.goTest provider health state changes +10/-0

Test provider health state changes

• Adds unit tests for provider health state behavior.

internal/provider/health_state_test.go

provider_integration_test.goUpdate provider integration tests +208/-3

Update provider integration tests

• Expands/updates integration tests for provider changes relied on by routing/retry.

internal/provider/provider_integration_test.go

provider_unit_test.goAdd/update provider unit tests +29/-0

Add/update provider unit tests

• Adds unit-level coverage for provider registry/service behaviors.

internal/provider/provider_unit_test.go

service_test.goUpdate provider service tests +557/-2

Update provider service tests

• Updates tests to reflect service registration and health monitor wiring changes.

internal/provider/service/service_test.go

file_durability_test.goAdd file store durability tests +91/-0

Add file store durability tests

• Adds tests validating persistence/durability expectations for the file-backed store.

internal/provider/store/file_durability_test.go

file_test.goUpdate file store unit tests +77/-0

Update file store unit tests

• Adds/updates unit tests for updated file store behavior.

internal/provider/store/file_test.go

suite_test.goAdd provider store test suite bootstrap +13/-0

Add provider store test suite bootstrap

• Adds suite wiring for provider store package tests.

internal/provider/store/suite_test.go

backoff_routing_test.goAdd routing retry/backoff tests +27/-0

Add routing retry/backoff tests

• Adds targeted tests for routing retry/backoff behavior on transient SP errors.

internal/routing/backoff_routing_test.go

concurrency_race_test.goAdd routing concurrency race tests +90/-0

Add routing concurrency race tests

• Adds tests covering main-topic vs retry-topic races and in-flight dispatch prevention.

internal/routing/concurrency_race_test.go

forwarder_test.goTest SP forwarder behavior +439/-0

Test SP forwarder behavior

• Adds tests for HTTP request construction, embedded routing, and error handling.

internal/routing/forwarder_test.go

keylock_test.goTest KeyLock primitive +37/-0

Test KeyLock primitive

• Adds unit tests for KeyLock AddIfAbsent/Remove/Len behavior.

internal/routing/keylock_test.go

resourceset_test.goTest ResourceSet behavior +71/-0

Test ResourceSet behavior

• Adds tests for LRU behavior, consume semantics, and set operations.

internal/routing/resourceset_test.go

deadline_integration_test.goRetry deadline integration tests +122/-0

Retry deadline integration tests

• Adds integration coverage ensuring handler timeout cancels hung forwarding during retry processing.

internal/routing/retry/deadline_integration_test.go

idempotency_integration_test.goRetry idempotency integration tests +115/-0

Retry idempotency integration tests

• Adds integration coverage validating stable Idempotency-Key behavior across retry forwards.

internal/routing/retry/idempotency_integration_test.go

inflight_integration_test.goRetry/main in-flight coordination integration tests +119/-0

Retry/main in-flight coordination integration tests

• Adds integration tests validating shared KeyLock prevents double-dispatch between router and retry processor.

internal/routing/retry/inflight_integration_test.go

nak_semantics_integration_test.goRetry Nak-in-place integration tests +125/-0

Retry Nak-in-place integration tests

• Adds integration tests asserting retry-topic messages are Nak’d in place rather than republished.

internal/routing/retry/nak_semantics_integration_test.go

processor_integration_test.goRetry processor integration tests +371/-0

Retry processor integration tests

• Adds end-to-end coverage for retry draining and transition-triggered processing.

internal/routing/retry/processor_integration_test.go

startup_integration_test.goRetry processor startup drain tests +191/-0

Retry processor startup drain tests

• Adds tests covering restart drain ordering and behavior before live consumption begins.

internal/routing/retry/startup_integration_test.go

suite_test.goAdd retry package test suite bootstrap +62/-0

Add retry package test suite bootstrap

• Adds suite wiring for retry processor tests.

internal/routing/retry/suite_test.go

router_publish_logging_test.goTest router publish logging/outcomes +91/-0

Test router publish logging/outcomes

• Adds tests ensuring CE publish results are logged consistently and safely.

internal/routing/router_publish_logging_test.go

routing_integration_test.goRouting integration test suite +926/-0

Routing integration test suite

• Adds broad integration coverage for create/delete/cancel flows including unhealthy queueing and retry paths.

internal/routing/routing_integration_test.go

helpers.goAdd routing test helpers and fakes +336/-0

Add routing test helpers and fakes

• Adds shared fakes/helpers used by routing and retry integration tests.

internal/routing/routingtest/helpers.go

suite_test.goAdd routing package test suite bootstrap +41/-0

Add routing package test suite bootstrap

• Adds suite wiring for routing tests.

internal/routing/suite_test.go

shutdown_test.goAdjust e2e shutdown test for new startup/shutdown sequence +8/-2

Adjust e2e shutdown test for new startup/shutdown sequence

• Updates e2e shutdown behavior expectations to match messaging drain and component lifecycle ordering.

test/e2e/httpserver/shutdown_test.go

[jordigilh]

Reviewed the full history across topics 7-9: all 4 Qodo bot findings, gciavarrini's spec-consistency pass, and jordigilh's test-coverage/design threads were addressed with concrete fixes and follow-up tests. Confirmed the key correctness fixes landed as described (lazy JetStream resolution via JSProvider, in-place Nak preserving NumDelivered on cancel purge, KeyLock/in-flight guard, Idempotency-Key forwarding). Left a note on the outstanding testing.mdc rewrite ask to track it as pending follow-up work, not a blocker. LGTM.

[gciavarrini]

Merging this PR.

@jordigilh i don't see the not on testing.mdc can you please double check? so we can track the work :)