src/genro_asgi/tasks/executor.py

Source from this local checkout, regenerated when the reader rebuilds. Line links use #L<number>; a GitHub line range opens its first line.

1 # Copyright 2025 Softwell S.r.l.2 #3 # Licensed under the Apache License, Version 2.0 (the "License");4 # you may not use this file except in compliance with the License.5 # You may obtain a copy of the License at6 #7 #     https://www.apache.org/licenses/LICENSE-2.08 #9 # Unless required by applicable law or agreed to in writing, software10 # distributed under the License is distributed on an "AS IS" BASIS,11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.12 # See the License for the specific language governing permissions and13 # limitations under the License.14 15 """LocalTaskExecutor — run one spool task in-process on the live server (◆D22).16 17 A batch task names its code by ``mount`` + ``node_path`` (the app it lives in,18 and the route path inside that app's router). Running it needs the app instance19 — which needs the whole server (storage, dbs, sibling apps). In this mono-process20 core the server is ALREADY live: the executor binds to it directly and reaches21 the target app through ``server.application_at``. It never serves22 HTTP; it only resolves and runs handlers. (Distributed execution — a worker23 process that rebuilds the server from a config path — belongs to the24 orchestration package, D22, and is out of scope here.)25 26 Running a handler needs NO request context: a ``@route`` handler is a bound27 method, so ``self.server`` / ``self.db`` are already reachable from the app28 instance. The executor resolves the callable with ``app.route.node(node_path)``29 (a ``RouterNode``, callable, no HTTP) and invokes it with the task's params —30 the same async/sync split the dispatcher uses: an async handler stays on the31 loop, a sync handler goes through ``server.run_sync`` (the Macro 1 pool32 protocol; ``routed_application.py``/``applications/mcp.py`` are the mirrors).33 34 The manager has already MOVED the task into ``active/<worker>/``; the executor35 reads it there, runs it, and settles it once: any outcome (ok / error) moves the36 folder to ``terminated`` / ``aborted`` and a ``batch_id`` never moves again37 (§5.7 terminal-by-position). There is a single logical worker in the core,38 ``WORKER_ID`` == ``"local"``; the per-worker ``active/<worker>/`` structure stays39 for D22 forward-compat.40 41 The A<->C bridge (core 1e Phase 6): a descriptor carrying the launching MCP42 ``session_id`` gets its lifecycle published on ``server.tasks.hub`` — ``started``43 before the run, ``settled`` (with outcome/error) after — so a subscribed SSE44 stream follows the task live. ``session_id`` ``None`` = no push channel, no-op.45 The spool stays the source of truth; the hub is only the live courier.46 """47 48 from __future__ import annotations49 50 import asyncio51 import logging52 from typing import TYPE_CHECKING, Any53 54 from .spool import TaskSpool55 56 if TYPE_CHECKING:57     from ..server import BaseServer58 59 __all__ = ["LocalTaskExecutor", "WORKER_ID"]60 61 WORKER_ID = "local"     # the single logical worker in the mono-process core62 63 64 class LocalTaskExecutor:65     """Resolve a spool task by ``mount``/``node_path`` and run it on the live server.66 67     Note:68         Bound to the live server (dual relationship: ``self.server``); owns the69         ``TaskSpool`` over the server's ``site`` storage. Both attributes are the70         seam the ``TaskManager`` reuses (the spool is stateless — the same storage71         yields an equivalent spool).72     """73 74     __slots__ = ("server", "spool")75 76     def __init__(self, server: BaseServer) -> None:77         """Bind to the live server and open the spool over its storage."""78         self.server = server79         self.spool = TaskSpool(server.storage)80 81     def resolve(self, descriptor: dict[str, Any]) -> Any:82         """Return the callable ``RouterNode`` for a task descriptor.83 84         Looks the app up by its ``mount`` (``""`` is the root app, exactly as in85         the request demux) and resolves ``node_path`` in that app's router. The86         node is callable — invoking it runs the handler, no HTTP.87 88         Raises:89             LookupError: if no application answers ``mount``.90         """91         mount = descriptor["mount"]92         app = self.server.application_at(mount)93         if app is None:94             raise LookupError(f"no app mounted at {mount!r}")95         return app.route.node(descriptor["node_path"])96 97     async def execute(self, task_id: str, worker_id: str) -> str:98         """Run the task active on ``worker_id`` and settle it; return the outcome.99 100         Reads the descriptor and params from the spool, resolves the handler, runs101         it (async on the loop, sync on the pool via ``server.run_sync``), writes the102         result, and settles the folder to ``terminated`` (ok) or ``aborted`` (error).103         The outcome is ``"ok"`` or ``"error"``; on error the exception text is104         stamped on the descriptor via ``settle``.105 106         Raises:107             LookupError: if no task ``task_id`` exists in the spool.108         """109         descriptor = self.spool.get(task_id)110         if descriptor is None:111             raise LookupError(f"task not found: {task_id}")112         params = self.spool.read_params(task_id)113         outcome, error = "ok", None114         self._publish(descriptor, {"type": "started"})115         try:116             node = self.resolve(descriptor)117             if asyncio.iscoroutinefunction(node):118                 result = await node(**params)119             else:120                 result = await self.server.run_sync(lambda: node(**params))121             self.spool.write_result(task_id, worker_id, result)122         except Exception as exc:123             outcome, error = "error", f"{type(exc).__name__}: {exc}"124             logging.getLogger(__name__).exception("batch task %r failed", task_id)125         self.spool.settle(task_id, worker_id, outcome, error=error)126         self._publish(descriptor, {"type": "settled", "outcome": outcome, "error": error})127         return outcome128 129     def _publish(self, descriptor: dict[str, Any], event: dict[str, Any]) -> None:130         """Publish a lifecycle event on the hub, keyed by the launching session.131 132         The A<->C bridge: ``session_id`` is the launching MCP session stamped on133         the descriptor at ``create`` time; ``None`` means the sender has no push134         channel and publishing is a no-op (the hub itself no-ops without135         subscribers). The event always carries the ``task_id``.136         """137         session_id = descriptor.get("session_id")138         if session_id is None:139             return140         self.server.tasks.hub.publish(141             session_id, {"task_id": descriptor["task_id"], **event}142         )