Celery Language Runtime

CeleryLanguage is the runtime heart of this repository.

Read this API page when working on:

• execution mode behavior (local, distributed, auto)
• orchestration lowering (simple vs compiled)
• retry profile configuration and policy propagation
• worker queue inference and routing decisions
• journal event emission and status transitions

lingo.celery.language

Celery-based language implementation.

CeleryLanguage

Bases: Language

Celery-based orchestration language for distributed task execution.

Source code in lingo/celery/language.py

class CeleryLanguage(Language):
    """Celery-based orchestration language for distributed task execution."""

    _RETRY_PROFILES: dict[str, dict[str, Any]] = {
        "none": {
            "retry_on_failure": False,
            "max_retries": 0,
            "backoff_seconds": 0.0,
            "exponential_backoff": False,
            "jitter_seconds": 0.0,
            "retry_exceptions": None,
        },
        "standard": {
            "retry_on_failure": True,
            "max_retries": 2,
            "backoff_seconds": 2.0,
            "exponential_backoff": True,
            "jitter_seconds": 0.0,
            "retry_exceptions": None,
        },
        "aggressive": {
            "retry_on_failure": True,
            "max_retries": 5,
            "backoff_seconds": 1.0,
            "exponential_backoff": True,
            "jitter_seconds": 0.5,
            "retry_exceptions": None,
        },
    }

    def __init__(
        self,
        archive: Optional[Archive] = None,
        journal: Optional[Journal] = None,
        broker: Optional[str] = None,
        backend: Optional[str] = None,
        channel: str = "lingo.default",
        node_id: Optional[str] = None,
        name: Optional[str] = None,
        routing_mode: str = "shared",
        execution_mode: str = "auto",
        orchestration_mode: str = "simple",
        restart_policy: Optional[RestartPolicy] = None,
        retry_profile: Optional[str] = None,
        retry_on_failure: Optional[bool] = None,
        max_retries: Optional[int] = None,
        backoff_seconds: Optional[float] = None,
        exponential_backoff: Optional[bool] = None,
        jitter_seconds: Optional[float] = None,
        retry_exceptions: Optional[list[str]] = None,
        compiled_retry_profile: Optional[str] = None,
        compiled_retry_on_failure: Optional[bool] = None,
        compiled_max_retries: Optional[int] = None,
        compiled_backoff_seconds: Optional[float] = None,
        compiled_exponential_backoff: Optional[bool] = None,
        compiled_jitter_seconds: Optional[float] = None,
        compiled_retry_exceptions: Optional[list[str]] = None,
        scribble_root: Optional[str] = None,
        debug: Optional[bool] = None,
    ):
        self.archive = archive
        self.journal = journal
        self.broker = broker or "redis://localhost:6379"
        self.backend = backend or self.broker
        self.channel = channel
        self.node_id = node_id or f"node-{uuid.uuid4().hex[:8]}"
        self.name = name
        if routing_mode not in {"shared", "task"}:
            raise ValueError("routing_mode must be 'shared' or 'task'")
        if orchestration_mode not in {"simple", "compiled"}:
            raise ValueError("orchestration_mode must be 'simple' or 'compiled'")
        self.routing_mode = routing_mode
        self.execution_mode = execution_mode
        self.orchestration_mode = orchestration_mode
        self.retry_profile = self._validate_retry_profile(retry_profile)
        self.compiled_retry_profile = self._validate_retry_profile(compiled_retry_profile)
        self._policy_overrides = {
            "retry_on_failure": retry_on_failure,
            "max_retries": max_retries,
            "backoff_seconds": backoff_seconds,
            "exponential_backoff": exponential_backoff,
            "jitter_seconds": jitter_seconds,
            "retry_exceptions": retry_exceptions,
        }
        self._compiled_policy_overrides = {
            "retry_on_failure": compiled_retry_on_failure,
            "max_retries": compiled_max_retries,
            "backoff_seconds": compiled_backoff_seconds,
            "exponential_backoff": compiled_exponential_backoff,
            "jitter_seconds": compiled_jitter_seconds,
            "retry_exceptions": compiled_retry_exceptions,
        }
        self.restart_policy = self._apply_policy_overrides(
            self._apply_retry_profile(restart_policy or RestartPolicy(), self.retry_profile),
            self._policy_overrides,
        )
        self.compiled_restart_policy = self._apply_policy_overrides(
            self._apply_retry_profile(self.restart_policy, self.compiled_retry_profile),
            self._compiled_policy_overrides,
        )
        self.scribble_root = scribble_root
        self.debug = self._resolve_debug_flag(debug)
        self.grammar_registry: Dict[str, Dict[str, Any]] = {}
        self.verb_registry: Dict[str, Callable] = {}
        self._started = False
        worker_hint = self.name or self.node_id
        self.private_queue = f"lingo.worker.{re.sub(r'[^A-Za-z0-9_.-]+', '_', worker_hint)}"

        self.celery_app = Celery(f"lingo.{self.node_id}", broker=self.broker, backend=self.backend)
        self.celery_app.conf.task_default_queue = self.channel
        self.celery_app.conf.task_track_started = True
        self.celery_app.conf.worker_send_task_events = True
        self.celery_app.conf.task_send_sent_event = True

        self._remote_task_name = "lingo.execute"
        self._compiled_seed_task_name = "lingo.compiled.seed"
        self._compiled_node_task_name = "lingo.compiled.node"
        self._compiled_join_task_name = "lingo.compiled.join"

        self._debug_log(
            "language.initialized",
            node_id=self.node_id,
            name=self.name,
            channel=self.channel,
            routing_mode=self.routing_mode,
            execution_mode=self.execution_mode,
            orchestration_mode=self.orchestration_mode,
        )

        @self.celery_app.task(name=self._remote_task_name, bind=True)
        def _execute(self_task, raw_envelope: dict[str, Any]) -> Any:
            envelope = ChannelEnvelope.model_validate(raw_envelope)
            envelope.meta.heartbeat_at = datetime.now(timezone.utc).isoformat()
            runtime_attempt = int(getattr(self_task.request, "retries", 0)) + 1
            self._debug_log(
                "task.pull",
                queue=envelope.meta.channel,
                job_id=envelope.meta.job_id,
                task=envelope.meta.root_task,
                attempt=runtime_attempt,
                source_node=envelope.meta.source_node,
                preferred_worker=envelope.meta.preferred_worker,
                local_required=envelope.meta.local_required,
                worker=self.name or self.node_id,
            )

            if self.journal is not None and hasattr(self.journal, "upsert_job"):
                self.journal.upsert_job(
                    envelope.meta.job_id,
                    status=JobStatusCode.RUNNING,
                    details=JobStatusDetails(
                        message=f"Executing task {envelope.meta.root_task}",
                        heartbeat_at=envelope.meta.heartbeat_at,
                        attempt=runtime_attempt,
                        progress=0.5,
                    ),
                    bump_attempt=True,
                )
            self._emit_event(
                envelope.meta.job_id,
                event="job.started",
                task_name=envelope.meta.root_task,
                attempt=runtime_attempt,
                message="Remote execution started",
            )

            phrase_obj = deserialize_phrase(envelope.payload)

            try:
                result = self._execute_phrase(
                    phrase_obj,
                    job_id=envelope.meta.job_id,
                    attempt=runtime_attempt,
                )
                transport_result = self._prepare_result_for_transport(
                    result,
                    job_id=envelope.meta.job_id,
                    task_name=envelope.meta.root_task,
                )
                encoded_result = serialize_value(transport_result)
                if self.journal is not None and hasattr(self.journal, "upsert_job"):
                    self.journal.upsert_job(
                        envelope.meta.job_id,
                        status=JobStatusCode.COMPLETED,
                        result=transport_result,
                        details=JobStatusDetails(
                            message="Completed successfully",
                            heartbeat_at=datetime.now(timezone.utc).isoformat(),
                            attempt=runtime_attempt,
                            progress=1.0,
                        ),
                    )
                self._emit_event(
                    envelope.meta.job_id,
                    event="job.completed",
                    task_name=envelope.meta.root_task,
                    attempt=runtime_attempt,
                    message="Remote execution completed",
                )
                return encoded_result
            except Exception as exc:
                retries_so_far = int(getattr(self_task.request, "retries", 0))
                if envelope.meta.policy.should_retry(exc, retries_so_far):
                    countdown = envelope.meta.policy.countdown_for_retry(retries_so_far)
                    self._emit_event(
                        envelope.meta.job_id,
                        event="job.retry_scheduled",
                        task_name=envelope.meta.root_task,
                        attempt=runtime_attempt,
                        message=str(exc),
                        data={"retry_in_seconds": countdown, "error": type(exc).__name__},
                    )
                    self._emit_event(
                        envelope.meta.job_id,
                        event="subjob.retry",
                        task_name=envelope.meta.root_task,
                        attempt=runtime_attempt,
                        message="Retry scheduled",
                    )
                    raise self_task.retry(exc=exc, countdown=countdown)
                self._emit_event(
                    envelope.meta.job_id,
                    event="job.failed",
                    task_name=envelope.meta.root_task,
                    attempt=runtime_attempt,
                    message=str(exc),
                    data={"error": type(exc).__name__},
                )
                raise

        @self.celery_app.task(name=self._compiled_seed_task_name)
        def _compiled_seed() -> Any:
            return None

        @self.celery_app.task(name=self._compiled_node_task_name, bind=True)
        def _execute_compiled_node(self_task, previous_result: Any, raw_node: dict[str, Any]) -> Any:
            job_id = str(raw_node.get("job_id") or "")
            task_name = str(raw_node["task"])
            inject_previous = bool(raw_node.get("inject_previous", False))
            serialized_args = list(raw_node.get("args") or [])
            serialized_kwargs = dict(raw_node.get("kwargs") or {})
            args = [deserialize_value(item) for item in serialized_args]
            kwargs = {key: deserialize_value(value) for key, value in serialized_kwargs.items()}

            if inject_previous:
                args = [deserialize_value(previous_result), *args]

            runtime_attempt = int(getattr(self_task.request, "retries", 0)) + 1
            policy = self._policy_from_payload(raw_node.get("policy"), fallback=self.compiled_restart_policy)
            self._debug_log(
                "task.pull",
                queue=getattr(self_task.request, "delivery_info", {}).get("routing_key"),
                job_id=job_id,
                task=task_name,
                attempt=runtime_attempt,
                node_id=raw_node.get("node_id"),
                inject_previous=inject_previous,
                compiled=True,
                worker=self.name or self.node_id,
            )
            try:
                result = self._execute_task(
                    task_name,
                    tuple(args),
                    kwargs,
                    job_id=job_id,
                    attempt=runtime_attempt,
                )
                transport_result = self._prepare_result_for_transport(
                    result,
                    job_id=job_id,
                    task_name=task_name,
                )
                return serialize_value(transport_result)
            except Exception as exc:
                retries_so_far = int(getattr(self_task.request, "retries", 0))
                if policy.should_retry(exc, retries_so_far):
                    countdown = policy.countdown_for_retry(retries_so_far)
                    self._emit_event(
                        job_id,
                        event="job.retry_scheduled",
                        task_name=task_name,
                        attempt=runtime_attempt,
                        message=str(exc),
                        data={"retry_in_seconds": countdown, "error": type(exc).__name__, "compiled": True},
                    )
                    self._emit_event(
                        job_id,
                        event="subjob.retry",
                        task_name=task_name,
                        attempt=runtime_attempt,
                        message="Retry scheduled",
                    )
                    raise self_task.retry(exc=exc, countdown=countdown)
                raise

        @self.celery_app.task(name=self._compiled_join_task_name, bind=True)
        def _execute_compiled_join(self_task, branch_results: list[Any], raw_join: dict[str, Any]) -> Any:
            job_id = str(raw_join.get("job_id") or "")
            task_name = str(raw_join["task"])
            decoded_branch_results = [deserialize_value(item) for item in branch_results]

            def decode_slot(slot: dict[str, Any]) -> Any:
                if slot.get("kind") == "dep":
                    return decoded_branch_results[int(slot["index"])]
                return deserialize_value(slot.get("value"))

            args = [decode_slot(slot) for slot in list(raw_join.get("args") or [])]
            kwargs = {
                key: decode_slot(slot)
                for key, slot in dict(raw_join.get("kwargs") or {}).items()
            }

            runtime_attempt = int(getattr(self_task.request, "retries", 0)) + 1
            policy = self._policy_from_payload(raw_join.get("policy"), fallback=self.compiled_restart_policy)
            self._debug_log(
                "task.pull",
                queue=getattr(self_task.request, "delivery_info", {}).get("routing_key"),
                job_id=job_id,
                task=task_name,
                attempt=runtime_attempt,
                branch_count=len(decoded_branch_results),
                compiled=True,
                worker=self.name or self.node_id,
            )
            try:
                result = self._execute_task(
                    task_name,
                    tuple(args),
                    kwargs,
                    job_id=job_id,
                    attempt=runtime_attempt,
                )
                transport_result = self._prepare_result_for_transport(
                    result,
                    job_id=job_id,
                    task_name=task_name,
                )
                return serialize_value(transport_result)
            except Exception as exc:
                retries_so_far = int(getattr(self_task.request, "retries", 0))
                if policy.should_retry(exc, retries_so_far):
                    countdown = policy.countdown_for_retry(retries_so_far)
                    self._emit_event(
                        job_id,
                        event="job.retry_scheduled",
                        task_name=task_name,
                        attempt=runtime_attempt,
                        message=str(exc),
                        data={"retry_in_seconds": countdown, "error": type(exc).__name__, "compiled": True},
                    )
                    self._emit_event(
                        job_id,
                        event="subjob.retry",
                        task_name=task_name,
                        attempt=runtime_attempt,
                        message="Retry scheduled",
                    )
                    raise self_task.retry(exc=exc, countdown=countdown)
                raise

        self._compiled_seed_task = _compiled_seed
        self._compiled_node_task = _execute_compiled_node
        self._compiled_join_task = _execute_compiled_join

    @staticmethod
    def _apply_policy_overrides(base_policy: RestartPolicy, overrides: dict[str, Any]) -> RestartPolicy:
        payload = base_policy.model_dump(mode="python")
        for key, value in overrides.items():
            if value is not None:
                payload[key] = value
        return RestartPolicy(**payload)

    @staticmethod
    def _env_flag_enabled(value: Optional[str]) -> bool:
        if value is None:
            return False
        return value.strip().lower() in {"1", "true", "yes", "on"}

    def _resolve_debug_flag(self, debug: Optional[bool]) -> bool:
        if debug is not None:
            return bool(debug)
        return self._env_flag_enabled(os.getenv("LINGO_DEBUG"))

    def _debug_log(self, event: str, **meta: Any) -> None:
        if not self.debug:
            return
        details = " ".join(
            f"{key}={meta[key]}"
            for key in sorted(meta.keys())
            if meta[key] is not None
        )
        if details:
            print(f"[lingo.debug] {event} {details}", flush=True)
        else:
            print(f"[lingo.debug] {event}", flush=True)

    @classmethod
    def _validate_retry_profile(cls, profile: Optional[str]) -> Optional[str]:
        if profile is None:
            return None
        normalized = profile.strip().lower()
        if normalized not in cls._RETRY_PROFILES:
            allowed = ", ".join(sorted(cls._RETRY_PROFILES.keys()))
            raise ValueError(f"retry_profile must be one of: {allowed}")
        return normalized

    @classmethod
    def _apply_retry_profile(cls, base_policy: RestartPolicy, profile: Optional[str]) -> RestartPolicy:
        if profile is None:
            return base_policy
        payload = base_policy.model_dump(mode="python")
        payload.update(cls._RETRY_PROFILES[profile])
        return RestartPolicy(**payload)

    @staticmethod
    def _policy_from_payload(raw_policy: Any, *, fallback: RestartPolicy) -> RestartPolicy:
        if isinstance(raw_policy, RestartPolicy):
            return raw_policy
        if isinstance(raw_policy, dict):
            return RestartPolicy(**raw_policy)
        return fallback

    def _resolve_policy(self, override: Optional[RestartPolicy]) -> RestartPolicy:
        if override is None:
            return self.restart_policy
        return self._apply_policy_overrides(override, self._policy_overrides)

    @staticmethod
    def _run_archive_coro(coro: Any) -> Any:
        try:
            running = asyncio.get_running_loop()
        except RuntimeError:
            running = None

        if running and running.is_running():
            loop = asyncio.new_event_loop()
            try:
                return loop.run_until_complete(coro)
            finally:
                loop.close()
        return asyncio.run(coro)

    def _archive_can_upload(self) -> bool:
        if self.archive is None:
            return False
        if not hasattr(self.archive, "ensure_reference_uploaded"):
            return False
        has_upload_client = getattr(self.archive, "has_upload_client", None)
        if callable(has_upload_client):
            return bool(has_upload_client())
        return True

    def _warn_unshared_corpus(self, *, task_name: str) -> None:
        warnings.warn(
            (
                f"Task '{task_name}' produced a Corpus without MinIO upload credentials. "
                "The corpus remains local to this worker and may be inaccessible to the server or other workers."
            ),
            RuntimeWarning,
            stacklevel=3,
        )

    def _upload_reference_if_available(self, reference: Reference[Any]) -> None:
        if self.archive is None or not hasattr(self.archive, "ensure_reference_uploaded"):
            return
        self._run_archive_coro(self.archive.ensure_reference_uploaded(reference))
        if hasattr(self.archive, "get_public_url"):
            reference.public_url = self._run_archive_coro(self.archive.get_public_url(reference))

    def _upload_references_in_value(self, value: Any) -> None:
        if isinstance(value, Reference):
            self._upload_reference_if_available(value)
            return
        if isinstance(value, Phrase):
            for arg in value.args:
                self._upload_references_in_value(arg)
            for arg in value.kwargs.values():
                self._upload_references_in_value(arg)
            for _task_name, args, kwargs in value.chain:
                for arg in args:
                    self._upload_references_in_value(arg)
                for arg in kwargs.values():
                    self._upload_references_in_value(arg)
            return
        if isinstance(value, (list, tuple, set)):
            for item in value:
                self._upload_references_in_value(item)
            return
        if isinstance(value, dict):
            for item in value.values():
                self._upload_references_in_value(item)

    async def _prepare_dispatch_value(self, value: Any, *, job_id: str) -> Any:
        if isinstance(value, Phrase):
            return await self._prepare_phrase_for_dispatch(value, job_id=job_id)

        if isinstance(value, Corpus):
            if value.source_url:
                return value
            if value.source_path and Path(value.source_path).exists():
                if self.archive is not None and hasattr(self.archive, "prepare_corpus_for_dispatch") and self._archive_can_upload():
                    prepared = await self.archive.prepare_corpus_for_dispatch(value, task_id=job_id)
                    if isinstance(prepared, Corpus):
                        return prepared
                warnings.warn(
                    (
                        "Distributed dispatch includes a local Corpus path but MinIO upload credentials are unavailable. "
                        "Workers may fail to access this local file."
                    ),
                    RuntimeWarning,
                    stacklevel=3,
                )
            return value

        if isinstance(value, Reference):
            if self.archive is not None and hasattr(self.archive, "create_task_reference") and self._archive_can_upload():
                if value.object_key:
                    return value
                path_name = Path(value.path).name
                return await self.archive.create_task_reference(
                    task_id=job_id,
                    content_type=value.content_type,
                    path_hint=path_name,
                )
            return value

        if isinstance(value, list):
            return [await self._prepare_dispatch_value(item, job_id=job_id) for item in value]
        if isinstance(value, tuple):
            return tuple([await self._prepare_dispatch_value(item, job_id=job_id) for item in value])
        if isinstance(value, dict):
            out: dict[Any, Any] = {}
            for key, item in value.items():
                out[key] = await self._prepare_dispatch_value(item, job_id=job_id)
            return out
        return value

    async def _prepare_phrase_for_dispatch(self, phrase: Phrase, *, job_id: str) -> Phrase:
        args = tuple([await self._prepare_dispatch_value(item, job_id=job_id) for item in phrase.args])
        kwargs: dict[str, Any] = {}
        for key, value in phrase.kwargs.items():
            kwargs[key] = await self._prepare_dispatch_value(value, job_id=job_id)

        cloned = Phrase(
            phrase.task_name,
            args=args,
            kwargs=kwargs,
            mock_reply=phrase._mock_reply,
            mock_delay=phrase._mock_delay,
        )
        cloned._is_local = phrase._is_local
        cloned._local_force = phrase._local_force

        for task_name, chain_args, chain_kwargs in phrase.chain:
            prepared_args = tuple([await self._prepare_dispatch_value(item, job_id=job_id) for item in chain_args])
            prepared_kwargs: dict[str, Any] = {}
            for key, value in chain_kwargs.items():
                prepared_kwargs[key] = await self._prepare_dispatch_value(value, job_id=job_id)
            cloned.chain.append((task_name, prepared_args, prepared_kwargs))

        return cloned

    def _prepare_result_for_transport(self, value: Any, *, job_id: str, task_name: str) -> Any:
        if isinstance(value, Corpus):
            if value.source_url:
                return value

            if value.source_path and Path(value.source_path).exists() and self.archive is not None and hasattr(self.archive, "prepare_corpus_for_dispatch") and self._archive_can_upload():
                prepared = self._run_archive_coro(self.archive.prepare_corpus_for_dispatch(value, task_id=job_id))
                if isinstance(prepared, Corpus):
                    return prepared

            if value.source_path and Path(value.source_path).exists():
                self._warn_unshared_corpus(task_name=task_name)
            return value

        if isinstance(value, Reference):
            self._upload_reference_if_available(value)
            return value

        if isinstance(value, list):
            return [self._prepare_result_for_transport(item, job_id=job_id, task_name=task_name) for item in value]
        if isinstance(value, tuple):
            return tuple([
                self._prepare_result_for_transport(item, job_id=job_id, task_name=task_name)
                for item in value
            ])
        if isinstance(value, dict):
            return {
                key: self._prepare_result_for_transport(item, job_id=job_id, task_name=task_name)
                for key, item in value.items()
            }
        return value

    def _resolve_compiled_policy(self, base_policy: RestartPolicy) -> RestartPolicy:
        return self._apply_policy_overrides(base_policy, self._compiled_policy_overrides)

    def _emit_event(
        self,
        job_id: str,
        *,
        event: str,
        task_name: Optional[str] = None,
        attempt: Optional[int] = None,
        message: Optional[str] = None,
        data: Optional[dict[str, Any]] = None,
    ) -> None:
        if not job_id:
            return
        if self.journal is None or not hasattr(self.journal, "append_event"):
            return
        self.journal.append_event(
            job_id,
            event=event,
            task_name=task_name,
            attempt=attempt,
            message=message,
            data=data,
        )

    def grammar(self, task_name: str, reply: Any = None) -> Callable:
        """
        Register the expected input/output signatures of a task.

        Usage:
            lang.grammar('normalize', reply=str)(str)
            lang.grammar('compare', reply=dict)(str, str)
        """
        def decorator(*input_types):
            self.grammar_registry[task_name] = {
                "reply": reply,
                "input_types": input_types,
            }
            return None

        return decorator

    def verb(self, task_name_or_func: Optional[Any] = None) -> Callable:
        """
        Decorator that defines the actual executable logic for a task.

        Usage:
            @lang.verb
            def normalize(text: str) -> str:
                ...

            @lang.verb('custom_name')
            def my_function(...):
                ...
        """
        def decorator(func: Callable) -> Callable:
            name = func.__name__
            if isinstance(task_name_or_func, str):
                name = task_name_or_func
            self.verb_registry[name] = func

            @wraps(func)
            def wrapper(*args, **kwargs):
                return func(*args, **kwargs)

            return wrapper

        if callable(task_name_or_func) and not isinstance(task_name_or_func, str):
            return decorator(task_name_or_func)
        return decorator

    def inferred_worker_queues(self) -> list[str]:
        """Infer worker queues from registered verbs with sensible fallback."""
        if self.routing_mode == "shared":
            queues = [self.channel]
            if self.private_queue not in queues:
                queues.append(self.private_queue)
            return queues
        if self.verb_registry:
            queues = sorted(self.verb_registry.keys())
            if self.private_queue not in queues:
                queues.append(self.private_queue)
            return queues
        if self.grammar_registry:
            queues = sorted(self.grammar_registry.keys())
            if self.private_queue not in queues:
                queues.append(self.private_queue)
            return queues
        queues = [self.channel]
        if self.private_queue not in queues:
            queues.append(self.private_queue)
        return queues

    def speak(self, block: bool = True) -> None:
        """Initialize the Celery language service."""
        self._started = True
        if self.journal is not None and hasattr(self.journal, "register_worker"):
            worker_id = self.name or self.node_id
            self.journal.register_worker(
                worker_id=worker_id,
                node_id=self.node_id,
                tasks=sorted(set(self.verb_registry.keys()) | set(self.grammar_registry.keys())),
                queues=self.inferred_worker_queues(),
                private_queue=self.private_queue,
            )

    def _job_status_from_celery(self, async_result: Any) -> JobStatusCode:
        state = (getattr(async_result, "state", "PENDING") or "PENDING").upper()
        if state in {"SUCCESS"}:
            return JobStatusCode.COMPLETED
        if state in {"FAILURE"}:
            return JobStatusCode.FAILED
        if state in {"REVOKED"}:
            return JobStatusCode.CANCELLED
        if state in {"STARTED", "RETRY", "RECEIVED"}:
            return JobStatusCode.RUNNING
        return JobStatusCode.PENDING

    def _build_envelope(
        self,
        job_id: str,
        phrase: Phrase,
        policy: Optional[RestartPolicy] = None,
        *,
        override_channel: Optional[str] = None,
        preferred_worker: Optional[str] = None,
        local_required: bool = False,
        payload_override: Optional[dict[str, Any]] = None,
    ) -> ChannelEnvelope:
        channel = override_channel or (self.channel if self.routing_mode == "shared" else phrase.task_name)
        return ChannelEnvelope(
            payload=payload_override or serialize_phrase(phrase),
            meta=ChannelMeta(
                job_id=job_id,
                root_task=phrase.task_name,
                channel=channel,
                source_node=self.node_id,
                preferred_worker=preferred_worker,
                local_required=local_required,
                policy=policy or self.restart_policy,
            ),
        )

    async def _resolve_local_worker(self, phrase: Phrase) -> Optional[dict[str, Any]]:
        if self.journal is None or not hasattr(self.journal, "find_worker_for_tasks"):
            return None
        worker = await self.journal.find_worker_for_tasks(phrase.task_names())
        if worker is None:
            return None
        return {
            "worker_id": worker.worker_id,
            "private_queue": worker.private_queue,
        }

    @staticmethod
    def _extract_phrase_values(values: list[Any]) -> list[Phrase]:
        out: list[Phrase] = []
        for value in values:
            if isinstance(value, Phrase):
                out.append(value)
        return out

    def _build_dag_payload(self, phrase: Phrase) -> dict[str, Any]:
        """Build a compact graph snapshot for a phrase DAG.

        Format:
        - nodes: [{"id": int, "task": str, "local": bool}]
        - edges: [[from_id, to_id], ...]
        """
        nodes: list[dict[str, Any]] = []
        edges: set[tuple[int, int]] = set()
        seen: dict[int, tuple[int, int]] = {}

        def add_node(task: str, local: bool) -> int:
            node_id = len(nodes)
            nodes.append({"id": node_id, "task": task, "local": local})
            return node_id

        def visit(node_phrase: Phrase) -> tuple[int, int]:
            identity = id(node_phrase)
            if identity in seen:
                return seen[identity]

            start_id = add_node(node_phrase.task_name, node_phrase._is_local)
            current_id = start_id

            # Resolve argument dependencies for the entry task first.
            arg_deps = self._extract_phrase_values(list(node_phrase.args) + list(node_phrase.kwargs.values()))
            for dep in arg_deps:
                _dep_start, dep_end = visit(dep)
                edges.add((dep_end, start_id))

            for chain_task, chain_args, chain_kwargs in node_phrase.chain:
                next_id = add_node(chain_task, node_phrase._is_local)
                edges.add((current_id, next_id))

                chain_deps = self._extract_phrase_values(list(chain_args) + list(chain_kwargs.values()))
                for dep in chain_deps:
                    _dep_start, dep_end = visit(dep)
                    edges.add((dep_end, next_id))

                current_id = next_id

            seen[identity] = (start_id, current_id)
            return start_id, current_id

        root_start, root_end = visit(phrase)
        indegree = {node["id"]: 0 for node in nodes}
        for src, dst in edges:
            indegree[dst] += 1

        return {
            "format": "compact-dag-v1",
            "root": root_start,
            "terminal": root_end,
            "roots": [node_id for node_id, degree in indegree.items() if degree == 0],
            "nodes": nodes,
            "edges": [[src, dst] for src, dst in sorted(edges)],
        }

    def _should_dispatch_distributed(self, phrase: Phrase) -> bool:
        if self.execution_mode == "local":
            return False
        if self.execution_mode == "distributed":
            return True
        return self.broker not in {"memory://", "inmemory://"}

    def _resolve_argument(self, arg: Any, *, job_id: Optional[str] = None, attempt: Optional[int] = None) -> Any:
        if isinstance(arg, Phrase):
            return self._execute_phrase(arg, job_id=job_id, attempt=attempt)
        return arg

    def _coerce_value(self, value: Any, expected_type: Any) -> Any:
        """Best-effort coercion from transport-friendly values to typed task inputs."""
        if expected_type in {None, Any, inspect._empty}:
            return value
        if value is None:
            return None

        origin = get_origin(expected_type)
        if origin is Union:
            # Optional[T] or Union[...] - pick first successful coercion.
            for option in get_args(expected_type):
                if option is type(None):
                    continue
                try:
                    return self._coerce_value(value, option)
                except Exception:
                    continue
            return value

        if origin is not None:
            expected_type = origin

        if isinstance(expected_type, type) and issubclass(expected_type, BaseModel):
            if isinstance(value, expected_type):
                return value
            if isinstance(value, dict):
                return expected_type.model_validate(value)
            return value

        if isinstance(expected_type, type):
            if isinstance(value, expected_type):
                return value
            try:
                return expected_type(value)
            except Exception:
                return value

        return value

    def _coerce_call_arguments(
        self,
        task_name: str,
        handler: Callable,
        args: tuple[Any, ...],
        kwargs: dict[str, Any],
    ) -> tuple[tuple[Any, ...], dict[str, Any]]:
        """Coerce args/kwargs using function annotations with grammar input fallback."""
        sig = inspect.signature(handler)
        bound = sig.bind_partial(*args, **kwargs)
        grammar_inputs = self.grammar_registry.get(task_name, {}).get("input_types") or ()
        resolved_hints = get_type_hints(handler)

        param_names = list(sig.parameters.keys())
        for idx, name in enumerate(param_names):
            if name not in bound.arguments:
                continue
            param = sig.parameters[name]
            expected = resolved_hints.get(name, param.annotation)
            if expected is inspect._empty and idx < len(grammar_inputs):
                expected = grammar_inputs[idx]
            bound.arguments[name] = self._coerce_value(bound.arguments[name], expected)

        return bound.args, bound.kwargs

    def _execute_task(
        self,
        task_name: str,
        args: tuple,
        kwargs: dict,
        phrase_obj: Optional[Phrase] = None,
        *,
        job_id: Optional[str] = None,
        attempt: Optional[int] = None,
    ) -> Any:
        if phrase_obj is not None and phrase_obj._mock_delay > 0:
            time.sleep(phrase_obj._mock_delay)
        if phrase_obj is not None and phrase_obj._mock_reply is not None:
            return phrase_obj._mock_reply

        handler = self.verb_registry.get(task_name)
        if handler is None:
            raise KeyError(f"Task '{task_name}' is not registered by any verb.")

        self._emit_event(job_id or "", event="subjob.started", task_name=task_name, attempt=attempt)
        try:
            resolved_args = tuple(self._resolve_argument(a, job_id=job_id, attempt=attempt) for a in args)
            resolved_kwargs = {
                k: self._resolve_argument(v, job_id=job_id, attempt=attempt) for k, v in kwargs.items()
            }
            coerced_args, coerced_kwargs = self._coerce_call_arguments(
                task_name,
                handler,
                resolved_args,
                resolved_kwargs,
            )
            out = handler(*coerced_args, **coerced_kwargs)
            self._upload_references_in_value(coerced_args)
            self._upload_references_in_value(coerced_kwargs)
            self._emit_event(job_id or "", event="subjob.completed", task_name=task_name, attempt=attempt)
            return out
        except Exception as exc:
            self._emit_event(
                job_id or "",
                event="subjob.failed",
                task_name=task_name,
                attempt=attempt,
                message=str(exc),
                data={"error": type(exc).__name__},
            )
            raise

    def _execute_phrase(self, phrase: Phrase, *, job_id: Optional[str] = None, attempt: Optional[int] = None) -> Any:
        current = self._execute_task(
            phrase.task_name,
            phrase.args,
            phrase.kwargs,
            phrase,
            job_id=job_id,
            attempt=attempt,
        )
        for task_name, args, kwargs in phrase.chain:
            resolved_args = tuple(self._resolve_argument(a, job_id=job_id, attempt=attempt) for a in args)
            current = self._execute_task(
                task_name,
                (current, *resolved_args),
                kwargs,
                job_id=job_id,
                attempt=attempt,
            )
        return current

    def _can_execute(self, phrase: Phrase) -> bool:
        def visit(node: Phrase) -> bool:
            # Mocked phrase nodes (e.g., echo()) are intentionally executable
            # without a registered verb.
            if node._mock_reply is None and node.task_name not in self.verb_registry:
                return False

            arg_deps = self._extract_phrase_values(list(node.args) + list(node.kwargs.values()))
            for dep in arg_deps:
                if not visit(dep):
                    return False

            for chain_task, chain_args, chain_kwargs in node.chain:
                if chain_task not in self.verb_registry:
                    return False
                chain_deps = self._extract_phrase_values(list(chain_args) + list(chain_kwargs.values()))
                for dep in chain_deps:
                    if not visit(dep):
                        return False
            return True

        return visit(phrase)

    def _compile_phrase_workflow(self, phrase: Phrase) -> dict[str, Any]:
        """Compile a phrase DAG into a workflow plan representation.

        This keeps orchestration optimization modular and isolated from transport,
        metadata, and storage concerns. The plan can later be lowered to Celery
        canvas primitives (chain/group/chord) without touching the rest of say().
        """
        dag = self._build_dag_payload(phrase)
        nodes = dag.get("nodes", [])
        edges = [tuple(edge) for edge in dag.get("edges", [])]

        node_by_id = {item["id"]: item for item in nodes}
        indegree = {item["id"]: 0 for item in nodes}
        outgoing: dict[int, list[int]] = defaultdict(list)
        for src, dst in edges:
            outgoing[src].append(dst)
            indegree[dst] = indegree.get(dst, 0) + 1

        q = deque(sorted([node_id for node_id, degree in indegree.items() if degree == 0]))
        topo: list[int] = []
        while q:
            node_id = q.popleft()
            topo.append(node_id)
            for nxt in outgoing.get(node_id, []):
                indegree[nxt] -= 1
                if indegree[nxt] == 0:
                    q.append(nxt)

        # Layered Kahn traversal to expose parallelizable steps.
        indegree2 = {item["id"]: 0 for item in nodes}
        for _src, dst in edges:
            indegree2[dst] = indegree2.get(dst, 0) + 1
        frontier = sorted([node_id for node_id, degree in indegree2.items() if degree == 0])
        layers: list[list[int]] = []
        while frontier:
            layers.append(frontier)
            next_frontier: list[int] = []
            for node_id in frontier:
                for nxt in outgoing.get(node_id, []):
                    indegree2[nxt] -= 1
                    if indegree2[nxt] == 0:
                        next_frontier.append(nxt)
            frontier = sorted(next_frontier)

        execution_order = [node_by_id[node_id]["task"] for node_id in topo if node_id in node_by_id]
        layer_tasks = [
            [node_by_id[node_id]["task"] for node_id in layer if node_id in node_by_id]
            for layer in layers
        ]

        return {
            "engine": "celery-canvas",
            "mode": self.orchestration_mode,
            "is_linear": all(len(layer) == 1 for layer in layer_tasks),
            "tasks": execution_order,
            "node_count": len(nodes),
            "edge_count": len(edges),
            "layers": layer_tasks,
            "max_parallelism": max((len(layer) for layer in layer_tasks), default=1),
        }

    @staticmethod
    def _has_phrase_dependencies(values: list[Any]) -> bool:
        return any(isinstance(value, Phrase) for value in values)

    def _is_linear_canvas_compatible(self, phrase: Phrase) -> bool:
        if self._has_phrase_dependencies(list(phrase.args) + list(phrase.kwargs.values())):
            return False
        for _task_name, args, kwargs in phrase.chain:
            if self._has_phrase_dependencies(list(args) + list(kwargs.values())):
                return False
        return True

    @staticmethod
    def _ordered_phrase_dependencies(values: list[Any]) -> list[Phrase]:
        out: list[Phrase] = []
        seen: set[int] = set()
        for value in values:
            if isinstance(value, Phrase):
                identity = id(value)
                if identity not in seen:
                    out.append(value)
                    seen.add(identity)
        return out

    def _extract_root_phrase_dependencies(self, phrase: Phrase) -> list[Phrase]:
        return self._ordered_phrase_dependencies(list(phrase.args) + list(phrase.kwargs.values()))

    def _is_fanin_canvas_compatible(self, phrase: Phrase) -> bool:
        if phrase.chain:
            return False
        deps = self._extract_root_phrase_dependencies(phrase)
        if not deps:
            return False
        return all(self._is_linear_canvas_compatible(dep) for dep in deps)

    def _build_linear_compiled_nodes(
        self,
        phrase: Phrase,
        job_id: str,
        *,
        policy: Optional[RestartPolicy] = None,
        node_prefix: str = "n",
    ) -> list[dict[str, Any]]:
        policy_payload = (policy or self.compiled_restart_policy).model_dump(mode="python")
        nodes: list[dict[str, Any]] = [
            {
                "node_id": f"{node_prefix}0",
                "job_id": job_id,
                "task": phrase.task_name,
                "inject_previous": False,
                "args": [serialize_value(item) for item in phrase.args],
                "kwargs": {key: serialize_value(value) for key, value in phrase.kwargs.items()},
                "policy": policy_payload,
            }
        ]
        for index, (task_name, args, kwargs) in enumerate(phrase.chain, start=1):
            nodes.append(
                {
                    "node_id": f"{node_prefix}{index}",
                    "job_id": job_id,
                    "task": task_name,
                    "inject_previous": True,
                    "args": [serialize_value(item) for item in args],
                    "kwargs": {key: serialize_value(value) for key, value in kwargs.items()},
                    "policy": policy_payload,
                }
            )
        return nodes

    @staticmethod
    def _extract_canvas_task_ids(async_result: Any) -> list[str]:
        ids: list[str] = []
        current = async_result
        while current is not None:
            task_id = getattr(current, "id", None)
            if task_id:
                ids.append(task_id)
            current = getattr(current, "parent", None)
        return list(reversed(ids))

    def _build_compiled_linear_canvas(
        self,
        nodes: list[dict[str, Any]],
        *,
        queue: str,
        task_id: str,
    ) -> Any:
        signatures: list[Any] = [self._compiled_seed_task.s()]
        signatures.extend(self._compiled_node_task.s(node) for node in nodes)
        for signature in signatures:
            signature.set(queue=queue)
        return celery_chain(*signatures).apply_async(task_id=task_id)

    def _build_compiled_fanin_canvas(
        self,
        branch_chains: list[Any],
        join_signature: Any,
    ) -> Any:
        return celery_chord(branch_chains)(join_signature)

    @staticmethod
    def _build_join_slot(value: Any, dep_index_by_identity: dict[int, int]) -> dict[str, Any]:
        if isinstance(value, Phrase):
            return {"kind": "dep", "index": dep_index_by_identity[id(value)]}
        return {"kind": "literal", "value": serialize_value(value)}

    def _build_fanin_join_payload(
        self,
        phrase: Phrase,
        dep_index_by_identity: dict[int, int],
        *,
        job_id: str,
        policy: Optional[RestartPolicy] = None,
    ) -> dict[str, Any]:
        return {
            "job_id": job_id,
            "task": phrase.task_name,
            "args": [self._build_join_slot(item, dep_index_by_identity) for item in phrase.args],
            "kwargs": {
                key: self._build_join_slot(value, dep_index_by_identity)
                for key, value in phrase.kwargs.items()
            },
            "policy": (policy or self.compiled_restart_policy).model_dump(mode="python"),
        }

    async def _dispatch_compiled_linear_remote(self, ctx: dict[str, Any], workflow: dict[str, Any]) -> Job:
        phrase: Phrase = ctx.get("transport_phrase", ctx["phrase"])
        job: Job = ctx["job"]
        job_id: str = ctx["job_id"]
        local_assignment = ctx["local_assignment"]
        compiled_policy: RestartPolicy = ctx["compiled_policy_obj"]

        if self.routing_mode == "task":
            default_queue = phrase.task_name
        else:
            default_queue = self.channel
        queue = local_assignment["private_queue"] if local_assignment else default_queue

        nodes = self._build_linear_compiled_nodes(phrase, job_id, policy=compiled_policy)
        self._debug_log(
            "task.push",
            queue=queue,
            job_id=job_id,
            task=phrase.task_name,
            node_count=len(nodes),
            compiled=True,
            preferred_worker=(local_assignment or {}).get("worker_id"),
        )
        async_result = self._build_compiled_linear_canvas(nodes, queue=queue, task_id=job_id)

        all_task_ids = self._extract_canvas_task_ids(async_result)
        node_task_ids = all_task_ids[-len(nodes):] if len(all_task_ids) >= len(nodes) else []
        node_task_map: dict[str, str] = {}
        compiled_node_specs: dict[str, dict[str, Any]] = {}
        for index, node in enumerate(nodes):
            if index < len(node_task_ids):
                node_task_map[node["node_id"]] = node_task_ids[index]
            compiled_node_specs[node["node_id"]] = {
                "kind": "node",
                "queue": queue,
                "payload": node,
            }

        if self.journal is not None and hasattr(self.journal, "upsert_job"):
            self.journal.upsert_job(
                job_id,
                status=JobStatusCode.PENDING,
                node_task_map=node_task_map,
                compiled_node_specs=compiled_node_specs,
            )
        if self.journal is not None and hasattr(self.journal, "register_providers"):
            self.journal.register_providers(
                job_id,
                status_provider=lambda: self._job_status_from_celery(async_result),
                result_provider=lambda: deserialize_value(getattr(async_result, "result", None)),
            )

        self._emit_event(
            job_id,
            event="workflow.lowered",
            task_name=phrase.task_name,
            attempt=1,
            message="Lowered linear workflow to Celery chain",
            data={
                "mode": "compiled",
                "queue": queue,
                "node_count": len(nodes),
                "max_parallelism": workflow.get("max_parallelism", 1),
            },
        )
        self._emit_event(
            job_id,
            event="job.queued_remote",
            task_name=phrase.task_name,
            attempt=1,
            message=f"Queued compiled workflow on '{queue}'",
            data={"local": phrase._is_local, "compiled": True},
        )

        job.id = job_id
        job.code = JobStatusCode.PENDING
        return job

    async def _dispatch_compiled_fanin_remote(self, ctx: dict[str, Any], workflow: dict[str, Any]) -> Job:
        phrase: Phrase = ctx.get("transport_phrase", ctx["phrase"])
        job: Job = ctx["job"]
        job_id: str = ctx["job_id"]
        local_assignment = ctx["local_assignment"]
        compiled_policy: RestartPolicy = ctx["compiled_policy_obj"]

        if self.routing_mode == "task":
            default_queue = phrase.task_name
        else:
            default_queue = self.channel
        queue = local_assignment["private_queue"] if local_assignment else default_queue

        dependencies = self._extract_root_phrase_dependencies(phrase)
        self._debug_log(
            "task.push",
            queue=queue,
            job_id=job_id,
            task=phrase.task_name,
            branch_count=len(dependencies),
            compiled=True,
            preferred_worker=(local_assignment or {}).get("worker_id"),
        )
        dep_index_by_identity = {id(dep): idx for idx, dep in enumerate(dependencies)}
        branch_chains: list[Any] = []
        node_task_map: dict[str, str] = {}
        compiled_node_specs: dict[str, dict[str, Any]] = {}

        for dep_index, dep_phrase in enumerate(dependencies):
            node_prefix = f"d{dep_index}.n"
            dep_nodes = self._build_linear_compiled_nodes(
                dep_phrase,
                job_id,
                policy=compiled_policy,
                node_prefix=node_prefix,
            )
            dep_signatures: list[Any] = [self._compiled_seed_task.s().set(queue=queue)]
            for node in dep_nodes:
                dep_task_id = str(uuid.uuid4())
                node_task_map[node["node_id"]] = dep_task_id
                compiled_node_specs[node["node_id"]] = {
                    "kind": "node",
                    "queue": queue,
                    "payload": node,
                }
                dep_signatures.append(
                    self._compiled_node_task.s(node).set(queue=queue, task_id=dep_task_id)
                )
            branch_chains.append(celery_chain(*dep_signatures))

        join_task_id = str(uuid.uuid4())
        node_task_map["join"] = join_task_id
        join_payload = self._build_fanin_join_payload(
            phrase,
            dep_index_by_identity,
            job_id=job_id,
            policy=compiled_policy,
        )
        compiled_node_specs["join"] = {
            "kind": "join",
            "queue": queue,
            "payload": join_payload,
        }
        join_signature = self._compiled_join_task.s(join_payload).set(queue=queue, task_id=join_task_id)

        async_result = self._build_compiled_fanin_canvas(branch_chains, join_signature)

        if self.journal is not None and hasattr(self.journal, "upsert_job"):
            self.journal.upsert_job(
                job_id,
                status=JobStatusCode.PENDING,
                node_task_map=node_task_map,
                compiled_node_specs=compiled_node_specs,
            )
        if self.journal is not None and hasattr(self.journal, "register_providers"):
            self.journal.register_providers(
                job_id,
                status_provider=lambda: self._job_status_from_celery(async_result),
                result_provider=lambda: deserialize_value(getattr(async_result, "result", None)),
            )

        self._emit_event(
            job_id,
            event="workflow.lowered",
            task_name=phrase.task_name,
            attempt=1,
            message="Lowered fan-in workflow to Celery chord",
            data={
                "mode": "compiled",
                "queue": queue,
                "node_count": workflow.get("node_count", len(node_task_map)),
                "max_parallelism": workflow.get("max_parallelism", 1),
            },
        )
        self._emit_event(
            job_id,
            event="job.queued_remote",
            task_name=phrase.task_name,
            attempt=1,
            message=f"Queued compiled fan-in workflow on '{queue}'",
            data={"local": phrase._is_local, "compiled": True},
        )

        job.id = job_id
        job.code = JobStatusCode.PENDING
        return job

    async def _prepare_dispatch_context(
        self,
        phrase: Phrase,
        info: Optional[BaseModel],
        policy: Optional[RestartPolicy],
    ) -> dict[str, Any]:
        if self.journal is not None and getattr(self.journal, "_current_id", None):
            job_id = self.journal._current_id
        else:
            job_id = str(uuid.uuid4())

        job = Job(job_id=job_id, info=info)
        policy_obj = self._resolve_policy(policy)
        compiled_policy_obj = self._resolve_compiled_policy(policy_obj)

        dag_payload = None
        if self.journal is not None and bool(getattr(self.journal, "store_full_dag", False)):
            dag_payload = self._build_dag_payload(phrase)

        local_assignment = None
        if phrase._is_local and self._should_dispatch_distributed(phrase):
            local_assignment = await self._resolve_local_worker(phrase)
            if local_assignment is None and phrase._local_force is False:
                raise RuntimeError(
                    "No registered worker can execute all tasks required by local() phrase"
                )

        transport_phrase = phrase
        if self._should_dispatch_distributed(phrase):
            transport_phrase = await self._prepare_phrase_for_dispatch(phrase, job_id=job_id)

        if self.journal is not None and hasattr(self.journal, "upsert_job"):
            self.journal.upsert_job(
                job_id,
                status=JobStatusCode.PENDING,
                info=info,
                phrase_payload=serialize_phrase(transport_phrase),
                dag_payload=dag_payload,
                policy_payload=policy_obj.model_dump(mode="python"),
                details=JobStatusDetails(message="Dispatched", progress=0.0),
            )
        self._emit_event(
            job_id,
            event="job.dispatched",
            task_name=phrase.task_name,
            attempt=1,
            message="Job accepted by language runtime",
        )

        return {
            "phrase": phrase,
            "job": job,
            "job_id": job_id,
            "info": info,
            "policy_obj": policy_obj,
            "compiled_policy_obj": compiled_policy_obj,
            "local_assignment": local_assignment,
            "transport_phrase": transport_phrase,
        }

    async def _dispatch_simple(self, ctx: dict[str, Any]) -> Job:
        phrase: Phrase = ctx["phrase"]
        transport_phrase: Phrase = ctx.get("transport_phrase", phrase)
        job: Job = ctx["job"]
        job_id: str = ctx["job_id"]
        info = ctx["info"]
        policy_obj: RestartPolicy = ctx["policy_obj"]
        local_assignment = ctx["local_assignment"]

        if self._should_dispatch_distributed(phrase):
            local_required = bool(phrase._is_local and phrase._local_force is False)
            preferred_worker = local_assignment["worker_id"] if local_assignment else None
            override_queue = local_assignment["private_queue"] if local_assignment else None
            payload = serialize_phrase(transport_phrase, prefer_local_paths=bool(phrase._is_local))
            envelope = self._build_envelope(
                job_id,
                transport_phrase,
                policy=policy_obj,
                override_channel=override_queue,
                preferred_worker=preferred_worker,
                local_required=local_required,
                payload_override=payload,
            )
            try:
                self._debug_log(
                    "task.push",
                    queue=envelope.meta.channel,
                    job_id=envelope.meta.job_id,
                    task=envelope.meta.root_task,
                    source_node=envelope.meta.source_node,
                    preferred_worker=envelope.meta.preferred_worker,
                    local_required=envelope.meta.local_required,
                    retry_max=envelope.meta.policy.max_retries,
                )
                self.celery_app.send_task(
                    self._remote_task_name,
                    args=[envelope.model_dump(mode="python")],
                    queue=envelope.meta.channel,
                    task_id=job_id,
                )
                job.id = job_id
                job.code = JobStatusCode.PENDING

                if self.journal is not None and hasattr(self.journal, "register_providers"):
                    self.journal.register_providers(
                        job_id,
                        status_provider=lambda: self._job_status_from_celery(self.celery_app.AsyncResult(job_id)),
                        result_provider=lambda: deserialize_value(self.celery_app.AsyncResult(job_id).result),
                    )
                self._emit_event(
                    job_id,
                    event="job.queued_remote",
                    task_name=envelope.meta.root_task,
                    attempt=1,
                    message=f"Queued on channel '{envelope.meta.channel}'",
                    data={"preferred_worker": preferred_worker, "local": phrase._is_local},
                )
                return job
            except Exception as exc:
                if self.journal is not None and hasattr(self.journal, "upsert_job"):
                    self.journal.upsert_job(
                        job_id,
                        status=JobStatusCode.RUNNING,
                        details=JobStatusDetails(
                            message=f"Broker unavailable, executing locally: {exc}",
                            error=type(exc).__name__,
                            progress=0.2,
                        ),
                    )
                self._emit_event(
                    job_id,
                    event="job.local_fallback",
                    task_name=phrase.task_name,
                    attempt=1,
                    message=str(exc),
                    data={"error": type(exc).__name__},
                )

        if not self._can_execute(phrase):
            job.code = JobStatusCode.PENDING
            return job

        if self.journal is not None and hasattr(self.journal, "upsert_job"):
            self.journal.upsert_job(job_id, status=JobStatusCode.RUNNING, info=info, bump_attempt=True)
        self._emit_event(
            job_id,
            event="job.running",
            task_name=phrase.task_name,
            attempt=1,
            message="Executing locally",
        )

        try:
            result = self._execute_phrase(phrase, job_id=job_id, attempt=1)
            job.set_result(result)
            if self.journal is not None and hasattr(self.journal, "upsert_job"):
                self.journal.upsert_job(
                    job_id,
                    status=JobStatusCode.COMPLETED,
                    info=info,
                    result=result,
                    details=JobStatusDetails(message="Completed locally", progress=1.0),
                )
            self._emit_event(
                job_id,
                event="job.completed",
                task_name=phrase.task_name,
                attempt=1,
                message="Local execution completed",
            )
        except Exception as exc:
            details = traceback.format_exc()
            failure_message = details or str(exc)
            job.set_failed(failure_message)
            if self.journal is not None and hasattr(self.journal, "upsert_job"):
                self.journal.upsert_job(
                    job_id,
                    status=JobStatusCode.FAILED,
                    info=info,
                    details=JobStatusDetails(
                        message=failure_message,
                        error=type(exc).__name__,
                        progress=1.0,
                    ),
                )
            self._emit_event(
                job_id,
                event="job.failed",
                task_name=phrase.task_name,
                attempt=1,
                message=failure_message,
                data={"error": type(exc).__name__},
            )

        return job

    async def _dispatch_compiled(self, ctx: dict[str, Any]) -> Job:
        """Dispatch through compiled workflow planner.

        Current implementation compiles a modular plan and routes through the
        simple dispatcher. This keeps an explicit extension point for future
        Celery canvas lowering (chain/chord/group) without reworking call sites.
        """
        phrase: Phrase = ctx["phrase"]
        workflow = self._compile_phrase_workflow(phrase)
        self._emit_event(
            ctx["job_id"],
            event="workflow.compiled",
            task_name=phrase.task_name,
            attempt=1,
            message="Compiled workflow plan",
            data=workflow,
        )
        phrase: Phrase = ctx["phrase"]
        if self._should_dispatch_distributed(phrase):
            if workflow.get("is_linear") and self._is_linear_canvas_compatible(phrase):
                try:
                    return await self._dispatch_compiled_linear_remote(ctx, workflow)
                except Exception as exc:
                    if self.journal is not None and hasattr(self.journal, "upsert_job"):
                        self.journal.upsert_job(
                            ctx["job_id"],
                            status=JobStatusCode.RUNNING,
                            details=JobStatusDetails(
                                message=f"Compiled lowering unavailable, executing locally: {exc}",
                                error=type(exc).__name__,
                                progress=0.2,
                            ),
                        )
                    self._emit_event(
                        ctx["job_id"],
                        event="workflow.lowering_failed",
                        task_name=phrase.task_name,
                        attempt=1,
                        message=str(exc),
                        data={"error": type(exc).__name__},
                    )
            elif self._is_fanin_canvas_compatible(phrase):
                try:
                    return await self._dispatch_compiled_fanin_remote(ctx, workflow)
                except Exception as exc:
                    if self.journal is not None and hasattr(self.journal, "upsert_job"):
                        self.journal.upsert_job(
                            ctx["job_id"],
                            status=JobStatusCode.RUNNING,
                            details=JobStatusDetails(
                                message=f"Compiled fan-in lowering unavailable, executing locally: {exc}",
                                error=type(exc).__name__,
                                progress=0.2,
                            ),
                        )
                    self._emit_event(
                        ctx["job_id"],
                        event="workflow.lowering_failed",
                        task_name=phrase.task_name,
                        attempt=1,
                        message=str(exc),
                        data={"error": type(exc).__name__, "shape": "fanin"},
                    )
            else:
                reason = "non_linear" if not workflow.get("is_linear") else "unsupported_dependencies"
                self._emit_event(
                    ctx["job_id"],
                    event="workflow.lowering_skipped",
                    task_name=phrase.task_name,
                    attempt=1,
                    message="Compiled lowering skipped",
                    data={"reason": reason},
                )
        return await self._dispatch_simple(ctx)

    async def say(
        self,
        phrase: Phrase,
        info: Optional[BaseModel] = None,
        policy: Optional[RestartPolicy] = None,
    ) -> Job:
        """
        Dispatch a phrase (DAG) to the message broker for execution.

        Returns a Job object with the job ID and status.
        """
        ctx = await self._prepare_dispatch_context(phrase, info, policy)
        if self.orchestration_mode == "compiled":
            return await self._dispatch_compiled(ctx)
        return await self._dispatch_simple(ctx)

    async def resubmit(self, job_record: Any, policy: Optional[Any] = None) -> Job:
        """Resubmit a job from a stored record using persisted phrase payload."""
        phrase_payload = getattr(job_record, "phrase_payload", None)
        if not phrase_payload:
            raise ValueError("Job record has no stored phrase payload for resubmission")

        phrase_obj = deserialize_phrase(phrase_payload)
        info = getattr(job_record, "info", None)

        retry_policy = None
        if policy is not None:
            retry_policy = policy if isinstance(policy, RestartPolicy) else RestartPolicy(**policy)
        else:
            raw = getattr(job_record, "policy_payload", None)
            if raw:
                retry_policy = RestartPolicy(**raw)

        return await self.say(phrase_obj, info=info, policy=retry_policy)

    async def requeue_compiled_node(
        self,
        job_record: Any,
        node_id: str,
        *,
        previous_result: Any = None,
        dependency_results: Optional[list[Any]] = None,
    ) -> Job:
        """Requeue a specific compiled workflow node by node id.

        For regular node tasks, provide previous_result when the node requires
        upstream chain output (inject_previous=True).
        For join nodes, provide dependency_results aligned to dependency order.
        """
        job_id = getattr(job_record, "id", None)
        if not job_id:
            raise ValueError("Job record does not include an id")

        specs = getattr(job_record, "compiled_node_specs", None) or {}
        if not specs:
            raise ValueError("Job record has no compiled node specs")

        spec = specs.get(node_id)
        if spec is None:
            raise ValueError(f"Unknown compiled node id: {node_id}")

        kind = str(spec.get("kind") or "")
        payload = spec.get("payload")
        queue = str(spec.get("queue") or self.channel)
        task_id = str(uuid.uuid4())

        if kind == "node":
            if not isinstance(payload, dict):
                raise ValueError("Compiled node payload is invalid")
            inject_previous = bool(payload.get("inject_previous", False))
            if inject_previous and previous_result is None:
                raise ValueError("previous_result is required when requeuing dependent node")
            args = [serialize_value(previous_result), payload]
            task_name = self._compiled_node_task_name
            display_name = str(payload.get("task") or node_id)
        elif kind == "join":
            if dependency_results is None:
                raise ValueError("dependency_results are required when requeuing join node")
            if not isinstance(payload, dict):
                raise ValueError("Compiled join payload is invalid")
            args = [[serialize_value(item) for item in dependency_results], payload]
            task_name = self._compiled_join_task_name
            display_name = str(payload.get("task") or "join")
        else:
            raise ValueError(f"Unsupported compiled node kind: {kind}")

        self.celery_app.send_task(
            task_name,
            args=args,
            queue=queue,
            task_id=task_id,
        )
        self._debug_log(
            "task.push",
            queue=queue,
            job_id=job_id,
            task=display_name,
            node_id=node_id,
            task_id=task_id,
            kind=kind,
            requeue=True,
            compiled=True,
        )

        current_map = dict(getattr(job_record, "node_task_map", None) or {})
        current_map[node_id] = task_id

        if self.journal is not None and hasattr(self.journal, "upsert_job"):
            self.journal.upsert_job(
                job_id,
                status=JobStatusCode.RUNNING,
                node_task_map=current_map,
                compiled_node_specs=specs,
                bump_attempt=True,
                details=JobStatusDetails(
                    message=f"Requeued compiled node {node_id}",
                    progress=None,
                ),
            )

        self._emit_event(
            job_id,
            event="subjob.retry",
            task_name=display_name,
            attempt=1,
            message="Node requeue scheduled",
            data={"node_id": node_id, "task_id": task_id},
        )
        self._emit_event(
            job_id,
            event="workflow.node_requeued",
            task_name=display_name,
            attempt=1,
            message="Compiled node requeued",
            data={"node_id": node_id, "task_id": task_id, "kind": kind},
        )

        node_job = Job(job_id=task_id, info={"parent_job_id": job_id, "node_id": node_id})
        node_job.code = JobStatusCode.PENDING
        return node_job

    async def attempt_cancel(self, job: Job) -> bool:
        """Attempt to cancel a running job."""
        job.set_cancelled()
        if self.journal is not None and hasattr(self.journal, "upsert_job"):
            self.journal.upsert_job(job.id, status=JobStatusCode.CANCELLED, info=job.info)
        self._emit_event(job.id, event="job.cancelled", message="Cancellation requested")
        return True

    def conversation(self, func: Callable) -> Callable:
        """
        Decorator that provides a Moderator to orchestrate test runs.

        Usage:
            @lang.conversation
            def test_something(mod: Moderator):
                ...
        """
        from lingo.moderator import Moderator
        import inspect
        from functools import wraps

        @wraps(func)
        def wrapper(*args, **kwargs):
            moderator = Moderator(self)
            return func(moderator, *args, **kwargs)

        # Preserve pytest fixture injection for parameters after the moderator argument.
        sig = inspect.signature(func)
        params = list(sig.parameters.values())
        if params:
            wrapper.__signature__ = sig.replace(parameters=params[1:])
        wrapper.__lingo_conversation__ = True
        wrapper.__lingo_language_name__ = self.name or self.node_id
        wrapper.__lingo_execution_mode__ = self.execution_mode

        return wrapper

