client

@traced
async def add_object(
    self,
    mgmt_name: str,
    obj_type: str,
    payload: dict[str, Any],
    *,
    domain: str | None = None,
    sid: str | None = None,
    server_ip: str | None = None,
) -> ApiCallResult:
    """Create a Check Point object (`add-<obj_type>`) using active `WriteSession` context, explicit `sid`, or an automatic isolated `write_session` (`C8`)."""
    clean_type = obj_type.strip().lower().removeprefix("add-")
    span_attrs(mgmt_name=mgmt_name, domain=domain, object_type=clean_type)
    dom_key = normalize_domain_key(domain) if domain is not None else None
    if sid is None and get_write_session(mgmt_name=mgmt_name, domain_name=dom_key) is None:
        async with self.write_session(mgmt_name, domain) as ws:
            return await self.api_call(
                mgmt_name,
                f"add-{clean_type}",
                payload,
                domain=domain,
                sid=ws.sid,
                server_ip=server_ip or ws.server_ip,
            )
    return await self.api_call(
        mgmt_name,
        f"add-{clean_type}",
        payload,
        domain=domain,
        sid=sid,
        server_ip=server_ip,
    )

@traced
async def api_call(
    self,
    mgmt_name: str,
    command: str,
    payload: dict[str, Any] | None = None,
    *,
    domain: str | None = None,
    sid: str | None = None,
    server_ip: str | None = None,
    wait_for_task: bool = True,
    timeout: float | None = None,
) -> ApiCallResult:
    """Execute a Check Point Management API command, updating active `WriteSession` overlay and invalidating domain on `publish`."""
    if not self._db.is_initialized:
        await self._db.initialize()

    clean_cmd = command.strip().lstrip("/")
    dom_key = normalize_domain_key(domain)
    active_ws = get_write_session(mgmt_name=mgmt_name, domain_name=dom_key if domain is not None else None)

    eff_sid = sid
    eff_ip = server_ip
    if eff_sid is None and active_ws is not None and self._amgmt is None:
        eff_sid = active_ws.sid
        eff_ip = eff_ip or active_ws.server_ip

    res = await self._api.api_call(
        mgmt_name,
        command,
        payload,
        domain=domain,
        sid=eff_sid,
        server_ip=eff_ip,
        wait_for_task=wait_for_task,
        timeout=timeout,
    )

    # If using a custom ApiPort implementation (not AMgmtClient), ensure overlay & invalidation still run
    if self._amgmt is None:
        if active_ws is not None and (sid is None or sid == active_ws.sid):
            _update_overlay_from_command(active_ws.overlay, clean_cmd, payload, res)
        if clean_cmd == "publish" and res.success:
            eff_domain = domain or (active_ws.domain_name if active_ws is not None else None)
            self.invalidate_domain(mgmt_name, eff_domain)
    elif clean_cmd == "publish" and res.success:
        eff_domain = domain or (active_ws.domain_name if active_ws is not None else None)
        self.invalidate_domain(mgmt_name, eff_domain)

    return res

@traced
async def api_query(
    self,
    mgmt_name: str,
    command: str,
    *,
    domain: str | None = None,
    details_level: str = "standard",
    payload: dict[str, Any] | None = None,
    sid: str | None = None,
    server_ip: str | None = None,
    limit: int = DEFAULT_PAGE_LIMIT,
) -> ApiQueryResult:
    """Execute a continuity-verified paginated Check Point object listing query."""
    if not self._db.is_initialized:
        await self._db.initialize()
    return await self._api.api_query(
        mgmt_name,
        command,
        domain=domain,
        details_level=details_level,
        payload=payload,
        sid=sid,
        server_ip=server_ip,
        limit=limit,
    )

@contextmanager
def auth_context(
    self,
    auth: AuthContext | None = None,
    *,
    username: str | None = None,
    password: str | None = None,
    api_key: str | None = None,
    user_id: str | None = None,
) -> Iterator[AuthContext]:
    """Bind runtime administrator credentials (`AuthContext`) in `contextvars`."""
    with _ctx_auth_context(
        auth,
        username=username,
        password=password,
        api_key=api_key,
        user_id=user_id,
    ) as resolved:
        yield resolved

