Reserve pure execution for lifespan loops and native workers

src/firefly_weave/connectors/dispatcher.py

@@ -34,6 +34,7 @@
 from firefly_weave.contracts.access import Scope
 from firefly_weave.contracts.definitions import ContractModel
 from firefly_weave.contracts.workers import InstanceRequest
+from firefly_weave.operations.execution import ReservedSlots
 from firefly_weave.sdk.worker import Worker
 from firefly_weave.workers.repository import WorkerRepository
 
@@ -118,6 +119,9 @@ def __init__(
             image_digest,
             shutdown_seconds,
         )
+        # One slot per execution the workers can run at once, plus a spare for a pure call that
+        # outlived its cancelled execution. Connector calls never hold a request work slot.
+        self.reservation = ReservedSlots(sum(entry.capacity for entry in entries) + 1)
         self.workers: list[Worker] = []
         self.tasks: list[asyncio.Task[None]] = []
         self.closing = False
@@ -172,7 +176,7 @@ async def open(self) -> None:
                 )
 
             async def execute(lease: Any, scope: Scope = entry.scope, principal: UUID = entry.principal_id) -> Any:
-                return await self.service.execute(scope, principal, lease)
+                return await self.service.execute(scope, principal, lease, reservation=self.reservation)
 
             transport = ServiceTransport(self.service, entry.scope, entry.principal_id, instance.id)
             self.workers.append(Worker(transport, {ref: execute for ref in entry.task_types}, entry.capacity))

src/firefly_weave/connectors/execution.py

@@ -38,7 +38,7 @@
     TaskLease,
 )
 from firefly_weave.email.service import EmailService
-from firefly_weave.operations.execution import request_execution
+from firefly_weave.operations.execution import ReservedSlots, request_execution
 from firefly_weave.persistence.uow import UnitOfWork
 from firefly_weave.runtime.repository import RuntimeRepository
 from firefly_weave.workers.leases import TaskService