attempt_cancel(job: Job) -> bool async

Attempt to cancel a running job.

Source code in lingo/celery/language.py

async def attempt_cancel(self, job: Job) -> bool:
    """Attempt to cancel a running job."""
    job.set_cancelled()
    if self.journal is not None and hasattr(self.journal, "upsert_job"):
        self.journal.upsert_job(job.id, status=JobStatusCode.CANCELLED, info=job.info)
    self._emit_event(job.id, event="job.cancelled", message="Cancellation requested")
    return True

conversation(func: Callable) -> Callable

Decorator that provides a Moderator to orchestrate test runs.

Usage

@lang.conversation def test_something(mod: Moderator): ...

Source code in lingo/celery/language.py

def conversation(self, func: Callable) -> Callable:
    """
    Decorator that provides a Moderator to orchestrate test runs.

    Usage:
        @lang.conversation
        def test_something(mod: Moderator):
            ...
    """
    from lingo.moderator import Moderator
    import inspect
    from functools import wraps

    @wraps(func)
    def wrapper(*args, **kwargs):
        moderator = Moderator(self)
        return func(moderator, *args, **kwargs)

    # Preserve pytest fixture injection for parameters after the moderator argument.
    sig = inspect.signature(func)
    params = list(sig.parameters.values())
    if params:
        wrapper.__signature__ = sig.replace(parameters=params[1:])
    wrapper.__lingo_conversation__ = True
    wrapper.__lingo_language_name__ = self.name or self.node_id
    wrapper.__lingo_execution_mode__ = self.execution_mode

    return wrapper

