@@ -111,6 +111,29 @@ curl -L -X POST 'http://0.0.0.0:4000/v1/images/generations' \
 </TabItem>
 </Tabs>
 
+## Amazon Nova Canvas - Image Edit
+
+Use OpenAI-compatible `image_edit()` with Bedrock Nova Canvas (`amazon.nova-canvas-v1:0`). Requests use the same `InvokeModel` API as generation; LiteLLM maps inputs to [Nova Canvas task types](https://docs.aws.amazon.com/nova/latest/userguide/image-gen-access.html):
+
+| Scenario | `taskType` sent to Bedrock |
+|----------|----------------------------|
+| Image + prompt (no mask) | `IMAGE_VARIATION` |
+| Image + prompt + mask | `INPAINTING` (`inPaintingParams.image`, `maskImage` or `maskPrompt`) |
+| `taskType: OUTPAINTING` + `mask` or `maskPrompt` | `OUTPAINTING` (Bedrock requires one; LiteLLM raises a clear error if both are missing) |
+| `taskType: BACKGROUND_REMOVAL` | `BACKGROUND_REMOVAL` |
+
+```python
+from litellm import image_edit
+
+response = image_edit(
+    image=open("photo.png", "rb"),
+    prompt="Add soft sunset lighting",
+    model="bedrock/amazon.nova-canvas-v1:0",
+)
+```
+
+For **`BACKGROUND_REMOVAL`**, the AWS request must not include `imageGenerationConfig`; LiteLLM omits it for that task even if you pass `size`, `n`, `seed`, etc. Additional Nova Canvas inference IDs for image edit should set **`supports_nova_canvas_image_edit`: true** in `model_prices_and_context_window.json` (see `amazon.nova-canvas-v1:0`).
+
 ## Using Inference Profiles with Image Generation
 
 For AWS Bedrock Application Inference Profiles with image generation, use the `model_id` parameter to specify the inference profile ARN:
@@ -147,4 +170,3 @@ model_list:
 ## Authentication
 
 All standard Bedrock authentication methods are supported for image generation. See [Bedrock Authentication](./bedrock#boto3---authentication) for details.
-
 
@@ -0,0 +1,12 @@
+-- AlterTable
+ALTER TABLE "LiteLLM_PromptTable" ADD COLUMN "environment" TEXT NOT NULL DEFAULT 'development';
+ALTER TABLE "LiteLLM_PromptTable" ADD COLUMN "created_by" TEXT;
+
+-- DropIndex (old unique constraint)
+DROP INDEX IF EXISTS "LiteLLM_PromptTable_prompt_id_version_key";
+
+-- CreateIndex (new unique constraint)
+CREATE UNIQUE INDEX "LiteLLM_PromptTable_prompt_id_version_environment_key" ON "LiteLLM_PromptTable"("prompt_id", "version", "environment");
+
+-- CreateIndex (new composite index)
+CREATE INDEX "LiteLLM_PromptTable_prompt_id_environment_idx" ON "LiteLLM_PromptTable"("prompt_id", "environment");
@@ -321,11 +321,12 @@ model LiteLLM_MCPServerTable {
   byok_description      String[] @default([])
   byok_api_key_help_url String?
   source_url            String?
-  approval_status       String?   @default("active")
-  submitted_by          String?
-  submitted_at          DateTime?
-  reviewed_at           DateTime?
-  review_notes          String?
+  // BYOM submission lifecycle
+  approval_status  String?   @default("active")
+  submitted_by     String?
+  submitted_at     DateTime?
+  reviewed_at      DateTime?
+  review_notes     String?
 
@@index([approval_status])
 }
@@ -1001,12 +1002,15 @@ model LiteLLM_PromptTable {
   id String @id @default(uuid())
   prompt_id String
   version Int @default(1)
+  environment String @default("development")
+  created_by String?
   litellm_params Json
   prompt_info Json?
   created_at DateTime @default(now())
   updated_at DateTime @updatedAt
 
-  @@unique([prompt_id, version])
+  @@unique([prompt_id, version, environment])
+  @@index([prompt_id, environment])
@@index([prompt_id])
 }
 
 
@@ -38,6 +38,102 @@ def __init__(self, response: AnthropicMessagesResponse):
         self.chunks = self._create_streaming_chunks()
         self.current_index = 0
 
+    def _create_content_block_chunks(
+        self, block_dict: Dict[str, Any], index: int
+    ) -> List[bytes]:
+        """Build SSE chunks for a single content block."""
+        chunks = []
+        block_type = block_dict.get("type")
+
+        if block_type == "text":
+            content_block_start = {
+                "type": "content_block_start",
+                "index": index,
+                "content_block": {"type": "text", "text": ""},
+            }
+            chunks.append(
+                f"event: content_block_start\ndata: {json.dumps(content_block_start)}\n\n".encode()
+            )
+            text = block_dict.get("text", "")
+            content_block_delta = {
+                "type": "content_block_delta",
+                "index": index,
+                "delta": {"type": "text_delta", "text": text},
+            }
+            chunks.append(
+                f"event: content_block_delta\ndata: {json.dumps(content_block_delta)}\n\n".encode()
+            )
+
+        elif block_type == "thinking":
+            content_block_start = {
+                "type": "content_block_start",
+                "index": index,
+                "content_block": {"type": "thinking", "thinking": "", "signature": ""},
+            }
+            chunks.append(
+                f"event: content_block_start\ndata: {json.dumps(content_block_start)}\n\n".encode()
+            )
+            thinking_text = block_dict.get("thinking", "")
+            if thinking_text:
+                content_block_delta = {
+                    "type": "content_block_delta",
+                    "index": index,
+                    "delta": {"type": "thinking_delta", "thinking": thinking_text},
+                }
+                chunks.append(
+                    f"event: content_block_delta\ndata: {json.dumps(content_block_delta)}\n\n".encode()
+                )
+            signature = block_dict.get("signature", "")
+            if signature:
+                signature_delta = {
+                    "type": "content_block_delta",
+                    "index": index,
+                    "delta": {"type": "signature_delta", "signature": signature},
+                }
+                chunks.append(
+                    f"event: content_block_delta\ndata: {json.dumps(signature_delta)}\n\n".encode()
+                )
+
+        elif block_type == "redacted_thinking":
+            content_block_start = {
+                "type": "content_block_start",
+                "index": index,
+                "content_block": {"type": "redacted_thinking"},
+            }
+            chunks.append(
+                f"event: content_block_start\ndata: {json.dumps(content_block_start)}\n\n".encode()
+            )
+
+        elif block_type == "tool_use":
+            content_block_start = {
+                "type": "content_block_start",
+                "index": index,
+                "content_block": {
+                    "type": "tool_use",
+                    "id": block_dict.get("id"),
+                    "name": block_dict.get("name"),
+                    "input": {},
+                },
+            }
+            chunks.append(
+                f"event: content_block_start\ndata: {json.dumps(content_block_start)}\n\n".encode()
+            )
+            input_data = block_dict.get("input", {})
+            content_block_delta = {
+                "type": "content_block_delta",
+                "index": index,
+                "delta": {"type": "input_json_delta", "partial_json": json.dumps(input_data)},
+            }
+            chunks.append(
+                f"event: content_block_delta\ndata: {json.dumps(content_block_delta)}\n\n".encode()
+            )
+
+        content_block_stop = {"type": "content_block_stop", "index": index}
+        chunks.append(
+            f"event: content_block_stop\ndata: {json.dumps(content_block_stop)}\n\n".encode()
+        )
+        return chunks
+
     def _create_streaming_chunks(self) -> List[bytes]:
         """Convert the non-streaming response to streaming chunks"""
         chunks = []
@@ -69,152 +165,34 @@ def _create_streaming_chunks(self) -> List[bytes]:
 
         # 2-4. For each content block, send start/delta/stop events
         content_blocks = response_dict.get("content", [])
-        if content_blocks:
-            for index, block in enumerate(content_blocks):
-                # Cast block to dict for easier access
-                block_dict = cast(Dict[str, Any], block)
-                block_type = block_dict.get("type")
-
-                if block_type == "text":
-                    # content_block_start
-                    content_block_start = {
-                        "type": "content_block_start",
-                        "index": index,
-                        "content_block": {"type": "text", "text": ""},
-                    }
-                    chunks.append(
-                        f"event: content_block_start\ndata: {json.dumps(content_block_start)}\n\n".encode()
-                    )
-
-                    # content_block_delta (send full text as one delta for simplicity)
-                    text = block_dict.get("text", "")
-                    content_block_delta = {
-                        "type": "content_block_delta",
-                        "index": index,
-                        "delta": {"type": "text_delta", "text": text},
-                    }
-                    chunks.append(
-                        f"event: content_block_delta\ndata: {json.dumps(content_block_delta)}\n\n".encode()
-                    )
-
-                    # content_block_stop
-                    content_block_stop = {"type": "content_block_stop", "index": index}
-                    chunks.append(
-                        f"event: content_block_stop\ndata: {json.dumps(content_block_stop)}\n\n".encode()
-                    )
-
-                elif block_type == "thinking":
-                    # content_block_start for thinking
-                    content_block_start = {
-                        "type": "content_block_start",
-                        "index": index,
-                        "content_block": {
-                            "type": "thinking",
-                            "thinking": "",
-                            "signature": "",
-                        },
-                    }
-                    chunks.append(
-                        f"event: content_block_start\ndata: {json.dumps(content_block_start)}\n\n".encode()
-                    )
-
-                    # content_block_delta for thinking text
-                    thinking_text = block_dict.get("thinking", "")
-                    if thinking_text:
-                        content_block_delta = {
-                            "type": "content_block_delta",
-                            "index": index,
-                            "delta": {
-                                "type": "thinking_delta",
-                                "thinking": thinking_text,
-                            },
-                        }
-                        chunks.append(
-                            f"event: content_block_delta\ndata: {json.dumps(content_block_delta)}\n\n".encode()
-                        )
-
-                    # content_block_delta for signature (if present)
-                    signature = block_dict.get("signature", "")
-                    if signature:
-                        signature_delta = {
-                            "type": "content_block_delta",
-                            "index": index,
-                            "delta": {
-                                "type": "signature_delta",
-                                "signature": signature,
-                            },
-                        }
-                        chunks.append(
-                            f"event: content_block_delta\ndata: {json.dumps(signature_delta)}\n\n".encode()
-                        )
-
-                    # content_block_stop
-                    content_block_stop = {"type": "content_block_stop", "index": index}
-                    chunks.append(
-                        f"event: content_block_stop\ndata: {json.dumps(content_block_stop)}\n\n".encode()
-                    )
-
-                elif block_type == "redacted_thinking":
-                    # content_block_start for redacted_thinking
-                    content_block_start = {
-                        "type": "content_block_start",
-                        "index": index,
-                        "content_block": {"type": "redacted_thinking"},
-                    }
-                    chunks.append(
-                        f"event: content_block_start\ndata: {json.dumps(content_block_start)}\n\n".encode()
-                    )
-
-                    # content_block_stop (no delta for redacted thinking)
-                    content_block_stop = {"type": "content_block_stop", "index": index}
-                    chunks.append(
-                        f"event: content_block_stop\ndata: {json.dumps(content_block_stop)}\n\n".encode()
-                    )
-
-                elif block_type == "tool_use":
-                    # content_block_start
-                    content_block_start = {
-                        "type": "content_block_start",
-                        "index": index,
-                        "content_block": {
-                            "type": "tool_use",
-                            "id": block_dict.get("id"),
-                            "name": block_dict.get("name"),
-                            "input": {},
-                        },
-                    }
-                    chunks.append(
-                        f"event: content_block_start\ndata: {json.dumps(content_block_start)}\n\n".encode()
-                    )
-
-                    # content_block_delta (send input as JSON delta)
-                    input_data = block_dict.get("input", {})
-                    content_block_delta = {
-                        "type": "content_block_delta",
-                        "index": index,
-                        "delta": {
-                            "type": "input_json_delta",
-                            "partial_json": json.dumps(input_data),
-                        },
-                    }
-                    chunks.append(
-                        f"event: content_block_delta\ndata: {json.dumps(content_block_delta)}\n\n".encode()
-                    )
-
-                    # content_block_stop
-                    content_block_stop = {"type": "content_block_stop", "index": index}
-                    chunks.append(
-                        f"event: content_block_stop\ndata: {json.dumps(content_block_stop)}\n\n".encode()
-                    )
+        for index, block in enumerate(content_blocks):
+            block_dict = cast(Dict[str, Any], block)
+            chunks.extend(self._create_content_block_chunks(block_dict, index))
 
         # 5. message_delta event (with final usage and stop_reason)
+        # Include cache usage fields so clients that only read message_delta
+        # (like Claude Code's SDK) see the full input token breakdown.
+        delta_usage: Dict[str, Any] = {
+            "output_tokens": usage.get("output_tokens", 0) if usage else 0,
+        }
+        if usage:
+            if usage.get("input_tokens") is not None:
+                delta_usage["input_tokens"] = usage["input_tokens"]
+            if usage.get("cache_creation_input_tokens") is not None:
+                delta_usage["cache_creation_input_tokens"] = usage[
+                    "cache_creation_input_tokens"
+                ]
+            if usage.get("cache_read_input_tokens") is not None:
+                delta_usage["cache_read_input_tokens"] = usage[
+                    "cache_read_input_tokens"
+                ]
         message_delta = {
             "type": "message_delta",
             "delta": {
                 "stop_reason": response_dict.get("stop_reason"),
                 "stop_sequence": response_dict.get("stop_sequence"),
             },
-            "usage": {"output_tokens": usage.get("output_tokens", 0) if usage else 0},
+            "usage": delta_usage,
         }
         chunks.append(
             f"event: message_delta\ndata: {json.dumps(message_delta)}\n\n".encode()