class ActivityContext:
    """Context object providing Pyramid integration for one activity execution.

    A context belongs to a single execution. The bound activity creates one per
    invocation, so concurrent executions never share a request, and therefore
    never share a dbsession or a transaction.

    The request has all the same properties and methods as a web request,
    including any configured via add_request_method (like dbsession, tm, etc.).

    Example:
        @activity.defn
        async def my_activity(context: ActivityContext, user_id: int) -> bool:
            # Real Pyramid request with all configured methods
            session = context.request.dbsession
            user = session.query(User).get(user_id)
            return user is not None
    """

    def __init__(self, env: PyramidEnvironment) -> None:
        """Initialize the activity context.

        Args:
            env: PyramidEnvironment instance
        """
        self._env = env
        self._request: Optional[Request] = None
        self._threadlocal_context: Optional[Union[RequestContext, RegistryContext]] = None

    @property
    def env(self) -> PyramidEnvironment:
        """Get the Pyramid environment."""
        return self._env

    @property
    def registry(self) -> "Registry":
        """Get the Pyramid registry."""
        return self._env.registry

    @property
    def settings(self) -> dict:
        """Get application settings (shortcut to registry.settings)."""
        return self._env.registry.settings

    @property
    def request(self) -> Request:
        """Get this execution's request.

        This is a real Pyramid Request object with all configured
        request methods (dbsession, tm, etc.) available.

        Raises:
            RuntimeError: If accessed outside of activity execution
        """
        if self._request is None:
            raise RuntimeError(
                "ActivityContext.request accessed outside of activity execution. "
                "The request is only available during activity execution."
            )
        return self._request

    def create_request(self, *, threadlocal_request: bool = True) -> Request:
        """Create the Pyramid Request for this activity execution.

        Uses Pyramid's request factory to create a real request, applies request
        extensions (add_request_method), and opens a threadlocal scope.

        Args:
            threadlocal_request: Publish the request on Pyramid's threadlocal
                stack, so ``get_current_request`` returns it. Only correct when
                the execution owns its thread, which is the case for a sync
                activity running in Temporal's activity executor. Concurrent
                async executions share the event loop thread, so they publish
                the registry alone instead.

        Returns:
            A real Pyramid Request instance

        Raises:
            RuntimeError: If this context already has a request
        """
        if self._request is not None:
            raise RuntimeError(
                "ActivityContext already has a request. A context belongs to a "
                "single activity execution and cannot be reused."
            )

        registry = self._env.registry
        request_factory = registry.queryUtility(IRequestFactory, default=Request)
        request = request_factory.blank("/")
        request.registry = registry

        if self._env.request is not None:
            request.environ.update(self._env.request.environ)

        self._threadlocal_context = RequestContext(request) if threadlocal_request else RegistryContext(registry)
        self._threadlocal_context.begin()
        apply_request_extensions(request)

        self._request = request

        logger.debug(
            "Created Pyramid Request for activity (request id: %s)",
            id(self._request),
        )
        return self._request

    def close_request(self) -> None:
        """Close this execution's request and clean up resources.

        Processes finished callbacks and tears down the threadlocal scope.
        """
        request = self._request
        threadlocal_context = self._threadlocal_context

        if request is None or threadlocal_context is None:
            return

        try:
            if request.finished_callbacks:
                request._process_finished_callbacks()
            threadlocal_context.end()
            logger.debug("Closed Pyramid Request context")
        except Exception as e:
            logger.warning("Error closing Pyramid Request context: %s", e)
        finally:
            self._threadlocal_context = None
            self._request = None