@traced
async def build_change_report(
    self,
    scopes: Sequence[ReportScope | dict[str, Any]],
    *,
    formats: Sequence[str] = ("html", "markdown", "json"),
    options: RenderOptions | None = None,
    max_uid_lookups: int = MAX_UID_LOOKUPS,
) -> RenderedReportBundle:
    """Collect and render a `ChangeReport` across `scopes` into `formats`."""
    if not self._db.is_initialized:
        await self._db.initialize()

    async def _warm_rulebase(m_name: str, d_name: str) -> Any:
        return await self.refresh_rulebases(
            mgmt_name=m_name,
            domains=[d_name] if d_name else None,
            mode="smart",
        )

    return await _build_change_report(
        self._api,
        self._cache,
        scopes,
        formats=formats,
        options=options,
        max_uid_lookups=max_uid_lookups,
        now_fn=self._now,
        refresh_rulebase_fn=_warm_rulebase,
    )

@traced
async def close(self, *, logout: bool | None = None) -> None:
    """Cancel running background warm-up tasks immediately, drain startup cleanup, close transport, and dispose owned engine."""
    # 1. Cancel any running background warm-up tasks immediately
    active_warmups = [t for t in self._warmup_tasks if not t.done()]
    for task in active_warmups:
        task.cancel()
    if active_warmups:
        await asyncio.gather(*active_warmups, return_exceptions=True)
    self._warmup_tasks.clear()

    # 2. Determine automatic logout policy: True if client auto-created in-memory DB, False if persistent DB or external engine
    is_ephemeral_db = self._owns_engine and ":memory:" in (self._settings.database_url or "")
    should_logout = is_ephemeral_db if logout is None else bool(logout)

    # 3. Close AMgmtClient / ApiPort (drains startup cleanup tasks & logs out according to policy)
    await self._api.close(logout=should_logout)

    # 4. Dispose SQLAlchemy engine only if owned (`self._db.close()` checks `owns_engine`)
    await self._db.close()

@traced
async def collect_change_report(
    self,
    scopes: Sequence[ReportScope | dict[str, Any]],
    *,
    options: RenderOptions | None = None,
    max_uid_lookups: int = MAX_UID_LOOKUPS,
) -> ChangeReport:
    """Collect read-only auditable session change evidence (`ChangeReport`) across `scopes`."""
    if not self._db.is_initialized:
        await self._db.initialize()

    async def _warm_rulebase(m_name: str, d_name: str) -> Any:
        return await self.refresh_rulebases(
            mgmt_name=m_name,
            domains=[d_name] if d_name else None,
            mode="smart",
        )

    return await _collect_change_report(
        self._api,
        self._cache,
        scopes,
        options=options,
        max_uid_lookups=max_uid_lookups,
        now_fn=self._now,
        refresh_rulebase_fn=_warm_rulebase,
    )

async def create_dedicated_session(
    self,
    mgmt_name: str,
    domain: str | None = None,
    *,
    auth: AuthContext | None = None,
    session_name: str | None = None,
    session_description: str | None = None,
) -> tuple[str, str]:
    """Open an isolated write session outside `sid_cache`, returning `(sid, server_ip)`."""
    if not self._db.is_initialized:
        await self._db.initialize()
    return await self._api.create_dedicated_session(
        mgmt_name,
        domain,
        auth=auth,
        session_name=session_name,
        session_description=session_description,
    )

@traced
async def delete_object(
    self,
    mgmt_name: str,
    obj_type: str,
    *,
    uid: str | None = None,
    name: str | None = None,
    payload: dict[str, Any] | None = None,
    domain: str | None = None,
    sid: str | None = None,
    server_ip: str | None = None,
) -> ApiCallResult:
    """Delete a Check Point object (`delete-<obj_type>`) by `uid` or `name` using active `WriteSession`, explicit `sid`, or automatic isolated `write_session` (`C8`)."""
    clean_type = obj_type.strip().lower().removeprefix("delete-")
    req_payload: dict[str, Any] = dict(payload or {})
    if uid is not None:
        req_payload["uid"] = uid
    if name is not None:
        req_payload["name"] = name
    span_attrs(mgmt_name=mgmt_name, domain=domain, object_type=clean_type)
    dom_key = normalize_domain_key(domain) if domain is not None else None
    if sid is None and get_write_session(mgmt_name=mgmt_name, domain_name=dom_key) is None:
        async with self.write_session(mgmt_name, domain) as ws:
            return await self.api_call(
                mgmt_name,
                f"delete-{clean_type}",
                req_payload,
                domain=domain,
                sid=ws.sid,
                server_ip=server_ip or ws.server_ip,
            )
    return await self.api_call(
        mgmt_name,
        f"delete-{clean_type}",
        req_payload,
        domain=domain,
        sid=sid,
        server_ip=server_ip,
    )

