class DbSessionInjector(BaseModel, Injector[AsyncSession]):
    persistence_dir: Path
    host: str | None = None
    port: int | None = None
    name: str | None = None
    user: str | None = None
    password: SecretStr | None = None
    echo: bool = False
    pool_size: int = 25
    max_overflow: int = 10
    pool_recycle: int = 1800
    pool_use_lifo: bool = True
    gcp_db_instance: str | None = None
    gcp_project: str | None = None
    gcp_region: str | None = None
    ssl_mode: str | None = None

    # Private attrs
    _engine: Engine | None = PrivateAttr(default=None)
    _async_engine: AsyncEngine | None = PrivateAttr(default=None)
    _session_maker: sessionmaker | None = PrivateAttr(default=None)
    _async_session_maker: async_sessionmaker | None = PrivateAttr(default=None)
    _gcp_connector: Any = PrivateAttr(default=None)

    @model_validator(mode='after')
    def fill_empty_fields(self):
        """Override any defaults with values from legacy environment variables"""
        if self.host is None:
            self.host = os.getenv('DB_HOST')
        if self.port is None:
            self.port = int(os.getenv('DB_PORT', '5432'))
        if self.name is None:
            self.name = os.getenv('DB_NAME', 'openhands')
        if self.user is None:
            self.user = os.getenv('DB_USER', 'postgres')
        if self.password is None:
            self.password = SecretStr(os.getenv('DB_PASS', 'postgres').strip())
        # ... GCP 和 SSL 配置同样从环境变量读取
        return self

    async def get_async_db_engine(self) -> AsyncEngine:
        async_engine = self._async_engine
        if async_engine:
            return async_engine
        if self.gcp_db_instance:  # GCP environments
            async_engine = await self._create_async_gcp_engine()
        else:
            url: str | URL
            if self.host:
                # PostgreSQL with asyncpg driver
                url = URL.create(
                    'postgresql+asyncpg',
                    username=self.user or '',
                    password=password,
                    host=self.host,
                    port=self.port,
                    database=self.name,
                )
            else:
                # SQLite fallback for local development
                url = f'sqlite+aiosqlite:///{str(self.persistence_dir)}/openhands.db'

            if self.host:
                async_engine = create_async_engine(
                    url,
                    connect_args=build_asyncpg_connect_args(self.ssl_mode),
                    pool_size=self.pool_size,
                    max_overflow=self.max_overflow,
                    pool_recycle=self.pool_recycle,
                    pool_pre_ping=True,
                    pool_use_lifo=self.pool_use_lifo,
                )
            else:
                async_engine = create_async_engine(
                    url,
                    poolclass=NullPool,   # SQLite 不需要连接池
                    pool_pre_ping=True,
                )

    async def inject(
        self, state: InjectorState, request: Request | None = None
    ) -> AsyncGenerator[AsyncSession, None]:
        """Dependency function that manages database sessions through request state."""
        db_session = getattr(state, DB_SESSION_ATTR, None)
        if db_session:
            yield db_session
        else:
            # Create a new session and store it in request state
            session_maker = await self.get_async_session_maker()
            db_session = session_maker()
            try:
                setattr(state, DB_SESSION_ATTR, db_session)
                yield db_session
                if not getattr(state, DB_SESSION_KEEP_OPEN_ATTR, False):
                    await db_session.commit()
            except Exception:
                _logger.exception('Rolling back SQL due to error', stack_info=True)
                await db_session.rollback()
                raise
            finally:
                if not getattr(state, DB_SESSION_KEEP_OPEN_ATTR, False):
                    if hasattr(state, DB_SESSION_ATTR):
                        delattr(state, DB_SESSION_ATTR)
                    with contextlib.suppress(Exception):
                        await db_session.close()

def get_db_session(
    state: InjectorState, request: Request | None = None
) -> AsyncContextManager[AsyncSession]:
    """Return an async context manager yielding the request-scoped AsyncSession."""
    from openhands.app_server.config import get_global_config
    return get_global_config().db_session.context(state, request)

async def _yield_db_session(request: Request) -> AsyncGenerator[AsyncSession, None]:
    """Per-request FastAPI dependency yielding the request-scoped AsyncSession.

    get_global_config is looked up at request time -- not at module load --
    so that this module is safe to evaluate from inside config_from_env.
    """
    from openhands.app_server.config import get_global_config
    async for db_session in get_global_config().db_session.depends(request):
        yield db_session

def depends_db_session():
    """FastAPI Depends(...) factory for the request-scoped AsyncSession."""
    return Depends(_yield_db_session)

from sqlalchemy.orm import DeclarativeBase

class Base(DeclarativeBase):
    """
    Base class for all SQLAlchemy models.

    Uses SQLAlchemy 2.0 DeclarativeBase for proper type inference with Mapped types.
    This is backward compatible with existing Column() definitions while enabling
    gradual migration to mapped_column() with Mapped[T] type annotations.
    """
    pass