class PyramidActivity:
    """Wrapper for pyramid-temporal activities.

    This class wraps an activity function and provides the ability to bind it to
    a Pyramid environment for execution.

    The wrapper is what workflow code references, so it carries a Temporal
    activity definition and is callable: ``workflow.execute_activity(my_activity,
    ...)`` reads the registered name from that definition. What actually runs is
    the bound wrapper ``bind`` gives the Worker, and both carry the same name.
    """

    def __init__(
        self,
        fn: Callable[..., Any],
        name: Optional[str] = None,
        no_thread_cancel_exception: bool = False,
    ) -> None:
        """Initialize the pyramid activity wrapper.

        Args:
            fn: The original activity function
            name: Optional custom activity name
            no_thread_cancel_exception: Thread cancellation setting, sync only
        """
        self._fn = fn
        self._name = name or fn.__name__
        self._no_thread_cancel_exception = no_thread_cancel_exception

        # Copy function metadata for better debugging
        functools.update_wrapper(self, fn)

        # Mark as pyramid-temporal activity
        setattr(self, PYRAMID_ACTIVITY_MARKER, True)

        # Let workflow code reference this activity instead of its name
        setattr(self, TEMPORAL_ACTIVITY_DEFINITION, self._temporal_definition())

    def _temporal_definition(self) -> temporal_activity._Definition:
        """Build the definition Temporal reads when a workflow references this.

        The types are passed explicitly rather than left to Temporal's own
        inference for two reasons: the context argument belongs to the binding
        and never travels over the wire, so it must not appear in ``arg_types``;
        and inference on a callable instance would look at ``__call__`` and lose
        the declared return type, which is what converts an activity result back
        into the type the body declared.
        """
        arg_types, ret_type = self._type_hints()
        self._check_context_parameter(arg_types)

        return temporal_activity._Definition(
            name=self._name,
            fn=self._fn,
            is_async=self.is_async,
            no_thread_cancel_exception=self._no_thread_cancel_exception,
            arg_types=None if arg_types is None else arg_types[1:],
            ret_type=ret_type,
        )

    def _type_hints(self) -> Tuple[Optional[List[type]], Optional[type]]:
        """Resolve the body's annotations, the way Temporal resolves an activity's.

        Raises:
            TypeError: If an annotation cannot be resolved, which is what a type
                imported only under ``TYPE_CHECKING`` leaves behind
        """
        try:
            return temporal_common._type_hints_from_func(self._fn)
        except NameError as error:
            raise TypeError(
                f"Activity '{self._name}' has an annotation that cannot be resolved: {error}. "
                "Temporal reads these to convert arguments and results, so every type an "
                "activity annotates must be importable at runtime, not only under TYPE_CHECKING."
            ) from error

    def _check_context_parameter(self, arg_types: Optional[List[type]]) -> None:
        """Refuse a body that cannot receive the injected context.

        Every execution calls the body with its context first, and the workflow
        facing definition describes only the arguments that follow it. A body
        without that parameter breaks both quietly: the definition would claim
        one argument fewer than the activity takes, and the mistake would
        surface as an activity failure rather than here, where it was made.

        Raises:
            TypeError: If the first parameter cannot be an ActivityContext
        """
        positional = {
            inspect.Parameter.POSITIONAL_ONLY,
            inspect.Parameter.POSITIONAL_OR_KEYWORD,
            inspect.Parameter.VAR_POSITIONAL,
        }
        parameters = inspect.signature(self._fn).parameters.values()

        if not any(param.kind in positional for param in parameters):
            raise TypeError(
                f"Activity '{self._name}' takes no positional argument, so it cannot "
                "receive an ActivityContext. Declare it as the first parameter."
            )

        # A first parameter annotated with nothing, or with Any, claims nothing to
        # contradict. Any needs saying explicitly because it became a class in 3.11,
        # and the check would otherwise refuse it there and accept it on 3.10.
        declared = arg_types[0] if arg_types else None
        if declared is None or declared is Any:
            return

        if isinstance(declared, type) and not issubclass(declared, ActivityContext):
            raise TypeError(
                f"Activity '{self._name}' declares {declared.__name__} as its first "
                "parameter, which must be an ActivityContext."
            )

    def __call__(self, context: ActivityContext, *args: Any, **kwargs: Any) -> Any:
        """Run the activity body inside an execution that already exists.

        This is how an activity body is exercised as a plain function, given a
        context from ``activity_execution``. Async bodies come back as their
        coroutine, for the caller to await.

        Raises:
            TypeError: If the first argument is not an ActivityContext, which
                means nothing bound the activity to a Pyramid environment.
        """
        if not isinstance(context, ActivityContext):
            raise TypeError(
                f"Activity '{self._name}' takes an ActivityContext as its first argument, "
                f"got {type(context).__name__}. Register it with pyramid_temporal.Worker, "
                "which binds the Pyramid environment, not with temporalio.worker.Worker."
            )

        return self._fn(context, *args, **kwargs)

    @property
    def name(self) -> str:
        """Get the activity name."""
        return self._name

    @property
    def fn(self) -> Callable[..., Any]:
        """Get the original function."""
        return self._fn

    @property
    def is_async(self) -> bool:
        """Whether the activity body is a coroutine function."""
        return inspect.iscoroutinefunction(self._fn)

    def bind(self, env: PyramidEnvironment) -> Callable[..., Any]:
        """Bind this activity to a Pyramid environment for Temporal registration.

        Every call of the returned activity is one execution, and owns its own
        ActivityContext, Pyramid request, and transaction. Nothing is shared
        between executions, so they are safe to run concurrently.

        An async activity binds to a coroutine function, which Temporal runs on
        the worker's event loop. A sync activity binds to a plain function, which
        Temporal runs in its activity executor, so a blocking body leaves the
        event loop free.

        Args:
            env: The PyramidEnvironment each execution builds its request from

        Returns:
            An activity that can be registered with a Temporal Worker
        """
        fn = self._fn
        name = self._name

        if self.is_async:

            @temporal_activity.defn(name=name)
            async def execute_async(*args: Any, **kwargs: Any) -> Any:
                """Execute one async activity execution with context injection."""
                with activity_execution(env, threadlocal_request=False) as context:
                    return await fn(context, *args, **kwargs)

            return execute_async

        @temporal_activity.defn(name=name, no_thread_cancel_exception=self._no_thread_cancel_exception)
        def execute_sync(*args: Any, **kwargs: Any) -> Any:
            """Execute one sync activity execution with context injection."""
            with activity_execution(env, threadlocal_request=True) as context:
                return fn(context, *args, **kwargs)

        return execute_sync

    def __repr__(self) -> str:
        return f"<PyramidActivity '{self._name}'>"