grammar(task_name: str, reply: Any = None) -> Callable

Register the expected input/output signatures of a task.

Usage

lang.grammar('normalize', reply=str)(str) lang.grammar('compare', reply=dict)(str, str)

Source code in lingo/celery/language.py

def grammar(self, task_name: str, reply: Any = None) -> Callable:
    """
    Register the expected input/output signatures of a task.

    Usage:
        lang.grammar('normalize', reply=str)(str)
        lang.grammar('compare', reply=dict)(str, str)
    """
    def decorator(*input_types):
        self.grammar_registry[task_name] = {
            "reply": reply,
            "input_types": input_types,
        }
        return None

    return decorator

inferred_worker_queues() -> list[str]

Infer worker queues from registered verbs with sensible fallback.

Source code in lingo/celery/language.py

def inferred_worker_queues(self) -> list[str]:
    """Infer worker queues from registered verbs with sensible fallback."""
    if self.routing_mode == "shared":
        queues = [self.channel]
        if self.private_queue not in queues:
            queues.append(self.private_queue)
        return queues
    if self.verb_registry:
        queues = sorted(self.verb_registry.keys())
        if self.private_queue not in queues:
            queues.append(self.private_queue)
        return queues
    if self.grammar_registry:
        queues = sorted(self.grammar_registry.keys())
        if self.private_queue not in queues:
            queues.append(self.private_queue)
        return queues
    queues = [self.channel]
    if self.private_queue not in queues:
        queues.append(self.private_queue)
    return queues

