# Bedrock Image Analysis Lambda

This Lambda function analyzes product images using AWS Bedrock's Claude 3 Vision model to determine if specific retail issues have been resolved.

## Features

- Analyzes product images using Claude 3 Vision
- Connects to Supabase for data storage and retrieval
- Handles image downloads from Supabase storage
- Updates analysis results and error states
- **Optional S3 artifacts:** save shelf image, prompt, and response per run (see below)
- Comprehensive error handling and logging
- REST API endpoint via API Gateway
- **Step Functions retry workflow:** lists incomplete GenAI work for tasks **fixed today (Africa/Johannesburg)** and runs the same Bedrock analysis in **parallel** (configurable concurrency; see below)
- **Optional fallback model:** if the primary inference profile returns throttling (`ThrottlingException` / `TooManyRequestsException`), the Lambda can retry **once** with `BEDROCK_FALLBACK_INFERENCE_PROFILE_ARN` (template parameter **`BedrockFallbackInferenceProfileArn`**) — e.g. **Sonnet 4.5** when **Sonnet 4.6** hits daily/RPM limits. Logs include `model_id` on each **`BedrockUsage`** line.

## Bedrock retry (Step Functions)

Use this when shelf images were slow to upload or earlier runs failed (`processed_genai` still false).

1. **State machine** (CloudFormation output `BedrockRetryWorkflowArn`): name `{StackName}-BedrockRetry`.
2. **Flow:** `ListBedrockRetryTasks` loads **one page** of candidates (default **500** per page, `RETRY_LIST_BATCH_SIZE`). If that page has work, **Map** invokes `BedrockImageAnalysis` for each item in parallel up to **`BedrockRetryMapMaxConcurrency`**. When **`has_more`** is true, the workflow lists the **next page** (same SAST `run_date`) until everything is processed. The Map state uses **`ResultPath: null`** so Step Functions does **not** aggregate huge Bedrock responses (avoids the **~256 KB** state size limit / `States.DataLimitExceeded`).
3. **Throughput vs Bedrock limits:** AWS enforces **per-model / per-region quotas** (on-demand **requests per minute** and **tokens per minute**). Parallel Step Functions branches each call **InvokeModel**; if **`ThrottlingException` / “Too many requests”** appears, **lower** **`BedrockRetryMapMaxConcurrency`** (deploy default is **2**) or request a **quota increase** in **Service Quotas** → *Amazon Bedrock* → your model / inference profile in **`eu-west-1`**. The analysis Lambda also **retries with backoff** on throttling, but sustained overload still requires lower concurrency or higher quota.
4. **Selection rules (list Lambda):**
   - Tasks with `fixed_at` in the **current SAST calendar day** (UTC window computed from `Africa/Johannesburg`); `fixed_at` not null.
   - Task not complete: `processed_genai` is not true.
   - **Task table only** (no `product_images` in the list Lambda). **Only `task_id` items** (one Map branch per task). **`task.image_url` must be non-empty**; tasks without a shelf URL are **skipped**.
   - **500 items per page** by default (`RETRY_LIST_BATCH_SIZE`); the workflow repeats until all matching tasks are processed.
5. **Manual start:** Step Functions console → **Start execution**. Input **`{}`** → first list uses **today** in **Africa/Johannesburg**. To process a **previous calendar day** (tasks whose **`fixed_at`** falls that SAST day), start with:
   ```json
   { "run_date": "2026-05-04" }
   ```
   Use **`YYYY-MM-DD`** for the **SAST date** you care about (same interpretation as **`ListBedrockRetryTasks`**). Later pages keep that **`run_date`** automatically. You can still invoke **`ListBedrockRetryTasks`** directly with **`{"offset":0,"run_date":"YYYY-MM-DD"}`** for ad-hoc checks without Step Functions.
6. **Optional schedule:** deploy with template parameter `EnableBedrockRetrySchedule=true`. EventBridge runs **daily at 22:00 South African time** (**Africa/Johannesburg**, UTC+2 → **20:00 UTC** in `template.yaml`). Adjust **`cron(…)`** there if you need a different local time.

