diff --git a/coaching/pulumi/__main__.py b/coaching/pulumi/__main__.py index 566716cf..2133a165 100644 --- a/coaching/pulumi/__main__.py +++ b/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( diff --git a/coaching/src/api/auth.py b/coaching/src/api/auth.py index 15bc0a02..228618c6 100644 --- a/coaching/src/api/auth.py +++ b/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: diff --git a/coaching/src/api/dependencies/async_execution.py b/coaching/src/api/dependencies/async_execution.py index 4cd15b8b..91bfa1ce 100644 --- a/coaching/src/api/dependencies/async_execution.py +++ b/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 diff --git a/coaching/src/api/models/ai_job_kickoff.py b/coaching/src/api/models/ai_job_kickoff.py index 4a098e3d..20f7b783 100644 --- a/coaching/src/api/models/ai_job_kickoff.py +++ b/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") diff --git a/coaching/src/api/models/async_ai.py b/coaching/src/api/models/async_ai.py index 2d69ab0f..bb401589 100644 --- a/coaching/src/api/models/async_ai.py +++ b/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", diff --git a/coaching/src/api/routes/ai_execute_async.py b/coaching/src/api/routes/ai_execute_async.py index f702b276..9f03125a 100644 --- a/coaching/src/api/routes/ai_execute_async.py +++ b/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) diff --git a/coaching/src/api/routes/coaching_sessions.py b/coaching/src/api/routes/coaching_sessions.py index 5c1f8bbd..4a7048c2 100644 --- a/coaching/src/api/routes/coaching_sessions.py +++ b/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, ) diff --git a/coaching/src/core/config_multitenant.py b/coaching/src/core/config_multitenant.py index b12fd671..f1ac4818 100644 --- a/coaching/src/core/config_multitenant.py +++ b/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 diff --git a/coaching/src/domain/entities/ai_job.py b/coaching/src/domain/entities/ai_job.py index fe707474..99733f52 100644 --- a/coaching/src/domain/entities/ai_job.py +++ b/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, diff --git a/coaching/src/infrastructure/repositories/dynamodb_job_repository.py b/coaching/src/infrastructure/repositories/dynamodb_job_repository.py index 085311f7..87de5416 100644 --- a/coaching/src/infrastructure/repositories/dynamodb_job_repository.py +++ b/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"), diff --git a/coaching/src/services/async_execution_service.py b/coaching/src/services/async_execution_service.py index b786ea76..5cd837f4 100644 --- a/coaching/src/services/async_execution_service.py +++ b/coaching/src/services/async_execution_service.py @@ -7,6 +7,7 @@ from __future__ import annotations import time +from datetime import UTC, datetime from decimal import Decimal from typing import Any from uuid import uuid4 @@ -124,6 +125,10 @@ def _validate_and_build_pending_job( correlation_id: str | None, idempotency_key: str | None, event_id: str | None, + request_id: str | None = None, + topic_category: str | None = None, + event_signal: str | None = None, + kickoff_transport: str | None = None, ) -> AIJob: """Validate topic/params and build a pending AIJob (no persistence).""" endpoint = get_topic_by_topic_id(topic_id) @@ -159,6 +164,10 @@ def _validate_and_build_pending_job( correlation_id=correlation_id, idempotency_key=idempotency_key, event_id=event_id, + request_id=request_id, + topic_category=topic_category, + event_signal=event_signal, + kickoff_transport=kickoff_transport, status=AIJobStatus.PENDING, estimated_duration_ms=estimated_duration, ) @@ -179,6 +188,87 @@ def _publish_job_created_trigger(self, job: AIJob) -> None: event_id=job.event_id, ) + @staticmethod + def _should_publish_email_insight_domain_terminal(job: AIJob) -> bool: + """v2.4 domain-bus terminals only for Api EventBridge kickoff (issue #305).""" + return ( + job.request_id is not None + and job.topic_category == TopicCategory.EMAIL_INSIGHT.value + and job.event_id is not None + and job.kickoff_transport == "eventbridge" + ) + + def _publish_async_job_completed_terminal( + self, + job: AIJob, + *, + result_dict: dict[str, Any], + processing_time_ms: int, + ) -> None: + """Publish completion: v2.4 terminal to domain bus or legacy WebSocket-oriented event.""" + if self._should_publish_email_insight_domain_terminal(job): + self._publisher.publish_email_insight_terminal_v24( + terminal_status="completed", + terminal_event_id=str(uuid4()), + occurred_at_utc=datetime.now(UTC), + job_id=job.job_id, + request_id=job.request_id or "", + kickoff_event_id=job.event_id or "", + tenant_id=job.tenant_id, + user_id=job.user_id, + correlation_id=job.correlation_id or "", + idempotency_key=job.idempotency_key or "", + topic_category=job.topic_category or TopicCategory.EMAIL_INSIGHT.value, + topic_id=job.topic_id, + event_signal=job.event_signal or job.topic_id, + data={"result": result_dict}, + ) + return + self._publisher.publish_ai_job_completed( + job_id=job.job_id, + tenant_id=job.tenant_id, + user_id=job.user_id, + topic_id=job.topic_id, + result=result_dict, + processing_time_ms=processing_time_ms, + ) + + def _publish_async_job_failed_terminal( + self, + job: AIJob, + *, + error: str, + error_code: AIJobErrorCode, + _processing_time_ms: int, + ) -> None: + """Publish failure: v2.4 terminal or legacy failed event.""" + if self._should_publish_email_insight_domain_terminal(job): + self._publisher.publish_email_insight_terminal_v24( + terminal_status="failed", + terminal_event_id=str(uuid4()), + occurred_at_utc=datetime.now(UTC), + job_id=job.job_id, + request_id=job.request_id or "", + kickoff_event_id=job.event_id or "", + tenant_id=job.tenant_id, + user_id=job.user_id, + correlation_id=job.correlation_id or "", + idempotency_key=job.idempotency_key or "", + topic_category=job.topic_category or TopicCategory.EMAIL_INSIGHT.value, + topic_id=job.topic_id, + event_signal=job.event_signal or job.topic_id, + data={"errorCode": error_code.value, "error": error}, + ) + return + self._publisher.publish_ai_job_failed( + job_id=job.job_id, + tenant_id=job.tenant_id, + user_id=job.user_id, + topic_id=job.topic_id, + error=error, + error_code=error_code.value, + ) + async def ingest_api_job_requested_event(self, detail: ApiAiJobRequestedDetail) -> None: """Persist job from PurposePath_Api EventBridge kickoff and trigger async execution (#302).""" if detail.stage is not None and detail.stage != settings.stage: @@ -200,6 +290,10 @@ async def ingest_api_job_requested_event(self, detail: ApiAiJobRequestedDetail) correlation_id=detail.correlation_id, idempotency_key=detail.idempotency_key, event_id=detail.event_id, + request_id=detail.request_id, + topic_category=detail.topic_category, + event_signal=detail.event_signal, + kickoff_transport="eventbridge", ) inserted = await self._repository.put_if_absent(job) @@ -258,6 +352,10 @@ async def create_job( correlation_id: str | None = None, idempotency_key: str | None = None, event_id: str | None = None, + request_id: str | None = None, + topic_category: str | None = None, + event_signal: str | None = None, + kickoff_transport: str | None = None, ) -> AIJob: """Create and validate a new async AI job. @@ -290,6 +388,10 @@ async def create_job( correlation_id=correlation_id, idempotency_key=idempotency_key, event_id=event_id, + request_id=request_id, + topic_category=topic_category, + event_signal=event_signal, + kickoff_transport=kickoff_transport, ) await self._repository.save(job) @@ -515,20 +617,19 @@ async def _execute_job(self, job: AIJob, *, already_processing: bool = False) -> processing_time_ms=processing_time_ms, ) - # Publish completed event + # Publish completed event (v2.4 domain terminal or legacy) try: - self._publisher.publish_ai_job_completed( - job_id=job.job_id, - tenant_id=job.tenant_id, - user_id=job.user_id, - topic_id=job.topic_id, - result=result_dict, + self._publish_async_job_completed_terminal( + job, + result_dict=result_dict, processing_time_ms=processing_time_ms, ) except EventBridgePublishError as e: logger.warning( "async_job.completed_event_failed", job_id=job.job_id, + request_id=job.request_id, + idempotency_key=job.idempotency_key, error=str(e), ) @@ -619,20 +720,20 @@ async def _handle_failure( processing_time_ms=processing_time_ms, ) - # Publish failed event + # Publish failed event (v2.4 domain terminal or legacy) try: - self._publisher.publish_ai_job_failed( - job_id=job.job_id, - tenant_id=job.tenant_id, - user_id=job.user_id, - topic_id=job.topic_id, + self._publish_async_job_failed_terminal( + job, error=error, - error_code=error_code.value, + error_code=error_code, + _processing_time_ms=processing_time_ms, ) except EventBridgePublishError as e: logger.warning( "async_job.failed_event_failed", job_id=job.job_id, + request_id=job.request_id, + idempotency_key=job.idempotency_key, error=str(e), ) diff --git a/coaching/tests/unit/api/handlers/test_eventbridge_handler.py b/coaching/tests/unit/api/handlers/test_eventbridge_handler.py index 0c4b5fdb..0b598b1c 100644 --- a/coaching/tests/unit/api/handlers/test_eventbridge_handler.py +++ b/coaching/tests/unit/api/handlers/test_eventbridge_handler.py @@ -191,6 +191,7 @@ def test_routes_api_ai_job_requested(self) -> None: "detail-type": "ai.job.requested", "detail": { "eventId": "e1", + "requestId": "req-1", "occurredAtUtc": now, "sourceService": "PurposePath.NotificationProcessor.Lambda", "schemaVersion": "2.0", diff --git a/coaching/tests/unit/api/models/test_async_ai.py b/coaching/tests/unit/api/models/test_async_ai.py index a396584f..535a8c64 100644 --- a/coaching/tests/unit/api/models/test_async_ai.py +++ b/coaching/tests/unit/api/models/test_async_ai.py @@ -25,6 +25,7 @@ def test_v2_request_is_valid(self) -> None: """V2 backend-triggered payload validates with required fields.""" request = AsyncAIRequest( eventId="evt-1", + requestId="req-1", occurredAtUtc=datetime.now(UTC), sourceService="PurposePath_Api", schemaVersion="2.0", @@ -71,6 +72,7 @@ def test_v2_missing_service_token_fails(self) -> None: with pytest.raises(ValidationError): AsyncAIRequest( eventId="evt-1", + requestId="req-1", occurredAtUtc=datetime.now(UTC), sourceService="PurposePath_Api", schemaVersion="2.0", @@ -97,6 +99,7 @@ def test_v2_expired_service_token_fails(self) -> None: with pytest.raises(ValidationError, match="expired"): AsyncAIRequest( eventId="evt-1", + requestId="req-1", occurredAtUtc=datetime.now(UTC), sourceService="PurposePath_Api", schemaVersion="2.0", @@ -124,6 +127,7 @@ def test_v2_invalid_token_type_fails(self) -> None: with pytest.raises(ValidationError, match="service_enrichment"): AsyncAIRequest( eventId="evt-1", + requestId="req-1", occurredAtUtc=datetime.now(UTC), sourceService="PurposePath_Api", schemaVersion="2.0", diff --git a/coaching/tests/unit/api/routes/test_ai_execute_async.py b/coaching/tests/unit/api/routes/test_ai_execute_async.py index fd89f862..143027be 100644 --- a/coaching/tests/unit/api/routes/test_ai_execute_async.py +++ b/coaching/tests/unit/api/routes/test_ai_execute_async.py @@ -70,6 +70,7 @@ def test_v2_flow_uses_payload_context_and_service_token( "/api/v1/ai/execute-async", json={ "eventId": "evt-1", + "requestId": "req-1", "occurredAtUtc": datetime.now(UTC).isoformat(), "sourceService": "PurposePath_Api", "schemaVersion": "2.0", @@ -103,6 +104,8 @@ def test_v2_flow_uses_payload_context_and_service_token( assert call_kwargs["correlation_id"] == "corr-1" assert call_kwargs["idempotency_key"] == "idem-1" assert call_kwargs["event_id"] == "evt-1" + assert call_kwargs["request_id"] == "req-1" + assert call_kwargs["kickoff_transport"] == "api" finally: app.dependency_overrides.clear() @@ -133,6 +136,7 @@ def test_v2_without_service_token_fails_validation( "/api/v1/ai/execute-async", json={ "eventId": "evt-1", + "requestId": "req-1", "occurredAtUtc": datetime.now(UTC).isoformat(), "sourceService": "PurposePath_Api", "schemaVersion": "2.0", diff --git a/coaching/tests/unit/services/test_async_execution_service.py b/coaching/tests/unit/services/test_async_execution_service.py index 5576cd35..1bd97a24 100644 --- a/coaching/tests/unit/services/test_async_execution_service.py +++ b/coaching/tests/unit/services/test_async_execution_service.py @@ -427,6 +427,7 @@ async def test_ingest_api_job_requested_inserts_and_publishes( detail = ApiAiJobRequestedDetail( event_id="evt-1", + request_id="req-kickoff-1", occurred_at_utc=datetime.now(UTC), source_service="PurposePath.NotificationProcessor.Lambda", schema_version="2.0", @@ -460,6 +461,9 @@ async def test_ingest_api_job_requested_inserts_and_publishes( assert written.correlation_id == "corr-1" assert written.idempotency_key == "idem-1" assert written.event_id == "evt-1" + assert written.request_id == "req-kickoff-1" assert written.jwt_token == "svc" mock_eventbridge.publish_ai_job_created.assert_called_once() - assert mock_eventbridge.publish_ai_job_created.call_args.kwargs["job_id"] == "backend-job-id-1" + assert ( + mock_eventbridge.publish_ai_job_created.call_args.kwargs["job_id"] == "backend-job-id-1" + ) diff --git a/coaching/tests/unit/shared/test_eventbridge_client.py b/coaching/tests/unit/shared/test_eventbridge_client.py index 8db91b0e..99d8ef20 100644 --- a/coaching/tests/unit/shared/test_eventbridge_client.py +++ b/coaching/tests/unit/shared/test_eventbridge_client.py @@ -92,3 +92,45 @@ def test_publish_ai_job_created_includes_trace_metadata(monkeypatch: Any) -> Non assert '"correlationId": "corr-1"' in detail assert '"idempotencyKey": "idem-1"' in detail assert '"eventId": "evt-1"' in detail + + +def test_publish_email_insight_terminal_v24_uses_domain_bus(monkeypatch: Any) -> None: + """v2.4 terminal events must use flat detail JSON on the domain bus.""" + from datetime import UTC, datetime + + dummy = _DummyClient() + monkeypatch.setattr( + "shared.services.eventbridge_client.get_eventbridge_client", + lambda *_args, **_kwargs: dummy, + ) + + publisher = EventBridgePublisher( + enabled=True, + stage="dev", + event_bus_name="default", + domain_event_bus_name="purposepath-domain-events-dev", + ) + publisher.publish_email_insight_terminal_v24( + terminal_status="completed", + terminal_event_id="term-1", + occurred_at_utc=datetime.now(UTC), + job_id="job-1", + request_id="req-1", + kickoff_event_id="kick-1", + tenant_id="t1", + user_id="u1", + correlation_id="c1", + idempotency_key="i1", + topic_category="email_insight", + topic_id="goal_created_email_insight", + event_signal="goal_created_email_insight", + data={"result": {"schemaVersion": "1.0.0"}}, + ) + + assert len(dummy.entries) == 1 + assert dummy.entries[0]["EventBusName"] == "purposepath-domain-events-dev" + assert dummy.entries[0]["DetailType"] == "ai.job.completed" + detail_json = dummy.entries[0]["Detail"] + assert '"schemaVersion": "2.4"' in detail_json + assert '"executionMode": "eventbridge_terminal"' in detail_json + assert '"requestId": "req-1"' in detail_json diff --git a/security/.secrets.baseline b/security/.secrets.baseline index 050489f5..04fda01a 100644 --- a/security/.secrets.baseline +++ b/security/.secrets.baseline @@ -90,6 +90,10 @@ { "path": "detect_secrets.filters.allowlist.is_line_allowlisted" }, + { + "path": "detect_secrets.filters.common.is_baseline_file", + "filename": "security/.secrets.baseline" + }, { "path": "detect_secrets.filters.common.is_ignored_due_to_verification_policies", "min_level": 2 @@ -129,30 +133,50 @@ } ], "results": { - "coaching/src/api/models/async_ai.py": [ + "coaching\\src\\api\\auth.py": [ + { + "type": "Secret Keyword", + "filename": "coaching\\src\\api\\auth.py", + "hashed_secret": "ce094fa09693604fb88de28e4876f8c38a5548d3", + "is_verified": false, + "line_number": 111, + "is_added": false, + "is_removed": false + }, + { + "type": "Secret Keyword", + "filename": "coaching\\src\\api\\auth.py", + "hashed_secret": "c03d694a32c7d9c2fc31c4826927eaa686a97c98", + "is_verified": false, + "line_number": 359, + "is_added": false, + "is_removed": false + } + ], + "coaching\\src\\api\\models\\async_ai.py": [ { "type": "Secret Keyword", - "filename": "coaching/src/api/models/async_ai.py", + "filename": "coaching\\src\\api\\models\\async_ai.py", "hashed_secret": "c03d694a32c7d9c2fc31c4826927eaa686a97c98", "is_verified": false, - "line_number": 128, + "line_number": 131, "is_added": false, "is_removed": false }, { "type": "Secret Keyword", - "filename": "coaching/src/api/models/async_ai.py", + "filename": "coaching\\src\\api\\models\\async_ai.py", "hashed_secret": "bcf6ba89ed8f1e9b4f504705a0cf4ce3605f4a87", "is_verified": false, - "line_number": 157, + "line_number": 161, "is_added": false, "is_removed": false } ], - "coaching/src/domain/entities/ai_job.py": [ + "coaching\\src\\domain\\entities\\ai_job.py": [ { "type": "Secret Keyword", - "filename": "coaching/src/domain/entities/ai_job.py", + "filename": "coaching\\src\\domain\\entities\\ai_job.py", "hashed_secret": "d527d64bcb9bf980310bdd61a944100e212c1133", "is_verified": false, "line_number": 55, @@ -161,7 +185,7 @@ }, { "type": "Secret Keyword", - "filename": "coaching/src/domain/entities/ai_job.py", + "filename": "coaching\\src\\domain\\entities\\ai_job.py", "hashed_secret": "ffa8fba99c65a6993d560b3ebc1baeec614f07d7", "is_verified": false, "line_number": 56, @@ -170,7 +194,7 @@ }, { "type": "Secret Keyword", - "filename": "coaching/src/domain/entities/ai_job.py", + "filename": "coaching\\src\\domain\\entities\\ai_job.py", "hashed_secret": "33b280288406e33b9f31d63869575252df739cb0", "is_verified": false, "line_number": 57, @@ -179,7 +203,7 @@ }, { "type": "Secret Keyword", - "filename": "coaching/src/domain/entities/ai_job.py", + "filename": "coaching\\src\\domain\\entities\\ai_job.py", "hashed_secret": "dd359584ac23098cbde8801396cdffc4c60d48ee", "is_verified": false, "line_number": 58, @@ -187,10 +211,10 @@ "is_removed": false } ], - "coaching/src/infrastructure/llm/bedrock_provider.py": [ + "coaching\\src\\infrastructure\\llm\\bedrock_provider.py": [ { "type": "Secret Keyword", - "filename": "coaching/src/infrastructure/llm/bedrock_provider.py", + "filename": "coaching\\src\\infrastructure\\llm\\bedrock_provider.py", "hashed_secret": "3d54973f528b01019a58a52d34d518405a01b891", "is_verified": false, "line_number": 321, @@ -198,10 +222,10 @@ "is_removed": false } ], - "coaching/src/infrastructure/llm/google_vertex_provider.py": [ + "coaching\\src\\infrastructure\\llm\\google_vertex_provider.py": [ { "type": "Secret Keyword", - "filename": "coaching/src/infrastructure/llm/google_vertex_provider.py", + "filename": "coaching\\src\\infrastructure\\llm\\google_vertex_provider.py", "hashed_secret": "3d54973f528b01019a58a52d34d518405a01b891", "is_verified": false, "line_number": 312, @@ -209,10 +233,10 @@ "is_removed": false } ], - "coaching/src/integration/sql_template/enums.py": [ + "coaching\\src\\integration\\sql_template\\enums.py": [ { "type": "Secret Keyword", - "filename": "coaching/src/integration/sql_template/enums.py", + "filename": "coaching\\src\\integration\\sql_template\\enums.py", "hashed_secret": "c954a24f64ff3f8d91c06117be58f86922c04ef0", "is_verified": false, "line_number": 59, @@ -220,10 +244,10 @@ "is_removed": false } ], - "deployment/account-service/template-dotnet-fixed.yaml": [ + "deployment\\account-service\\template-dotnet-fixed.yaml": [ { "type": "Secret Keyword", - "filename": "deployment/account-service/template-dotnet-fixed.yaml", + "filename": "deployment\\account-service\\template-dotnet-fixed.yaml", "hashed_secret": "f0caa14e43b1e9df508fac74c15c1b7f3c695840", "is_verified": false, "line_number": 57, @@ -232,7 +256,7 @@ }, { "type": "Secret Keyword", - "filename": "deployment/account-service/template-dotnet-fixed.yaml", + "filename": "deployment\\account-service\\template-dotnet-fixed.yaml", "hashed_secret": "55dc7e06ffb295a95e827349abcff1fcf632d87e", "is_verified": false, "line_number": 58, @@ -241,7 +265,7 @@ }, { "type": "Secret Keyword", - "filename": "deployment/account-service/template-dotnet-fixed.yaml", + "filename": "deployment\\account-service\\template-dotnet-fixed.yaml", "hashed_secret": "099b0bcdf64466ba253fee8d6cb53e8c9db9f6c2", "is_verified": false, "line_number": 59, @@ -250,7 +274,7 @@ }, { "type": "Secret Keyword", - "filename": "deployment/account-service/template-dotnet-fixed.yaml", + "filename": "deployment\\account-service\\template-dotnet-fixed.yaml", "hashed_secret": "3f37ae4c43bf9ed4ded9cc65a44b45fc138eccf8", "is_verified": false, "line_number": 84, @@ -258,10 +282,10 @@ "is_removed": false } ], - "deployment/account-service/template-dotnet-lambda-only.yaml": [ + "deployment\\account-service\\template-dotnet-lambda-only.yaml": [ { "type": "Secret Keyword", - "filename": "deployment/account-service/template-dotnet-lambda-only.yaml", + "filename": "deployment\\account-service\\template-dotnet-lambda-only.yaml", "hashed_secret": "f9d93d0a9293f0bf023729ddc98f56f0578135c7", "is_verified": false, "line_number": 41, @@ -270,7 +294,7 @@ }, { "type": "Secret Keyword", - "filename": "deployment/account-service/template-dotnet-lambda-only.yaml", + "filename": "deployment\\account-service\\template-dotnet-lambda-only.yaml", "hashed_secret": "3f37ae4c43bf9ed4ded9cc65a44b45fc138eccf8", "is_verified": false, "line_number": 84, @@ -278,10 +302,10 @@ "is_removed": false } ], - "deployment/account-service/template-dotnet.yaml": [ + "deployment\\account-service\\template-dotnet.yaml": [ { "type": "Secret Keyword", - "filename": "deployment/account-service/template-dotnet.yaml", + "filename": "deployment\\account-service\\template-dotnet.yaml", "hashed_secret": "f0caa14e43b1e9df508fac74c15c1b7f3c695840", "is_verified": false, "line_number": 86, @@ -290,7 +314,7 @@ }, { "type": "Secret Keyword", - "filename": "deployment/account-service/template-dotnet.yaml", + "filename": "deployment\\account-service\\template-dotnet.yaml", "hashed_secret": "55dc7e06ffb295a95e827349abcff1fcf632d87e", "is_verified": false, "line_number": 87, @@ -299,7 +323,7 @@ }, { "type": "Secret Keyword", - "filename": "deployment/account-service/template-dotnet.yaml", + "filename": "deployment\\account-service\\template-dotnet.yaml", "hashed_secret": "099b0bcdf64466ba253fee8d6cb53e8c9db9f6c2", "is_verified": false, "line_number": 88, @@ -308,7 +332,7 @@ }, { "type": "Secret Keyword", - "filename": "deployment/account-service/template-dotnet.yaml", + "filename": "deployment\\account-service\\template-dotnet.yaml", "hashed_secret": "b63bf00edb07af6ffba7f7ceb7ed573a913271f7", "is_verified": false, "line_number": 140, @@ -316,10 +340,10 @@ "is_removed": false } ], - "deployment/account-service/template-python.yaml": [ + "deployment\\account-service\\template-python.yaml": [ { "type": "Secret Keyword", - "filename": "deployment/account-service/template-python.yaml", + "filename": "deployment\\account-service\\template-python.yaml", "hashed_secret": "b63bf00edb07af6ffba7f7ceb7ed573a913271f7", "is_verified": false, "line_number": 152, @@ -327,10 +351,10 @@ "is_removed": false } ], - "scripts/get_e2e_token.py": [ + "scripts\\get_e2e_token.py": [ { "type": "Secret Keyword", - "filename": "scripts/get_e2e_token.py", + "filename": "scripts\\get_e2e_token.py", "hashed_secret": "b44dda1dadd351948fcace1856ed97366e679239", "is_verified": false, "line_number": 13, @@ -338,10 +362,21 @@ "is_removed": false } ], - "shared/models/schemas.py": [ + "scripts\\analyze_slow_llm_calls.py": [ { "type": "Secret Keyword", - "filename": "shared/models/schemas.py", + "filename": "scripts\\analyze_slow_llm_calls.py", + "hashed_secret": "50d8b4a941c26b89482c94ab324b5a274f9ced66", + "is_verified": false, + "line_number": 64, + "is_added": false, + "is_removed": false + } + ], + "shared\\models\\schemas.py": [ + { + "type": "Secret Keyword", + "filename": "shared\\models\\schemas.py", "hashed_secret": "d8ea47c4fe663448bc4847803eb921f7d149dde8", "is_verified": false, "line_number": 38, @@ -350,7 +385,7 @@ }, { "type": "Secret Keyword", - "filename": "shared/models/schemas.py", + "filename": "shared\\models\\schemas.py", "hashed_secret": "9f12480c3c9dc372b40128b1becb69d9b4324d72", "is_verified": false, "line_number": 39, @@ -359,7 +394,7 @@ }, { "type": "Secret Keyword", - "filename": "shared/models/schemas.py", + "filename": "shared\\models\\schemas.py", "hashed_secret": "d06c8bd6d2da6dfd0963c433ffd393733c6e07b4", "is_verified": false, "line_number": 40, @@ -368,7 +403,7 @@ }, { "type": "Secret Keyword", - "filename": "shared/models/schemas.py", + "filename": "shared\\models\\schemas.py", "hashed_secret": "4c9c4ba968615cd9aa4d173eefb6672ba6591fa2", "is_verified": false, "line_number": 49, @@ -377,36 +412,14 @@ }, { "type": "Secret Keyword", - "filename": "shared/models/schemas.py", + "filename": "shared\\models\\schemas.py", "hashed_secret": "f4991d9f176172ed58e57601a72127318a839f4d", "is_verified": false, "line_number": 65, "is_added": false, "is_removed": false } - ], - "coaching/src/api/auth.py": [ - { - "type": "Secret Keyword", - "filename": "coaching/src/api/auth.py", - "hashed_secret": "ce094fa09693604fb88de28e4876f8c38a5548d3", - "is_verified": false, - "line_number": 99, - "is_added": false, - "is_removed": false - } - ], - "scripts/analyze_slow_llm_calls.py": [ - { - "type": "Secret Keyword", - "filename": "scripts/analyze_slow_llm_calls.py", - "hashed_secret": "50d8b4a941c26b89482c94ab324b5a274f9ced66", - "is_verified": false, - "line_number": 64, - "is_added": false, - "is_removed": false - } ] }, - "generated_at": "2026-04-06T03:16:33Z" + "generated_at": "2026-04-10T01:09:14Z" } diff --git a/security/pip_audit_allowlist.txt b/security/pip_audit_allowlist.txt index 99fbdeda..da927cb4 100644 --- a/security/pip_audit_allowlist.txt +++ b/security/pip_audit_allowlist.txt @@ -21,3 +21,5 @@ GHSA-3936-cmfr-pm3m PYSEC-2024-48 # langchain-openai SSRF in token counting (not in hot path for this service) GHSA-2g6r-c272-w58r +# langchain-core — f-string / DictPromptTemplate validation (GHSA-926x); bump 0.3.84+ when ecosystem allows +GHSA-926x-3r5x-gfhw diff --git a/shared/services/eventbridge_client.py b/shared/services/eventbridge_client.py index 736a9dcb..103960bb 100644 --- a/shared/services/eventbridge_client.py +++ b/shared/services/eventbridge_client.py @@ -8,7 +8,7 @@ import json from datetime import UTC, datetime -from typing import Any, cast +from typing import Any, Literal, cast import boto3 import structlog @@ -67,18 +67,21 @@ def __init__( source: str = AI_EVENT_SOURCE, stage: str = "dev", enabled: bool = True, + domain_event_bus_name: str | None = None, ) -> None: """Initialize the EventBridge publisher. Args: region_name: AWS region - event_bus_name: EventBridge bus name (default: "default") + event_bus_name: EventBridge bus for internal events (ai.job.created, WebSocket, etc.) source: Event source identifier stage: Environment stage (dev/staging/production) for event filtering enabled: Whether publishing is enabled for this environment + domain_event_bus_name: Bus for email-insight v2.4 terminal events; defaults to event_bus_name """ self._client: Any = get_eventbridge_client(region_name) self._event_bus_name = event_bus_name + self._domain_event_bus_name = domain_event_bus_name or event_bus_name self._source = source self._stage = stage self._enabled = enabled @@ -171,6 +174,118 @@ def publish(self, event: DomainEvent) -> str: ) raise EventBridgePublishError(f"EventBridge error: {e}") from e + def publish_email_insight_terminal_v24( + self, + *, + terminal_status: Literal["completed", "failed"], + terminal_event_id: str, + occurred_at_utc: datetime, + job_id: str, + request_id: str, + kickoff_event_id: str, + tenant_id: str, + user_id: str, + correlation_id: str, + idempotency_key: str, + topic_category: str, + topic_id: str, + event_signal: str, + data: dict[str, Any], + ) -> str: + """Publish normative email-insight terminal event (spec v2.4 §3.7) to the domain bus. + + Detail JSON is the flat contract shape (not wrapped in the legacy DomainEvent envelope). + """ + detail_type = "ai.job.completed" if terminal_status == "completed" else "ai.job.failed" + detail: dict[str, Any] = { + "schemaVersion": "2.4", + "eventId": terminal_event_id, + "occurredAtUtc": occurred_at_utc.replace(tzinfo=UTC) + if occurred_at_utc.tzinfo is None + else occurred_at_utc.astimezone(UTC), + "sourceService": "PurposePath.AI", + "status": terminal_status, + "jobId": job_id, + "requestId": request_id, + "kickoffEventId": kickoff_event_id, + "tenantId": tenant_id, + "userId": user_id, + "correlationId": correlation_id, + "idempotencyKey": idempotency_key, + "topicCategory": topic_category, + "topicId": topic_id, + "eventSignal": event_signal, + "kickoffTransport": "eventbridge", + "executionMode": "eventbridge_terminal", + "data": data, + } + # ISO format for occurredAtUtc in JSON + detail["occurredAtUtc"] = detail["occurredAtUtc"].isoformat().replace("+00:00", "Z") + + if not self._enabled: + logger.warning( + "eventbridge.email_insight_terminal_skipped_disabled", + detail_type=detail_type, + job_id=job_id, + request_id=request_id, + idempotency_key=idempotency_key, + ) + return f"disabled-{detail_type}" + + entry = { + "Source": self._source, + "DetailType": detail_type, + "Detail": json.dumps(detail, default=str), + "EventBusName": self._domain_event_bus_name, + "Time": datetime.now(UTC), + } + + logger.info( + "eventbridge.email_insight_terminal_publishing", + detail_type=detail_type, + job_id=job_id, + request_id=request_id, + correlation_id=correlation_id, + idempotency_key=idempotency_key, + tenant_id=tenant_id, + event_bus=self._domain_event_bus_name, + ) + + try: + response = self._client.put_events(Entries=[entry]) + if response.get("FailedEntryCount", 0) > 0: + failed = response.get("Entries", [{}])[0] + error_code = failed.get("ErrorCode", "Unknown") + error_message = failed.get("ErrorMessage", "Unknown error") + logger.error( + "eventbridge.email_insight_terminal_publish_failed", + detail_type=detail_type, + error_code=error_code, + error_message=error_message, + job_id=job_id, + request_id=request_id, + ) + raise EventBridgePublishError( + f"Failed to publish terminal event: {error_code} - {error_message}" + ) + event_id: str = str(response.get("Entries", [{}])[0].get("EventId", "")) + logger.info( + "eventbridge.email_insight_terminal_published", + detail_type=detail_type, + eventbridge_event_id=event_id, + job_id=job_id, + request_id=request_id, + ) + return event_id + except ClientError as e: + logger.error( + "eventbridge.email_insight_terminal_internal_error", + detail_type=detail_type, + error=str(e), + job_id=job_id, + ) + raise EventBridgePublishError(f"EventBridge error: {e}") from e + def publish_ai_job_started( self, job_id: str,