@traced
async def describe_objects(
    self,
    *,
    mgmt_name: str | None = None,
    domain_name: str | None = None,
    include_global: bool = True,
) -> dict[str, Any]:
    """Return progressive-disclosure schema and summary counts for cached objects and services."""
    if not self._db.is_initialized:
        await self._db.initialize()
    return await self._search_service.describe_objects(
        mgmt_name=mgmt_name,
        domain_name=domain_name,
        include_global=include_global,
    )

@traced
async def discard_staged_session(
    self,
    mgmt_name: str,
    domain: str | None = None,
    *,
    session_uid: str,
    sid: str | None = None,
    server_ip: str | None = None,
    auth: AuthContext | None = None,
    force_disconnect: bool = False,
) -> ApiCallResult:
    """Adopt `session_uid`, discard uncommitted changes, and log out."""
    return await self._cpcrud.discard_staged_session(
        mgmt_name,
        domain,
        session_uid=session_uid,
        sid=sid,
        server_ip=server_ip,
        auth=auth,
        force_disconnect=force_disconnect,
    )

@traced
async def filter_rulebase(
    self,
    mgmt_name: str,
    domain_name: str | None = None,
    *,
    package: str | None = None,
    layer: str | None = None,
    rulebase_type: RulebaseType = "access",
    filter_query: str = "",
    output_format: RuleFormat = "model_friendly",
    cache_mode: CacheMode = "cache",
    cache_ttl: float | None = None,
) -> FilteredRulebaseResult:
    """Evaluate `filter_query` locally against cached rulebase and cached objects/services."""
    if not self._db.is_initialized:
        await self._db.initialize()
    return await self._rulebase_source.filter_rulebase(
        mgmt_name,
        domain_name,
        package=package,
        layer=layer,
        rulebase_type=rulebase_type,
        filter_query=filter_query,
        output_format=output_format,
        cache_mode=cache_mode,
        cache_ttl=cache_ttl,
    )