@@ -67,9 +67,13 @@ async def operation(self, name: str, scope: Scope, principal_id: UUID, *args: An
             actor = await self.access.load_principal(principal_id, tx=tx)
             return await getattr(self.tasks, name)(tx, *args, actor=actor, scope=scope, context=AuditContext())
 
-    async def execute(self, scope: Scope, principal_id: UUID, lease: TaskLease) -> JsonValue:
+    async def execute(
+        self, scope: Scope, principal_id: UUID, lease: TaskLease, *, reservation: ReservedSlots | None = None
+    ) -> JsonValue:
         self.registry.require_operational()
-        async with request_execution():
+        # A request lease covers the whole connector call. In-process native workers bring their
+        # own reservation instead, whose slots only the call's pure work takes.
+        async with request_execution() if reservation is None else reservation.execution():
             return await self._execute(scope, principal_id, lease)
 
     async def _execute(self, scope: Scope, principal_id: UUID, lease: TaskLease) -> JsonValue:

src/firefly_weave/operations/execution.py

@@ -70,6 +70,7 @@ def release(self) -> None:
 
 _request_lease: ContextVar[_Lease | None] = ContextVar("weave_request_execution", default=None)
 _debug_lease: ContextVar[_Lease | None] = ContextVar("weave_debug_execution", default=None)
+_reserved_slots: ContextVar[threading.BoundedSemaphore | None] = ContextVar("weave_reserved_execution", default=None)
 
 
 @asynccontextmanager
@@ -95,6 +96,31 @@ async def guarded(*args: P.args, **kwargs: P.kwargs) -> R:
     return guarded
 
 
+class ReservedSlots:
+    """Pure execution reserved for one lifespan owner, apart from request admission.
+
+    In-flight requests can hold every work and control slot. Background work that borrowed them
+    was refused and read the refusal as a failed recovery quantum, a blocked receipt, or a failed
+    consumer turn. Inside `execution()`, each pure call takes one slot from this reservation
+    instead, and any request lease the task inherited is set aside. A call still running after its
+    caller was cancelled keeps its slot until it ends, so the size bounds the owner's pure threads;
+    a spare slot keeps one such call from refusing the owner's next one.
+    """
+
+    def __init__(self, slots: int) -> None:
+        self.size = slots
+        self.slots = threading.BoundedSemaphore(slots)
+
+    @asynccontextmanager
+    async def execution(self) -> AsyncIterator[None]:
+        request, reserved = _request_lease.set(None), _reserved_slots.set(self.slots)
+        try:
+            yield
+        finally:
+            _reserved_slots.reset(reserved)
+            _request_lease.reset(request)
+
+
 @asynccontextmanager
 async def request_execution(*, control: bool = False) -> AsyncIterator[None]:
     lease = _Lease(control)
@@ -125,7 +151,7 @@ async def inventory_execution() -> AsyncIterator[None]:
 
 async def execute_pure[Result](call: Callable[[], Result], *, control: bool = False) -> Result:
     inherited = _request_lease.get()
-    lease = inherited or _Lease(control)
+    lease = inherited or _Lease(control, slots=_reserved_slots.get())
     try:
         lease.enter()
         debug = _debug_lease.get()

src/firefly_weave/providers/loop.py

@@ -25,11 +25,16 @@
 
 from firefly_weave.access.scheduler import next_scope
 from firefly_weave.email.source import EmailSourceService
+from firefly_weave.operations.execution import ReservedSlots
 from firefly_weave.persistence.migrations import check_schema
 from firefly_weave.persistence.uow import UnitOfWork
 from firefly_weave.providers.dispatcher import ProviderDispatcher
 from firefly_weave.settings import Settings
 
+# Requests can hold every work and control slot. A provider cycle runs one pure call at a time;
+# the spare slot absorbs a call that outlived its timed-out receipt, as the work pool did.
+PROVIDER_EXECUTION = ReservedSlots(2)
+
 
 class ProviderLoop:
     def __init__(
@@ -75,18 +80,19 @@ async def open(self) -> None:
 
     async def cycle(self) -> None:
         assert self.engine is not None
-        async with asyncio.timeout(5):
-            async with self.engine.begin() as connection:
-                await connection.execute(text("SET LOCAL statement_timeout='4000ms'"))
-                await connection.execute(text("SET LOCAL lock_timeout='1000ms'"))
-                tenant = await connection.scalar(text("SELECT weave_provider_tenants(1)"))
-            if tenant is None:
-                return
-            authority = await next_scope(self.uow, tenant, provider=True)
-        if authority is not None:
-            await self.dispatcher.scan(authority, 10)
-            if self.email_sources is not None:
-                await self.email_sources.scan(authority, 1)
+        async with PROVIDER_EXECUTION.execution():
+            async with asyncio.timeout(5):
+                async with self.engine.begin() as connection:
+                    await connection.execute(text("SET LOCAL statement_timeout='4000ms'"))
+                    await connection.execute(text("SET LOCAL lock_timeout='1000ms'"))
+                    tenant = await connection.scalar(text("SELECT weave_provider_tenants(1)"))
+                if tenant is None:
+                    return
+                authority = await next_scope(self.uow, tenant, provider=True)
+            if authority is not None:
+                await self.dispatcher.scan(authority, 10)
+                if self.email_sources is not None:
+                    await self.email_sources.scan(authority, 1)
 
     async def poll(self) -> None:
         while True:

src/firefly_weave/runtime/scheduler.py

@@ -26,12 +26,25 @@
 from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
 
 from firefly_weave.access.scheduler import next_scope, tenant_page
+from firefly_weave.definitions.models import CatalogError
+from firefly_weave.operations.execution import ReservedSlots
 from firefly_weave.persistence.migrations import check_schema
 from firefly_weave.persistence.uow import UnitOfWork
 from firefly_weave.runtime.recovery import RecoveryService
 from firefly_weave.settings import Settings
 from firefly_weave.triggers.scheduler import Scheduler
 
+# Requests can hold every work and control slot. A recovery cycle runs one pure call at a time;
+# the spare slot absorbs a call that outlived its cancelled quantum, as the work pool did.
+RECOVERY_EXECUTION = ReservedSlots(2)
+
+
+def failure_cause(error: Exception) -> str:
+    """Name a failed scan by error class and catalog code, never by message or values."""
+    if isinstance(error, CatalogError):
+        return f"{type(error).__name__} {error.code}"
+    return type(error).__name__
+
 
 class RecoveryLoop:
     def __init__(
@@ -86,39 +99,46 @@ async def open(self) -> None:
 
     async def cycle(self) -> None:
         assert self.engine is not None
-        budget = min(self.settings.database_timeout_seconds, 5.0)
-        async with asyncio.timeout(budget):
-            tenants = await tenant_page(self.engine)
-        for tenant in tenants:
-            try:
-                async with asyncio.timeout(budget):
-                    authority = await next_scope(self.uow, tenant)
-                if authority is None:
+        async with RECOVERY_EXECUTION.execution():
+            budget = min(self.settings.database_timeout_seconds, 5.0)
+            async with asyncio.timeout(budget):
+                tenants = await tenant_page(self.engine)
+            for tenant in tenants:
+                try:
+                    async with asyncio.timeout(budget):
+                        authority = await next_scope(self.uow, tenant)
+                    if authority is None:
+                        continue
+                    # Separate transactions/time budgets reserve recovery even when
+                    # starts are backpressured. A tenant gets one environment quantum.
+                    async with asyncio.timeout(budget):
+                        if self.operational():
+                            await self.recovery._scan_scheduled(authority, 10)
+                        else:
+                            await self.recovery._scan_terminal_scheduled(authority, 10)
+                except Exception as error:
+                    logging.getLogger(__name__).error(
+                        "Tenant recovery scan failed (%s); traversal will continue", failure_cause(error)
+                    )
                     continue
-                # Separate transactions/time budgets reserve recovery even when
-                # starts are backpressured. A tenant gets one environment quantum.
-                async with asyncio.timeout(budget):
-                    if self.operational():
-                        await self.recovery._scan_scheduled(authority, 10)
-                    else:
-                        await self.recovery._scan_terminal_scheduled(authority, 10)
-            except Exception:
-                logging.getLogger(__name__).error("Tenant recovery scan failed; traversal will continue")
-                continue
-            if not self.operational():
-                continue
-            try:
-                async with asyncio.timeout(budget):
-                    await self.schedules._scan_scheduled(authority, 10)
-            except Exception:
-                logging.getLogger(__name__).error("Tenant schedule scan failed; traversal will continue")
+                if not self.operational():
+                    continue
+                try:
+                    async with asyncio.timeout(budget):
+                        await self.schedules._scan_scheduled(authority, 10)
+                except Exception as error:
+                    logging.getLogger(__name__).error(
+                        "Tenant schedule scan failed (%s); traversal will continue", failure_cause(error)
+                    )
 
     async def poll(self) -> None:
         while True:
             try:
                 await self.cycle()
-            except Exception:
-                logging.getLogger(__name__).error("Recovery catalog scan failed; retrying on next poll")
+            except Exception as error:
+                logging.getLogger(__name__).error(
+                    "Recovery catalog scan failed (%s); retrying on next poll", failure_cause(error)
+                )
             await asyncio.sleep(self.settings.scheduler_poll_seconds)
 
     async def close(self) -> None:

src/firefly_weave/triggers/kafka_loop.py

@@ -27,11 +27,16 @@
 from firefly_weave.access.scheduler import next_scope
 from firefly_weave.connectors.broker import BrokerConnectionConfig
 from firefly_weave.connectors.kafka_transport import BrokerClients, BrokerOwner
+from firefly_weave.operations.execution import ReservedSlots
 from firefly_weave.persistence.migrations import check_schema
 from firefly_weave.persistence.uow import UnitOfWork
 from firefly_weave.settings import Settings
 from firefly_weave.triggers.kafka import KafkaSource, KafkaTrigger
 
+# Requests can hold every work and control slot. At most `max_clients - 1` consumer turns run
+# one pure call each, and the broker policy caps `max_clients` at 8, which leaves a spare slot.
+TURN_EXECUTION = ReservedSlots(8)
+
 
 class KafkaLoop:
     def __init__(self, settings: Settings, uow: UnitOfWork, triggers: KafkaTrigger, clients: BrokerClients) -> None:
@@ -101,7 +106,7 @@ async def turn(self, source: KafkaSource) -> None:
         current = asyncio.current_task()
         assert current is not None
         try:
-            async with asyncio.timeout(60):
+            async with asyncio.timeout(60), TURN_EXECUTION.execution():
                 revision, secrets = await self.triggers.credentials(source.authority)
                 import json
 

tests/unit/connectors/test_local_build.py

@@ -268,6 +268,28 @@ async def test_local_development_executor_uses_local_attestation_not_the_image(a
     }
 
 
+async def test_native_handlers_execute_through_the_dispatcher_reservation(admission):
+    digest = "sha256:" + "f" * 64
+    service = _Service(_release(digest))
+    executions = []
+
+    async def execute(scope, principal_id, lease, *, reservation=None):
+        executions.append(reservation)
+        return {"ok": True}
+
+    service.execute = execute  # type: ignore[attr-defined]
+    native = NativeDispatcher(service, (_executor(),), digest, 1)  # type: ignore[arg-type]
+    await native.open()
+    try:
+        [worker] = native.workers
+        [handler] = set(worker.handlers.values())  # type: ignore[attr-defined]
+        assert await handler(object()) == {"ok": True}
+    finally:
+        await native.close()
+    # Connector calls take the native reservation, never a request work slot.
+    assert executions == [native.reservation]
+
+
 async def test_image_executor_keeps_the_packaged_image_check(admission):
     digest = "sha256:" + "b" * 64
     native = NativeDispatcher(_Service(_release(digest)), (_executor("image"),), digest, 1)