requeue_compiled_node(job_record: Any, node_id: str, *, previous_result: Any = None, dependency_results: Optional[list[Any]] = None) -> Job async

Requeue a specific compiled workflow node by node id.

For regular node tasks, provide previous_result when the node requires upstream chain output (inject_previous=True). For join nodes, provide dependency_results aligned to dependency order.

Source code in lingo/celery/language.py

async def requeue_compiled_node(
    self,
    job_record: Any,
    node_id: str,
    *,
    previous_result: Any = None,
    dependency_results: Optional[list[Any]] = None,
) -> Job:
    """Requeue a specific compiled workflow node by node id.

    For regular node tasks, provide previous_result when the node requires
    upstream chain output (inject_previous=True).
    For join nodes, provide dependency_results aligned to dependency order.
    """
    job_id = getattr(job_record, "id", None)
    if not job_id:
        raise ValueError("Job record does not include an id")

    specs = getattr(job_record, "compiled_node_specs", None) or {}
    if not specs:
        raise ValueError("Job record has no compiled node specs")

    spec = specs.get(node_id)
    if spec is None:
        raise ValueError(f"Unknown compiled node id: {node_id}")

    kind = str(spec.get("kind") or "")
    payload = spec.get("payload")
    queue = str(spec.get("queue") or self.channel)
    task_id = str(uuid.uuid4())

    if kind == "node":
        if not isinstance(payload, dict):
            raise ValueError("Compiled node payload is invalid")
        inject_previous = bool(payload.get("inject_previous", False))
        if inject_previous and previous_result is None:
            raise ValueError("previous_result is required when requeuing dependent node")
        args = [serialize_value(previous_result), payload]
        task_name = self._compiled_node_task_name
        display_name = str(payload.get("task") or node_id)
    elif kind == "join":
        if dependency_results is None:
            raise ValueError("dependency_results are required when requeuing join node")
        if not isinstance(payload, dict):
            raise ValueError("Compiled join payload is invalid")
        args = [[serialize_value(item) for item in dependency_results], payload]
        task_name = self._compiled_join_task_name
        display_name = str(payload.get("task") or "join")
    else:
        raise ValueError(f"Unsupported compiled node kind: {kind}")

    self.celery_app.send_task(
        task_name,
        args=args,
        queue=queue,
        task_id=task_id,
    )
    self._debug_log(
        "task.push",
        queue=queue,
        job_id=job_id,
        task=display_name,
        node_id=node_id,
        task_id=task_id,
        kind=kind,
        requeue=True,
        compiled=True,
    )

    current_map = dict(getattr(job_record, "node_task_map", None) or {})
    current_map[node_id] = task_id

    if self.journal is not None and hasattr(self.journal, "upsert_job"):
        self.journal.upsert_job(
            job_id,
            status=JobStatusCode.RUNNING,
            node_task_map=current_map,
            compiled_node_specs=specs,
            bump_attempt=True,
            details=JobStatusDetails(
                message=f"Requeued compiled node {node_id}",
                progress=None,
            ),
        )

    self._emit_event(
        job_id,
        event="subjob.retry",
        task_name=display_name,
        attempt=1,
        message="Node requeue scheduled",
        data={"node_id": node_id, "task_id": task_id},
    )
    self._emit_event(
        job_id,
        event="workflow.node_requeued",
        task_name=display_name,
        attempt=1,
        message="Compiled node requeued",
        data={"node_id": node_id, "task_id": task_id, "kind": kind},
    )

    node_job = Job(job_id=task_id, info={"parent_job_id": job_id, "node_id": node_id})
    node_job.code = JobStatusCode.PENDING
    return node_job