**Deploy tuning:** After confirming quotas in Service Quotas, you can raise parallelism, e.g. `sam deploy --parameter-overrides BedrockRetryMapMaxConcurrency=5`. If throttling persists, use **`1`** (strictly serial Bedrock calls).

## Bedrock batch (product-centric Step Functions)

Use this when **many pending rows share the same product** (same `product_id`) but **different shelf photos** — e.g. multiple stores or visits for one SKU. One Bedrock call sends the **product reference image and instructions once**, then **N shelf images** (each labeled with `task_id` / `image_id`), and returns **JSON** with one result block per shelf. This cuts repeated reference-image and prompt tokens versus **`BedrockRetry`** (which invokes **`BedrockImageAnalysis` once per **task** via **`task_id`**).

1. **State machine** (output `BedrockBatchWorkflowArn`): name `{StackName}-BedrockBatch`.
2. **Flow:** `ListBedrockBatchGroups` returns **pages** of **`groups`**. Each group is `{ product_id, items }` where **`items`** are **`{ task_id }` only** (built from the **`task`** table), up to **`BedrockBatchMaxShelfImages`** per chunk. **Map** invokes **`BedrockBatchAnalyze`** per group with **`BedrockBatchMapMaxConcurrency`** (default **2**). **`ResultPath: null`** on Map (same 256 KB state rationale as retry). Pagination: **`has_more`** / **`next_offset`** / **`run_date`** (same SAST “fixed today” window as retry).
3. **Eligibility:** Same task filter as retry (`fixed_at` today SAST, `processed_genai` incomplete). Tasks **without `product_id`** or **without `image_url`** (shelf photo) are **skipped** here; use **`BedrockRetry`** where needed. The **product** must still have **`image_path`** set (reference packshot) or the batch Lambda fails that group.
4. **Bedrock usage:** **`BedrockBatchAnalyze`** performs **one `InvokeModel` per group** (primary profile only; no fallback profile and no Step Functions retry on that Map branch), so quotas are not multiplied by retries.
5. **Parameters (template):** `BedrockBatchMaxShelfImages`, `BedrockBatchListPageSize`, `BedrockBatchMapMaxConcurrency`.
6. **Manual start:** Step Functions → **Start execution** → **`{}`** for today SAST, or **`{ "run_date": "YYYY-MM-DD" }`** for a past day (same as **`BedrockRetry`**).

## Setup

1. Install dependencies:
```bash
pip install -r app/requirements.txt
```

2. Deploy using SAM:
```bash
sam build
sam deploy
```

3. **Optional – S3 artifacts:** To save each run’s shelf image, prompt, and response to S3, set the `ArtifactsBucketName` parameter to an existing S3 bucket name at deploy time. Leave it empty (default) to disable. When enabled, objects are written under `artifacts/{image_id_or_task_id}/{timestamp}/` as `shelf_image.jpg`, `product_image.jpg` (when product has an image), `prompt.txt`, and `response.txt`. To remove this behaviour, redeploy with `ArtifactsBucketName` empty or omit the parameter; no code change is required.

## API Usage

Two routes on the same API (see stack output `BedrockAnalysisApi` and `BedrockAnalysisTestApiUrl`).

### `POST /analyze-image-test` (dry-run / model test)

Same request fields as **`/analyze-image`** (`image_id` or `task_id`), plus optional **`model_id`**:

- Bedrock runs only through **`test_app.invoke_bedrock_test_model`** — **not** `analyze_image_with_bedrock` (production). Use this route to compare models without touching the prod invoke path.
- **`model_id`** resolution: JSON **`model_id`** / **`modelId`** → env **`BEDROCK_TEST_MODEL_ID`** (template defaults to **Qwen**). If you omit **`model_id`** in the body, the test Lambda uses Qwen (**`qwen.qwen3-vl-235b-a22b`**) via **Converse** unless you change **`BEDROCK_TEST_MODEL_ID`**.
- **`model_id`** starting with **`qwen.`** → **Converse**; otherwise → **InvokeModel** (Anthropic Messages body). Response includes **`test_invoke_path`:** **`test_converse`** | **`test_invoke_model`**.
- Prompt is the shorter harness in **`test_app`** (not byte-identical to production **`/analyze-image`**). IAM: add **`foundation-model`** ARNs for any extra models you pass.

