feat: propagate audio metadata and idempotency

[murilloimparavel]

Summary\n- Carry stable CRM message IDs through debounce/recovery into A2A calls.\n- Retry only the processor's explicit in-progress conflict; do not repeat payload conflicts or expired claims.\n- Carry deterministic audio response notices through the pipeline and preserve pre-existing debounce recovery.\n\n## Validation\n- Targeted Go tests for AI adapter, debounce, and pipeline passed.\n\nDepends on evolution-foundation/evo-ai-processor-community#78. Merge after that PR so runtime retries can safely rely on processor idempotency.

Summary by Sourcery

Propagate deterministic message identity and audio notices through the pipeline while making idempotency retries safe and preserving recovery behavior.

New Features:

• Propagate stable CRM message IDs through debouncing, recovery, and A2A requests for idempotent processing.
• Carry validated audio response notices through debounce state and append them deterministically to dispatched AI responses.

Bug Fixes:

• Restrict retries to explicit active idempotency conflicts while treating payload mismatches and expired claims as permanent errors.
• Preserve debounce recovery by generating deterministic legacy idempotency keys when message IDs are unavailable.

Enhancements:

• Require inbound message IDs and enforce a maximum response-notice length.
• Persist, deduplicate, sort, and clear message IDs alongside pipeline state.

Tests:

• Add coverage for message ID propagation, selective idempotency retries, response-notice handling, debounce persistence, repository cleanup, validation, and legacy recovery.

[sourcery-ai]

Reviewer's Guide

The PR threads CRM message IDs through debounce persistence, restart recovery, and A2A JSON-RPC calls, deriving deterministic fallback keys for legacy state and limiting retries to explicit active-claim conflicts. It also persists and deduplicates audio response notices so they survive debounce/recovery and are appended to the final dispatch without altering AI input, with Redis storage cleanup and targeted validation/tests.

Sequence diagram for idempotent debounced A2A processing

sequenceDiagram
    participant CRM as CRM
    participant Pipeline as PipelineService
    participant Redis as RedisPipelineRepository
    participant Debounce as DebounceEngine
    participant Adapter as aiAdapter
    participant Processor as AIProcessor

    CRM->>Pipeline: Process(MessageEvent)
    Pipeline->>Debounce: Start or Reset(content, messageID)
    Debounce->>Redis: AppendMessageID(messageID)
    Pipeline->>Redis: GetMessageIDs()
    Redis-->>Pipeline: Sorted message IDs
    Pipeline->>Pipeline: stableMessageID(messageIDs)
    Pipeline->>Adapter: Call(A2ARequest.MessageID)
    Adapter->>Processor: JSON-RPC messageId
    Processor-->>Adapter: 200 response
    Adapter-->>Pipeline: NormalResponse

Loading

Sequence diagram for selective in-progress retry

sequenceDiagram
    participant Adapter as aiAdapter
    participant Processor as AIProcessor

    Adapter->>Processor: A2A JSON-RPC request with messageId
    Processor-->>Adapter: 409 Request already in progress
    Adapter->>Adapter: doOnce()
    Adapter->>Processor: Retry same idempotent request
    Processor-->>Adapter: 200 response

    Adapter->>Processor: A2A JSON-RPC request
    Processor-->>Adapter: 409 payload conflict or expired claim
    Adapter-->>Adapter: Return non-retryable error

Loading

Flow diagram for debounce recovery and response notices

flowchart TD
    Event[MessageEvent with messageID and responseNotice] --> Start[Start or Reset debounce]
    Start --> Persist[Persist message ID and response notice]
    Persist --> Recover{Restart recovery?}
    Recover -->|yes| Load[Load state and message IDs]
    Recover -->|no| Aggregate[Aggregate debounce buffers]
    Load --> Aggregate
    Aggregate --> Stable[Derive stable message ID]
    Stable --> AI[Call AI processor]
    AI --> Notice[appendResponseNotice to AI response]
    Notice --> Dispatch[Dispatch final response]
    Dispatch --> Clear[Clear state and message ID buffer]

Loading

File-Level Changes

Change	Details	Files
Propagate stable message identifiers into A2A requests and make retry behavior idempotency-aware.	Add message ID fields to A2A request and JSON-RPC message models. Forward CRM IDs through debounce storage, pipeline state, recovery, and AI-stage invocation. Retry only 409 responses explicitly indicating an active in-progress request; leave payload mismatches non-retryable. Generate deterministic legacy IDs when recovering state created before message ID persistence. Add adapter tests covering ID propagation and conflict classification.	pkg/ai/model/a2a.go pkg/ai/service/ai_adapter.go pkg/ai/service/ai_adapter_retry_test.go pkg/ai/service/ai_adapter_test.go pkg/debounce/service/debounce_engine.go pkg/debounce/service/debounce_engine_test.go pkg/pipeline/repository/pipeline_repository.go pkg/pipeline/repository/redis_pipeline_repository.go pkg/pipeline/repository/redis_pipeline_repository_test.go pkg/pipeline/service/pipeline_service.go pkg/pipeline/service/pipeline_service_test.go pkg/pipeline/model/pipeline.go pkg/pipeline/model/pipeline_test.go pkg/pipeline/handler/handler_test.go
Carry audio response notices through debounce and recovery, then append them deterministically to dispatched AI responses.	Validate optional notices at a 512-byte limit. Persist notices in pipeline state and retain them in in-memory entries across debounce resets and restart recovery. Deduplicate repeated notices and append them after AI content without including them in the AI input. Handle nil AI responses before dispatch cleanup. Add tests for single-turn dispatch, debounce deduplication, and legacy recovery behavior.	pkg/pipeline/model/pipeline.go pkg/pipeline/model/pipeline_test.go pkg/pipeline/service/pipeline_service.go pkg/pipeline/service/pipeline_service_test.go
Extend Redis debounce storage to track all message IDs across a debounce window and clean them up with pipeline state.	Store IDs in a Redis set with a bounded TTL. Return IDs in deterministic sorted order for stable hashing. Clear the ID set when pipeline state is cleared. Update debounce interfaces and implementations to persist IDs on start and reset. Add uniqueness, ordering, and cleanup coverage.	pkg/debounce/service/debounce_engine.go pkg/pipeline/repository/pipeline_repository.go pkg/pipeline/repository/redis_pipeline_repository.go pkg/pipeline/repository/redis_pipeline_repository_test.go pkg/debounce/service/debounce_engine_test.go pkg/pipeline/handler/handler_test.go pkg/pipeline/service/pipeline_service_test.go
Require inbound events to provide an idempotency-capable message ID.	Reject events without message_id during validation. Add handler coverage asserting missing IDs return HTTP 400. Update test fixtures and repository mocks for the expanded interfaces.	pkg/pipeline/model/pipeline.go pkg/pipeline/handler/handler_test.go pkg/pipeline/service/pipeline_service_test.go

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[sourcery-ai]

Hey - I've reviewed your changes and they look great!

Sourcery assessment

Needs a human reviewer. The change alters retry decisions for the AI Processor and carries stable IDs through persisted Redis debounce state. If the conflict classification or generated idempotency key is wrong, reverting will not undo a duplicate processor request or any external side effects it already caused; the stored debounce data itself is bounded and clearable.

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

[murilloimparavel]

Closing: this change is intended for the author's personal fork, not the evolution-foundation upstream. No merge performed.