class StoredSecretStr(TypeDecorator):
    """TypeDecorator for secret strings. Encrypts before storing."""
    impl = String
    cache_ok = True

    def process_bind_param(self, value, dialect):
        if value is not None:
            jwt_service = get_global_config().jwt.get_jwt_service()
            token = jwt_service.create_jwe_token({'v': value.get_secret_value()})
            return token
        return None

    def process_result_param(self, value, dialect):
        if value is not None:
            jwt_service = get_global_config().jwt.get_jwt_service()
            token = jwt_service.decrypt_jwe_token(value)
            return SecretStr(token['v'])
        return None

class StoredConversationMetadata(Base):
    __tablename__ = 'conversation_metadata'

    conversation_id: Mapped[str] = mapped_column(
        String, primary_key=True, default=lambda: str(uuid.uuid4())
    )
    selected_repository: Mapped[str | None] = mapped_column(String, nullable=True)
    selected_branch: Mapped[str | None] = mapped_column(String, nullable=True)
    git_provider: Mapped[str | None] = mapped_column(String, nullable=True)
    title: Mapped[str | None] = mapped_column(String, nullable=True)
    last_updated_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), default=utc_now
    )
    created_at: Mapped[datetime | None] = mapped_column(
        DateTime(timezone=True), default=utc_now
    )
    trigger: Mapped[str | None] = mapped_column(String, nullable=True)
    pr_number: Mapped[list[int] | None] = mapped_column(
        create_json_type_decorator(list[int])
    )
    # Cost and token metrics
    accumulated_cost: Mapped[float | None] = mapped_column(default=0.0)
    prompt_tokens: Mapped[int | None] = mapped_column(default=0)
    completion_tokens: Mapped[int | None] = mapped_column(default=0)
    total_tokens: Mapped[int | None] = mapped_column(default=0)
    max_budget_per_task: Mapped[float | None] = mapped_column(nullable=True)
    cache_read_tokens: Mapped[int | None] = mapped_column(default=0)
    cache_write_tokens: Mapped[int | None] = mapped_column(default=0)
    reasoning_tokens: Mapped[int | None] = mapped_column(default=0)
    context_window: Mapped[int | None] = mapped_column(default=0)
    per_turn_token: Mapped[int | None] = mapped_column(default=0)
    llm_model: Mapped[str | None] = mapped_column(String, nullable=True)
    agent_kind: Mapped[str | None] = mapped_column(String, nullable=True)
    conversation_version: Mapped[str] = mapped_column(
        String, nullable=False, default='V0', index=True
    )
    sandbox_id: Mapped[str | None] = mapped_column(String, nullable=True, index=True)
    parent_conversation_id: Mapped[str | None] = mapped_column(
        String, nullable=True, index=True
    )
    public: Mapped[bool | None] = mapped_column(nullable=True, index=True)
    execution_status: Mapped[str | None] = mapped_column(
        String, nullable=True, index=True
    )
    tags: Mapped[dict[str, str] | None] = mapped_column(
        create_json_type_decorator(dict[str, str]), nullable=True
    )

class EncryptionKey(BaseModel):
    id: str = Field(default_factory=lambda: base62.encodebytes(os.urandom(32)))
    key: SecretStr
    active: bool = True
    notes: str | None = None
    created_at: datetime.datetime = Field(default_factory=utc_now)

def get_default_encryption_keys(workspace_dir: Path) -> list[EncryptionKey]:
    # Priority 1: JWT_SECRET env var (deterministic key ID for multi-pod)
    jwt_secret = os.getenv('JWT_SECRET')
    if jwt_secret:
        key_id = base62.encodebytes(hashlib.sha256(jwt_secret.encode()).digest())
        return [EncryptionKey(id=key_id, key=SecretStr(jwt_secret), ...)]
    # Priority 2: .keys file (persisted key rotation)
    # Priority 3: .jwt_secret file (legacy fallback)
    # Priority 4: Generate new random key

SUPPORTED_DB_SSL_MODES: frozenset[str] = frozenset({'prefer', 'require', 'disable'})

def build_pg8000_connect_args(db_ssl_mode: str | None) -> dict[str, bool]:
    # sync driver: ssl_context = True/False
    mode = normalize_db_ssl_mode(db_ssl_mode)
    if mode == 'require':
        return {'ssl_context': True}
    if mode == 'disable':
        return {'ssl_context': False}
    return {}

def build_asyncpg_connect_args(db_ssl_mode: str | None) -> dict[str, str]:
    # async driver: ssl = 'require'/'disable'
    mode = normalize_db_ssl_mode(db_ssl_mode)
    if mode:
        return {'ssl': mode}
    return {}