```json
{
  "task_id": "YOUR-TASK-UUID",
  "model_id": "eu.anthropic.claude-sonnet-4-5"
}
```

Omit **`model_id`** to use the default **Qwen** model (**Converse**). Pass **`model_id`** to try another model (e.g. **`eu.anthropic.claude-sonnet-4-5`**) or set **`BEDROCK_TEST_MODEL_ID`** on the test Lambda.

**No** updates to **`task`** / **`product_images`**, **no** S3 artifacts, **no** `genai_error` writes on failure (errors are only in the JSON response).

**Lambda console test:** Put **`task_id`** / **`image_id`** at the **root** of the test JSON (`{"task_id":"..."}`). API Gateway sends the same fields inside a string **`body`**; both shapes are accepted.

**Qwen / Converse payload size:** Bedrock documents (e.g. images) are often capped around **4.5 MiB per item**, but **Qwen’s gateway** can still reject when **two large images + prompt** exceed its **total request body** buffer (`validation_error` / “length limit exceeded”). The test Lambda compresses each image to **`BEDROCK_TEST_CONVERSE_MAX_IMAGE_BYTES`** (default **1 500 000** bytes ~1.5 MiB per image). Lower that env var if errors persist.

### `POST /analyze-image` (production)

Persists GenAI results to Supabase (and optional S3 artifacts when configured).

### Request Format
```json
{
    "image_id": "uuid-of-image-to-analyze"
}
```

### Response Format
```json
{
    "message": "Image analysis completed successfully",
    "result": {
        "result": true|false,
        "message": "Explanation of the analysis result",
        "confidence": "75%",
        "facings_count": 10,
        "position": "Shelf 4, left (4); Shelf 5, right (3); Shelf 7, center-right (3)",
        "positions": [{"shelf": 4, "position": "left", "facings": 4}, {"shelf": 5, "position": "right", "facings": 3}, {"shelf": 7, "position": "center-right", "facings": 3}],
        "has_pi_label": true
    }
}
```
Positions: Shelf 1 = first (topmost) visible row, then left|center|right. The model outputs POSITIONS as a JSON array of {shelf, position, facings} per location. The API returns both "position" (backward-compatible string) and "positions" (array). Only the EXACT product counts (e.g. Fruit Cocktail not Peach Slices); same brand but different product is excluded. The prompt requires counting total visible shelves and scanning every shelf 1..N so none are missed, and a double-check that FACINGS_COUNT equals the sum of per-location facings. See BEDROCK_SETUP.md for setup.

### Error Response
```json
{
    "error": "Error message describing what went wrong"
}
```

## Database Schema

The function interacts with the following Supabase tables:

### product_images
- id: UUID
- product_id: UUID (foreign key to products)
- task_id: UUID (foreign key to tasks)
- image_path: string
- processed_genai: boolean
- genai_result: boolean
- genai_message: string
- genai_error: string

### products
- id: UUID
- name: string
- ... other fields

### tasks
- id: UUID
- name: string
- ... other fields

## Error Handling

- All errors are logged with full stack traces
- Database errors are captured and stored in the genai_error field
- HTTP 500 responses include error details in the response body

## CORS

The API endpoint supports CORS with the following configuration:
- Allowed Methods: POST, OPTIONS
- Allowed Headers: Content-Type, X-Amz-Date, Authorization, X-Api-Key, X-Amz-Security-Token
- Allowed Origins: * (all origins)

## Monitoring

- CloudWatch logs contain detailed execution information
- Supabase tables track processing status and results
- Error states are preserved for debugging
- API Gateway metrics available in CloudWatch