refactor: the introduced types somewhere central

Author: bencegadanyi1-nhs

packages/slackBotFunction/app/core/types.py

@@ -0,0 +1,86 @@
+"""
+types for direct lambda invocation - defines contracts for bypassing slack
+
+centralizes all type definitions for direct invocation flow to avoid scattered
+inline type hints across handlers and processors.
+"""
+
+from typing import Any, TypedDict, Literal
+from datetime import datetime, timezone
+
+
+class DirectInvocationRequest(TypedDict, total=False):
+    """payload contract for direct lambda calls - bypasses slack entirely"""
+
+    invocation_type: Literal["direct"]
+    query: str
+    session_id: str | None  # conversation continuity across calls
+
+
+class DirectInvocationResponseData(TypedDict):
+    """successful ai response payload - matches slack handler output format"""
+
+    text: str
+    session_id: str | None
+    citations: list[dict[str, str]]  # [{title: str, uri: str}, ...]
+    timestamp: str  # iso8601 with Z suffix
+
+
+class DirectInvocationErrorData(TypedDict):
+    """error response payload - consistent structure for all failure modes"""
+
+    error: str
+    timestamp: str  # iso8601 with Z suffix
+
+
+class DirectInvocationResponse(TypedDict):
+    """complete lambda response envelope - includes status code + payload"""
+
+    statusCode: int
+    response: DirectInvocationResponseData | DirectInvocationErrorData
+
+
+class AIProcessorResponse(TypedDict):
+    """ai processor output - shared between slack and direct invocation"""
+
+    text: str
+    session_id: str | None
+    citations: list[dict[str, str]]
+    # TODO: ensure proper typing for bedrock response when refactoring other types in the future
+    kb_response: dict[str, Any]  # raw bedrock data for slack session handling
+
+
+# type guards for runtime validation
+def is_valid_direct_request(event: dict[str, Any]) -> bool:
+    """validate direct invocation payload structure"""
+    return (
+        event.get("invocation_type") == "direct"
+        and isinstance(event.get("query"), str)
+        and bool(event.get("query", "").strip())  # non-empty after whitespace removal
+    )
+
+
+def create_success_response(
+    text: str, session_id: str | None, citations: list[dict[str, str]]
+) -> DirectInvocationResponse:
+    """factory for successful direct invocation responses"""
+    return {
+        "statusCode": 200,
+        "response": {
+            "text": text,
+            "session_id": session_id,
+            "citations": citations,
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+        },
+    }
+
+
+def create_error_response(status_code: int, error_message: str) -> DirectInvocationResponse:
+    """factory for error responses - ensures consistent timestamp format"""
+    return {
+        "statusCode": status_code,
+        "response": {
+            "error": error_message,
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+        },
+    }

packages/slackBotFunction/app/handler.py

