Implement AI-side EventBridge handshake for email insight (PurposePath_Api #912 alignment)

[mottych]

Context

PurposePath_Api has completed and deployed the NotificationProcessor side for issue #912 with EventBridge-first AI kickoff + API fallback.

Backend references:

• Implementation PR: https://github.com/mottych/PurposePath_Api/pull/913
• Deploy/e2e stabilization PRs: https://github.com/mottych/PurposePath_Api/pull/914, https://github.com/mottych/PurposePath_Api/pull/915
• Deployed verification run (dev): https://github.com/mottych/PurposePath_Api/actions/runs/24195932995
• Contract spec update PR: https://github.com/mottych/PurposePath_Api/pull/916

Goal

Implement the AI-side async handshake so email insight requests can be consumed via EventBridge and resolved via existing job-status API semantics, while preserving canonical payload semantics across transports.

Required Handshake Contract (AI side)

1) EventBridge kickoff consumption

Consume EventBridge entries with routing keys:

• source = configured backend source (default purposepath.api)
• detail-type = configured kickoff type (default ai.job.requested)
• event bus = configured backend bus (default default)

Expected EventBridge detail payload shape (minimum):

{
  "eventId": "uuid",
  "occurredAtUtc": "2026-04-09T13:00:00.0000000Z",
  "sourceService": "PurposePath.NotificationProcessor.Lambda",
  "schemaVersion": "2.0",
  "correlationId": "corr-123",
  "idempotencyKey": "req-123:goal_created_email_insight:uuid",
  "retryAttempt": 0,
  "tenantId": "tenant-1",
  "userId": "user-1",
  "topicCategory": "email_insight",
  "topicId": "goal_created_email_insight",
  "eventSignal": "goal_created_email_insight",
  "locale": "en-US",
  "timezone": "UTC",
  "activityData": {
    "goal_id": "goal-1"
  },
  "authContext": {
    "serviceToken": "opaque-token",
    "expiresAtUtc": "2026-04-09T13:05:00Z",
    "issuer": "purposepath-api",
    "tokenType": "service_enrichment"
  },
  "jobId": "job-uuid",
  "eventType": "goal_created_email_insight",
  "kickoffTransport": "eventbridge"
}

2) Job status contract (must stay compatible)

Backend polls AI status endpoint and expects:

{
  "data": {
    "status": "completed|failed|cancelled|timed_out|running|queued",
    "result": {
      "schemaVersion": "1.0.0",
      "title": "...",
      "summary": "...",
      "blocks": [],
      "generationMeta": {}
    }
  }
}

Contract rules:

• data.status is required.
• data.result is required only when status=completed.
• completed with missing/empty result is treated by backend as terminal non-success (no fallback).

3) API fallback compatibility

Backend fallback kickoff still calls POST /api/v1/ai/execute-async with the same canonical envelope semantics.
Response should remain:

{ "data": { "jobId": "job-uuid" } }

(data.job_id may be tolerated but canonical is data.jobId.)

AI Implementation Requirements

1. Add EventBridge consumer for above routing keys and detail payload.
2. Use detail.jobId as the authoritative async job identity.
3. Persist and propagate correlation fields unchanged:

• correlationId
• idempotencyKey
• eventId

4. Ensure existing status API returns the exact wrapper shape expected by backend polling.
5. Ensure completed status returns valid result object matching email insight schema expectations.
6. Ensure enrichment requests use authContext.serviceToken as opaque bearer token.

Fallback/Behavior Expectations from backend side

Backend will fallback to API kickoff when EventBridge path cannot resolve quickly enough:

• EventBridge publish failure
• status request HTTP failure
• status payload missing data.status
• EventBridge await window timeout

Backend does NOT fallback on:

• completed without payload
• terminal failed/cancelled/timed_out status

Acceptance Criteria

• EventBridge kickoff request is consumed and processed for email_insight topics.
• Job lifecycle is visible via existing status endpoint using detail.jobId.
• Completed jobs return valid data.result payload compatible with current backend parser.
• correlationId/idempotencyKey/eventId preserved in telemetry and lifecycle records.
• End-to-end verification with PurposePath_Api: EventBridge kickoff path resolves without fallback for at least one goal_created_email_insight case.
• Fallback path remains operational and semantically identical.

Suggested Test Matrix