class PyramidEnvironment:
    """Wrapper for Pyramid bootstrap environment.

    This class wraps the output of pyramid.paster.bootstrap, providing
    structured access to the Pyramid application components.

    Attributes:
        registry: The Pyramid registry
        app: The WSGI application
        request: The base request object
        root: The root object (for traversal-based applications)

    Example:
        from pyramid.paster import bootstrap
        from pyramid_temporal import PyramidEnvironment

        # Create from bootstrap output
        env_dict = bootstrap('development.ini')
        env = PyramidEnvironment.from_bootstrap(env_dict)

        # Access components
        settings = env.settings
        registry = env.registry

        # Clean up when done
        env.close()
    """

    def __init__(
        self,
        registry: "Registry",
        app: Optional[Any] = None,
        request: Optional[Any] = None,
        root: Optional[Any] = None,
        closer: Optional[Callable[[], None]] = None,
    ) -> None:
        """Initialize the Pyramid environment.

        Args:
            registry: Pyramid registry instance (required)
            app: WSGI application instance
            request: Base request object from bootstrap
            root: Root object for traversal-based applications
            closer: Cleanup callable from bootstrap
        """
        self._registry = registry
        self._app = app
        self._request = request
        self._root = root
        self._closer = closer

        logger.debug("Created PyramidEnvironment with registry: %s", registry)

    @classmethod
    def from_bootstrap(cls, env: dict) -> "PyramidEnvironment":
        """Create a PyramidEnvironment from bootstrap output.

        This is the preferred way to create a PyramidEnvironment when
        using pyramid.paster.bootstrap.

        Args:
            env: Dictionary returned by pyramid.paster.bootstrap()

        Returns:
            PyramidEnvironment instance

        Example:
            from pyramid.paster import bootstrap
            from pyramid_temporal import PyramidEnvironment

            env = PyramidEnvironment.from_bootstrap(bootstrap('development.ini'))
        """
        return cls(
            registry=env["registry"],
            app=env.get("app"),
            request=env.get("request"),
            root=env.get("root"),
            closer=env.get("closer"),
        )

    @property
    def registry(self) -> "Registry":
        """Get the Pyramid registry."""
        return self._registry

    @property
    def app(self) -> Optional[Any]:
        """Get the WSGI application."""
        return self._app

    @property
    def request(self) -> Optional[Any]:
        """Get the base request object from bootstrap."""
        return self._request

    @property
    def root(self) -> Optional[Any]:
        """Get the root object for traversal-based applications."""
        return self._root

    @property
    def settings(self) -> dict:
        """Get application settings (shortcut to registry.settings)."""
        return self._registry.settings

    def close(self) -> None:
        """Clean up resources.

        This calls the closer function from bootstrap to properly
        clean up the Pyramid application.
        """
        if self._closer is not None:
            logger.debug("Closing PyramidEnvironment")
            self._closer()

    def __repr__(self) -> str:
        """Return string representation."""
        return f"<PyramidEnvironment registry={self._registry}>"

