diff --git a/.github/workflows/deploy.yml b/.github/workflows/deploy.yml index 6e3370c..49a52cc 100644 --- a/.github/workflows/deploy.yml +++ b/.github/workflows/deploy.yml @@ -110,25 +110,37 @@ jobs: # Create deployment package for each Lambda mkdir -p .lambda-packages - # Package stripe_webhook - cd src/lambdas/stripe_webhook - zip -r ../../../.lambda-packages/stripe_webhook.zip . - cd ../../.. - - # Package stripe_checkout - cd src/lambdas/stripe_checkout - zip -r ../../../.lambda-packages/stripe_checkout.zip . - cd ../../.. - - # Package nb_oauth_callback - cd src/lambdas/nb_oauth_callback - zip -r ../../../.lambda-packages/nb_oauth_callback.zip . - cd ../../.. - - # Package token_refresh - cd src/lambdas/token_refresh - zip -r ../../../.lambda-packages/token_refresh.zip . - cd ../../.. + # Package stripe_webhook (includes shared modules for metrics/observability) + mkdir -p .lambda-packages/stripe_webhook_pkg + cp -r src/lambdas/stripe_webhook/* .lambda-packages/stripe_webhook_pkg/ + cp -r src/lambdas/shared .lambda-packages/stripe_webhook_pkg/ + cd .lambda-packages/stripe_webhook_pkg + zip -r ../stripe_webhook.zip . + cd ../.. + + # Package stripe_checkout (includes shared modules for metrics/observability) + mkdir -p .lambda-packages/stripe_checkout_pkg + cp -r src/lambdas/stripe_checkout/* .lambda-packages/stripe_checkout_pkg/ + cp -r src/lambdas/shared .lambda-packages/stripe_checkout_pkg/ + cd .lambda-packages/stripe_checkout_pkg + zip -r ../stripe_checkout.zip . + cd ../.. + + # Package nb_oauth_callback (includes shared modules for session token + observability) + mkdir -p .lambda-packages/nb_oauth_callback_pkg + cp -r src/lambdas/nb_oauth_callback/* .lambda-packages/nb_oauth_callback_pkg/ + cp -r src/lambdas/shared .lambda-packages/nb_oauth_callback_pkg/ + cd .lambda-packages/nb_oauth_callback_pkg + zip -r ../nb_oauth_callback.zip . + cd ../.. + + # Package token_refresh (includes shared modules for metrics/observability) + mkdir -p .lambda-packages/token_refresh_pkg + cp -r src/lambdas/token_refresh/* .lambda-packages/token_refresh_pkg/ + cp -r src/lambdas/shared .lambda-packages/token_refresh_pkg/ + cd .lambda-packages/token_refresh_pkg + zip -r ../token_refresh.zip . + cd ../.. # Package nat_agent (includes shared modules and nat source) mkdir -p .lambda-packages/nat_agent_staging @@ -222,25 +234,37 @@ jobs: # Create deployment package for each Lambda mkdir -p .lambda-packages - # Package stripe_webhook - cd src/lambdas/stripe_webhook - zip -r ../../../.lambda-packages/stripe_webhook.zip . - cd ../../.. - - # Package stripe_checkout - cd src/lambdas/stripe_checkout - zip -r ../../../.lambda-packages/stripe_checkout.zip . - cd ../../.. - - # Package nb_oauth_callback - cd src/lambdas/nb_oauth_callback - zip -r ../../../.lambda-packages/nb_oauth_callback.zip . - cd ../../.. - - # Package token_refresh - cd src/lambdas/token_refresh - zip -r ../../../.lambda-packages/token_refresh.zip . - cd ../../.. + # Package stripe_webhook (includes shared modules for metrics/observability) + mkdir -p .lambda-packages/stripe_webhook_pkg + cp -r src/lambdas/stripe_webhook/* .lambda-packages/stripe_webhook_pkg/ + cp -r src/lambdas/shared .lambda-packages/stripe_webhook_pkg/ + cd .lambda-packages/stripe_webhook_pkg + zip -r ../stripe_webhook.zip . + cd ../.. + + # Package stripe_checkout (includes shared modules for metrics/observability) + mkdir -p .lambda-packages/stripe_checkout_pkg + cp -r src/lambdas/stripe_checkout/* .lambda-packages/stripe_checkout_pkg/ + cp -r src/lambdas/shared .lambda-packages/stripe_checkout_pkg/ + cd .lambda-packages/stripe_checkout_pkg + zip -r ../stripe_checkout.zip . + cd ../.. + + # Package nb_oauth_callback (includes shared modules for session token + observability) + mkdir -p .lambda-packages/nb_oauth_callback_pkg + cp -r src/lambdas/nb_oauth_callback/* .lambda-packages/nb_oauth_callback_pkg/ + cp -r src/lambdas/shared .lambda-packages/nb_oauth_callback_pkg/ + cd .lambda-packages/nb_oauth_callback_pkg + zip -r ../nb_oauth_callback.zip . + cd ../.. + + # Package token_refresh (includes shared modules for metrics/observability) + mkdir -p .lambda-packages/token_refresh_pkg + cp -r src/lambdas/token_refresh/* .lambda-packages/token_refresh_pkg/ + cp -r src/lambdas/shared .lambda-packages/token_refresh_pkg/ + cd .lambda-packages/token_refresh_pkg + zip -r ../token_refresh.zip . + cd ../.. # Package nat_agent (includes shared modules and nat source) mkdir -p .lambda-packages/nat_agent_staging diff --git a/infrastructure/template.yaml b/infrastructure/template.yaml index f0a6675..185a1fc 100644 --- a/infrastructure/template.yaml +++ b/infrastructure/template.yaml @@ -11,12 +11,30 @@ Parameters: - prod Description: Deployment environment + AlarmEmail: + Type: String + Default: '' + Description: >- + Optional email address subscribed to the CloudWatch alarm SNS topic. + Leave blank to create the topic without an email subscription. + + SentryDsnSecretName: + Type: String + Default: '' + Description: >- + Optional Secrets Manager secret name holding the Sentry DSN. When set, + Lambda handlers initialise Sentry error tracking; when blank, error + tracking is a no-op. + # Placeholder for future custom domain support - not wired up yet # CustomDomainName: # Type: String # Default: '' # Description: Optional custom domain name for the API (e.g., api.natassistant.com) +Conditions: + HasAlarmEmail: !Not [!Equals [!Ref AlarmEmail, '']] + Resources: # ============================================================================= # DynamoDB Tables @@ -365,6 +383,8 @@ Resources: USERS_TABLE: !Ref UsersTable STRIPE_EVENTS_TABLE: !Ref StripeEventsTable STRIPE_WEBHOOK_SECRET_NAME: !Sub 'nat/stripe-webhook-secret-${Environment}' + ENVIRONMENT: !Ref Environment + SENTRY_DSN_SECRET: !Ref SentryDsnSecretName Tags: - Key: Environment Value: !Ref Environment @@ -403,6 +423,8 @@ Resources: STRIPE_PRICE_STARTER: !Sub 'price_starter_${Environment}' STRIPE_PRICE_TEAM: !Sub 'price_team_${Environment}' STRIPE_PRICE_ORG: !Sub 'price_org_${Environment}' + ENVIRONMENT: !Ref Environment + SENTRY_DSN_SECRET: !Ref SentryDsnSecretName Tags: - Key: Environment Value: !Ref Environment @@ -445,6 +467,8 @@ Resources: SESSION_JWT_SECRET_NAME: !Sub 'nat/session-jwt-secret-${Environment}' SUCCESS_REDIRECT_URL: !Sub 'https://natassistant.com/connected?env=${Environment}' ERROR_REDIRECT_URL: !Sub 'https://natassistant.com/connection-error?env=${Environment}' + ENVIRONMENT: !Ref Environment + SENTRY_DSN_SECRET: !Ref SentryDsnSecretName Tags: - Key: Environment Value: !Ref Environment @@ -521,6 +545,8 @@ Resources: NB_CLIENT_ID_SECRET: !Sub 'nat/nb-client-id-${Environment}' NB_CLIENT_SECRET_SECRET: !Sub 'nat/nb-client-secret-${Environment}' TOKEN_EXPIRY_WINDOW_HOURS: '12' + ENVIRONMENT: !Ref Environment + SENTRY_DSN_SECRET: !Ref SentryDsnSecretName Tags: - Key: Environment Value: !Ref Environment @@ -828,6 +854,8 @@ Resources: ANTHROPIC_API_KEY_SECRET: !Sub 'nat/anthropic-api-key-${Environment}' SESSION_JWT_SECRET_NAME: !Sub 'nat/session-jwt-secret-${Environment}' CLAUDE_MODEL: 'claude-haiku-4-5-20251001' + ENVIRONMENT: !Ref Environment + SENTRY_DSN_SECRET: !Ref SentryDsnSecretName Tags: - Key: Environment Value: !Ref Environment @@ -934,6 +962,8 @@ Resources: ANTHROPIC_API_KEY_SECRET: !Sub 'nat/anthropic-api-key-${Environment}' SESSION_JWT_SECRET_NAME: !Sub 'nat/session-jwt-secret-${Environment}' CLAUDE_MODEL: 'claude-haiku-4-5-20251001' + ENVIRONMENT: !Ref Environment + SENTRY_DSN_SECRET: !Ref SentryDsnSecretName Tags: - Key: Environment Value: !Ref Environment @@ -1007,6 +1037,225 @@ Resources: - Key: Project Value: nat + # ============================================================================= + # Observability - SNS alarm topic, CloudWatch alarms, dashboard + # ============================================================================= + # Custom metrics are emitted by the Lambda handlers via the CloudWatch + # Embedded Metric Format (EMF) into the "Nat" namespace with an "Environment" + # dimension. Metric names must stay in sync with src/lambdas/shared/metrics.py. + + AlarmTopic: + Type: AWS::SNS::Topic + Properties: + TopicName: !Sub 'nat-alarms-${Environment}' + DisplayName: !Sub 'Nat Alarms (${Environment})' + Tags: + - Key: Environment + Value: !Ref Environment + - Key: Project + Value: nat + + AlarmEmailSubscription: + Type: AWS::SNS::Subscription + Condition: HasAlarmEmail + Properties: + TopicArn: !Ref AlarmTopic + Protocol: email + Endpoint: !Ref AlarmEmail + + # Elevated agent error rate (custom AgentError metric). + AgentErrorAlarm: + Type: AWS::CloudWatch::Alarm + Properties: + AlarmName: !Sub 'nat-agent-errors-${Environment}' + AlarmDescription: Elevated Nat agent error rate + Namespace: Nat + MetricName: AgentError + Dimensions: + - Name: Environment + Value: !Ref Environment + Statistic: Sum + Period: 300 + EvaluationPeriods: 1 + Threshold: 5 + ComparisonOperator: GreaterThanThreshold + TreatMissingData: notBreaching + AlarmActions: + - !Ref AlarmTopic + OKActions: + - !Ref AlarmTopic + + # Spike in "nation not found" lookups (custom NationNotFound metric). + NationNotFoundAlarm: + Type: AWS::CloudWatch::Alarm + Properties: + AlarmName: !Sub 'nat-nation-not-found-${Environment}' + AlarmDescription: Spike in nation-not-found subscription lookups + Namespace: Nat + MetricName: NationNotFound + Dimensions: + - Name: Environment + Value: !Ref Environment + Statistic: Sum + Period: 300 + EvaluationPeriods: 1 + Threshold: 10 + ComparisonOperator: GreaterThanThreshold + TreatMissingData: notBreaching + AlarmActions: + - !Ref AlarmTopic + + # Token-refresh failures (custom TokenRefreshFailure metric). + TokenRefreshFailureAlarm: + Type: AWS::CloudWatch::Alarm + Properties: + AlarmName: !Sub 'nat-token-refresh-failures-${Environment}' + AlarmDescription: NationBuilder token refresh failures + Namespace: Nat + MetricName: TokenRefreshFailure + Dimensions: + - Name: Environment + Value: !Ref Environment + Statistic: Sum + Period: 300 + EvaluationPeriods: 1 + Threshold: 3 + ComparisonOperator: GreaterThanThreshold + TreatMissingData: notBreaching + AlarmActions: + - !Ref AlarmTopic + + # Stripe webhook processing failures (custom StripeWebhookFailure metric). + StripeWebhookFailureAlarm: + Type: AWS::CloudWatch::Alarm + Properties: + AlarmName: !Sub 'nat-stripe-webhook-failures-${Environment}' + AlarmDescription: Stripe webhook processing failures + Namespace: Nat + MetricName: StripeWebhookFailure + Dimensions: + - Name: Environment + Value: !Ref Environment + Statistic: Sum + Period: 300 + EvaluationPeriods: 1 + Threshold: 1 + ComparisonOperator: GreaterThanThreshold + TreatMissingData: notBreaching + AlarmActions: + - !Ref AlarmTopic + + # Built-in Lambda runtime errors for the agent function (catches crashes / + # unhandled exceptions that never reach a custom metric). + AgentLambdaErrorsAlarm: + Type: AWS::CloudWatch::Alarm + Properties: + AlarmName: !Sub 'nat-agent-lambda-errors-${Environment}' + AlarmDescription: Unhandled errors in the Nat agent Lambda + Namespace: AWS/Lambda + MetricName: Errors + Dimensions: + - Name: FunctionName + Value: !Ref NatAgentFunction + Statistic: Sum + Period: 300 + EvaluationPeriods: 1 + Threshold: 5 + ComparisonOperator: GreaterThanThreshold + TreatMissingData: notBreaching + AlarmActions: + - !Ref AlarmTopic + + StreamingLambdaErrorsAlarm: + Type: AWS::CloudWatch::Alarm + Properties: + AlarmName: !Sub 'nat-agent-streaming-lambda-errors-${Environment}' + AlarmDescription: Unhandled errors in the Nat streaming agent Lambda + Namespace: AWS/Lambda + MetricName: Errors + Dimensions: + - Name: FunctionName + Value: !Ref NatAgentStreamingFunction + Statistic: Sum + Period: 300 + EvaluationPeriods: 1 + Threshold: 5 + ComparisonOperator: GreaterThanThreshold + TreatMissingData: notBreaching + AlarmActions: + - !Ref AlarmTopic + + ObservabilityDashboard: + Type: AWS::CloudWatch::Dashboard + Properties: + DashboardName: !Sub 'nat-observability-${Environment}' + DashboardBody: !Sub | + { + "widgets": [ + { + "type": "metric", + "x": 0, "y": 0, "width": 12, "height": 6, + "properties": { + "title": "Subscription & access", + "region": "${AWS::Region}", + "stat": "Sum", + "period": 300, + "metrics": [ + [ "Nat", "SubscriptionVerification", "Environment", "${Environment}" ], + [ "Nat", "SubscriptionVerificationFailure", "Environment", "${Environment}" ], + [ "Nat", "QueryLimitHit", "Environment", "${Environment}" ], + [ "Nat", "NationNotFound", "Environment", "${Environment}" ] + ] + } + }, + { + "type": "metric", + "x": 12, "y": 0, "width": 12, "height": 6, + "properties": { + "title": "Errors", + "region": "${AWS::Region}", + "stat": "Sum", + "period": 300, + "metrics": [ + [ "Nat", "AgentError", "Environment", "${Environment}" ], + [ "Nat", "NBApiError", "Environment", "${Environment}" ], + [ "Nat", "TokenRefreshFailure", "Environment", "${Environment}" ], + [ "Nat", "StripeWebhookFailure", "Environment", "${Environment}" ] + ] + } + }, + { + "type": "metric", + "x": 0, "y": 6, "width": 12, "height": 6, + "properties": { + "title": "Prompt cache effectiveness", + "region": "${AWS::Region}", + "stat": "Sum", + "period": 300, + "metrics": [ + [ "Nat", "CacheReadInputTokens", "Environment", "${Environment}" ], + [ "Nat", "CacheCreationInputTokens", "Environment", "${Environment}" ], + [ "Nat", "CacheHit", "Environment", "${Environment}" ], + [ "Nat", "CacheMiss", "Environment", "${Environment}" ] + ] + } + }, + { + "type": "metric", + "x": 12, "y": 6, "width": 12, "height": 6, + "properties": { + "title": "Agent latency (ms)", + "region": "${AWS::Region}", + "period": 300, + "metrics": [ + [ "Nat", "AgentLatencyMs", "Environment", "${Environment}", { "stat": "Average" } ], + [ "Nat", "AgentLatencyMs", "Environment", "${Environment}", { "stat": "p90" } ] + ] + } + } + ] + } + Outputs: NationsTableName: Description: Name of the Nations DynamoDB table diff --git a/pyproject.toml b/pyproject.toml index 29b3cc1..729051b 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -12,6 +12,7 @@ dependencies = [ "httpx>=0.27.0", "python-dotenv>=1.0.0", "typing-extensions>=4.9.0", + "sentry-sdk>=2.0.0", ] [project.optional-dependencies] diff --git a/requirements.txt b/requirements.txt index 5394198..cfb0953 100644 --- a/requirements.txt +++ b/requirements.txt @@ -9,3 +9,6 @@ python-dotenv>=1.0.0 # Type hints typing-extensions>=4.9.0 + +# Error tracking (optional at runtime: handlers no-op when SENTRY_DSN is unset) +sentry-sdk>=2.0.0 diff --git a/src/lambdas/nat_agent/handler.py b/src/lambdas/nat_agent/handler.py index 371609f..b8a675d 100644 --- a/src/lambdas/nat_agent/handler.py +++ b/src/lambdas/nat_agent/handler.py @@ -32,11 +32,18 @@ SessionTokenError, authenticate_request, ) + from src.lambdas.shared import metrics + from src.lambdas.shared.observability import capture_exception, init_sentry except ModuleNotFoundError: # pragma: no cover - exercised only in Lambda from shared.session_token import ( # type: ignore[no-redef] SessionTokenError, authenticate_request, ) + from shared import metrics # type: ignore[no-redef] + from shared.observability import ( # type: ignore[no-redef] + capture_exception, + init_sentry, + ) try: # Resolve in both pytest (repo root) and flattened Lambda packages. from src.lambdas.shared.validation import is_valid_nation_slug @@ -284,7 +291,19 @@ async def run_agent_query( "input": block.input, }) elif isinstance(message, ResultMessage): + # Record prompt-cache effectiveness and agent latency. + metrics.record_cache_usage(message.usage, nb_slug) + if message.duration_ms is not None: + metrics.emit_metric( + metrics.AGENT_LATENCY_MS, + float(message.duration_ms), + metrics.UNIT_MILLISECONDS, + {"nation_slug": nb_slug}, + ) if message.is_error: + metrics.emit_count( + metrics.AGENT_ERROR, {"nation_slug": nb_slug} + ) logger.warning(f"Tool error: {message.result}") return { @@ -295,6 +314,8 @@ async def run_agent_query( except Exception as e: logger.error(f"Agent query failed: {e}") + metrics.emit_count(metrics.AGENT_ERROR, {"nation_slug": nb_slug}) + capture_exception(e, nation_slug=nb_slug) return { "response": "", "error": str(e), @@ -339,6 +360,8 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: "Access-Control-Allow-Headers": "Content-Type,Authorization", } + init_sentry() + try: # Parse request body body_str = event.get("body", "") @@ -498,6 +521,7 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: except ClientError as e: logger.error(f"AWS service error: {e}") + capture_exception(e) return { "statusCode": 500, "body": json.dumps({"error": "Internal server error"}), @@ -505,6 +529,7 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: } except Exception as e: logger.error(f"Unexpected error: {e}") + capture_exception(e) return { "statusCode": 500, "body": json.dumps({"error": "Internal server error"}), diff --git a/src/lambdas/nat_agent_streaming/handler.py b/src/lambdas/nat_agent_streaming/handler.py index 60c7c66..f85938d 100644 --- a/src/lambdas/nat_agent_streaming/handler.py +++ b/src/lambdas/nat_agent_streaming/handler.py @@ -33,6 +33,8 @@ SessionTokenError, authenticate_request, ) + from src.lambdas.shared import metrics + from src.lambdas.shared.observability import capture_exception, init_sentry from src.lambdas.shared.session_state import ( append_undo_entry, compute_tool_id, @@ -48,6 +50,11 @@ SessionTokenError, authenticate_request, ) + from shared import metrics # type: ignore[no-redef] + from shared.observability import ( # type: ignore[no-redef] + capture_exception, + init_sentry, + ) from shared.session_state import ( # type: ignore[no-redef] append_undo_entry, compute_tool_id, @@ -522,6 +529,21 @@ async def stream_agent_response( yield format_sse_event(SSE_EVENT_TOOL_USE, tool_info) elif isinstance(message, ResultMessage): + # Record observability signals from the final result: + # prompt-cache effectiveness and end-to-end agent latency. + metrics.record_cache_usage(message.usage, nb_slug) + if message.duration_ms is not None: + metrics.emit_metric( + metrics.AGENT_LATENCY_MS, + float(message.duration_ms), + metrics.UNIT_MILLISECONDS, + {"nation_slug": nb_slug}, + ) + if message.is_error: + metrics.emit_count( + metrics.AGENT_ERROR, {"nation_slug": nb_slug} + ) + # Send tool result notification yield format_sse_event(SSE_EVENT_TOOL_RESULT, { "result": str(message.result)[:500], # Truncate long results @@ -539,6 +561,8 @@ async def stream_agent_response( except Exception as e: logger.error(f"Agent streaming error: {e}") + metrics.emit_count(metrics.AGENT_ERROR, {"nation_slug": nb_slug}) + capture_exception(e, nation_slug=nb_slug) yield format_sse_event(SSE_EVENT_ERROR, { "error": str(e), "error_code": "AGENT_ERROR", @@ -723,6 +747,7 @@ def handler(event: dict[str, Any], context: Any) -> dict[str, Any]: """ # For Lambda Function URL with response streaming, we need to handle # the request and return a streaming response + init_sentry() # Parse request body - Lambda Function URL may have different event structure body_str = "" @@ -807,6 +832,8 @@ async def streaming_handler(event: dict[str, Any], response_stream: Any) -> None event: Lambda event response_stream: awslambda.ResponseStream for writing chunks """ + init_sentry() + # Write SSE headers await response_stream.write(b"") # Initialize stream diff --git a/src/lambdas/nb_oauth_callback/handler.py b/src/lambdas/nb_oauth_callback/handler.py index 017a5e9..528f17a 100644 --- a/src/lambdas/nb_oauth_callback/handler.py +++ b/src/lambdas/nb_oauth_callback/handler.py @@ -31,6 +31,7 @@ get_session_secret, mint_session_token, ) + from src.lambdas.shared.observability import capture_exception, init_sentry except ModuleNotFoundError: # pragma: no cover - exercised only in Lambda from shared.oauth_state import ( # type: ignore[no-redef] OAuthStateError, @@ -40,6 +41,10 @@ get_session_secret, mint_session_token, ) + from shared.observability import ( # type: ignore[no-redef] + capture_exception, + init_sentry, + ) logger = logging.getLogger() logger.setLevel(logging.INFO) @@ -421,6 +426,8 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: "Access-Control-Allow-Origin": "*", } + init_sentry() + try: # Extract query parameters query_params = event.get("queryStringParameters") or {} @@ -548,9 +555,11 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: except ClientError as e: logger.error(f"AWS service error: {e}") + capture_exception(e) error_url = f"{ERROR_REDIRECT_URL}?error=service_error" return create_redirect_response(error_url) except Exception as e: logger.error(f"Unexpected error: {e}") + capture_exception(e) error_url = f"{ERROR_REDIRECT_URL}?error=unexpected" return create_redirect_response(error_url) diff --git a/src/lambdas/shared/__init__.py b/src/lambdas/shared/__init__.py index f3422ab..887d148 100644 --- a/src/lambdas/shared/__init__.py +++ b/src/lambdas/shared/__init__.py @@ -1,11 +1,23 @@ """Shared utilities for Lambda functions.""" -from src.lambdas.shared.subscription_middleware import ( - SubscriptionError, - SubscriptionMiddleware, - SubscriptionStatus, - verify_subscription, -) +# Lambda packages are built by flattening ``src/lambdas/shared`` to a top-level +# ``shared`` package, so the absolute ``src.lambdas.shared`` path does not +# resolve at runtime. Fall back to a relative import so this package imports +# cleanly in both the repo (pytest) layout and the flattened Lambda layout. +try: + from src.lambdas.shared.subscription_middleware import ( + SubscriptionError, + SubscriptionMiddleware, + SubscriptionStatus, + verify_subscription, + ) +except ModuleNotFoundError: # pragma: no cover - exercised only in Lambda + from .subscription_middleware import ( + SubscriptionError, + SubscriptionMiddleware, + SubscriptionStatus, + verify_subscription, + ) __all__ = [ "SubscriptionError", diff --git a/src/lambdas/shared/metrics.py b/src/lambdas/shared/metrics.py new file mode 100644 index 0000000..7f18433 --- /dev/null +++ b/src/lambdas/shared/metrics.py @@ -0,0 +1,162 @@ +""" +CloudWatch custom metrics via the Embedded Metric Format (EMF). + +Emitting a metric here means writing a single structured JSON line to the +Lambda's CloudWatch Logs. CloudWatch automatically extracts those lines into +custom metrics in the ``Nat`` namespace -- no extra IAM permissions, no +synchronous ``PutMetricData`` API call (which would add latency to every +request, including the SSE streaming path), and the raw event is still queryable +in CloudWatch Logs Insights. + +See: https://docs.aws.amazon.com/AmazonCloudWatch/latest/monitoring/CloudWatch_Embedded_Metric_Format_Specification.html + +Design notes +------------ +* The only **dimension** is ``Environment`` (dev/staging/prod). Keeping the + dimension set tiny and low-cardinality means alarms aggregate across all + nations and stay cheap. Higher-cardinality context (nation_slug, error_type, + the failing API path) is attached as non-dimension *properties*: it shows up + in the log event for Logs Insights queries but does not multiply the number + of billed metric streams. +* This module imports only the standard library, so it is safe to bundle into + every Lambda package without pulling in extra dependencies. +* Set ``NAT_DISABLE_METRICS=true`` to silence emission (used in tests / local). +""" + +from __future__ import annotations + +import json +import logging +import os +import time +from typing import Any + +logger = logging.getLogger() + +# Namespace and dimension values +METRICS_NAMESPACE = os.environ.get("METRICS_NAMESPACE", "Nat") +ENVIRONMENT = os.environ.get("ENVIRONMENT", "dev") + +# --------------------------------------------------------------------------- +# Metric names. Defined once here so handlers (which emit) and the +# CloudFormation alarms / dashboard (which reference) cannot drift apart on +# spelling. If you rename one of these, update infrastructure/template.yaml. +# --------------------------------------------------------------------------- +SUBSCRIPTION_VERIFICATION = "SubscriptionVerification" +SUBSCRIPTION_VERIFICATION_FAILURE = "SubscriptionVerificationFailure" +QUERY_LIMIT_HIT = "QueryLimitHit" +NATION_NOT_FOUND = "NationNotFound" +NB_API_ERROR = "NBApiError" +TOKEN_REFRESH_FAILURE = "TokenRefreshFailure" +STRIPE_WEBHOOK_FAILURE = "StripeWebhookFailure" +AGENT_LATENCY_MS = "AgentLatencyMs" +AGENT_ERROR = "AgentError" +CACHE_READ_TOKENS = "CacheReadInputTokens" +CACHE_CREATION_TOKENS = "CacheCreationInputTokens" +CACHE_HIT = "CacheHit" +CACHE_MISS = "CacheMiss" + +# Units understood by CloudWatch (subset we use) +UNIT_COUNT = "Count" +UNIT_MILLISECONDS = "Milliseconds" + + +def _metrics_disabled() -> bool: + """Metrics are emitted unless explicitly disabled (e.g. in tests).""" + return os.environ.get("NAT_DISABLE_METRICS", "").lower() == "true" + + +def emit_metric( + name: str, + value: float = 1.0, + unit: str = UNIT_COUNT, + properties: dict[str, Any] | None = None, +) -> None: + """ + Emit a single custom metric in EMF format to the Lambda logs. + + Args: + name: Metric name (use the constants defined in this module). + value: Metric value. Defaults to 1.0 (a simple "it happened" counter). + unit: CloudWatch unit, e.g. ``"Count"`` or ``"Milliseconds"``. + properties: Optional high-cardinality context (nation_slug, error_type, + …). Attached to the log event for Logs Insights but NOT used as a + metric dimension. + + Emission never raises: observability must not be able to break a request. + """ + if _metrics_disabled(): + return + + try: + emf: dict[str, Any] = { + "_aws": { + "Timestamp": int(time.time() * 1000), + "CloudWatchMetrics": [ + { + "Namespace": METRICS_NAMESPACE, + "Dimensions": [["Environment"]], + "Metrics": [{"Name": name, "Unit": unit}], + } + ], + }, + "Environment": ENVIRONMENT, + name: value, + } + if properties: + for key, prop_value in properties.items(): + # Never let a property clobber the reserved keys. + if key not in ("_aws", "Environment", name): + emf[key] = prop_value + logger.info(json.dumps(emf, default=str)) + except Exception as exc: # pragma: no cover - defensive only + logger.warning(f"Failed to emit metric {name}: {exc}") + + +def emit_count(name: str, properties: dict[str, Any] | None = None) -> None: + """Convenience wrapper: emit a count of 1 for ``name``.""" + emit_metric(name, 1.0, UNIT_COUNT, properties) + + +def record_cache_usage(usage: dict[str, Any] | None, nation_slug: str | None = None) -> None: + """ + Record prompt-cache effectiveness from an Anthropic ``usage`` payload. + + The Claude Agent SDK surfaces token usage on ``ResultMessage.usage``. The + Anthropic API reports cache activity via ``cache_read_input_tokens`` (tokens + served from cache -- the savings) and ``cache_creation_input_tokens`` (tokens + written to the cache on a miss). A non-zero ``cache_read_input_tokens`` is the + ground-truth confirmation that prompt caching is actually active. + + Emits ``CacheReadInputTokens`` / ``CacheCreationInputTokens`` (token counts) + plus a ``CacheHit`` or ``CacheMiss`` counter, and logs a human-readable + summary so cache effectiveness is visible in the logs even before the metric + aggregates. + """ + if not usage: + return + + try: + cache_read = int(usage.get("cache_read_input_tokens", 0) or 0) + cache_creation = int(usage.get("cache_creation_input_tokens", 0) or 0) + except (TypeError, ValueError): + logger.warning(f"Unparseable usage payload for cache metrics: {usage!r}") + return + + props = {"nation_slug": nation_slug} if nation_slug else None + + emit_metric(CACHE_READ_TOKENS, float(cache_read), UNIT_COUNT, props) + emit_metric(CACHE_CREATION_TOKENS, float(cache_creation), UNIT_COUNT, props) + + if cache_read > 0: + emit_count(CACHE_HIT, props) + else: + emit_count(CACHE_MISS, props) + + logger.info( + "Prompt cache usage: cache_read_input_tokens=%s cache_creation_input_tokens=%s " + "caching_active=%s", + cache_read, + cache_creation, + cache_read > 0, + ) diff --git a/src/lambdas/shared/observability.py b/src/lambdas/shared/observability.py new file mode 100644 index 0000000..f1b7a5a --- /dev/null +++ b/src/lambdas/shared/observability.py @@ -0,0 +1,119 @@ +""" +Optional Sentry error tracking for Lambda handlers. + +Every handler calls :func:`init_sentry` at entry and :func:`capture_exception` +in its top-level ``except`` block. Both functions are **no-ops** unless: + +1. the ``sentry-sdk`` package is importable, and +2. a DSN is configured (``SENTRY_DSN`` env var, or the name of a Secrets + Manager secret in ``SENTRY_DSN_SECRET``). + +This keeps the wiring inert in tests/local and in any environment where Sentry +has not been provisioned, while making it a one-secret change to turn on in +production. Nothing here can raise into a request path. +""" + +from __future__ import annotations + +import json +import logging +import os +from typing import Any + +logger = logging.getLogger() + +ENVIRONMENT = os.environ.get("ENVIRONMENT", "dev") +SENTRY_DSN_ENV = "SENTRY_DSN" +SENTRY_DSN_SECRET_ENV = "SENTRY_DSN_SECRET" +SENTRY_TRACES_SAMPLE_RATE = float(os.environ.get("SENTRY_TRACES_SAMPLE_RATE", "0.0")) + +# Tracks whether init has run so we only initialise once per warm container. +_initialized = False +_active = False + + +def _resolve_dsn() -> str | None: + """Resolve the Sentry DSN from env var or a Secrets Manager secret name.""" + dsn = os.environ.get(SENTRY_DSN_ENV, "").strip() + if dsn: + return dsn + + secret_name = os.environ.get(SENTRY_DSN_SECRET_ENV, "").strip() + if not secret_name: + return None + + try: + import boto3 # imported lazily; available in the Lambda runtime + + client = boto3.client("secretsmanager") + response = client.get_secret_value(SecretId=secret_name) + secret = response.get("SecretString", "") or "" + try: + data = json.loads(secret) + return str(data.get("dsn") or data.get("sentry_dsn") or secret) or None + except json.JSONDecodeError: + return secret or None + except Exception as exc: # pragma: no cover - defensive only + logger.warning(f"Could not resolve Sentry DSN from secret: {exc}") + return None + + +def init_sentry() -> bool: + """ + Initialise Sentry if a DSN is configured and the SDK is installed. + + Idempotent and safe to call on every invocation. Returns ``True`` when Sentry + is active, ``False`` otherwise (which is the normal case when no DSN is set). + """ + global _initialized, _active + if _initialized: + return _active + + _initialized = True + + dsn = _resolve_dsn() + if not dsn: + logger.debug("Sentry DSN not configured; error tracking disabled") + return False + + try: + import sentry_sdk + from sentry_sdk.integrations.aws_lambda import AwsLambdaIntegration + + sentry_sdk.init( + dsn=dsn, + environment=ENVIRONMENT, + traces_sample_rate=SENTRY_TRACES_SAMPLE_RATE, + integrations=[AwsLambdaIntegration(timeout_warning=True)], + ) + _active = True + logger.info("Sentry error tracking initialised") + except ImportError: + logger.warning("sentry-sdk not installed; error tracking disabled") + except Exception as exc: # pragma: no cover - defensive only + logger.warning(f"Failed to initialise Sentry: {exc}") + + return _active + + +def capture_exception(exc: BaseException, **context: Any) -> None: + """ + Report an exception to Sentry when active; otherwise a no-op. + + Extra keyword arguments are attached as Sentry tags (e.g. ``nation_slug=...``) + to aid triage. Never raises. + """ + if not _active: + return + try: + import sentry_sdk + + if context: + with sentry_sdk.push_scope() as scope: + for key, value in context.items(): + scope.set_tag(key, str(value)) + sentry_sdk.capture_exception(exc) + else: + sentry_sdk.capture_exception(exc) + except Exception as report_exc: # pragma: no cover - defensive only + logger.warning(f"Failed to report exception to Sentry: {report_exc}") diff --git a/src/lambdas/shared/subscription_middleware.py b/src/lambdas/shared/subscription_middleware.py index 4c75875..de72066 100644 --- a/src/lambdas/shared/subscription_middleware.py +++ b/src/lambdas/shared/subscription_middleware.py @@ -19,6 +19,7 @@ import boto3 from botocore.exceptions import ClientError +from . import metrics from .session_token import ( SessionTokenError, authenticate_request, @@ -180,6 +181,9 @@ def get_nation_subscription(nation_slug: str) -> dict[str, Any]: item = response.get("Item") if not item: + metrics.emit_count( + metrics.NATION_NOT_FOUND, {"nation_slug": nation_slug} + ) raise SubscriptionError( code=SubscriptionErrorCode.NATION_NOT_FOUND, message=f"Nation {nation_slug} not found. Please connect your NationBuilder account first.", @@ -223,6 +227,10 @@ def verify_nation_subscription( # Check subscription status if subscription_status not in ACTIVE_STATUSES: + metrics.emit_count( + metrics.SUBSCRIPTION_VERIFICATION_FAILURE, + {"nation_slug": nation_slug, "reason": "inactive", "status": subscription_status}, + ) raise SubscriptionError( code=SubscriptionErrorCode.SUBSCRIPTION_INACTIVE, message=f"Nation subscription is not active (status: {subscription_status}). Please subscribe at https://natassistant.com/pricing", @@ -231,12 +239,22 @@ def verify_nation_subscription( # Check query limit (0 means unlimited, skip check) if queries_limit > 0 and queries_used >= queries_limit: + metrics.emit_count(metrics.QUERY_LIMIT_HIT, {"nation_slug": nation_slug}) + metrics.emit_count( + metrics.SUBSCRIPTION_VERIFICATION_FAILURE, + {"nation_slug": nation_slug, "reason": "query_limit"}, + ) raise SubscriptionError( code=SubscriptionErrorCode.QUERY_LIMIT_EXCEEDED, message=f"Monthly query limit of {queries_limit} exceeded for nation {nation_slug}. Upgrade your plan for more queries.", http_status=403, ) + # Subscription is active and within limits. + metrics.emit_count( + metrics.SUBSCRIPTION_VERIFICATION, {"nation_slug": nation_slug, "plan": plan} + ) + return NationSubscriptionStatus( valid=True, nation_slug=nation_slug, diff --git a/src/lambdas/stripe_checkout/handler.py b/src/lambdas/stripe_checkout/handler.py index 9900c9c..55975af 100644 --- a/src/lambdas/stripe_checkout/handler.py +++ b/src/lambdas/stripe_checkout/handler.py @@ -16,6 +16,14 @@ import boto3 from botocore.exceptions import ClientError +try: # Resolve in both pytest (repo root) and flattened Lambda packages. + from src.lambdas.shared.observability import capture_exception, init_sentry +except ModuleNotFoundError: # pragma: no cover - exercised only in Lambda + from shared.observability import ( # type: ignore[no-redef] + capture_exception, + init_sentry, + ) + logger = logging.getLogger() logger.setLevel(logging.INFO) @@ -154,6 +162,8 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: "Access-Control-Allow-Headers": "Content-Type", } + init_sentry() + # Handle OPTIONS preflight if event.get("httpMethod") == "OPTIONS": return { @@ -237,6 +247,7 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: } except ClientError as e: logger.error(f"AWS service error: {e}") + capture_exception(e) return { "statusCode": 500, "body": json.dumps({"error": "Internal server error"}), @@ -244,6 +255,7 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: } except RuntimeError as e: logger.error(f"Stripe API error: {e}") + capture_exception(e) return { "statusCode": 502, "body": json.dumps({"error": str(e)}), @@ -251,6 +263,7 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: } except Exception as e: logger.error(f"Unexpected error: {e}") + capture_exception(e) return { "statusCode": 500, "body": json.dumps({"error": "Internal server error"}), diff --git a/src/lambdas/stripe_webhook/handler.py b/src/lambdas/stripe_webhook/handler.py index dac1301..48f9ae7 100644 --- a/src/lambdas/stripe_webhook/handler.py +++ b/src/lambdas/stripe_webhook/handler.py @@ -23,6 +23,16 @@ import boto3 from botocore.exceptions import ClientError +try: # Resolve in both pytest (repo root) and flattened Lambda packages. + from src.lambdas.shared import metrics + from src.lambdas.shared.observability import capture_exception, init_sentry +except ModuleNotFoundError: # pragma: no cover - exercised only in Lambda + from shared import metrics # type: ignore[no-redef] + from shared.observability import ( # type: ignore[no-redef] + capture_exception, + init_sentry, + ) + logger = logging.getLogger() logger.setLevel(logging.INFO) @@ -492,6 +502,8 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: "Access-Control-Allow-Origin": "*", } + init_sentry() + try: # Get request body and signature body = event.get("body", "") @@ -575,6 +587,8 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: } except ClientError as e: logger.error(f"AWS service error: {e}") + metrics.emit_count(metrics.STRIPE_WEBHOOK_FAILURE, {"reason": "aws_error"}) + capture_exception(e) return { "statusCode": 500, "body": json.dumps({"error": "Internal server error"}), @@ -582,6 +596,8 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: } except Exception as e: logger.error(f"Unexpected error: {e}") + metrics.emit_count(metrics.STRIPE_WEBHOOK_FAILURE, {"reason": "processing_error"}) + capture_exception(e) return { "statusCode": 500, "body": json.dumps({"error": "Internal server error"}), diff --git a/src/lambdas/token_refresh/handler.py b/src/lambdas/token_refresh/handler.py index fb80420..106bde6 100644 --- a/src/lambdas/token_refresh/handler.py +++ b/src/lambdas/token_refresh/handler.py @@ -24,6 +24,16 @@ from boto3.dynamodb.conditions import Attr from botocore.exceptions import ClientError +try: # Resolve in both pytest (repo root) and flattened Lambda packages. + from src.lambdas.shared import metrics + from src.lambdas.shared.observability import capture_exception, init_sentry +except ModuleNotFoundError: # pragma: no cover - exercised only in Lambda + from shared import metrics # type: ignore[no-redef] + from shared.observability import ( # type: ignore[no-redef] + capture_exception, + init_sentry, + ) + logger = logging.getLogger() logger.setLevel(logging.INFO) @@ -208,6 +218,10 @@ def refresh_access_token( logger.error( f"Token refresh failed: {response.status} - {response.data.decode('utf-8')}" ) + metrics.emit_count( + metrics.NB_API_ERROR, + {"nation_slug": nb_slug, "status_code": response.status, "endpoint": "oauth/token"}, + ) raise ValueError(f"Token refresh failed with status {response.status}") token_data: TokenResponse = json.loads(response.data.decode("utf-8")) @@ -431,6 +445,7 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: Scans for expiring tokens and refreshes them proactively. """ logger.info("Starting token refresh job") + init_sentry() try: # Get NB OAuth credentials from Secrets Manager @@ -469,9 +484,14 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: logger.info(f"Token refresh complete: {succeeded} succeeded, {failed} failed") - # Log failures for debugging + # Log failures for debugging and emit a metric per failure so the + # token-refresh-failure alarm can fire on an elevated rate. for result in results: if not result["success"]: + metrics.emit_count( + metrics.TOKEN_REFRESH_FAILURE, + {"user_id": result["user_id"], "reason": result["error"]}, + ) logger.warning( f"Failed to refresh token for user {result['user_id']}: {result['error']}" ) @@ -489,12 +509,14 @@ def handler(event: dict[str, Any], context: Any) -> LambdaResponse: except ClientError as e: logger.error(f"AWS service error: {e}") + capture_exception(e) return { "statusCode": 500, "body": json.dumps({"error": "AWS service error"}), } except Exception as e: logger.error(f"Unexpected error: {e}") + capture_exception(e) return { "statusCode": 500, "body": json.dumps({"error": "Unexpected error"}), diff --git a/src/nat/client.py b/src/nat/client.py index 2eec6c5..518e15c 100644 --- a/src/nat/client.py +++ b/src/nat/client.py @@ -7,10 +7,36 @@ from __future__ import annotations +import logging from dataclasses import dataclass, field from typing import Any, List, Dict, cast import httpx +logger = logging.getLogger(__name__) + + +def _emit_nb_api_error(slug: str, status_code: int) -> None: + """ + Emit an ``NBApiError`` CloudWatch metric for a failing NationBuilder call. + + The ``nat`` package is also used by the standalone CLI, so the dependency on + the Lambda ``shared.metrics`` module is kept soft: the import is attempted in + both the repo and the flattened-Lambda layout and any failure is swallowed. + Observability must never break an API call. + """ + try: + try: + from src.lambdas.shared import metrics # repo / CLI layout + except ModuleNotFoundError: + from shared import metrics # type: ignore[no-redef] # flattened Lambda layout + + metrics.emit_count( + metrics.NB_API_ERROR, + {"nation_slug": slug, "status_code": status_code}, + ) + except Exception: # pragma: no cover - metrics must not raise into requests + pass + @dataclass class NationBuilderV2Client: @@ -47,9 +73,15 @@ def __post_init__(self) -> None: "Content-Type": "application/vnd.api+json", "Accept": "application/vnd.api+json" }, - timeout=self.timeout + timeout=self.timeout, + event_hooks={"response": [self._on_response]}, ) + async def _on_response(self, response: httpx.Response) -> None: + """Emit an NBApiError metric for any 4xx/5xx NationBuilder response.""" + if response.status_code >= 400: + _emit_nb_api_error(self.slug, response.status_code) + @property def client(self) -> httpx.AsyncClient: """Get the HTTP client, raising if not initialized.""" diff --git a/tests/test_metrics.py b/tests/test_metrics.py new file mode 100644 index 0000000..32849fa --- /dev/null +++ b/tests/test_metrics.py @@ -0,0 +1,144 @@ +""" +Tests for the EMF-based CloudWatch metrics helper. +""" + +from __future__ import annotations + +import json +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from src.lambdas.shared import metrics + + +@pytest.fixture(autouse=True) +def _enable_metrics(monkeypatch: pytest.MonkeyPatch) -> None: + """Ensure metrics are enabled for these tests regardless of the environment.""" + monkeypatch.delenv("NAT_DISABLE_METRICS", raising=False) + + +def _capture(monkeypatch: pytest.MonkeyPatch) -> MagicMock: + """Replace the module logger so we can inspect emitted EMF lines.""" + fake_logger = MagicMock() + monkeypatch.setattr(metrics, "logger", fake_logger) + return fake_logger + + +def _emf_payloads(fake_logger: MagicMock) -> list[dict[str, Any]]: + """Return parsed EMF dicts, ignoring non-JSON (human-readable) log lines.""" + payloads: list[dict[str, Any]] = [] + for call in fake_logger.info.call_args_list: + # EMF lines are emitted as a single positional JSON string with no + # logging format args; the human-readable summary uses %s args. + if len(call.args) != 1: + continue + try: + payloads.append(json.loads(call.args[0])) + except (TypeError, ValueError): + continue + return payloads + + +class TestEmitMetric: + def test_emits_valid_emf(self, monkeypatch: pytest.MonkeyPatch) -> None: + fake_logger = _capture(monkeypatch) + monkeypatch.setattr(metrics, "ENVIRONMENT", "prod") + monkeypatch.setattr(metrics, "METRICS_NAMESPACE", "Nat") + + metrics.emit_metric("AgentError", 1.0) + + fake_logger.info.assert_called_once() + payload = json.loads(fake_logger.info.call_args[0][0]) + assert payload["Environment"] == "prod" + assert payload["AgentError"] == 1.0 + cw = payload["_aws"]["CloudWatchMetrics"][0] + assert cw["Namespace"] == "Nat" + assert cw["Dimensions"] == [["Environment"]] + assert cw["Metrics"][0] == {"Name": "AgentError", "Unit": "Count"} + assert isinstance(payload["_aws"]["Timestamp"], int) + + def test_properties_are_attached_but_not_dimensions( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + fake_logger = _capture(monkeypatch) + + metrics.emit_metric( + "NationNotFound", 1.0, properties={"nation_slug": "acme"} + ) + + payload = json.loads(fake_logger.info.call_args[0][0]) + assert payload["nation_slug"] == "acme" + # nation_slug must NOT become a metric dimension + assert payload["_aws"]["CloudWatchMetrics"][0]["Dimensions"] == [["Environment"]] + + def test_property_cannot_clobber_reserved_keys( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + fake_logger = _capture(monkeypatch) + + metrics.emit_metric( + "AgentError", 7.0, properties={"AgentError": 999, "Environment": "x"} + ) + + payload = json.loads(fake_logger.info.call_args[0][0]) + assert payload["AgentError"] == 7.0 # not clobbered by property + + def test_disabled_emits_nothing(self, monkeypatch: pytest.MonkeyPatch) -> None: + fake_logger = _capture(monkeypatch) + monkeypatch.setenv("NAT_DISABLE_METRICS", "true") + + metrics.emit_metric("AgentError") + + fake_logger.info.assert_not_called() + + def test_custom_unit(self, monkeypatch: pytest.MonkeyPatch) -> None: + fake_logger = _capture(monkeypatch) + + metrics.emit_metric( + metrics.AGENT_LATENCY_MS, 1234.0, metrics.UNIT_MILLISECONDS + ) + + payload = json.loads(fake_logger.info.call_args[0][0]) + assert payload["_aws"]["CloudWatchMetrics"][0]["Metrics"][0]["Unit"] == "Milliseconds" + + +class TestRecordCacheUsage: + def test_cache_hit(self, monkeypatch: pytest.MonkeyPatch) -> None: + fake_logger = _capture(monkeypatch) + + metrics.record_cache_usage( + {"cache_read_input_tokens": 19000, "cache_creation_input_tokens": 0}, + "acme", + ) + + emitted = _emf_payloads(fake_logger) + names = {k for p in emitted for k in p if k not in ("_aws", "Environment", "nation_slug")} + assert metrics.CACHE_READ_TOKENS in names + assert metrics.CACHE_HIT in names + assert metrics.CACHE_MISS not in names + + def test_cache_miss(self, monkeypatch: pytest.MonkeyPatch) -> None: + fake_logger = _capture(monkeypatch) + + metrics.record_cache_usage( + {"cache_read_input_tokens": 0, "cache_creation_input_tokens": 19000} + ) + + emitted = _emf_payloads(fake_logger) + names = {k for p in emitted for k in p if k not in ("_aws", "Environment")} + assert metrics.CACHE_MISS in names + assert metrics.CACHE_HIT not in names + + def test_none_usage_is_noop(self, monkeypatch: pytest.MonkeyPatch) -> None: + fake_logger = _capture(monkeypatch) + metrics.record_cache_usage(None) + fake_logger.info.assert_not_called() + + def test_unparseable_usage_does_not_raise( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + _capture(monkeypatch) + # Should swallow the bad value and not raise. + metrics.record_cache_usage({"cache_read_input_tokens": object()}) # type: ignore[dict-item] diff --git a/tests/test_nat_agent_streaming.py b/tests/test_nat_agent_streaming.py index 77acf3b..1c464e9 100644 --- a/tests/test_nat_agent_streaming.py +++ b/tests/test_nat_agent_streaming.py @@ -869,9 +869,18 @@ def __init__(self, content: list[Any]) -> None: class _FakeResultMessage: - def __init__(self, result: str, is_error: bool = False) -> None: + def __init__( + self, + result: str, + is_error: bool = False, + usage: dict[str, Any] | None = None, + duration_ms: int = 1234, + ) -> None: self.result = result self.is_error = is_error + # Mirror the real SDK ResultMessage: usage may be None; duration_ms is int. + self.usage = usage + self.duration_ms = duration_ms def _make_fake_client(messages: list[Any]) -> Any: @@ -936,6 +945,45 @@ async def _run() -> list[str]: return run_async(_run()) +class TestStreamingObservability: + """The ResultMessage branch emits cache + latency metrics (issue #17).""" + + SESSION = "testnation#user-test-12345" + + def test_result_message_emits_cache_and_latency_metrics(self) -> None: + """A ResultMessage in the stream drives record_cache_usage + latency.""" + from src.lambdas.shared import metrics + + table = _SessionStateFakeTable() + usage = { + "cache_read_input_tokens": 100, + "cache_creation_input_tokens": 20, + } + result_msg = _FakeResultMessage("done", usage=usage, duration_ms=4242) + + with _patch_streaming_sdk([result_msg], table): + with patch.object(metrics, "record_cache_usage") as rec, patch.object( + metrics, "emit_metric" + ) as emit: + _collect_stream( + query="hello", + nb_slug=TEST_NB_SLUG, + nb_token=TEST_NB_TOKEN, + model="claude-haiku-4-5-20251001", + confirmed_tools=set(), + session_id=self.SESSION, + ) + + # The usage dict is passed straight through to record_cache_usage. + rec.assert_called_once() + assert rec.call_args.args[0] == usage + # A latency metric is emitted with the message's duration. + assert any( + len(call.args) >= 2 and call.args[1] == float(4242) + for call in emit.call_args_list + ), "expected a latency metric emit with duration_ms=4242" + + class TestServerSideConfirmation: """The confirmation gate must be authoritative server-side (issue #12).""" diff --git a/tests/test_observability.py b/tests/test_observability.py new file mode 100644 index 0000000..7818564 --- /dev/null +++ b/tests/test_observability.py @@ -0,0 +1,55 @@ +""" +Tests for the optional Sentry error-tracking helper. + +These verify the no-op behaviour (the common case where no DSN is configured) +without requiring a live Sentry project. +""" + +from __future__ import annotations + +import pytest + +from src.lambdas.shared import observability + + +@pytest.fixture(autouse=True) +def _reset_state(monkeypatch: pytest.MonkeyPatch) -> None: + """Reset the module-level init flags and clear DSN env between tests.""" + monkeypatch.setattr(observability, "_initialized", False) + monkeypatch.setattr(observability, "_active", False) + monkeypatch.delenv("SENTRY_DSN", raising=False) + monkeypatch.delenv("SENTRY_DSN_SECRET", raising=False) + + +class TestInitSentry: + def test_no_dsn_is_noop(self) -> None: + assert observability.init_sentry() is False + assert observability._active is False + + def test_idempotent(self, monkeypatch: pytest.MonkeyPatch) -> None: + calls = {"n": 0} + + def fake_resolve() -> None: + calls["n"] += 1 + return None + + monkeypatch.setattr(observability, "_resolve_dsn", fake_resolve) + observability.init_sentry() + observability.init_sentry() + # _resolve_dsn must only be consulted once (init is cached). + assert calls["n"] == 1 + + +class TestResolveDsn: + def test_env_var_takes_precedence(self, monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("SENTRY_DSN", "https://abc@example.com/1") + assert observability._resolve_dsn() == "https://abc@example.com/1" + + def test_returns_none_when_unconfigured(self) -> None: + assert observability._resolve_dsn() is None + + +class TestCaptureException: + def test_noop_when_inactive(self) -> None: + # Must not raise even though Sentry was never initialised. + observability.capture_exception(ValueError("boom"), nation_slug="acme")