feat: PLTF-3553 Add org condenser max_tokens rollout configuration

server/app_lifespan/saas_app_lifespan_service.py

@@ -6,14 +6,57 @@
 
 from __future__ import annotations
 
+import asyncio
 import os
+import random
+
+from sqlalchemy.exc import DBAPIError, OperationalError
 
 from openhands.analytics import get_analytics_service, init_analytics_service
 from openhands.app_server.app_lifespan.app_lifespan_service import AppLifespanService
 from openhands.app_server.utils.logger import openhands_logger as logger
 from openhands.server.types import AppMode
 from server.constants import DEPLOYMENT_MODE, IS_FEATURE_ENV
 
+_ORG_CONDENSER_RECONCILIATION_LOCK_ID = 865115708052677401
+_TRANSIENT_SQLSTATES = {
+    '40001',  # serialization_failure
+    '40P01',  # deadlock_detected
+    '55P03',  # lock_not_available / lock timeout
+    '57014',  # query_canceled
+}
+_TRANSIENT_SQLSTATE_PREFIXES = ('08',)  # connection exception class
+
+
+class TransientReconciliationError(Exception):
+    """Retryable org defaults reconciliation failure."""
+
+
+def _sqlstate(exc: BaseException) -> str | None:
+    orig = getattr(exc, 'orig', None)
+    for attr in ('sqlstate', 'pgcode'):
+        value = getattr(orig, attr, None)
+        if isinstance(value, str):
+            return value
+    return None
+
+
+def _is_transient_reconciliation_error(exc: BaseException) -> bool:
+    if isinstance(exc, OperationalError):
+        return True
+    if not isinstance(exc, DBAPIError):
+        return False
+    if exc.connection_invalidated:
+        return True
+    state = _sqlstate(exc)
+    return bool(
+        state
+        and (
+            state in _TRANSIENT_SQLSTATES
+            or state.startswith(_TRANSIENT_SQLSTATE_PREFIXES)
+        )
+    )
+
 
 class SaasAppLifespanService(AppLifespanService):
     """Lifespan service for the SaaS server.
@@ -37,8 +80,102 @@ async def __aenter__(self):
             app_mode=AppMode.SAAS,
             is_feature_env=IS_FEATURE_ENV,
         )
+        await self._reconcile_org_condenser_defaults()
         return self
 
+    async def _reconcile_org_condenser_defaults(self) -> None:
+        from server.org_defaults_config import get_org_defaults_condenser_config
+
+        config = get_org_defaults_condenser_config()
+        if config.max_tokens is None or not config.apply_to_existing:
+            return
+
+        max_tokens = config.max_tokens
+        try:
+            result = await self._reconcile_org_condenser_defaults_once(
+                max_tokens=max_tokens,
+                overwrite_existing=config.overwrite_existing,
+            )
+        except TransientReconciliationError:
+            await asyncio.sleep(random.uniform(0.25, 1.5))
+            try:
+                result = await self._reconcile_org_condenser_defaults_once(
+                    max_tokens=max_tokens,
+                    overwrite_existing=config.overwrite_existing,
+                )
+            except Exception:
+                logger.exception(
+                    'org_condenser_defaults_reconciliation_failed',
+                    extra={
+                        'max_tokens': max_tokens,
+                        'apply_to_existing': config.apply_to_existing,
+                        'overwrite_existing': config.overwrite_existing,
+                    },
+                    stack_info=True,
+                )
+                return
+        except Exception:
+            logger.exception(
+                'org_condenser_defaults_reconciliation_failed',
+                extra={
+                    'max_tokens': max_tokens,
+                    'apply_to_existing': config.apply_to_existing,
+                    'overwrite_existing': config.overwrite_existing,
+                },
+                stack_info=True,
+            )
+            return
+
+        logger.info(
+            'org_condenser_defaults_reconciliation_succeeded',
+            extra={
+                'max_tokens': max_tokens,
+                'apply_to_existing': config.apply_to_existing,
+                'overwrite_existing': config.overwrite_existing,
+                'updated_count': result.updated_count,
+                'skipped_agent_variant_count': result.skipped_agent_variant_count,
+                'skipped_condenser_variant_count': result.skipped_condenser_variant_count,
+                'malformed_repaired_count': result.malformed_repaired_count,
+            },
+        )
+
+    async def _reconcile_org_condenser_defaults_once(
+        self,
+        *,
+        max_tokens: int,
+        overwrite_existing: bool,
+    ):
+        from sqlalchemy import text
+
+        from storage.database import a_session_maker
+        from storage.org_store import OrgStore
+
+        try:
+            async with a_session_maker() as session:
+                async with session.begin():
+                    await session.execute(text("SET LOCAL lock_timeout = '5s'"))
+                    await session.execute(
+                        text('SELECT pg_advisory_xact_lock(:lock_id)'),
+                        {'lock_id': _ORG_CONDENSER_RECONCILIATION_LOCK_ID},
+                    )
+                    return await OrgStore.reconcile_applicable_org_condenser_max_tokens(
+                        session,
+                        max_tokens=max_tokens,
+                        overwrite_existing=overwrite_existing,
+                    )
+        except Exception as exc:
+            if _is_transient_reconciliation_error(exc):
+                logger.warning(
+                    'org_condenser_defaults_reconciliation_transient_failure',
+                    extra={
+                        'max_tokens': max_tokens,
+                        'overwrite_existing': overwrite_existing,
+                        'error': str(exc),
+                    },
+                )
+                raise TransientReconciliationError from exc
+            raise
+
     async def __aexit__(self, exc_type, exc_value, traceback):
         try:
             svc = get_analytics_service()

server/org_defaults_config.py

@@ -0,0 +1,61 @@
+"""Deployment-configured organization defaults."""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+
+_BOOLEAN_ENV_VALUES = {
+    'true': True,
+    '1': True,
+    'false': False,
+    '0': False,
+}
+
+
+@dataclass(frozen=True)
+class OrgDefaultsCondenserConfig:
+    """Deployment defaults for organization condenser settings."""
+
+    max_tokens: int | None
+    apply_to_existing: bool
+    overwrite_existing: bool
+
+
+def _parse_bool_env(name: str, *, default: bool) -> bool:
+    raw = os.getenv(name)
+    if raw is None or raw.strip() == '':
+        return default
+    normalized = raw.strip().lower()
+    if normalized not in _BOOLEAN_ENV_VALUES:
+        raise ValueError(f'{name} must be one of true, false, 1, or 0; got {raw!r}')
+    return _BOOLEAN_ENV_VALUES[normalized]
+
+
+def _parse_positive_int_env(name: str) -> int | None:
+    raw = os.getenv(name)
+    if raw is None or raw.strip() == '':
+        return None
+    value = raw.strip()
+    if not value.isdecimal():
+        raise ValueError(f'{name} must be a positive base-10 integer; got {raw!r}')
+    parsed = int(value)
+    if parsed <= 0:
+        raise ValueError(f'{name} must be greater than 0; got {raw!r}')
+    return parsed
+
+
+def get_org_defaults_condenser_config() -> OrgDefaultsCondenserConfig:
+    """Read Helm/Replicated-controlled organization condenser defaults."""
+
+    return OrgDefaultsCondenserConfig(
+        max_tokens=_parse_positive_int_env(
+            'OPENHANDS_ORG_DEFAULTS_CONDENSER_MAX_TOKENS'
+        ),
+        apply_to_existing=_parse_bool_env(
+            'OPENHANDS_ORG_DEFAULTS_CONDENSER_APPLY_TO_EXISTING', default=False
+        ),
+        overwrite_existing=_parse_bool_env(
+            'OPENHANDS_ORG_DEFAULTS_CONDENSER_OVERWRITE_EXISTING', default=False
+        ),
+    )

storage/org_default_settings.py

@@ -0,0 +1,119 @@
+"""Helpers for deployment-configured organization default settings."""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from typing import Any, TypeVar
+
+from openhands.sdk.settings import OpenHandsAgentSettings, default_agent_settings
+from server.constants import get_default_llm_base_url, get_default_llm_model
+from server.org_defaults_config import get_org_defaults_condenser_config
+
+_APPLICABLE_AGENT_KINDS = {None, 'openhands', 'llm'}
+_APPLICABLE_CONDENSER_KINDS = {None, 'llm_summarizing'}
+_SettingsT = TypeVar('_SettingsT', dict[str, Any], OpenHandsAgentSettings)
+
+
+def _configured_max_tokens() -> int | None:
+    return get_org_defaults_condenser_config().max_tokens
+
+
+def create_base_org_agent_settings() -> OpenHandsAgentSettings:
+    """Create default org agent settings with deployment defaults applied."""
+
+    agent_settings = default_agent_settings()
+    agent_settings.llm.model = get_default_llm_model()
+    agent_settings.llm.base_url = get_default_llm_base_url()
+    return apply_org_condenser_max_tokens_default(
+        agent_settings,
+        max_tokens=_configured_max_tokens(),
+    )
+
+
+def _dict_agent_kind(agent_settings: dict[str, Any]) -> str | None:
+    value = agent_settings.get('agent_kind')
+    return value if isinstance(value, str) else None
+
+
+def _dict_condenser_kind(condenser: dict[str, Any]) -> str | None:
+    value = condenser.get('condenser_kind')
+    return value if isinstance(value, str) else None
+
+
+def apply_org_condenser_max_tokens_default(
+    agent_settings: _SettingsT,
+    *,
+    max_tokens: int | None = None,
+    overwrite: bool = False,
+) -> _SettingsT:
+    """Apply deployment default ``condenser.max_tokens`` when applicable.
+
+    Only OpenHands/legacy-LLM agent settings with the legacy/default
+    LLM-summarizing condenser support this setting. ACP settings and no-op
+    condensers are left unchanged.
+    """
+
+    if max_tokens is None:
+        return agent_settings
+
+    if isinstance(agent_settings, OpenHandsAgentSettings):
+        model_condenser = agent_settings.condenser
+        condenser_kind = getattr(model_condenser, 'condenser_kind', None)
+        if condenser_kind not in _APPLICABLE_CONDENSER_KINDS:
+            return agent_settings
+        current = getattr(model_condenser, 'max_tokens', None)
+        if overwrite or current is None:
+            setattr(model_condenser, 'max_tokens', max_tokens)
+        return agent_settings
+
+    data = deepcopy(agent_settings)
+    agent_kind = _dict_agent_kind(data)
+    if agent_kind not in _APPLICABLE_AGENT_KINDS:
+        return data
+
+    raw_condenser = data.get('condenser')
+    if raw_condenser is None:
+        condenser: dict[str, Any] = {'condenser_kind': 'llm_summarizing'}
+    elif isinstance(raw_condenser, dict):
+        condenser = deepcopy(raw_condenser)
+    else:
+        condenser = {'condenser_kind': 'llm_summarizing'}
+
+    condenser_kind = _dict_condenser_kind(condenser)
+    if condenser_kind not in _APPLICABLE_CONDENSER_KINDS:
+        return data
+
+    if overwrite or condenser.get('max_tokens') is None:
+        condenser['max_tokens'] = max_tokens
+    data['condenser'] = condenser
+    return data
+
+
+def apply_configured_org_condenser_default(
+    agent_settings: _SettingsT,
+    *,
+    overwrite: bool = False,
+) -> _SettingsT:
+    """Apply the configured org condenser default, if one exists."""
+
+    return apply_org_condenser_max_tokens_default(
+        agent_settings,
+        max_tokens=_configured_max_tokens(),
+        overwrite=overwrite,
+    )
+
+
+def strip_unset_condenser_max_tokens(agent_settings_diff: dict[str, Any]) -> None:
+    """Remove JSON-null member ``condenser.max_tokens`` overrides.
+
+    Legacy serialized defaults can contain ``max_tokens: null``. Treat that as
+    unset so member diffs do not mask org-level deployment defaults.
+    """
+
+    condenser = agent_settings_diff.get('condenser')
+    if not isinstance(condenser, dict):
+        return
+    if condenser.get('max_tokens') is None and 'max_tokens' in condenser:
+        condenser.pop('max_tokens', None)
+    if not condenser:
+        agent_settings_diff.pop('condenser', None)

storage/org_member_store.py

@@ -17,6 +17,7 @@
 )
 from storage.database import a_session_maker
 from storage.mcp_config import serialize_mcp_config
+from storage.org_default_settings import strip_unset_condenser_max_tokens
 from storage.org_member import OrgMember
 from storage.user import User
 from storage.user_settings import UserSettings
@@ -178,6 +179,7 @@ def get_kwargs_from_user_settings(user_settings: UserSettings) -> dict[str, Any]
         """Return kwargs for OrgMember construction (keys match column names)."""
         agent_settings_diff = dict(user_settings.agent_settings or {})
         nested_mcp_config = _pop_mcp_config(agent_settings_diff)
+        strip_unset_condenser_max_tokens(agent_settings_diff)
         mcp_config = (
             nested_mcp_config
             if nested_mcp_config is not _MISSING