@traced
async def find_object(
    self,
    mgmt_name: str,
    domain: str | None = None,
    name_or_uid: str = "",
    *,
    obj_type: str | None = None,
    cache_mode: CacheMode = "smart",
    cache_ttl: float | None = None,
) -> CPObject | CPService | Asset | None:
    """Find a single `CPObject`, `CPService`, or `Asset` by exact `uid` or `name` in `(mgmt_name, domain)` (or across all domains when `domain` is `None`), including active `WriteSession` overlay."""
    clean_query = name_or_uid.strip() if name_or_uid else ""
    if not clean_query:
        raise ValidationError("name_or_uid must not be empty or whitespace")

    if not self._db.is_initialized:
        await self._db.initialize()

    clean_type = normalize_any_type(obj_type) if obj_type else None
    is_service_type = clean_type is not None and (
        clean_type in SERVICE_COMMAND_TYPES or clean_type.startswith("service-")
    )
    is_asset_type = clean_type is not None and clean_type in {t.lower() for t in CANONICAL_ASSET_TYPES}

    eff_domain = domain.strip() if domain and domain.strip() else None
    lookup_key = clean_query
    if eff_domain:
        composite_prefix = f"{mgmt_name}:{eff_domain}:"
        if clean_query.startswith(composite_prefix):
            lookup_key = clean_query.removeprefix(composite_prefix)
    elif clean_query.startswith(f"{mgmt_name}:"):
        parts = clean_query.split(":", 2)
        if len(parts) == 3 and parts[1].strip() and parts[2].strip():
            eff_domain = parts[1].strip()
            lookup_key = parts[2].strip()
    lookup_names = list(dict.fromkeys([clean_query, lookup_key]))

    if eff_domain is not None:
        if not is_service_type and not is_asset_type:
            by_uid = await self._object_service.get_object_by_uid(
                mgmt_name,
                eff_domain,
                lookup_key,
                cache_mode=cache_mode,
                cache_ttl=cache_ttl,
            )
            if by_uid is not None and (clean_type is None or by_uid.type.lower() == clean_type):
                return by_uid

            objs = await self._object_service.get_objects(
                mgmt_name=mgmt_name,
                domain=eff_domain,
                types=[clean_type] if clean_type else None,
                names=lookup_names,
                include_global=(eff_domain == "Global"),
                cache_mode="cache" if cache_mode != "force" else cache_mode,
                cache_ttl=cache_ttl,
            )
            if objs:
                return objs[0]

        if is_service_type or clean_type is None:
            svc_by_uid = await self._service_cache.get_service_by_uid(
                mgmt_name,
                eff_domain,
                lookup_key,
                cache_mode=cache_mode,
                cache_ttl=cache_ttl,
            )
            if svc_by_uid is not None and (clean_type is None or svc_by_uid.type.lower() == clean_type):
                return svc_by_uid

            svcs = await self._service_cache.get_services(
                mgmt_name=mgmt_name,
                domain=eff_domain,
                types=[clean_type] if clean_type else None,
                names=lookup_names,
                include_global=(eff_domain == "Global"),
                cache_mode="cache" if cache_mode != "force" else cache_mode,
                cache_ttl=cache_ttl,
            )
            if svcs:
                return svcs[0]

        if is_asset_type or clean_type is None:
            assets = await self._asset_service.get_assets(
                mgmt_name=mgmt_name,
                domain=eff_domain,
                types=[clean_type] if clean_type else None,
                include_global=(eff_domain == "Global"),
                cache_mode=cache_mode,
                cache_ttl=cache_ttl,
            )
            needle_lower = clean_query.lower()
            lookup_lower = lookup_key.lower()
            for asset in assets:
                if (
                    asset.uid in (clean_query, lookup_key)
                    or asset.name in (clean_query, lookup_key)
                    or asset.asset_id == clean_query
                    or asset.id == clean_query
                ):
                    return asset
            for asset in assets:
                if (
                    asset.name.lower() in (needle_lower, lookup_lower)
                    or asset.asset_id.lower() == needle_lower
                    or asset.id.lower() == needle_lower
                ):
                    return asset

        return None

    # Cross-domain lookup (eff_domain is None): strictly cache-only and includes Global
    needle_lower = clean_query.lower()
    lookup_lower = lookup_key.lower()
    exact_matches: list[CPObject | CPService | Asset] = []
    ci_matches: list[CPObject | CPService | Asset] = []

    if not is_service_type and not is_asset_type:
        objs = await self._object_service.get_objects(
            mgmt_name=mgmt_name,
            domain=None,
            types=[clean_type] if clean_type else None,
            names=None,
            include_global=True,
            cache_mode="cache",
            cache_ttl=cache_ttl,
        )
        for o in objs:
            if (
                o.uid in (clean_query, lookup_key)
                or o.name in (clean_query, lookup_key)
                or o.object_id == clean_query
                or o.id == clean_query
            ):
                exact_matches.append(o)
            elif (
                o.uid.lower() in (needle_lower, lookup_lower)
                or o.name.lower() in (needle_lower, lookup_lower)
                or o.object_id.lower() == needle_lower
                or o.id.lower() == needle_lower
            ):
                ci_matches.append(o)

    if is_service_type or clean_type is None:
        svcs = await self._service_cache.get_services(
            mgmt_name=mgmt_name,
            domain=None,
            types=[clean_type] if clean_type else None,
            names=None,
            include_global=True,
            cache_mode="cache",
            cache_ttl=cache_ttl,
        )
        for s in svcs:
            if (
                s.uid in (clean_query, lookup_key)
                or s.name in (clean_query, lookup_key)
                or s.service_id == clean_query
                or s.id == clean_query
            ):
                exact_matches.append(s)
            elif (
                s.uid.lower() in (needle_lower, lookup_lower)
                or s.name.lower() in (needle_lower, lookup_lower)
                or s.service_id.lower() == needle_lower
                or s.id.lower() == needle_lower
            ):
                ci_matches.append(s)

    if is_asset_type or clean_type is None:
        assets = await self._asset_service.get_assets(
            mgmt_name=mgmt_name,
            domain=None,
            types=[clean_type] if clean_type else None,
            include_global=True,
            cache_mode="cache",
            cache_ttl=cache_ttl,
        )
        for asset in assets:
            if (
                asset.uid in (clean_query, lookup_key)
                or asset.name in (clean_query, lookup_key)
                or asset.asset_id == clean_query
                or asset.id == clean_query
            ):
                exact_matches.append(asset)
            elif (
                asset.name.lower() in (needle_lower, lookup_lower)
                or asset.asset_id.lower() == needle_lower
                or asset.id.lower() == needle_lower
            ):
                ci_matches.append(asset)

    chosen_pool = exact_matches if exact_matches else ci_matches
    if not chosen_pool:
        return None

    deduped: list[CPObject | CPService | Asset] = []
    seen_keys: set[tuple[str, str]] = set()
    for item in chosen_pool:
        key = (item.domain_name, item.uid)
        if key not in seen_keys:
            seen_keys.add(key)
            deduped.append(item)

    distinct_domains = {item.domain_name for item in deduped}
    if len(distinct_domains) > 1:
        raise AmbiguousObjectError(
            mgmt_name,
            clean_query,
            [item.model_dump(mode="json") for item in deduped],
        )
    return deduped[0]