# Manager

A **Manager** is a core component responsible for handling {doc}`/event` instances delivered by the Router and executing the required application logic.

Managers are single-threaded by default and act as deterministic event processors. Concurrency is achieved explicitly via Workers.

There are three base classes:

* `ManagerBase` – FIFO processing
* `ManagerPriorityBase` – Priority FIFO processing (extended behavior)
* `AsyncManagerBase` – **Experimental:** asyncio-driven event loop variant that supports `async def` handlers

---

## Responsibilities

A Manager is responsible for:

1. Subscribing to events
2. Handling incoming events
3. Producing responses (if applicable)
4. Emitting new events when needed
5. Delegating heavy work to Workers

Managers **do not route events**. They receive events from the Router and may emit new events back to the Router via `self.router_queue.put(...)`.

---

## Event Subscription

Managers subscribe to events using decorators:

* `@request`
* `@response`

At startup, the framework scans decorated methods and registers the Manager as a subscriber for the corresponding event types.

### Example

```python
from edri.abstract import ManagerBase, request
from edri.dataclass.response import ResponseStatus
from my_events import Ping

class ExampleManager(ManagerBase):

    @request
    def ping(self, event: Ping) -> None:
        event.response.message = "pong"
```

If any field in `event.response` is set, the response status is automatically set to `OK`.

If `event.response.set_status(...)` is called explicitly, the status is set immediately, but the event is still sent only **after the handler function finishes execution**.

This design intentionally prevents developers from running background work in the manager thread after finishing event handling — once the handler returns, the event lifecycle continues.

---

## Handling Events

When a Manager receives an event:

1. The correct decorated handler method is selected
2. The method is executed synchronously
3. If the event has a response and it is modified, the response is sent automatically

To emit a new event manually:

```python
self.router_queue.put(new_event)
```

Managers must explicitly emit new events using `self.router_queue.put(...)`.

New events are **never** sent automatically — only responses of the currently handled event are sent automatically at the end of the handler.

### Unhandled events (`resolve_unknown`)

If an event arrives from the Router that matches **no** `@request` or `@response` handler on this Manager, it is passed to `resolve_unknown`:

```python
def resolve_unknown(self, event: Event) -> None:
    # default: log a warning
    ...
```

The default implementation only logs a warning, so unrecognized events are otherwise dropped. Override it to provide a catch-all — for example to route events handled dynamically rather than by decorator, or to raise instead of silently ignoring them.