resubmit(job_record: Any, policy: Optional[Any] = None) -> Job async

Resubmit a job from a stored record using persisted phrase payload.

Source code in lingo/celery/language.py

async def resubmit(self, job_record: Any, policy: Optional[Any] = None) -> Job:
    """Resubmit a job from a stored record using persisted phrase payload."""
    phrase_payload = getattr(job_record, "phrase_payload", None)
    if not phrase_payload:
        raise ValueError("Job record has no stored phrase payload for resubmission")

    phrase_obj = deserialize_phrase(phrase_payload)
    info = getattr(job_record, "info", None)

    retry_policy = None
    if policy is not None:
        retry_policy = policy if isinstance(policy, RestartPolicy) else RestartPolicy(**policy)
    else:
        raw = getattr(job_record, "policy_payload", None)
        if raw:
            retry_policy = RestartPolicy(**raw)

    return await self.say(phrase_obj, info=info, policy=retry_policy)

say(phrase: Phrase, info: Optional[BaseModel] = None, policy: Optional[RestartPolicy] = None) -> Job async

Dispatch a phrase (DAG) to the message broker for execution.

Returns a Job object with the job ID and status.

Source code in lingo/celery/language.py

async def say(
    self,
    phrase: Phrase,
    info: Optional[BaseModel] = None,
    policy: Optional[RestartPolicy] = None,
) -> Job:
    """
    Dispatch a phrase (DAG) to the message broker for execution.

    Returns a Job object with the job ID and status.
    """
    ctx = await self._prepare_dispatch_context(phrase, info, policy)
    if self.orchestration_mode == "compiled":
        return await self._dispatch_compiled(ctx)
    return await self._dispatch_simple(ctx)