• Happy path: EventBridge kickoff -> status completed -> payload returned.
• EventBridge publish succeeds, status returns running/queued then completed.
• Status returns missing data.status (backend should fallback).
• Status returns terminal failed/cancelled/timed_out (backend should not fallback).
• completed with malformed result (backend should treat as no payload).

[mottych]

[cu] Insufficient Evidence

Please answer:
Can you provide specific file paths or log traces from PurposePath_AI to support the findings?
Is there a CloudWatch target endpoint or log group available for the AI service?

[mottych]

Thanks for the review. I agreed with the ambiguity gap and updated the contract spec to make routing identity defaults normative and drift-safe.

Spec update is in PurposePath_Api PR #916 (latest commit):

• Added normative defaults for EventBridge routing identity:
  • source = purposepath.api
  • detail-type = ai.job.requested
  • eventBusName = default
• Added environment baseline table (dev/staging/prod) with override policy.
• Added source-of-truth rule requiring API publisher config + AI rule config to change together and spec update in same change set.
• Added explicit mapping to current NotificationProcessor runtime defaults.

PR link:

• https://github.com/mottych/PurposePath_Api/pull/916

If you want, I can also add a short checklist issue in PurposePath_AI for infra owners to verify these three values in AI EventBridge rules per environment before rollout.

[mottych]

Update: the contract normalization is now merged in PurposePath_Api.

Merged PR: https://github.com/mottych/PurposePath_Api/pull/916

Spec now makes routing identity values normative (not just issue hints):

• source = purposepath.api
• detail-type = �i.job.requested
• eventBusName = default

Also added:

• environment baseline table for dev/staging/prod
• override policy + source-of-truth rule requiring API publisher and AI rule changes to stay coupled

Please use the merged spec as the authoritative handshake source for AI implementation work in this issue.

[mottych]

Implementation (branch feature/issue-302-eventbridge-email-kickoff)

Done

• EventBridge consumer: purposepath.api + ai.job.requested (configurable via AI_KICKOFF_EVENT_SOURCE / AI_KICKOFF_DETAIL_TYPE, defaults per email-insights spec §3.3.1).
• Detail model ApiAiJobRequestedDetail: canonical envelope + jobId, eventType, kickoffTransport; optional stage must match STAGE when present.
• Flow: put_if_absent job row with backend jobId → publish internal purposepath.ai / ai.job.created (same worker path as API kickoff).
• Idempotency / concurrency: claim_pending_job (conditional pending→processing) before _execute_job so duplicate EventBridge or double-trigger does not run two LLM executions.
• Pulumi: rule api-ai-job-requested-{stack} + EventBridge invoke permission.
• API contract §3.5–3.6: POST /ai/execute-async and GET /ai/jobs/{jobId} use camelCase (jobId, topicId, …) via response_model_by_alias=True; pending/processing exposed as queued / running.

Tests

• Unit: handler routing, ingest + publish, claim-lost path, status mapping; existing async job tests updated for claim/reload.
• uv run pytest on touched suites: 37 passed.
• mypy on changed modules: clean.

Follow-up for you

• Open PR → merge to dev, deploy Pulumi so the new rule exists in AWS.
• E2E with PurposePath_Api: confirm dev bus publishes with defaults; if multi-env shares an account, consider adding stage to Event detail and tightening the EventBridge pattern.

Label: in-progress — remove after merge/verify.

[mottych]

PR merge & deployment

• PR #303 merged to dev (feature). First Deploy Dev run failed mypy on AsyncJobData construction.
• PR #304 merged to dev (fix: model_validate with camelCase). Deploy Dev run 24205479491 succeeded: Ruff, MyPy, tests, Pulumi up (dev), smoke tests.

Pulumi / EventBridge (verified in AWS)

• Rule api-ai-job-requested-dev exists on the default bus and is ENABLED (account 380276784420, us-east-1).

E2E with PurposePath_Api (your side)

1. Trigger a real goal_created_email_insight (or publish purposepath.api / ai.job.requested with the spec detail + valid authContext.serviceToken).
2. Confirm Lambda logs show eventbridge.api_kickoff.accepted then normal async job completion.
3. Poll GET {AI_BASE}/api/v1/ai/jobs/{jobId} with the same user JWT + tenant the job was created under; expect data.status queued → running → completed and camelCase data.jobId / data.result when done.

Full cross-system validation requires a live service token from Api and matching tenant/user context; infra + deploy path is green.

Closing this issue as completed for the AI-side contract; reopen if e2e finds a gap.