@@ -6,11 +6,18 @@
 2. Processes async operations when invoked by itself to avoid timeouts
 """
 
-from datetime import datetime
 from slack_bolt.adapter.aws_lambda import SlackRequestHandler
 from aws_lambda_powertools.utilities.typing import LambdaContext
 
+from typing import Any
+
 from app.core.config import get_logger
+from app.core.types import (
+    DirectInvocationResponse,
+    is_valid_direct_request,
+    create_success_response,
+    create_error_response,
+)
 from app.services.app import get_app
 from app.slack.slack_events import process_pull_request_slack_action, process_pull_request_slack_event
 
@@ -62,38 +69,26 @@ def handler(event: dict, context: LambdaContext) -> dict:
     return slack_handler.handle(event=event, context=context)
 
 
-def handle_direct_invocation(event: dict, context: LambdaContext) -> dict:
+def handle_direct_invocation(event: dict[str, Any], context: LambdaContext) -> DirectInvocationResponse:
     """direct lambda invocation for ai assistance - bypasses slack entirely"""
     try:
-        query = event.get("query")
-        session_id = event.get("session_id")
+        # validate request structure using type guard
+        if not is_valid_direct_request(event):
+            return create_error_response(400, "Missing required field: query")
 
-        if not query or not query.strip():
-            return {
-                "statusCode": 400,
-                "response": {
-                    "error": "Missing required field: query",
-                    "timestamp": datetime.utcnow().isoformat() + "Z",
-                },
-            }
+        query = event["query"]
+        session_id = event.get("session_id")
 
         # shared logic: same AI processing as slack handlers use
         from app.services.ai_processor import process_ai_query
 
         ai_response = process_ai_query(query, session_id)
 
-        return {
-            "statusCode": 200,
-            "response": {
-                "text": ai_response["text"],
-                "session_id": ai_response["session_id"],
-                "citations": ai_response["citations"],
-                "timestamp": datetime.utcnow().isoformat() + "Z",
-            },
-        }
+        return create_success_response(
+            text=ai_response["text"],
+            session_id=ai_response["session_id"],
+            citations=ai_response["citations"],
+        )
     except Exception as e:
         logger.error(f"Error in direct invocation: {e}")
-        return {
-            "statusCode": 500,
-            "response": {"error": "Internal server error", "timestamp": datetime.utcnow().isoformat() + "Z"},
-        }
+        return create_error_response(500, "Internal server error")

packages/slackBotFunction/app/services/ai_processor.py

@@ -5,15 +5,15 @@
 reformulation and bedrock interaction. single source of truth for AI flows.
 """
 
-from typing import Dict, Any, Optional
 from app.services.bedrock import query_bedrock
 from app.services.query_reformulator import reformulate_query
 from app.core.config import get_logger
+from app.core.types import AIProcessorResponse
 
 logger = get_logger()
 
 
-def process_ai_query(user_query: str, session_id: Optional[str] = None) -> Dict[str, Any]:
+def process_ai_query(user_query: str, session_id: str | None = None) -> AIProcessorResponse:
     """shared AI processing logic for both slack and direct invocation"""
     # reformulate: improves vector search quality in knowledge base
     reformulated_query = reformulate_query(user_query)

packages/slackBotFunction/tests/test_direct_invocation.py

@@ -2,6 +2,10 @@
 
 from unittest.mock import Mock, patch
 from app.handler import handle_direct_invocation
+from app.core.types import (
+    DirectInvocationRequest,
+    DirectInvocationResponse,
+)
 
 
 class TestDirectInvocation:
@@ -16,9 +20,9 @@ def test_successful_direct_invocation_without_session(self, mock_process_ai_quer
             "kb_response": {"sessionId": "new-session-123"},
         }
 
-        event = {"invocation_type": "direct", "query": "How do I authenticate with EPS API?"}
+        event: DirectInvocationRequest = {"invocation_type": "direct", "query": "How do I authenticate with EPS API?"}
 
-        result = handle_direct_invocation(event, Mock())
+        result: DirectInvocationResponse = handle_direct_invocation(event, Mock())
 
         assert result["statusCode"] == 200
         assert result["response"]["text"] == "AI response about EPS API authentication"
@@ -40,7 +44,7 @@ def test_successful_direct_invocation_with_session(self, mock_process_ai_query):
 
         event = {"invocation_type": "direct", "query": "What about rate limits?", "session_id": "existing-session-456"}
 
-        result = handle_direct_invocation(event, Mock())
+        result: DirectInvocationResponse = handle_direct_invocation(event, Mock())
 
         assert result["statusCode"] == 200
         assert result["response"]["text"] == "Follow-up response"
@@ -149,9 +153,9 @@ def test_direct_invocation_timestamp_format(self, mock_process_ai_query):
 
         timestamp = result["response"]["timestamp"]
         # iso8601 validation: parseable datetime with utc marker
-        assert timestamp.endswith("Z")
+        assert timestamp.endswith("+00:00")
         assert "T" in timestamp
         # format verification: datetime parsing confirms structure
         from datetime import datetime
 
-        datetime.fromisoformat(timestamp.rstrip("Z"))
+        datetime.fromisoformat(timestamp)