feat(routing): separate task routing from broker transport

Introduce router-owned task routing, flow subscriptions, and shared task
declarations while preserving the existing broker-first task API.

Add transport-neutral Flow contracts, explicit route/subscription semantics,
multi-broker router ownership, task_builder base_cls support, prepared invocation
route snapshots, and scheduler/requeue compatibility behavior.

Document migration guidance and add executable routing examples for multiple
brokers and shared task packages.

Author: GefMar

docs/examples/router/multiple_brokers.py

@@ -2,7 +2,7 @@
 
 import asyncio
 
-from taskiq import Flow, InMemoryBroker, TaskiqRouter
+from taskiq import Flow, InMemoryBroker, TaskiqRoute, TaskiqRouter
 
 router = TaskiqRouter()
 
@@ -35,6 +35,23 @@ async def send_email(user_id: int, template: str) -> str:
     broker=priority_broker,
     flow=priority_email_flow,
 )
+priority_subscription = router.subscribe(
+    priority_broker,
+    priority_email_flow,
+    send_email,
+)
+
+
+def _format_route(task_name: str, route: TaskiqRoute) -> str:
+    """Return a readable route diagnostic for the example output."""
+    flow_name = route.flow.name if route.flow is not None else "<default>"
+    return f"{task_name} -> broker={route.broker_name}, flow={flow_name}"
+
+
+def _format_listen_plan() -> str:
+    """Return flows that the priority broker should subscribe to."""
+    flow_names = ", ".join(flow.name for flow in priority_broker.get_subscribed_flows())
+    return f"priority listens to: {flow_names}"
 
 
 async def _main() -> None:
@@ -43,22 +60,34 @@ async def _main() -> None:
     try:
         direct_result = await send_email(7, "welcome")
 
-        routed_task = await send_email.kiq(7, "welcome")
-        routed_result = await routed_task.wait_result(timeout=2)
+        declared_route = router.resolve_route(send_email)
+        assert declared_route == priority_route
 
-        bulk_task = (
+        routed_task = (
             await send_email.kicker()
-            .with_route(
-                default_broker,
-                bulk_email_flow,
+            .with_route(declared_route)
+            .kiq(
+                7,
+                "welcome",
             )
-            .kiq(8, "digest")
         )
+        routed_result = await routed_task.wait_result(timeout=2)
+
+        bulk_route = router.resolve_route(
+            send_email,
+            broker=default_broker,
+            flow=bulk_email_flow,
+        )
+        bulk_task = await send_email.kicker().with_route(bulk_route).kiq(8, "digest")
         bulk_result = await bulk_task.wait_result(timeout=2)
 
         print(f"Direct call: {direct_result}")
-        print(f"Declared route: {priority_route.broker_name}")
+        print(f"Router rule: {_format_route(send_email.task_name, priority_route)}")
+        print(f"Subscription tasks: {sorted(priority_subscription.task_names)}")
+        print(_format_listen_plan())
+        print(f"Resolved route: {_format_route(send_email.task_name, declared_route)}")
         print(f"Routed call: {routed_result.return_value}")
+        print(f"Override route: {_format_route(send_email.task_name, bulk_route)}")
         print(f"Route override: {bulk_result.return_value}")
     finally:
         await priority_broker.shutdown()

docs/examples/router/shared_task_package.py

@@ -3,8 +3,15 @@
 import asyncio
 from collections.abc import Mapping
 from dataclasses import dataclass
-
-from taskiq import Flow, InMemoryBroker, TaskiqRouter, task_builder
+from typing import Any
+
+from taskiq import (
+    AsyncTaskiqDecoratedTask,
+    Flow,
+    InMemoryBroker,
+    TaskiqRouter,
+    task_builder,
+)
 
 
 @dataclass(frozen=True, slots=True)
@@ -14,15 +21,22 @@ class BillingQueue:
     name: str
     priority: int
 
-    def broker_options(self, broker_name: str) -> Mapping[str, object]:
+    def broker_options(self) -> Mapping[str, object]:
         """Return options that a billing broker adapter can understand."""
         return {
-            "broker": broker_name,
             "priority": self.priority,
         }
 
 
-@task_builder("billing.calculate_total", domain="billing")
+class BillingTask(AsyncTaskiqDecoratedTask[Any, Any]):
+    """Custom task class shared by billing package tasks."""
+
+    def billing_name(self) -> str:
+        """Return a billing-specific task name."""
+        return self.task_name
+
+
+@task_builder("billing.calculate_total", base_cls=BillingTask, domain="billing")
 async def calculate_total(price: int, quantity: int) -> int:
     """Package-level task definition that is not bound to any broker."""
     return price * quantity
@@ -44,18 +58,27 @@ async def calculate_total(price: int, quantity: int) -> int:
     broker=billing_broker,
     flow=billing_flow,
 )
+router.subscribe(
+    billing_broker,
+    billing_flow,
+    registered_calculate_total,
+)
 
 
 async def _main() -> None:
     await billing_broker.startup()
     try:
         direct_result = await calculate_total.call(19, 3)
 
+        priority_route = router.resolve_route(
+            registered_calculate_total,
+            broker=billing_broker,
+            flow=priority_billing_flow,
+        )
         prepared_task = (
             registered_calculate_total.kicker()
             .with_route(
-                billing_broker,
-                priority_billing_flow,
+                priority_route,
             )
             .prepare(19, 3)
         )
@@ -64,6 +87,9 @@ async def _main() -> None:
         queued_result = await queued_task.wait_result(timeout=2)
 
         print(f"Shared task direct call: {direct_result}")
+        print(f"Registered task class: {registered_calculate_total.billing_name()}")
+        listen_flow = billing_broker.get_subscribed_flows()[0]
+        print(f"Registered listen flow: {listen_flow.name}")
         print(f"Prepared message: {prepared_task.message.task_name}")
         print(f"Registered queued call: {queued_result.return_value}")
     finally:

docs/guide/architecture-overview.md

@@ -82,6 +82,28 @@ asyncio.run(main())
 
 ```
 
+## Router and flows
+
+Taskiq can use a `TaskiqRouter` to keep routing rules outside of broker
+implementations. Brokers remain transport adapters, while the router owns task
+registration, route resolution and flow subscriptions.
+This section describes the `experiment/separate_broker` branch contract. The
+old `@broker.task(...)`, `.kiq()`, labels, scheduler and result backend behavior
+remain compatible, while router/flow APIs are additive review material for the
+branch.
+
+`Flow` is a transport-neutral delivery address. Broker packages may provide
+their own flow classes for queue, topic, subject or stream options, as long as
+they implement the same flow protocol. The router deduplicates subscriptions by
+flow name and rejects same-name flows with incompatible broker options.
+
+Routing and subscribing are separate responsibilities. `route_task(...)`
+chooses the outbound broker and flow for task invocations. `subscribe(...)`
+adds flows to a broker listen plan for flow-aware broker adapters. Worker task
+lookup still uses `task_name`; flow does not select the Python task.
+
+Read more in the [Routing and flows](./routing-and-flows.md) section.
+
 ## Messages
 
 Every message has labels. You can define labels