async def add_prompt(
        self,
        name: str,
        template: str,
        input_types: dict[str, str],
        preserve_existing: bool = False,
        overwrite_on_diff: bool = False,  # <-- new param
    ) -> None:
        """Add or update a prompt.

        If `preserve_existing` is True and prompt already exists, we skip updating.

        If `overwrite_on_diff` is True and an existing prompt differs from what is provided,
        we overwrite and log a warning. Otherwise, we skip if the prompt differs.
        """
        # Check if prompt is in-memory
        existing_prompt = self.prompts.get(name)

        # If preserving existing and it already exists, skip entirely
        if preserve_existing and existing_prompt:
            logger.debug(
                f"Preserving existing prompt: {name}, skipping update."
            )
            return

        # If an existing prompt is found, check for diffs
        if existing_prompt:
            existing_template = existing_prompt["template"]
            existing_input_types = existing_prompt["input_types"]

            # If there's a difference in template or input_types, decide to overwrite or skip
            if (
                existing_template != template
                or existing_input_types != input_types
            ):
                if overwrite_on_diff:
                    logger.warning(
                        f"Overwriting existing prompt '{name}' due to detected diff."
                    )
                else:
                    logger.info(
                        f"Prompt '{name}' differs from existing but overwrite_on_diff=False. Skipping update."
                    )
                    return

        prompt_id = generate_default_prompt_id(name)

        # Ensure input_types is properly serialized
        input_types_json = (
            json.dumps(input_types)
            if isinstance(input_types, dict)
            else input_types
        )

        # Upsert logic
        query = f"""
        INSERT INTO {self._get_table_name("prompts")} (id, name, template, input_types)
        VALUES ($1, $2, $3, $4)
        ON CONFLICT (name) DO UPDATE
        SET template = EXCLUDED.template,
            input_types = EXCLUDED.input_types,
            updated_at = CURRENT_TIMESTAMP
        RETURNING id, created_at, updated_at;
        """

        result = await self.connection_manager.fetchrow_query(
            query, [prompt_id, name, template, input_types_json]
        )

        self.prompts[name] = {
            "id": result["id"],
            "template": template,
            "input_types": input_types,
            "created_at": result["created_at"],
            "updated_at": result["updated_at"],
        }

        # Update template cache
        self._template_cache.set(
            name,
            {
                "id": prompt_id,
                "template": template,
                "input_types": input_types,
            },
        )

        # Invalidate any cached formatted prompts
        for key in list(self._prompt_cache._cache.keys()):
            if key.startswith(f"{name}:"):
                self._prompt_cache.invalidate(key)
Add or update a prompt.

        If `preserve_existing` is True and prompt already exists, we skip updating.

        If `overwrite_on_diff` is True and an existing prompt differs from what is provided,
        we overwrite and log a warning. Otherwise, we skip if the prompt differs.
        
py/core/providers/database/prompts_handler.py