class Worker:
    """Pyramid-aware Temporal Worker.

    This worker wraps the standard Temporal Worker and binds pyramid-temporal
    activities to the Pyramid environment. Each activity execution gets its own
    Pyramid request and its own transaction, so ``max_concurrent_activities``
    greater than 1 is safe.

    Activities written as plain ``def`` functions run in Temporal's activity
    executor. The worker creates a thread pool for them unless the caller passes
    its own ``activity_executor``.

    Example:
        from pyramid.paster import bootstrap
        from pyramid_temporal import Worker, activity, PyramidEnvironment

        @activity.defn
        async def my_activity(context: ActivityContext, user_id: int) -> bool:
            session = context.request.dbsession
            # ... do work ...
            return True

        # Create environment from bootstrap
        env = PyramidEnvironment.from_bootstrap(bootstrap('development.ini'))

        worker = Worker(
            client,
            env,
            task_queue="my-queue",
            activities=[my_activity],
            workflows=[MyWorkflow],
            max_concurrent_activities=10,
        )

        # Run the worker
        await worker.run()
    """

    def __init__(
        self,
        client: Client,
        env: PyramidEnvironment,
        *,
        task_queue: str,
        activities: Sequence[Any] = (),
        workflows: Sequence[type] = (),
        interceptors: Sequence["Interceptor"] = (),
        **kwargs: Any,
    ) -> None:
        """Initialize the Pyramid-aware worker.

        Args:
            client: Temporal client instance
            env: PyramidEnvironment instance (from bootstrap)
            task_queue: Name of the task queue to poll
            activities: List of activities (both pyramid-temporal and plain Temporal)
            workflows: List of workflow classes
            interceptors: Interceptors to include
            **kwargs: Additional arguments passed to Temporal Worker
        """
        self._client = client
        self._env = env
        self._task_queue = task_queue
        self._workflows = workflows
        self._interceptors = list(interceptors)
        self._extra_kwargs = kwargs

        # Bind pyramid activities to the environment and pass others through
        self._bound_activities = self._bind_activities(activities)

        # Sync activities need an executor, and the worker owns the one it creates
        self._owned_activity_executor = self._create_activity_executor()

        # Create the underlying Temporal worker
        self._worker = self._create_worker()

        logger.info(
            "Created Pyramid Worker for task queue '%s' with %d activities and %d workflows",
            task_queue,
            len(self._bound_activities),
            len(workflows),
        )

    def _bind_activities(self, activities: Sequence[Any]) -> list:
        """Bind pyramid-temporal activities to the environment, pass through others.

        Args:
            activities: List of activities (mixed pyramid and plain)

        Returns:
            List of bound/processed activities
        """
        bound = []
        for act in activities:
            if is_pyramid_activity(act):
                pyramid_act: PyramidActivity = act
                bound_act = pyramid_act.bind(self._env)
                logger.debug(
                    "Bound pyramid activity: %s (%s)",
                    pyramid_act.name,
                    "async" if pyramid_act.is_async else "sync",
                )
                bound.append(bound_act)
            else:
                # Pass through plain Temporal activity
                logger.debug("Passing through plain activity: %s", getattr(act, "__name__", act))
                bound.append(act)
        return bound

    def _create_activity_executor(self) -> Optional[ThreadPoolExecutor]:
        """Create the executor Temporal requires for sync activities.

        Temporal refuses to register a non-async activity without an
        ``activity_executor``, so the worker provides one. Callers that pass
        their own executor keep full control and own its lifetime.

        Returns:
            The created executor, or None when none is needed or wanted
        """
        if "activity_executor" in self._extra_kwargs:
            return None

        if all(_is_async_activity(act) for act in self._bound_activities):
            return None

        max_workers = self._extra_kwargs.get("max_concurrent_activities") or DEFAULT_ACTIVITY_SLOTS
        logger.info("Creating activity executor with %d threads for sync activities", max_workers)
        return ThreadPoolExecutor(max_workers=max_workers, thread_name_prefix=ACTIVITY_THREAD_NAME_PREFIX)

    def _activity_executor_kwargs(self) -> dict:
        """Return the executor keyword argument, when the worker created one."""
        if self._owned_activity_executor is None:
            return {}
        return {"activity_executor": self._owned_activity_executor}

    def _create_worker(self) -> TemporalWorker:
        """Create the underlying Temporal worker.

        Returns:
            Configured Temporal Worker instance
        """
        return TemporalWorker(
            self._client,
            task_queue=self._task_queue,
            activities=self._bound_activities,
            workflows=list(self._workflows),
            interceptors=self._interceptors,
            **self._activity_executor_kwargs(),
            **self._extra_kwargs,
        )

    def _shutdown_activity_executor(self) -> None:
        """Shut down the executor the worker created, if any."""
        if self._owned_activity_executor is not None:
            self._owned_activity_executor.shutdown(wait=True)
            logger.debug("Activity executor shut down")

    @property
    def env(self) -> PyramidEnvironment:
        """Get the Pyramid environment."""
        return self._env

    @property
    def task_queue(self) -> str:
        """Get the task queue name."""
        return self._task_queue

    @property
    def activity_executor(self) -> Optional[ThreadPoolExecutor]:
        """Get the activity executor the worker created for sync activities."""
        return self._owned_activity_executor

    async def run(self) -> None:
        """Run the worker until shutdown is requested.

        This is the main entry point for running the worker.
        It will poll the task queue and execute activities/workflows.
        """
        logger.info("Starting Pyramid Worker on task queue '%s'", self._task_queue)
        try:
            await self._worker.run()
        finally:
            self._shutdown_activity_executor()

    async def __aenter__(self) -> "Worker":
        """Async context manager entry."""
        await self._worker.__aenter__()
        return self

    async def __aexit__(self, *args: Any) -> None:
        """Async context manager exit."""
        try:
            await self._worker.__aexit__(*args)
        finally:
            self._shutdown_activity_executor()