> This is the fallback for events on the **router pipe**. Events arriving on a custom pipe instead go to [`resolve_callback_pipe`](#listening-on-custom-pipes-additional-pipes), and a Worker's events go to [`resolve_callback_worker`](#receiving-events-back-from-a-worker-resolve-callback-worker).

---

## Concurrency with Workers

Managers remain single-threaded. For parallel or long-running work, use Workers.

To start a worker:

```python
self.workers.start(event, WorkerClass)
```

Workers can be:

* `WorkerThread`
* `WorkerProcess`

(receiving-events-back-from-a-worker-resolve-callback-worker)=
### Receiving events back from a Worker (`resolve_callback_worker`)

Every event a Worker emits with `self.event_send(...)` arrives at the Manager through a single callback:

```python
def resolve_callback_worker(self, event: Event, worker: Worker) -> None:
    ...
```

It is called **once per event, for every worker** — there is no other path by which a Worker's output enters the system.

**What the default does.** Out of the box, `resolve_callback_worker` tags the event with the worker's name (`event._worker`), records a timing stamp, and puts the event on the **router queue**. From there the Router dispatches it by type and direction — typically to a [`@response`](#event-subscription) handler (in this or another Manager) or back to the waiting API client.

> A Worker's event always travels **Worker → callback → Router** first; it never lands directly in a `@response` handler. The Manager is the only mediator of Worker traffic, which is exactly what makes this callback a useful interception point.

**The `worker` argument.** Alongside the event you receive the originating `Worker`, which lets you correlate the result with the task that produced it:

* `worker.event` — the event that started this worker (the original request)
* `worker.runtime.name` — the worker runtime's generated name (once spawned)

**When to override.** Override it to post-process worker output before it leaves the Manager — for example to aggregate several streamed results into one, tag or enrich events, or drop/filter some of them:

```python
def resolve_callback_worker(self, event: Event, worker: Worker) -> None:
    if isinstance(event, PartialResult):
        self._buffer.append(event)          # collect, emit nothing yet
        return
    super().resolve_callback_worker(event, worker)   # default: forward to the Router
```

> Events arriving on a **custom pipe** (not from a Worker) go to a different callback, `resolve_callback_pipe` — see [Listening on custom pipes](#listening-on-custom-pipes-additional-pipes).

---

(outgoing-event-hook)=
## Outgoing-event hook: `prepare_outgoing_event()`

Override `prepare_outgoing_event(event)` to inspect or modify an event immediately before a Manager places it on the Router queue:

```python
from edri.abstract import ManagerBase
from edri.dataclass.event import Event


class AuditedManager(ManagerBase):
    def prepare_outgoing_event(self, event: Event) -> None:
        self.logger.info("Sending %s", type(event).__name__)
```

The hook is called for every Router-bound event emitted by `ManagerBase` or its framework-provided subclasses. This includes resolved responses, Worker callback output, scheduled events, Store callbacks, API Broker output, and framework subscription events. `AsyncManagerBase` applies the same hook to both synchronous and asynchronous handler output.

For resolved responses, framework response preparation happens first, so the hook sees the final event state, including cache directives and updated Switch receive state. Any changes made by the hook are visible to the Router.

Keep the hook fast and non-blocking because it runs inline with the Manager's event processing. If the hook needs event-specific fields, check the event type first: it also receives framework control events such as subscriptions. Override this hook instead of `send_to_router()` so the framework retains a single, consistent queue-output path.

---

## Lifecycle hook: `after_start()`

Managers provide an `after_start()` hook that runs once, after the manager process starts.

Use it to perform initialization tasks that must happen inside the manager process (for example: warming caches, preparing in-memory state, seeding timestamps, opening long-lived resources, etc.).

```python
def after_start(self) -> None:
    """Hook method called after the manager process starts. Can be overridden to perform initialization tasks."""
```

---

## Reporting health

Every Manager answers the Router's periodic `HealthCheck` automatically. Two hooks let a Manager enrich what it reports:

* `health()` — return a `HealthReport(status, ready, reason)` to self-assess health and readiness (e.g. downgrade to `DEGRADED` when a dependency is down).
* `collect_custom_vitals()` — return a `dict[str, Any]` of application-defined metrics that ride along on the same round-trip and show up under `vitals.custom`.

Both run inline on the main loop on every check, so keep them cheap. See {doc}`/health` for the full contract, the status endpoint, and the vitals reference.

---

## Message Processing Strategies

Managers process incoming events in one of two modes:

### FIFO (`ManagerBase`)

Events are processed in arrival order.

### Priority FIFO (`ManagerPriorityBase`)

Events are processed by priority (lower value = higher priority).
Events with equal priority follow FIFO order.

Detailed priority configuration is described separately.

---

(manager-level-caching)=
## Manager-Level Caching

Managers support HTTP cache integration via the `@request` decorator.

Example:

```python
@request(cache="information")
def get_information(self, event: InformationEvent) -> None:
    ...
```

### How it works

* The `cache` argument defines a **cache key namespace**.
* When used with a **GET HTTP request**, EDRI integrates with HTTP caching semantics.

#### ETag handling

To support ETag-based caching, the event must expose the `If-None-Match` header as a field.

Example:

```python
etag: list[str] | None = Header(name="If-None-Match")
```

* An `ETag` is generated in the format:

  ```
  <cache-key>-<last-change-timestamp-in-microseconds>
  ```

  The timestamp carries microsecond resolution, so two changes within the same second produce distinct ETags.

  > This cache ETag (identifier + timestamp, validated via `If-None-Match`) is a **separate mechanism** from the file-download ETag, which is a content fingerprint validated via `If-Range`. See {ref}`two-unrelated-etag-mechanisms`.

* If the incoming request contains `If-None-Match` (provided via the `etag` field) and it matches the current ETag,
  the response status is set to **304 Not Modified**.

* If it does not match:

  * The regular response is returned.
  * The following `HeaderResponseDirective`s are added automatically:

    * `ETag`
    * `Cache-Control`
    * `Vary`

#### Cache-Control

`Cache-Control` is controlled by the environment variable:

```
EDRI_API_CACHE_CONTROL
```

#### Vary header

By default, `Vary` contains:

```
Accept,Accept-Encoding
```

Additional headers can be appended using:

```
EDRI_API_CACHE_HEADERS
```

### Cache invalidation behavior

Automatic: a non-GET request (POST, PUT, PATCH, DELETE) whose handler carries the same `cache="..."` key and returns an OK response bumps the "last change" timestamp for that key. The next GET for that key then produces a new ETag and returns fresh data.

Manual: the resolver cache is exposed as `self.cache` on the Manager. Call `renew()` to invalidate a key from anywhere — a handler that does not share the `cache="..."` key, a worker callback, or in response to an external event:

```python
@response
def on_upstream_changed(self, event: UpstreamChanged) -> None:
    self.cache.renew("information")
```

### Cache: setting an initial timestamp

To pre-seed a cache key's timestamp at startup, override `after_start()` and call `seed()`:

```python
def after_start(self) -> None:
    super().after_start()
    self.cache.seed("information", datetime(2024, 1, 1, tzinfo=timezone.utc))
```

Caching is automatic once `cache="..."` is defined on the handler.

> `self._cache` is a deprecated alias for `self.cache` and will be removed in a future release.

---

## Key-Value Store (Specialized Manager)

EDRI provides a built-in Key-Value Store implemented as a specialized Manager.

It enables:

* Data caching
* Cross-manager data sharing
* Temporary storage of computed results

It is started explicitly via:

```python
edri.start_store()
```

Full details are described in a dedicated document.

---

## Scheduler (Specialized Manager)

The Scheduler is a time-based event emitter implemented as a specialized Manager.

It can:

* Emit events at specific times
* Repeat events at defined intervals

It is started via:

```python
edri.start_scheduler()
```

Full scheduling behavior is documented separately.

---

## Priority-based processing (`ManagerPriorityBase`)

`ManagerPriorityBase` is an extension of `ManagerBase` that processes events in **priority order** rather than strict FIFO.

### When to use

Use `ManagerPriorityBase` when some event types must be handled before others regardless of arrival order — for example, cancellation signals should preempt normal work, or health-check events must always respond promptly.

### How to subclass

You must implement the abstract `get_priority(event_type)` method. It returns an integer — **lower value = higher priority** (processed first).

`ManagerPriorityBase.__init__` accepts the same parameters as `ManagerBase`:

```python
OrderManager(
    router_queue=edri.router_queue,
    logger=None,          # optional custom Logger
    from_time=None,       # optional datetime — ignore events older than this
    max_workers=-1,       # optional concurrency limit (-1 = unlimited)
)
```

```python
from typing import Type
from edri.abstract import ManagerPriorityBase, request
from edri.dataclass.event import Event
from events.orders import UrgentOrder, NormalOrder, LowPriorityReport


class OrderManager(ManagerPriorityBase):

    def get_priority(self, event_type: Type[Event]) -> int:
        priority_map = {
            UrgentOrder: 1,
            NormalOrder: 5,
            LowPriorityReport: 10,
        }
        return priority_map.get(event_type, 5)

    @request
    def handle_urgent(self, event: UrgentOrder) -> None:
        event.response.status = "dispatched"

    @request
    def handle_normal(self, event: NormalOrder) -> None:
        event.response.status = "queued"

    @request
    def handle_report(self, event: LowPriorityReport) -> None:
        event.response.generated = True
```

### How it works internally

`ManagerPriorityBase` introduces a second internal thread (`command_handler`) that reads from a `PriorityQueue[QueueItem]`. The main event loop moves every incoming event into that priority queue. The `command_handler` thread then processes events in priority order.

```mermaid
flowchart TD
    In["Incoming events from Router"] --> PQ["PriorityQueue<br/>(sorted by get_priority())"]
    PQ --> CH["command_handler thread"]
    CH --> Res["resolve(event) → @request handler"]
```

Events with equal priority are processed in FIFO order among themselves.

### Caveats

* `ManagerPriorityBase` uses an additional background thread. Keep `get_priority()` fast — it is called on every incoming event in the main loop.
* Worker runtimes created via `workers.start(...)` send a wake signal to the priority loop so that worker events do not block the queue. This happens automatically.
* All other `ManagerBase` features (`after_start()`, caching, `@response`, Workers) work identically.

---

## Worker concurrency limit (`max_workers`)

Both `ManagerBase` and `ManagerPriorityBase` accept a `max_workers` parameter that limits how many workers can run simultaneously.

```python
class MyManager(ManagerBase):
    def __init__(self, router_queue):
        super().__init__(router_queue, max_workers=4)
```

* Default is `-1` (unlimited).
* When the limit is reached, additional `workers.start(...)` records remain `PENDING` in the controller's ordered storage.
* As running workers finish (`WorkerQuit`), queued worker starts are drained automatically.
* Setting `max_workers = 0` or any value below `-1` raises `ValueError`.

---

(listening-on-custom-pipes-additional-pipes)=
## Listening on custom pipes (`additional_pipes`)

A Manager's event loop normally waits on two kinds of pipes: the router pipe and its Workers' pipes. To make it also react to an external `multiprocessing.connection.Connection` — a hardware device, a socket bridge, a background thread, another process — override the `additional_pipes` property:

```python
from multiprocessing.connection import Connection

class DeviceManager(ManagerBase):
    @property
    def additional_pipes(self) -> set[Connection]:
        return {self._device_pipe}

    def resolve_callback_pipe(self, event, pipe: Connection) -> None:
        # Called for every object read off a pipe that is neither the
        # router pipe nor a known Worker pipe.
        self.handle_device_message(event)
```

The property is re-read on every loop iteration, so the set may change over time. Two things to keep in mind:

* **You must also override `resolve_callback_pipe`.** Objects arriving on a custom pipe are not Worker events, so the loop routes them to `resolve_callback_pipe(obj, pipe)` — whose default implementation only logs a warning. Without an override the messages are effectively dropped.
* Anything read off these pipes is whatever the other end sent; it does not have to be an `Event`. The dispatch by `@request` / `@response` does not apply here.

This is the same extension point `ManagerPriorityBase` uses internally (for its wake-up pipe) and that the `Broker` uses to listen to the Listener.

---

## Execution Model Summary

* Managers are single-threaded event processors
* Subscriptions are declared via decorators
* Responses are auto-sent when modified
* New events must be emitted manually
* Concurrency is achieved via Workers
* Priority handling is available through `ManagerPriorityBase`

Managers are deterministic coordination units in the EDRI architecture — Workers perform the heavy lifting.

---

## See also

* {doc}`/event` — event and response definitions
* {doc}`/worker` — Worker types, communication, SharedMemoryPipe
* {doc}`/middleware` — Middleware pipeline
* {doc}`/specialized_managers` — Scheduler and Key-Value Store
* {doc}`/testing` — how to unit-test Managers