feat(#305): EventBridge v2.4 terminals and service-token job polling

coaching/pulumi/__main__.py

@@ -255,21 +255,26 @@
     ),
 )
 
-# EventBridge access for publishing AI job events
+# EventBridge access for publishing AI job events (default + shared domain bus for v2.4 terminals)
 aws.iam.RolePolicy(
     "coaching-eventbridge-policy",
     role=lambda_role.id,
-    policy=json.dumps(
-        {
-            "Version": "2012-10-17",
-            "Statement": [
-                {
-                    "Effect": "Allow",
-                    "Action": ["events:PutEvents"],
-                    "Resource": ["arn:aws:events:us-east-1:*:event-bus/default"],
-                }
-            ],
-        }
+    policy=pulumi.Output.all(aws.get_caller_identity().account_id).apply(
+        lambda args: json.dumps(
+            {
+                "Version": "2012-10-17",
+                "Statement": [
+                    {
+                        "Effect": "Allow",
+                        "Action": ["events:PutEvents"],
+                        "Resource": [
+                            f"arn:aws:events:us-east-1:{args[0]}:event-bus/default",
+                            f"arn:aws:events:us-east-1:{args[0]}:event-bus/purposepath-domain-events-{stack}",
+                        ],
+                    }
+                ],
+            }
+        )
     ),
 )
 
@@ -539,7 +544,7 @@
         "api-ai-job-requested-rule",
         name=f"api-ai-job-requested-{stack}",
         description="Triggers coaching Lambda when Api publishes ai.job.requested (email_insight kickoff)",
-        event_bus_name="default",
+        event_bus_name=f"purposepath-domain-events-{stack}",
         event_pattern=json.dumps(
             {
                 "source": ["purposepath.api"],
@@ -553,7 +558,7 @@
         "api-ai-job-requested-target",
         rule=api_ai_job_requested_rule.name,
         arn=coaching_lambda.arn,
-        event_bus_name="default",
+        event_bus_name=f"purposepath-domain-events-{stack}",
     )
 
     aws.lambda_.Permission(

coaching/src/api/auth.py

@@ -340,6 +340,79 @@ async def get_current_user(authorization: str = Header(...)) -> UserContext:
         raise HTTPException(status_code=401, detail="Token validation failed") from e
 
 
+async def get_tenant_for_async_job_access(
+    authorization: str | None = Header(None),
+) -> str:
+    """Resolve tenant for async job polling per email-insights v2.4 §4.6 (user JWT or service token)."""
+    if not authorization or not authorization.startswith("Bearer "):
+        logger.warning("async_job_auth.missing_bearer")
+        raise HTTPException(
+            status_code=401,
+            detail="Missing authorization header",
+        )
+
+    token = authorization.split(" ", 1)[1].strip()
+
+    if token == "test_token":
+        return "tenant_test"
+
+    jwt_signing_key = _get_jwt_secret()
+    try:
+        try:
+            payload = jwt.decode(
+                token,
+                jwt_signing_key,
+                algorithms=[settings.jwt_algorithm],
+                options={
+                    "verify_aud": settings.stage != "dev",
+                    "verify_iss": settings.stage != "dev",
+                },
+                issuer=None if settings.stage == "dev" else settings.jwt_issuer,
+                audience=None if settings.stage == "dev" else settings.jwt_audience,
+            )
+        except JWTError as jwt_err:
+            if settings.stage == "dev":
+                payload = jwt.decode(
+                    token,
+                    "change-me-in-prod",
+                    algorithms=[settings.jwt_algorithm],
+                    options={"verify_aud": False, "verify_iss": False},
+                )
+            else:
+                logger.warning(f"async_job_auth.jwt_invalid: {jwt_err}")
+                raise HTTPException(
+                    status_code=401,
+                    detail="Invalid or expired token",
+                ) from jwt_err
+    except JWTError as e:
+        logger.warning(f"async_job_auth.jwt_decode_failed: {e}")
+        raise HTTPException(status_code=401, detail="Invalid or expired token") from e
+
+    token_type = payload.get("token_type") or payload.get("tokenType")
+    role = payload.get("role")
+    tenant_id = payload.get("tenant_id")
+
+    if token_type == "service_enrichment" and role == "service" and tenant_id:
+        logger.info(
+            f"async_job_auth.service_token_accepted tenant_id={tenant_id!s} "
+            f"issuer={payload.get('iss')!s}"
+        )
+        return str(tenant_id)
+
+    user_id = payload.get("user_id") or payload.get("sub")
+    if user_id and tenant_id:
+        return str(tenant_id)
+
+    logger.warning(
+        f"async_job_auth.claims_rejected has_tenant={bool(tenant_id)} "
+        f"has_user={bool(user_id)} token_type={token_type!r} role={role!r}"
+    )
+    raise HTTPException(
+        status_code=401,
+        detail="Token missing required fields for async job access",
+    )
+
+
 async def get_optional_context(
     authorization: str | None = Header(None),
 ) -> RequestContext | None:

coaching/src/api/dependencies/async_execution.py

@@ -48,7 +48,8 @@ async def get_event_publisher() -> EventBridgePublisher:
     if _event_publisher is None:
         _event_publisher = EventBridgePublisher(
             region_name=settings.aws_region,
-            event_bus_name="default",  # Using default EventBridge bus
+            event_bus_name="default",
+            domain_event_bus_name=settings.resolved_domain_event_bus_name,
             source="purposepath.ai",
             stage=settings.stage,
             enabled=settings.ai_async_jobs_enabled,
@@ -57,6 +58,7 @@ async def get_event_publisher() -> EventBridgePublisher:
             "EventBridgePublisher initialized",
             source="purposepath.ai",
             stage=settings.stage,
+            domain_event_bus=settings.resolved_domain_event_bus_name,
         )
 
     return _event_publisher

coaching/src/api/models/ai_job_kickoff.py

@@ -18,6 +18,7 @@ class ApiAiJobRequestedDetail(BaseModel):
     model_config = ConfigDict(populate_by_name=True, extra="forbid")
 
     event_id: str = Field(alias="eventId")
+    request_id: str = Field(alias="requestId", min_length=1)
     occurred_at_utc: datetime = Field(alias="occurredAtUtc")
     source_service: str = Field(alias="sourceService")
     schema_version: str = Field(alias="schemaVersion")

coaching/src/api/models/async_ai.py

@@ -75,6 +75,7 @@ class AsyncAIRequest(BaseModel):
 
     # Generic v2 backend-triggered request envelope (optional for backwards compatibility)
     event_id: str | None = Field(default=None, alias="eventId")
+    request_id: str | None = Field(default=None, alias="requestId")
     occurred_at_utc: datetime | None = Field(default=None, alias="occurredAtUtc")
     source_service: str | None = Field(default=None, alias="sourceService")
     schema_version: str | None = Field(default=None, alias="schemaVersion")
@@ -105,6 +106,7 @@ def validate_v2_contract_requirements(self) -> "AsyncAIRequest":
 
         required_fields = {
             "eventId": self.event_id,
+            "requestId": self.request_id,
             "occurredAtUtc": self.occurred_at_utc,
             "sourceService": self.source_service,
             "schemaVersion": self.schema_version,
@@ -140,6 +142,7 @@ def validate_v2_contract_requirements(self) -> "AsyncAIRequest":
                 },
                 {
                     "eventId": "evt-123",
+                    "requestId": "req-123",
                     "occurredAtUtc": "2026-03-27T16:00:00Z",
                     "sourceService": "PurposePath_Api",
                     "schemaVersion": "2.0",

coaching/src/api/routes/ai_execute_async.py

@@ -11,7 +11,7 @@
 """
 
 import structlog
-from coaching.src.api.auth import get_current_user
+from coaching.src.api.auth import get_current_user, get_tenant_for_async_job_access
 from coaching.src.api.dependencies.async_execution import get_async_execution_service
 from coaching.src.api.models.async_ai import (
     AsyncAIRequest,
@@ -144,6 +144,8 @@ async def execute_async(
         user_id = str(request_body.user_id)
         parameters = request_body.activity_data or {}
         jwt_token = request_body.auth_context.service_token if request_body.auth_context else None
+        if jwt_token is None and authorization and authorization.startswith("Bearer "):
+            jwt_token = authorization.split(" ", 1)[1].strip()
     else:
         if user is None:
             raise HTTPException(
@@ -178,6 +180,10 @@ async def execute_async(
             correlation_id=request_body.correlation_id if contract_v2 else None,
             idempotency_key=request_body.idempotency_key if contract_v2 else None,
             event_id=request_body.event_id if contract_v2 else None,
+            request_id=request_body.request_id if contract_v2 else None,
+            topic_category=request_body.topic_category if contract_v2 else None,
+            event_signal=request_body.event_signal if contract_v2 else None,
+            kickoff_transport="api" if contract_v2 else None,
         )
 
         logger.info(
@@ -288,7 +294,7 @@ async def get_job_status(
         description="Unique job identifier",
         examples=["550e8400-e29b-41d4-a716-446655440000"],
     ),
-    user: UserContext = Depends(get_current_user),
+    tenant_id: str = Depends(get_tenant_for_async_job_access),
     service: AsyncAIExecutionService = Depends(get_async_execution_service),
 ) -> JobStatusResponse:
     """Get job status by ID.
@@ -298,7 +304,7 @@ async def get_job_status(
 
     Args:
         job_id: Unique job identifier
-        user: Authenticated user context from JWT
+        tenant_id: Tenant from user JWT or service token (v2.4 §4.6)
         service: Async execution service from DI
 
     Returns:
@@ -307,9 +313,6 @@ async def get_job_status(
     Raises:
         HTTPException: 404 if job not found or tenant mismatch
     """
-    # Extract tenant from authenticated context
-    tenant_id = user.tenant_id
-
     try:
         job = await service.get_job(job_id=job_id, tenant_id=tenant_id)
 

coaching/src/api/routes/coaching_sessions.py

@@ -222,7 +222,8 @@ async def get_job_repository() -> DynamoDBJobRepository:
 async def get_event_publisher() -> EventBridgePublisher:
     """Get EventBridge publisher instance."""
     return EventBridgePublisher(
-        region_name="us-east-1",
+        region_name=settings.aws_region,
+        domain_event_bus_name=settings.resolved_domain_event_bus_name,
         stage=settings.stage,
         enabled=settings.ai_async_jobs_enabled,
     )

coaching/src/core/config_multitenant.py

@@ -173,6 +173,19 @@ def parse_cors_origins(cls, v: Any) -> list[str]:
     ai_kickoff_detail_type: str = Field(
         default="ai.job.requested", validation_alias="AI_KICKOFF_DETAIL_TYPE"
     )
+    # Shared domain bus for email-insight kickoff + terminal events (spec v2.4 §3.3.1 / §3.7)
+    domain_event_bus_name: str = Field(
+        default="",
+        validation_alias="DOMAIN_EVENT_BUS_NAME",
+        description="Override; default purposepath-domain-events-{STAGE}",
+    )
+
+    @property
+    def resolved_domain_event_bus_name(self) -> str:
+        """EventBridge bus for Api kickoff consumption and AI terminal events."""
+        if self.domain_event_bus_name.strip():
+            return self.domain_event_bus_name.strip()
+        return f"purposepath-domain-events-{self.stage}"
 
     # LLM Configuration
     llm_temperature: float = 0.7

coaching/src/domain/entities/ai_job.py

@@ -137,7 +137,23 @@ class AIJob(BaseModel):
     )
     event_id: str | None = Field(
         default=None,
-        description="Upstream event/request identifier",
+        description="Kickoff message identity (event lineage); maps to terminal kickoffEventId",
+    )
+    request_id: str | None = Field(
+        default=None,
+        description="Notification request identity for terminal correlation (spec v2.4)",
+    )
+    topic_category: str | None = Field(
+        default=None,
+        description="Topic category from kickoff/API contract (e.g. email_insight)",
+    )
+    event_signal: str | None = Field(
+        default=None,
+        description="Business event signal from kickoff/API contract",
+    )
+    kickoff_transport: str | None = Field(
+        default=None,
+        description="eventbridge (Api EB kickoff) vs api (POST execute-async); domain terminals only for eventbridge",
     )
     status: AIJobStatus = Field(
         default=AIJobStatus.PENDING,

coaching/src/infrastructure/repositories/dynamodb_job_repository.py

@@ -362,6 +362,14 @@ def _to_dynamodb_item(self, job: AIJob) -> dict[str, Any]:
             item["idempotency_key"] = job.idempotency_key
         if job.event_id is not None:
             item["event_id"] = job.event_id
+        if job.request_id is not None:
+            item["request_id"] = job.request_id
+        if job.topic_category is not None:
+            item["topic_category"] = job.topic_category
+        if job.event_signal is not None:
+            item["event_signal"] = job.event_signal
+        if job.kickoff_transport is not None:
+            item["kickoff_transport"] = job.kickoff_transport
 
         if job.result is not None:
             item["result"] = job.result
@@ -408,6 +416,10 @@ def _from_dynamodb_item(self, item: dict[str, Any]) -> AIJob:
             correlation_id=item.get("correlation_id"),
             idempotency_key=item.get("idempotency_key"),
             event_id=item.get("event_id"),
+            request_id=item.get("request_id"),
+            topic_category=item.get("topic_category"),
+            event_signal=item.get("event_signal"),
+            kickoff_transport=item.get("kickoff_transport"),
             status=AIJobStatus(item["status"]),
             result=item.get("result"),
             error=item.get("error"),