# Technical Architecture Breakdown
## Pydantic V2 Schema Firewall
The schema firewall defined in `src/schemas.py` validates incoming data models at runtime. Any extraction payload that violates data field types or enum bounds raises a `ValidationError`.
## Two-Tiered Resilience Pipeline
- **Tier 1 (Transport & Rate Limits)**: Uses `tenacity` exponential backoff with jitter to handle HTTP 429/5xx status codes.
- **Tier 2 (Schema Correction Loop)**: Captures Pydantic validation errors and passes failure tracebacks back to the model for self-correction.
## Instructor Async Client Integration & Domain Exception Firewall
- **`src/exceptions.py` (`ExtractionError`)**: Defines the domain exception hierarchy. All microservice errors inherit from `ExtractionError` (`ConfigurationError`, `NetworkTransportError`, `SchemaValidationError`), isolating internal third-party dependencies.
- **`src/engine.py` (`ExtractionEngine`)**: Manages the Instructor-patched `AsyncOpenAI` client. It prepares XML-delimited prompts, executes zero-temperature completions targeting `ArticleExtractionRecord`, handles network retries, and translates third-party errors into domain exceptions.
- **`src/main.py` (`extract_single`, `extract_batch`)**: FastAPI REST endpoints that invoke `ExtractionEngine` and handle `ExtractionError`, converting exceptions into sanitized HTTP responses without exposing raw stack traces.
## Tier-1 Network Transport Retry Deep-Dive (Step 4.2)
### Retry Predicate: `_is_retryable_error`
The retry policy is driven by a declarative predicate rather than type-based matching. This centralizes retry semantics and decouples the engine from SDK version internals:
| Exception Class | HTTP Semantics | Retryable |
|---|---|---|
| `TimeoutError`, `ConnectionError` | Transport-level drop | ✅ |
| `APIConnectionError` | Transport-level connection failure | ✅ |
| `APITimeoutError` | Transport-level timeout | ✅ |
| `RateLimitError` | HTTP 429 | ✅ |
| `InternalServerError` | HTTP 5xx (e.g. 500, 502, 503) | ✅ |
| `APIStatusError` with `status_code >= 500` | Any 5xx fallback | ✅ |
| `AuthenticationError` (4xx) | HTTP 401 — fatal config | ❌ |
| `BadRequestError` (4xx) | HTTP 400 — fatal input | ❌ |
| `NotFoundError` (4xx) | HTTP 404 — fatal lookup | ❌ |
### Tenacity Retry Configuration
```python
@retry(
stop=stop_after_attempt(settings.max_retries_network), # 3 default
wait=wait_exponential(multiplier=1, min=2, max=10), # 2s, 4s, 8s capped
retry=retry_if_exception(_is_retryable_error),
reraise=True, # surface original exn
)
```
Backoff timeline before the final attempt: `2s` before attempt 2, `4s` before attempt 3 (persistent outage = ~6s total sleep, then `NetworkTransportError`).
### Call Flow
1. `extract_record` prepares XML-delimited messages (`SYSTEM_PROMPT` + `FEW_SHOT_MESSAGES` + sanitized ``).
2. Calls decorated `_call_llm_with_retry(messages)`.
3. Tenacity evaluates `_is_retryable_error(exc)`:
- Retryable + attempts < `max_retries_network` → sleep exponential backoff → retry.
- Success → return `ArticleExtractionRecord`.
- Exhaustion → `reraise=True` propagates original exception.
4. `extract_record` catches `OpenAIError`/`TimeoutError`/`ConnectionError` → wraps in `NetworkTransportError`.
5. `InstructorRetryException` with transport root/cause → `NetworkTransportError`; otherwise → `SchemaValidationError`.
### Why Not Retry 4xx?
Client-side errors (`401`, `400`, `404`) are fatal: they indicate misconfiguration, invalid input, or missing resources. Retrying them only burns retry budget and incurs misleading latency. The predicate returns `False` for all such cases, triggering a fast-fail on the first attempt.
### Unit Test Coverage
- `test_tier1_retry_429_rate_limit_succeeds_after_retries` — 2 transient 429s then success → asserts 3 calls.
- `test_tier1_retry_5xx_succeeds_after_retries` — 1 transient 503 then success → asserts 2 calls.
- `test_tier1_retry_exhausts_after_max_attempts` — persistent 429 → asserts `NetworkTransportError` and exactly 3 calls.
## Tier-2 Schema Self-Correction Loop Deep-Dive (Step 4.3)
### Instructor `max_retries` Integration
The Tier-2 self-correction loop is enabled by passing `max_retries=settings.max_retries_schema` to the Instructor `chat.completions.create` call. When the LLM produces output that fails Pydantic `ArticleExtractionRecord` validation, Instructor captures the `ValidationError` traceback and sends it back as corrective feedback in the next attempt. This continues up to `max_retries_schema` (default 2) before raising `InstructorRetryException`.
### Recursive Network Error Classification: `_is_network_error`
Instructor's retry machinery nests exceptions. A shallow `isinstance` check on the top-level exception fails to detect network errors wrapped inside nested `InstructorRetryException` objects. The `_is_network_error` function recursively inspects:
1. The top-level exception type (`OpenAIError`, `TimeoutError`, `ConnectionError`).
2. The `failed_attempts` list of any `InstructorRetryException` (recursively).
3. The `__cause__` chain, including `last_attempt._exception` attributes.
This ensures `RateLimitError` (429) and `APIConnectionError` wrapped by Instructor are correctly classified as Tier-1 network errors, not Tier-2 schema errors.
### Feedback Message Format: `_build_self_correction_messages`
The utility formats `ValidationError` details into a corrective user message:
```
Your previous response failed schema validation. Please correct the following errors and return a valid JSON object matching the required schema exactly:
- confidence_score: confidence_score must be bounded between 0.0 and 1.0
- primary_category: Input should be 'politics', 'technology', 'business', ...
```
The message history is extended with an assistant acknowledgment and the corrective user message, mirroring Instructor's internal feedback format.
### Call Flow
1. `extract_record` builds messages and calls `_call_llm_with_schema_self_correction`.
2. `_call_llm_with_schema_self_correction` delegates to `_call_llm_with_retry` (Tier-1 Tenacity retries).
3. Instructor internally retries up to `max_retries_schema` times on `ValidationError`, sending feedback messages.
4. On success, returns validated `ArticleExtractionRecord`.
5. On `InstructorRetryException` exhaustion, `_is_network_error` classifies:
- Network error → `NetworkTransportError`.
- Schema error → `SchemaValidationError`.
### Unit Test Coverage
- `test_build_self_correction_messages_formats_validation_errors` — Verifies feedback message formatting.
- `test_tier2_schema_self_correction_passes_max_retries` — Verifies `max_retries` forwarding.
- `test_tier2_schema_self_correction_recovers_after_validation_error` — Verifies recovery after first failure.
- `test_tier2_schema_self_correction_exhausts_retries` — Verifies `SchemaValidationError` after exhaustion.
## SHA-256 Deduplication Cache Deep-Dive (Step 4.4)
### ContentHashCache Architecture
The `ContentHashCache` class in `src/utils/cache.py` provides a thread-safe local KV store keyed by a deterministic SHA-256 content hash. It uses an `OrderedDict` for LRU eviction, a `threading.Lock` for concurrent access safety, and optional JSON file persistence for restart survival.
### Cache Key Computation: `_compute_key`
The key is computed by serializing `[raw_text, model]` as a compact JSON array, encoding it as UTF-8, and hashing with SHA-256:
```python
@staticmethod
def _compute_key(raw_text: str, model: str) -> str:
"""Compute SHA-256 content hash over raw text + model (collision-proof)."""
payload = json.dumps([raw_text, model], separators=(",", ":")).encode("utf-8")
return hashlib.sha256(payload).hexdigest()
```
**Why JSON serialization over pipe concatenation?** Pipe concatenation (`f"{raw_text}|{model}"`) is collision-prone: `raw_text='a', model='b|c'` and `raw_text='a|b', model='c'` both produce `'a|b|c'`. JSON array serialization is unambiguous and preserves the exact boundary between components.
**Why include the model in the key?** Different models can produce different extraction results for the same input. Including the model name ensures cache isolation per model, preventing cross-model contamination.
### LRU Eviction: `_evict_lru`
The cache is bounded by a configurable `max_entries` (default 1000). On `set()`, entries are moved to the end of the `OrderedDict` (most-recently-used). When the store exceeds `max_entries`, `popitem(last=False)` removes the least-recently-used entry:
```python
def _evict_lru(self) -> None:
"""Evict least-recently-used entries when store exceeds max_entries."""
while len(self._store) > self._max_entries:
self._store.popitem(last=False)
```
On `get()`, a cache hit calls `move_to_end(key)` to mark the entry as recently used, maintaining accurate LRU ordering.
### Cache Read Path: `get`
1. Compute SHA-256 key from `raw_text + model`.
2. Acquire lock, look up store, move entry to end on hit (LRU).
3. Validate stored JSON against `ArticleExtractionRecord`.
4. On success, set `is_cached=True` and return the record.
5. On `ValidationError`, evict the invalid entry and return `None` (schema integrity).
### Cache Write Path: `set`
1. Compute SHA-256 key from `raw_text + model`.
2. Acquire lock, serialize record to JSON, store in `OrderedDict`.
3. Move entry to end (most-recently-used).
4. Evict least-recently-used entries if over `max_entries`.
5. Persist to JSON file if configured (best-effort).
### Engine Integration Flow
1. `extract_record` logs the request and checks `cache.get(raw_text, model)`.
2. **Cache hit** → returns cached `ArticleExtractionRecord` with `is_cached=True`, bypassing the entire LLM pipeline (Tier-1 retries and Tier-2 self-correction).
3. **Cache miss** → builds messages, calls LLM with Tier-1 retry + Tier-2 self-correction.
4. On successful extraction, `cache.set(raw_text, model, record)` stores the result.
5. Returns the record with `is_cached=False`.
### JSON Persistence & Corruption Resilience
The cache optionally persists to a JSON file. On initialization, `_load()` reads the file and filters to string-string pairs, gracefully handling missing or corrupt files. `_persist()` writes the store to disk (best-effort), logging a warning on `OSError`.
### Unit Test Coverage
- `test_compute_key_is_deterministic_sha256` — Identical inputs produce identical SHA-256 keys.
- `test_compute_key_differs_by_model` — Different models yield different keys for same text.
- `test_compute_key_differs_by_text` — Different text yields different keys for same model.
- `test_set_and_get_roundtrip` — Record stored via `set()` is retrievable via `get()`.
- `test_get_miss_returns_none` — Unknown key returns `None`.
- `test_get_returns_none_after_clear` — `clear()` empties the store.
- `test_persistence_to_json_file` — Cache persists to JSON file and reloads on new instance.
- `test_persistence_handles_corrupt_file` — Corrupt JSON file is ignored gracefully.
- `test_evicts_invalid_cached_record` — Invalid cached JSON is evicted and `get()` returns `None`.
- `test_lru_eviction_removes_oldest_entries` — LRU eviction removes least-recently-used entries.
- `test_compute_key_is_collision_proof` — JSON-serialized key prevents pipe-separator collisions.
- `test_engine_uses_cache_to_skip_llm_calls` — Duplicate extraction returns cached record without invoking LLM.
## FastAPI Application & CORS Setup Deep-Dive (Step 5.1)
### Architecture & Middleware Stack
The HTTP ingestion service defined in [`src/main.py`](file:///home/michael/Code/ai-engineering/projets/4_ASCGenerator/src/main.py) wraps the `ExtractionEngine` domain component within a FastAPI ASGI framework.
```python
app = FastAPI(
title=settings.app_name,
description="Automated Structured Content Generator Microservice",
version="0.1.0",
docs_url="/docs",
redoc_url="/redoc",
)
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
```
### Health Check Endpoint (`GET /health`)
The `/health` route provides a lightweight liveness probe for container orchestrators (Docker / Kubernetes):
```python
@app.get("/health", status_code=status.HTTP_200_OK)
async def health_check() -> Dict[str, str]:
"""Liveness & health check endpoint."""
return {
"status": "healthy",
"service": settings.app_name,
"timestamp": datetime.utcnow().isoformat(),
}
```
### Unit Test Coverage (`tests/test_api.py`)
- `test_health_endpoint_returns_200_and_healthy_status` — Verifies `/health` returns status code 200, `"status": "healthy"`, and ISO timestamp.
- `test_cors_middleware_headers` — Verifies CORS headers are attached when requests specify an `Origin` header.
- `test_cors_preflight_options_request` — Verifies HTTP `OPTIONS` preflight requests succeed with allowed methods.
- `test_openapi_documentation_accessible` — Verifies Swagger documentation UI is accessible at `/docs`.
- `test_openapi_schema_json_accessible` — Verifies raw OpenAPI JSON schema is accessible at `/openapi.json`.
## Live Single Extraction Endpoint Deep-Dive (Step 5.2)
### Architecture & Endpoint Implementation
The single extraction route in `src/main.py` exposes `POST /v1/extract` to client applications. It receives a `SingleExtractionRequest` Pydantic model, validates input parameters (minimum length, required fields), and forwards the request to `ExtractionEngine.extract_record()`.
```python
@app.post("/v1/extract", response_model=ArticleExtractionRecord, status_code=status.HTTP_200_OK)
async def extract_single(payload: SingleExtractionRequest) -> ArticleExtractionRecord:
"""Extract structured article record from single untrusted news payload."""
try:
record = await engine.extract_record(
raw_text=payload.raw_text,
source_name=payload.source_name,
source_url=str(payload.source_url) if payload.source_url else None,
)
return record
except ExtractionError as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=exc.message,
) from exc
except Exception as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Extraction service encountered an internal error.",
) from exc
```
### Input Validation & Exception Handling Flow
1. **Pydantic DTO Interception:** FastAPI automatically validates `SingleExtractionRequest`. Requests violating field constraints (e.g. `raw_text` < 10 characters or missing `source_name`) fail fast with HTTP 422 Unprocessable Entity.
2. **Engine Invocation:** Valid requests are passed to `engine.extract_record()`, triggering the content deduplication cache and LLM resilience pipeline.
3. **Domain Exception Mapping:** Any domain error inheriting from `ExtractionError` (such as `SchemaValidationError` or `NetworkTransportError`) is caught and re-raised as an HTTP 500 with its user-friendly message string.
4. **Generic Fallback:** Unhandled unexpected exceptions are caught and sanitized to an HTTP 500 internal error response, protecting internal stack traces.
### Unit Test Coverage (`tests/test_api.py`)
- `test_extract_single_endpoint_success` — Verifies valid extraction request returns HTTP 200 with complete `ArticleExtractionRecord`.
- `test_extract_single_endpoint_without_source_url` — Verifies optional `source_url` field behaves correctly when omitted.
- `test_extract_single_endpoint_validation_error` — Verifies HTTP 422 is returned when `raw_text` fails minimum length constraints.
- `test_extract_single_endpoint_missing_fields` — Verifies HTTP 422 is returned when required `source_name` is missing.
- `test_extract_single_endpoint_domain_exception` — Verifies `ExtractionError` is caught and returned as HTTP 500 with error detail.
- `test_extract_single_endpoint_unexpected_exception` — Verifies unexpected exceptions return generic HTTP 500 detail.
## Async Concurrency Batch Extraction Endpoint Deep-Dive (Step 5.3)
### Architecture & Concurrency Control
The batch extraction endpoint `POST /v1/extract/batch` processes arrays of up to 100 news payloads (`SingleExtractionRequest`) concurrently. Rather than executing requests sequentially or using unthrottled parallelism, `ExtractionEngine.extract_batch` uses an `asyncio.Semaphore` bounded by `settings.max_batch_concurrency` (default 10) to cap parallel LLM API calls.
```python
async def extract_batch(
self,
requests: List[SingleExtractionRequest],
concurrency_limit: Optional[int] = None,
) -> List[ArticleExtractionRecord]:
"""Process batch extractions concurrently with asyncio.Semaphore rate limiting."""
limit = concurrency_limit or settings.max_batch_concurrency
semaphore = asyncio.Semaphore(limit)
async def _extract_with_semaphore(
req: SingleExtractionRequest,
) -> Optional[ArticleExtractionRecord]:
async with semaphore:
try:
return await self.extract_record(
raw_text=req.raw_text,
source_name=req.source_name,
source_url=str(req.source_url) if req.source_url else None,
)
except Exception as exc:
logger.error(
"Batch item extraction failed",
source_name=req.source_name,
error=str(exc),
)
return None
tasks = [_extract_with_semaphore(req) for req in requests]
results = await asyncio.gather(*tasks)
return [record for record in results if record is not None]
```
### Partial Failure Aggregation & Response Metrics
In `src/main.py`, the endpoint handler receives the array of extracted records and computes execution metrics:
```python
@app.post(
"/v1/extract/batch",
response_model=BatchExtractionResponse,
status_code=status.HTTP_200_OK,
)
async def extract_batch(payload: BatchExtractionRequest) -> BatchExtractionResponse:
"""Extract structured article records concurrently from a batch of up to 100 news payloads."""
try:
records = await engine.extract_batch(payload.articles)
return BatchExtractionResponse(
total_processed=len(payload.articles),
successful=len(records),
failed=len(payload.articles) - len(records),
records=records,
)
except ExtractionError as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=exc.message,
) from exc
except Exception as exc:
raise HTTPException(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail="Batch extraction service encountered an internal error.",
) from exc
```
### Input Boundary Validation
The `BatchExtractionRequest` Pydantic model enforces batch bounds:
```python
class BatchExtractionRequest(BaseModel):
articles: List[SingleExtractionRequest] = Field(..., min_length=1, max_length=100)
```
If a client sends an empty payload (`[]`) or exceeds 100 items (`101`+), FastAPI automatically intercepts the request and returns an HTTP 422 Unprocessable Entity error prior to engine invocation.
### Unit Test Coverage (`tests/test_batch.py`)
- `test_extract_batch_concurrency_semaphore` — Verifies `extract_batch` strictly enforces the concurrency limit using `asyncio.Semaphore`.
- `test_extract_batch_100_articles_concurrently` — Verifies processing a batch of 100 articles succeeds without errors.
- `test_extract_batch_partial_failures` — Verifies extraction failures on individual items are logged and filtered out gracefully.
- `test_extract_batch_endpoint_success` — Verifies `POST /v1/extract/batch` returns HTTP 200 with accurate `BatchExtractionResponse` metrics.
- `test_extract_batch_endpoint_exceeds_max_limit` — Verifies HTTP 422 is returned when payload exceeds 100 articles.
- `test_extract_batch_endpoint_empty_articles` — Verifies HTTP 422 is returned when `articles` array is empty.
## Phase 6 Monitoring, Schema Inspection & Telemetry Deep-Dive
### Module Architecture (`src/monitoring.py`)
Phase 6 introduces a decoupled telemetry and diagnostic inspection subsystem composed of three core components:
1. **`ExtractionMetricsRegistry`**:
- In-memory tracker for operational metrics: request counts, success/failure rates, cache hit ratio, average latency (ms), average confidence score, total tokens (prompt + completion), and estimated USD cost.
- Microsecond aggregation with zero-I/O overhead.
2. **`SchemaInspector`**:
- Uses Pydantic V2 model reflection (`ArticleExtractionRecord.model_json_schema()`) to export JSON schema definitions and inspect field constraint rules.
- Provides `validate_payload(raw_dict)` to perform field-level diagnostics, reporting validation error strings, missing required fields, and parsed records.
3. **`InspectionLogBuffer`**:
- Maintains a fixed-capacity FIFO circular queue (default 100 entries) storing recent execution trace logs for live UI visualization.
### REST Endpoints (`src/main.py`)
- `GET /v1/monitoring/metrics`: Returns aggregated operational metrics summary (`success_rate`, `cache_hit_rate`, `avg_latency_ms`, `avg_confidence_score`, `total_cost_usd`).
- `POST /v1/monitoring/inspect`: Validates raw input payloads against Pydantic schema firewall with detailed diagnostic output.
- `GET /v1/monitoring/schema`: Returns target Pydantic model structure, field metadata, and JSON schema.
- `GET /v1/monitoring/logs`: Returns recent execution trace logs from the circular log buffer.
- `DELETE /v1/monitoring/reset`: Resets telemetry state and clears the inspection log buffer.
### Integration & Automated Telemetry Capture
Extraction endpoints (`POST /v1/extract`, `POST /v1/extract/batch`) automatically invoke `metrics_registry.record_extraction()` and append log entries to `inspection_logger`.
### Unit Test Coverage (`tests/test_monitoring.py`)
- `test_metrics_registry_records_telemetry` — Validates metric accumulation and average calculation.
- `test_metrics_registry_reset` — Validates counter reset behavior.
- `test_schema_inspector_details` — Validates JSON schema structural inspection.
- `test_schema_inspector_validate_valid_payload` — Validates compliance for valid payload models.
- `test_schema_inspector_validate_invalid_payload` — Identifies missing required fields and validation errors.
- `test_inspection_log_buffer` — Tests log entry buffer appending, retrieval order, capacity limit, and clearing.
- `test_monitoring_metrics_endpoint` — Validates `GET /v1/monitoring/metrics` REST endpoint.
- `test_monitoring_schema_endpoint` — Validates `GET /v1/monitoring/schema` REST endpoint.
- `test_monitoring_inspect_endpoint` — Validates `POST /v1/monitoring/inspect` payload inspection endpoint.
- `test_monitoring_logs_endpoint` — Validates `GET /v1/monitoring/logs` execution log feed endpoint.
- `test_monitoring_reset_endpoint` — Validates `DELETE /v1/monitoring/reset` state reset endpoint.
- `test_extraction_updates_monitoring_telemetry` — Validates that extraction calls automatically record telemetry metrics and log trace entries.
## Schema Unit Tests Suite Deep-Dive (Step 7.1)
### Test Architecture (`tests/test_schemas.py`)
Step 7.1 implements dedicated schema unit tests covering all Pydantic V2 models, value enums, literal domains, custom field validators, boundary constraints, and API DTOs in `src/schemas.py`.
### Key Validation Cases Covered:
1. **Model Instantiation & Field Types:**
- `ArticleAuthor`: Validates string length constraints and email string format validation (`ArticleAuthor(name="", email="invalid")` raises `ValidationError`).
- `NamedEntity`: Validates category `Literal["ORGANIZATION", "PERSON", "LOCATION", "PRODUCT"]` and `SentimentLabel` enums (`NamedEntity(category="INVALID")` raises `ValidationError`).
- `FinancialMetric`: Validates numerical float values, units, and timeframe strings.
- `ArticleExtractionRecord`: Validates master extraction schema defaults (`schema_version="1.0"`, UUID generation for `article_id`, datetime generation for `extracted_at`, default empty lists for nested models, `is_cached=False`).
2. **Confidence Score Boundary Validation:**
- `@field_validator("confidence_score")`: Verifies custom validation enforcing strict range `0.0 <= score <= 1.0`.
- `test_confidence_score_out_of_bounds`: Asserts `confidence_score=1.5` raises `ValidationError` with message `"confidence_score must be bounded between 0.0 and 1.0"`.
- `test_confidence_score_lower_bound_violation`: Asserts `confidence_score=-0.1` raises `ValidationError`.
- `test_confidence_score_boundary_values`: Asserts boundary scores `0.0` and `1.0` pass validation cleanly.
3. **Field Length Bounds & Domain Enums:**
- Title minimum length constraint (`min_length=5`) and summary minimum length constraint (`min_length=10`).
- Strict value assertions for taxonomy enums (`ArticleCategory`, `SentimentLabel`, `ImpactLevel`).
4. **API DTOs & Inspection Wrappers:**
- `SingleExtractionRequest`: Validates minimum raw text length (`min_length=10`).
- `BatchExtractionRequest`: Validates list item bounds (`min_length=1`, `max_length=100`).
- `BatchExtractionResponse`: Verifies serialization of batch result metrics and record lists.
- `InspectionPayloadRequest`: Verifies wrapping of raw payload dictionaries for inspection endpoints.
### Dashboard Integration
All test functions in `tests/test_schemas.py` are mapped into `TEST_DESCRIPTIONS` in `dashboard/src/app/page.tsx` for dynamic discovery and single-click execution via the Next.js test runner UI.
## Defensive Security Tests Suite Deep-Dive (Step 7.2)
### Technical Architecture (`src/security.py` & `tests/test_security.py`)
Step 7.2 establishes defensive prompt security unit tests and payload sanitization mechanisms to protect the extraction pipeline against direct and indirect prompt injection attacks.
```python
def prepare_user_payload(raw_article_text: str) -> str:
"""Sanitize raw input text and enclose inside untrusted content XML delimiters."""
sanitized_text = re.sub(
r"",
"[TAG_REMOVED]",
raw_article_text,
flags=re.IGNORECASE,
)
return f"""
{sanitized_text}
"""
```
### Core Security Defenses Implemented:
1. **Developer Authority Isolation (`SYSTEM_PROMPT`):**
- System instructions establish top-priority developer authority, instructing the LLM to process content *only* inside `` tags and treat embedded override attempts as plain data strings.
2. **Regex XML Tag Neutralization (`prepare_user_payload`):**
- Uses case-insensitive regex pattern matching (``) to convert malicious closing tags into `[TAG_REMOVED]`, preventing tag breakout attacks regardless of casing or whitespace padding.
3. **Native Message-History Few-Shot Structuring (`FEW_SHOT_MESSAGES`):**
- Uses native `user` and `assistant` message pairs to ground JSON schema expectations and target format output without cluttering developer system instructions.
4. **Heuristic Keyword Trigger Scanner (`detect_injection_keywords`):**
- Scans untrusted input text for known prompt injection triggers (`"ignore previous instructions"`, `"system override"`, `"dan mode"`), exposing an observational risk function.
### Unit Test Suite Breakdown (`tests/test_security.py`):
- `test_prepare_user_payload_sanitization` — Verifies replacement of closing XML tags `` with `[TAG_REMOVED]`.
- `test_prepare_user_payload_case_insensitive_sanitization` — Verifies neutralization of mixed-case and padded tags (``).
- `test_prepare_user_payload_multiple_and_nested_tags` — Verifies neutralization of multiple consecutive or nested tag breakout attempts.
- `test_prepare_user_payload_structure_enclosure` — Verifies raw article text is safely enclosed within `` delimiters.
- `test_system_prompt_developer_authority_and_constraints` — Asserts `SYSTEM_PROMPT` contains developer authority rules and override prohibitions.
- `test_few_shot_messages_format_and_roles` — Validates native message turn structures and assistant JSON payload compliance.
- `test_direct_prompt_injection_payload_sanitization` — Verifies direct injection payloads ("System Override", "DAN Mode", SQL injections) are wrapped safely without breaking execution.
- `test_indirect_prompt_injection_payload_sanitization` — Verifies indirect prompt injections embedded in news quotes are treated strictly as passive data.
- `test_detect_injection_keywords` — Tests heuristic risk scanner pattern matching accuracy on clean vs adversarial text.
- `test_security_payload_unicode_and_special_chars` — Verifies Unicode characters, currency symbols, and XML special entities (`&`, `<`, `>`) remain intact.
### Dashboard Test Runner Registration
All 10 security test cases are mapped directly in `TEST_DESCRIPTIONS` within `dashboard/src/app/page.tsx` for dynamic discovery and UI test execution.
## Live Engine & Resilience Tests Suite Deep-Dive (Step 7.3)
### Technical Architecture (`src/engine.py` & `tests/test_engine.py`)
Step 7.3 implements a comprehensive unit and integration test suite for the `ExtractionEngine`, verifying live engine execution, Tier-1 network retries, Tier-2 schema self-repair feedback loops, deduplication short-circuiting, concurrent batch processing, and telemetry integration.
### Core Engine & Resilience Mechanisms Tested:
1. **Tier-1 Transport Retries & Network Recovery (`_is_retryable_error`):**
- Verifies Tenacity exponential backoff (`multiplier=1`, `min=2s`, `max=10s`) automatically recovers from transient HTTP `429 RateLimitError`, HTTP `5xx InternalServerError`, and `APITimeoutError`.
- Asserts persistent network outages raise `NetworkTransportError` after max retry attempts.
2. **Tier-2 Schema Self-Repair Loop (`_build_self_correction_messages`):**
- Verifies Instructor `max_retries=settings.max_retries_schema` forwards Pydantic `ValidationError` tracebacks to the LLM as corrective feedback user messages.
- Asserts self-correction turns recover valid schema output or raise `SchemaValidationError` when retries are exhausted.
3. **SHA-256 Deduplication Caching Integration (`ContentHashCache`):**
- Verifies identical raw article text returns cached `ArticleExtractionRecord` on subsequent calls without making an external LLM call (`calls["count"] == 1`).
4. **Async Batch Concurrency & Semaphore Rate Limiting (`extract_batch`):**
- Verifies concurrent processing of article lists using `asyncio.Semaphore`, ensuring high throughput without rate limit saturation.
5. **Integrated Real-Time Telemetry & Log Inspection (`src/monitoring.py`):**
- Verifies `ExtractionEngine.extract_record` automatically records latency, success/failure counts, cache hit flags, and confidence scores into `metrics_registry` and appends trace logs into `inspection_logger`.
### Unit & Integration Test Suite Breakdown (`tests/test_engine.py`):
- `test_health_endpoint` — Validates `GET /health` REST endpoint returning HTTP 200 OK.
- `test_extraction_engine_init` — Validates `ExtractionEngine` initialization with API key, model selection, and cache.
- `test_extraction_engine_missing_key_raises_config_error` — Asserts missing OpenAI API key raises `ConfigurationError`.
- `test_extraction_engine_network_error_wrapping` — Asserts API connection drops raise `NetworkTransportError`.
- `test_tier1_retry_429_rate_limit_succeeds_after_retries` — Verifies recovery from HTTP 429 Rate Limits after transient retries.
- `test_tier1_retry_5xx_succeeds_after_retries` — Verifies recovery from 5xx server errors after transient retries.
- `test_tier1_retry_timeout_error_recovery` — Verifies recovery from `APITimeoutError` via Tenacity Tier-1 retries.
- `test_tier1_retry_exhausts_after_max_attempts` — Asserts persistent 429 errors raise `NetworkTransportError` upon exhaustion.
- `test_build_self_correction_messages_formats_validation_errors` — Validates formatting of `ValidationError` feedback messages.
- `test_tier2_schema_self_correction_passes_max_retries` — Verifies `max_retries_schema` parameter forwarding to Instructor.
- `test_tier2_schema_self_correction_recovers_after_validation_error` — Verifies recovery following schema self-repair turn.
- `test_tier2_schema_self_correction_exhausts_retries` — Asserts persistent validation failures raise `SchemaValidationError`.
- `test_engine_cache_deduplication_hit_and_miss` — Verifies deduplication cache short-circuiting on identical raw text.
- `test_engine_extract_batch_concurrency` — Verifies concurrent batch extractions bounded by `asyncio.Semaphore`.
- `test_engine_telemetry_metrics_tracking` — Verifies automatic telemetry metric accumulation and inspection log entry appending.
- `test_extraction_engine_generic_exception_wrapping` — Asserts unexpected exceptions map to `ExtractionError`.
- `test_extraction_engine_extract_record_mock` — Validates complete end-to-end mocked extraction pipeline.
- `test_extract_single_endpoint` — Validates `POST /v1/extract` FastAPI REST endpoint.
- `test_extract_batch_endpoint` — Validates `POST /v1/extract/batch` FastAPI REST endpoint.
### Dashboard Test Runner Integration
All 19 engine test cases are dynamically discovered and registered into `TEST_DESCRIPTIONS` within `dashboard/src/app/page.tsx` for interactive execution via the Next.js test runner UI.
---
## 8. Step 7.4 — 100-Article Benchmark Dataset Evaluation
### Component & Data Flow Overview
```text
Dataset File (data/dataset_100_articles.jsonl)
│
▼
BenchmarkEvaluator.load_dataset()
│
▼
asyncio.Semaphore(10) Concurrent Batch Processing
│
▼
ExtractionEngine.extract_record()
│
├── Schema Compliance Bounding (ArticleExtractionRecord)
├── Prompt Injection Resilience Check (Adversarial Override Defense)
└── Metric Calculation (P50/P90/P95 Latency, Compliance Rate, Throughput)
│
▼
BenchmarkRunResponse (POST /v1/benchmark/run)
```
### Technical Function Breakdown & Modules
1. **`data/dataset_100_articles.jsonl`**: Standardized 100-article corpus containing ground-truth news payloads across technology, finance, healthcare, science, cybersecurity, and adversarial prompt injection attack vectors.
2. **`src/schemas.py`**:
- `BenchmarkItemResult`: Detailed per-article evaluation status record.
- `BenchmarkMetrics`: Aggregated benchmark statistics (schema compliance rate, prompt injection resilience rate, confidence score validity rate, latency P50/P90/P95, throughput).
- `BenchmarkRunRequest`: Input configuration model for benchmark runs.
- `BenchmarkRunResponse`: Master response wrapper returning evaluation metrics and itemized results.
3. **`src/evaluator.py`**:
- `calculate_percentile(values, percentile)`: Linear interpolation percentile computation for latency analysis.
- `BenchmarkEvaluator`: Core evaluation runner managing dataset loading, async semaphore concurrency, metric computation, inspection logging, and error handling.
4. **`src/main.py`**:
- `POST /v1/benchmark/run`: REST API endpoint triggering benchmark dataset evaluations.
5. **`tests/test_benchmark.py`**:
- Automated unit and integration test suite asserting dataset completeness, percentile accuracy, evaluator execution, file error handling, and REST endpoint integration.
---
## 9. Step 8 — Hardened Containerization & Production Release
### Architecture & Security Overview
Step 8 packages the microservice into a production-ready, security-hardened Docker container and provides orchestration scripts for deployment:
```text
[ Docker Build Context ]
│
├── Stage 1: builder (python:3.11-slim + Poetry) ──► Install locked deps into .venv
│
└── Stage 2: runtime (python:3.11-slim)
├── Copy /app/.venv from builder
├── Create Non-Root User (appuser, UID 10001)
├── HEALTHCHECK --interval=30s CMD curl -f http://localhost:8000/health
└── USER 10001:10001
```
### Component Breakdown & Implementation Details
1. **`Dockerfile`**:
- Multi-stage build structure separating build toolchains (`gcc`, `curl`, `poetry`) from the final runtime image.
- Enforces execution under unprivileged non-root user `appuser` (UID `10001`, GID `10001`).
- Includes container-native `HEALTHCHECK` directive invoking `/health` every 30s.
2. **`docker-compose.yml`**:
- Multi-container service orchestration binding `api` (port 8000) and `dashboard` (port 3000).
- Enforces `security_opt: [no-new-privileges:true]` and healthcheck dependency conditions (`service_healthy`).
3. **`.dockerignore`**:
- Explicitly excludes `.venv`, `.git`, `tests`, `docs`, `dashboard/node_modules`, and cache directories to optimize image context build size (<250 MB).
4. **`src/container.py` & `src/schemas.py` (`ContainerInfoResponse`)**:
- `inspect_container_security()`: Introspects running container environment, user UID, and security flags (`multi-stage-build`, `non-root-uid-10001`, `no-new-privileges`).
- `get_container_diagnostics()`: Provides diagnostic dictionary for monitoring dashboards and automated health checks.
5. **`src/main.py` (`GET /v1/container/info` & `GET /health`)**:
- Exposes REST API endpoint `GET /v1/container/info` returning `ContainerInfoResponse`.
- Enriches `GET /health` with release version `1.0.0`, `user_uid`, and `is_non_root` status.
6. **`tests/test_containerization.py`**:
- Automated unit test suite verifying `Dockerfile` stage markers, `docker-compose.yml` healthchecks and security options, `.dockerignore` exclusion rules, runtime container security functions, and REST API endpoints.