speak(block: bool = True) -> None

Initialize the Celery language service.

Source code in lingo/celery/language.py

def speak(self, block: bool = True) -> None:
    """Initialize the Celery language service."""
    self._started = True
    if self.journal is not None and hasattr(self.journal, "register_worker"):
        worker_id = self.name or self.node_id
        self.journal.register_worker(
            worker_id=worker_id,
            node_id=self.node_id,
            tasks=sorted(set(self.verb_registry.keys()) | set(self.grammar_registry.keys())),
            queues=self.inferred_worker_queues(),
            private_queue=self.private_queue,
        )

verb(task_name_or_func: Optional[Any] = None) -> Callable

Decorator that defines the actual executable logic for a task.

Usage

@lang.verb def normalize(text: str) -> str: ...

@lang.verb('custom_name') def my_function(...): ...

Source code in lingo/celery/language.py

def verb(self, task_name_or_func: Optional[Any] = None) -> Callable:
    """
    Decorator that defines the actual executable logic for a task.

    Usage:
        @lang.verb
        def normalize(text: str) -> str:
            ...

        @lang.verb('custom_name')
        def my_function(...):
            ...
    """
    def decorator(func: Callable) -> Callable:
        name = func.__name__
        if isinstance(task_name_or_func, str):
            name = task_name_or_func
        self.verb_registry[name] = func

        @wraps(func)
        def wrapper(*args, **kwargs):
            return func(*args, **kwargs)

        return wrapper

    if callable(task_name_or_func) and not isinstance(task_name_or_func, str):
        return decorator(task_name_or_func)
    return decorator