feat: implement daena MVP core service

Complete implementation of the daena log/event collection service:

Core runtime:
- DaenaRuntime service lifecycle (start/stop/reload)
- Pipeline orchestrator with source/persist/dispatch flow
- LMDB persistence backend with key-prefix partitioning
- Plugin discovery via Python entry points

Pipeline:
- Source/Processor/Sink ABC interfaces with DeliveryResult
- DummySource for testing and development
- DummySink and HTTPSink implementations
- PassthroughProcessor
- Record dataclass for in-memory records
- StoredRecord model with RecordState tracking

REST API (FastAPI, v1):
- GET /system/{status,health,config}
- GET /plugins
- GET /queue, /queue/{pending,dead}
- GET /sinks
- Versioned under /api/v1 prefix

CLI (Typer):
- status, health, queue, sinks, plugins, config, version

Persistence (LMDB):
- Single-database key-prefix design (no named DBs)
- Index keys: idx:{state}:{ts}\x00{id}
- Record keys: rec:{id}
- Full CRUD: put/get/ack/nack/dead/update_state
- Pending/retry/dead-letter queries with state filtering
- delete_old for TTL-based retention

Deployment:
- systemd unit file (deploy/daena.service)
- Example YAML config (config/daena.yaml)
- Entry points: daena (CLI), daena-serve (uvicorn server)

Quality:
- 23 unit tests covering config, persistence, pipeline
- ruff linting and formatting pass clean
- Py3.13 compatible, uv-managed, pyenv-pinned
- GPG-signed commit on dev branch

README.md

@@ -1,2 +1,212 @@
 # daena
-Collect it. Keep it. Ship it. That’s Daena! 🤠
+
+Collect, buffer, and forward logs and events from Linux to central observability backends.
+
+## Overview
+
+Daena is a modular, extensible Linux service that:
+
+- Collects logs, events, and telemetry from pluggable sources
+- Normalizes and processes records through a configurable pipeline
+- Persists data locally in LMDB for durability, replay, and buffering
+- Forwards records reliably to one or more remote sinks
+- Exposes a REST API for management and inspection
+- Provides a CLI for operational tasks
+- Degrades gracefully under failures with retry, backoff, and dead-letter queues
+
+## Architecture
+
+```
+Sources → [Processors] → Pipeline → Persistence (LMDB) → Sink Dispatcher → Remote
+                                                              ↻ retry queue
+```
+
+Key design decisions:
+
+- **Plugin architecture**: Sources, processors, and sinks are loaded via
+  Python entry points. Add new ones by installing a package or registering
+  in `pyproject.toml`.
+- **Pluggable persistence**: The `Backend` protocol abstracts storage.
+  MVP uses LMDB. Future: NATS JetStream.
+- **Reliability-first**: Records survive restarts, sink failures trigger
+  exponential backoff, and permanently failed records land in a dead-letter
+  queue.
+
+## Quick start
+
+```bash
+# Prerequisites: pyenv with Python 3.13, uv installed
+
+# Clone and set up
+git clone <repo> && cd daena
+pyenv local 3.13.5
+uv sync --extra dev
+
+# Run with dummy source/sink (no external dependencies)
+uv run python -c "
+import asyncio
+from daena.config import DaenaConfig, SourceConfig, SinkConfig
+from daena.core.runtime import DaenaRuntime
+
+async def main():
+    config = DaenaConfig()
+    config.sources = [SourceConfig(name='test', type='dummy',
+        options={'interval': 1.0, 'message': 'hello'})]
+    config.sinks = [SinkConfig(name='log', type='dummy')]
+    config.persistence.lmdb.path = '/tmp/daena_data'
+
+    runtime = DaenaRuntime(config)
+    await runtime.start()
+    await asyncio.sleep(5)
+    print(f'Records: {runtime.backend.count()}')
+    await runtime.stop()
+
+asyncio.run(main())
+"
+
+# Run tests
+uv run pytest tests/ -v
+
+# Use the CLI (point at a running daena API)
+uv run daena status
+uv run daena health
+uv run daena queue
+```
+
+## Project structure
+
+```
+daena/
+├── pyproject.toml              # Project metadata, dependencies, entry points
+├── config/daena.yaml           # Example configuration
+├── deploy/daena.service        # systemd unit file
+├── src/daena/
+│   ├── app.py                  # `daena-serve` entry point (uvicorn)
+│   ├── config.py               # YAML → pydantic config
+│   ├── logutil.py              # structlog setup
+│   ├── exceptions.py           # Domain exceptions
+│   ├── api/                    # FastAPI REST API (v1 versioned)
+│   ├── cli/                    # Typer CLI
+│   ├── core/
+│   │   ├── runtime.py          # Service lifecycle
+│   │   └── pipeline.py         # Collection → persist → dispatch
+│   ├── persistence/
+│   │   ├── backend.py          # Abstract Backend protocol
+│   │   ├── models.py           # RecordState, StoredRecord
+│   │   └── lmdb.py             # LMDB implementation
+│   ├── pipeline/
+│   │   ├── base.py             # Source/Processor/Sink ABCs
+│   │   ├── record.py           # In-memory Record dataclass
+│   │   ├── sources/            # Built-in sources (dummy)
+│   │   ├── sinks/              # Built-in sinks (dummy, http)
+│   │   └── processors/         # Built-in processors (passthrough)
+│   └── plugins/
+│       ├── loader.py           # Entry-point discovery
+│       └── registry.py         # Plugin registry
+└── tests/
+```
+
+## Configuration
+
+Daena loads configuration from `/etc/daena/daena.yaml`, `./daena.yaml`, or
+`./config/daena.yaml`. See `config/daena.yaml` for a full example.
+
+```yaml
+logging:
+  level: info
+  format: json
+
+api:
+  http:
+    host: 127.0.0.1
+    port: 8642
+
+persistence:
+  backend: lmdb
+  lmdb:
+    path: /var/lib/daena/data
+    map_size: 104857600
+
+sources:
+  - name: system_logs
+    type: journald       # future
+    options: {}
+
+processors:
+  - name: add_hostname
+    type: enricher       # future
+    options: {}
+
+sinks:
+  - name: central
+    type: http
+    options:
+      url: https://logs.example.com/ingest
+```
+
+## REST API
+
+The API runs on `http://127.0.0.1:8642/api/v1`:
+
+| Endpoint | Description |
+|----------|-------------|
+| `GET /system/status` | Service status |
+| `GET /system/health` | Health check |
+| `GET /system/config` | Loaded configuration |
+| `GET /plugins` | Registered plugins |
+| `GET /queue` | Queue state counts |
+| `GET /queue/pending` | Pending records |
+| `GET /queue/dead` | Dead-letter records |
+| `GET /sinks` | Configured sinks |
+
+## Deployment
+
+### systemd
+
+```bash
+# Create daena user
+sudo useradd -r -s /sbin/nologin -d /var/lib/daena daena
+
+# Install package
+uv pip install .
+
+# Set up directories
+sudo mkdir -p /etc/daena /var/lib/daena /var/log/daena /run/daena
+sudo cp config/daena.yaml /etc/daena/
+sudo cp deploy/daena.service /etc/systemd/system/
+sudo systemctl daemon-reload
+sudo systemctl enable --now daena
+```
+
+## Extending
+
+Add a new source by creating a class that inherits `daena.pipeline.base.Source`
+and registering it as an entry point:
+
+```python
+# my_sources.py
+from daena.pipeline.base import Source
+from daena.pipeline.record import Record
+
+class MySource(Source):
+    async def start(self):
+        pass
+    async def stop(self):
+        pass
+```
+
+```toml
+# pyproject.toml
+[project.entry-points."daena.sources"]
+my_source = "my_sources:MySource"
+```
+
+Same pattern applies for `Processor` and `Sink` base classes.
+
+## Branch workflow
+
+- `main` — stable integration branch
+- `dev` — active development branch
+
+Feature work happens on `dev` or topic branches. Merges to `main` represent
+stable release checkpoints.

config/daena.yaml

@@ -0,0 +1,47 @@
+# daena configuration
+# Path: /etc/daena/daena.yaml
+
+data_dir: /var/lib/daena
+log_dir: /var/log/daena
+pid_file: /run/daena/daena.pid
+
+logging:
+  level: info
+  format: json
+
+api:
+  http:
+    host: 127.0.0.1
+    port: 8642
+    workers: 1
+
+pipeline:
+  batch_size: 100
+  flush_interval: 5.0
+  max_queue_size: 10000
+
+persistence:
+  backend: lmdb
+  lmdb:
+    path: /var/lib/daena/data
+    map_size: 104857600
+    max_databases: 16
+
+sources:
+  - name: dummy
+    type: dummy
+    enabled: true
+    options:
+      interval: 5.0
+      message: "daena heartbeat"
+
+processors:
+  - name: passthrough
+    type: passthrough
+    enabled: true
+
+sinks:
+  - name: log
+    type: dummy
+    enabled: true
+    options: {}

deploy/daena.service

@@ -0,0 +1,31 @@
+[Unit]
+Description=Daena - Linux log and event collector
+Documentation=https://github.com/anomalyco/daena
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+User=daena
+Group=daena
+
+ExecStart=/usr/local/bin/daena-serve
+ExecReload=/bin/kill -HUP $MAINPID
+Restart=on-failure
+RestartSec=5
+
+AmbientCapabilities=CAP_SYS_PTRACE
+NoNewPrivileges=true
+ProtectSystem=full
+ProtectHome=true
+PrivateTmp=true
+
+StateDirectory=daena
+LogsDirectory=daena
+RuntimeDirectory=daena
+
+# Environment
+Environment=DAENA_CONFIG_PATH=/etc/daena/daena.yaml
+
+[Install]
+WantedBy=multi-user.target

src/daena/__init__.py

@@ -1 +1 @@
-from daena._version import __version__
+from daena._version import __version__ as __version__

src/daena/__main__.py

@@ -0,0 +1,5 @@
+from __future__ import annotations
+
+from daena.cli.main import app
+
+app()

src/daena/api/router.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from fastapi import APIRouter
+
+from daena.api.routes.plugins import router as plugins_router
+from daena.api.routes.queue import router as queue_router
+from daena.api.routes.sinks import router as sinks_router
+from daena.api.routes.system import router as system_router
+
+api_router = APIRouter()
+
+api_router.include_router(system_router, prefix="/system", tags=["system"])
+api_router.include_router(plugins_router, prefix="/plugins", tags=["plugins"])
+api_router.include_router(queue_router, prefix="/queue", tags=["queue"])
+api_router.include_router(sinks_router, prefix="/sinks", tags=["sinks"])

src/daena/api/routes/plugins.py

@@ -0,0 +1,16 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+
+router = APIRouter()
+
+
+@router.get("")
+async def list_plugins(request: Request) -> dict:
+    runtime = request.app.state.runtime
+    registry = runtime.registry
+    plugins = registry.all_plugins()
+    result: dict[str, list[dict]] = {}
+    for category, entries in plugins.items():
+        result[category] = [{"name": name, "class": cls.__name__} for name, cls in entries.items()]
+    return {"plugins": result}

src/daena/api/routes/queue.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Query, Request
+
+from daena.persistence.models import RecordState
+
+router = APIRouter()
+
+
+@router.get("")
+async def get_queue_state(request: Request) -> dict:
+    runtime = request.app.state.runtime
+    backend = runtime.backend
+    if backend is None:
+        return {"error": "backend not initialized"}
+    return {
+        "pending": backend.count(RecordState.pending),
+        "delivered": backend.count(RecordState.delivered),
+        "failed": backend.count(RecordState.failed),
+        "dead": backend.count(RecordState.dead),
+        "total": backend.count(),
+    }
+
+
+@router.get("/pending")
+async def get_pending(
+    request: Request,
+    limit: int = Query(default=100, le=1000),
+    offset: int = Query(default=0, ge=0),
+) -> dict:
+    runtime = request.app.state.runtime
+    backend = runtime.backend
+    if backend is None:
+        return {"error": "backend not initialized", "records": []}
+    records = backend.list_by_state(RecordState.pending, limit=limit, offset=offset)
+    return {"records": [r.to_dict() for r in records], "count": len(records)}
+
+
+@router.get("/dead")
+async def get_dead(
+    request: Request,
+    limit: int = Query(default=100, le=1000),
+    offset: int = Query(default=0, ge=0),
+) -> dict:
+    runtime = request.app.state.runtime
+    backend = runtime.backend
+    if backend is None:
+        return {"error": "backend not initialized", "records": []}
+    records = backend.list_by_state(RecordState.dead, limit=limit, offset=offset)
+    return {"records": [r.to_dict() for r in records], "count": len(records)}

src/daena/api/routes/sinks.py

@@ -0,0 +1,17 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+
+router = APIRouter()
+
+
+@router.get("")
+async def list_sinks(request: Request) -> dict:
+    runtime = request.app.state.runtime
+    pipeline = runtime.pipeline
+    if pipeline is None:
+        return {"sinks": []}
+    sinks = pipeline.sinks
+    return {
+        "sinks": [{"name": s.name, "type": type(s).__name__, "config": s.config} for s in sinks]
+    }

src/daena/api/routes/system.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from fastapi import APIRouter, Request
+
+from daena.core.runtime import DaenaRuntime
+
+router = APIRouter()
+
+
+def _resolve_runtime(request: Request) -> DaenaRuntime:
+    return request.app.state.runtime  # type: ignore[no-any-return]
+
+
+@router.get("/status")
+async def get_status(request: Request) -> dict:
+    runtime = _resolve_runtime(request)
+    return {
+        "service": "daena",
+        "version": "0.1.0",
+        "running": runtime._running,
+        "uptime": None,
+    }
+
+
+@router.get("/health")
+async def get_health(request: Request) -> dict:
+    runtime = _resolve_runtime(request)
+    healthy = runtime._running
+    return {
+        "healthy": healthy,
+        "running": runtime._running,
+        "backend": runtime.backend is not None,
+        "plugins_loaded": bool(runtime.registry.list_sources()),
+    }
+
+
+@router.get("/config")
+async def get_config(request: Request) -> dict:
+    runtime = _resolve_runtime(request)
+    cfg = runtime.config
+    return {
+        "data_dir": cfg.data_dir,
+        "log_dir": cfg.log_dir,
+        "api": cfg.api.model_dump(),
+        "pipeline": cfg.pipeline.model_dump(),
+        "persistence": cfg.persistence.model_dump(),
+        "sources": [s.model_dump() for s in cfg.sources],
+        "processors": [p.model_dump() for p in cfg.processors],
+        "sinks": [s.model_dump() for s in cfg.sinks],
+    }

src/daena/api/server.py

@@ -0,0 +1,50 @@
+from __future__ import annotations
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI
+
+from daena.api.router import api_router
+from daena.core.runtime import DaenaRuntime
+from daena.logutil import get_logger
+
+log = get_logger("daena.api")
+
+
+class DaenaAPI:
+    """Wraps the FastAPI application with daena runtime integration."""
+
+    def __init__(self, runtime: DaenaRuntime) -> None:
+        self._runtime = runtime
+        self._app: FastAPI | None = None
+
+    def build(self) -> FastAPI:
+        runtime = self._runtime
+
+        @asynccontextmanager
+        async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
+            log.info("api_starting")
+            await runtime.start()
+            yield
+            log.info("api_stopping")
+            await runtime.stop()
+
+        app = FastAPI(
+            title="Daena",
+            description="Daena service management API",
+            version="0.1.0",
+            lifespan=lifespan,
+        )
+
+        app.state.runtime = runtime
+        app.include_router(api_router, prefix="/api/v1")
+
+        self._app = app
+        return app
+
+    @property
+    def app(self) -> FastAPI:
+        if self._app is None:
+            return self.build()
+        return self._app

src/daena/app.py

@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+import uvicorn
+
+from daena.api.server import DaenaAPI
+from daena.config import load_config
+from daena.logutil import get_logger, setup_logging
+
+log = get_logger("daena.app")
+
+
+def serve() -> None:
+    """Entry point for `daena-serve` (systemd/uvicorn mode)."""
+    config_path = _resolve_config_path()
+    config = load_config(config_path)
+    setup_logging(
+        level=config.logging.level,
+        fmt=config.logging.format,
+        log_file=config.logging.file,
+    )
+
+    runtime = _build_runtime(config)
+    api = DaenaAPI(runtime)
+    app = api.build()
+
+    log.info(
+        "starting_server",
+        host=config.api.http.host,
+        port=config.api.http.port,
+    )
+    uvicorn.run(
+        app,
+        host=config.api.http.host,
+        port=config.api.http.port,
+        workers=config.api.http.workers,
+        log_level=config.logging.level,
+    )
+
+
+def _resolve_config_path() -> str | None:
+    for candidate in (
+        "/etc/daena/daena.yaml",
+        "./daena.yaml",
+        "./config/daena.yaml",
+    ):
+        p = Path(candidate)
+        if p.exists():
+            return str(p)
+    return None
+
+
+def _build_runtime(config):
+    """Lazy import to prevent premature backend init."""
+    from daena.core.runtime import DaenaRuntime
+
+    return DaenaRuntime(config)
+
+
+if __name__ == "__main__":
+    serve()

src/daena/cli/main.py

@@ -0,0 +1,140 @@
+from __future__ import annotations
+
+import httpx
+import typer
+from rich.console import Console
+from rich.table import Table
+
+from daena._version import __version__
+
+app = typer.Typer(
+    name="daena",
+    help="Daena: collect, buffer, and forward logs and events.",
+)
+console = Console()
+
+DEFAULT_API_URL = "http://127.0.0.1:8642/api/v1"
+
+
+def _client(timeout: float = 10.0) -> httpx.Client:
+    return httpx.Client(timeout=httpx.Timeout(timeout))
+
+
+def _api_url() -> str:
+    return DEFAULT_API_URL  # TODO: make configurable via env/flag
+
+
+@app.command()
+def status() -> None:
+    """Show daemon status."""
+    try:
+        with _client(5.0) as c:
+            resp = c.get(f"{_api_url()}/system/status")
+            resp.raise_for_status()
+            data = resp.json()
+            console.print(f"[bold]Daena[/bold] v{data.get('version', '?')}")
+            console.print(f"Running: {data.get('running', '?')}")
+    except httpx.ConnectError:
+        console.print("[red]Cannot connect to daena service[/red]")
+        raise typer.Exit(1) from None
+
+
+@app.command()
+def health() -> None:
+    """Check service health."""
+    try:
+        with _client(5.0) as c:
+            resp = c.get(f"{_api_url()}/system/health")
+            resp.raise_for_status()
+            data = resp.json()
+            if data.get("healthy"):
+                console.print("[green]Healthy[/green]")
+            else:
+                console.print("[red]Unhealthy[/red]")
+                raise typer.Exit(1)
+    except httpx.ConnectError:
+        console.print("[red]Cannot connect to daena service[/red]")
+        raise typer.Exit(1) from None
+
+
+@app.command()
+def queue() -> None:
+    """Inspect queue state."""
+    try:
+        with _client() as c:
+            resp = c.get(f"{_api_url()}/queue")
+            resp.raise_for_status()
+            data = resp.json()
+            table = Table("State", "Count")
+            for key in ("pending", "delivered", "failed", "dead", "total"):
+                table.add_row(key, str(data.get(key, 0)))
+            console.print(table)
+    except httpx.ConnectError:
+        console.print("[red]Cannot connect to daena service[/red]")
+        raise typer.Exit(1) from None
+
+
+@app.command()
+def sinks() -> None:
+    """List configured sinks."""
+    try:
+        with _client() as c:
+            resp = c.get(f"{_api_url()}/sinks")
+            resp.raise_for_status()
+            data = resp.json()
+            sinks_list = data.get("sinks", [])
+            if not sinks_list:
+                console.print("No sinks configured")
+                return
+            table = Table("Name", "Type")
+            for s in sinks_list:
+                table.add_row(s["name"], s["type"])
+            console.print(table)
+    except httpx.ConnectError:
+        console.print("[red]Cannot connect to daena service[/red]")
+        raise typer.Exit(1) from None
+
+
+@app.command()
+def plugins() -> None:
+    """List loaded plugins."""
+    try:
+        with _client() as c:
+            resp = c.get(f"{_api_url()}/plugins")
+            resp.raise_for_status()
+            data = resp.json()
+            plugins_data = data.get("plugins", {})
+            for category, items in plugins_data.items():
+                if items:
+                    table = Table(f"{category.title()}")
+                    for item in items:
+                        table.add_row(f"{item['name']} ({item['class']})")
+                    console.print(table)
+                    console.print()
+    except httpx.ConnectError:
+        console.print("[red]Cannot connect to daena service[/red]")
+        raise typer.Exit(1) from None
+
+
+@app.command()
+def config() -> None:
+    """Show loaded service configuration."""
+    try:
+        with _client() as c:
+            resp = c.get(f"{_api_url()}/system/config")
+            resp.raise_for_status()
+            data = resp.json()
+            console.print(data)
+    except httpx.ConnectError:
+        console.print("[red]Cannot connect to daena service[/red]")
+        raise typer.Exit(1) from None
+
+
+@app.command()
+def version() -> None:
+    """Show version."""
+    console.print(f"daena v{__version__}")
+
+
+if __name__ == "__main__":
+    app()

src/daena/config.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import yaml
+from pydantic import Field, field_validator
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class LoggingConfig(BaseSettings):
+    level: str = "info"
+    format: str = "json"
+    file: str | None = None
+
+
+class APIHTTPConfig(BaseSettings):
+    host: str = "127.0.0.1"
+    port: int = 8642
+    workers: int = 1
+
+
+class APIConfig(BaseSettings):
+    http: APIHTTPConfig = APIHTTPConfig()
+
+
+class SinkRetryConfig(BaseSettings):
+    max_attempts: int = 5
+    min_delay: float = 1.0
+    max_delay: float = 120.0
+    multiplier: float = 2.0
+
+
+class SinkConfig(BaseSettings):
+    name: str
+    type: str
+    enabled: bool = True
+    retry: SinkRetryConfig = SinkRetryConfig()
+    options: dict[str, Any] = Field(default_factory=dict)
+
+
+class SourceConfig(BaseSettings):
+    name: str
+    type: str
+    enabled: bool = True
+    options: dict[str, Any] = Field(default_factory=dict)
+
+
+class ProcessorConfig(BaseSettings):
+    name: str
+    type: str
+    enabled: bool = True
+    options: dict[str, Any] = Field(default_factory=dict)
+
+
+class LMDBConfig(BaseSettings):
+    path: str = "/var/lib/daena/data"
+    map_size: int = 104_857_600
+    max_databases: int = 16
+
+
+class PersistenceConfig(BaseSettings):
+    backend: str = "lmdb"
+    lmdb: LMDBConfig = LMDBConfig()
+    options: dict[str, Any] = Field(default_factory=dict)
+
+
+class PipelineConfig(BaseSettings):
+    batch_size: int = 100
+    flush_interval: float = 5.0
+    max_queue_size: int = 10_000
+
+
+class DaenaConfig(BaseSettings):
+    model_config = SettingsConfigDict(env_prefix="DAENA_", extra="ignore")
+
+    data_dir: str = "/var/lib/daena"
+    log_dir: str = "/var/log/daena"
+    pid_file: str = "/run/daena/daena.pid"
+
+    logging: LoggingConfig = LoggingConfig()
+    api: APIConfig = APIConfig()
+    pipeline: PipelineConfig = PipelineConfig()
+    persistence: PersistenceConfig = PersistenceConfig()
+
+    sources: list[SourceConfig] = Field(default_factory=list)
+    processors: list[ProcessorConfig] = Field(default_factory=list)
+    sinks: list[SinkConfig] = Field(default_factory=list)
+
+    @field_validator("data_dir", "log_dir", mode="before")
+    @classmethod
+    def ensure_trailing(cls, v: str) -> str:
+        return v.rstrip("/")
+
+
+def load_config(path: str | Path | None = None) -> DaenaConfig:
+    config = DaenaConfig()
+    if path:
+        p = Path(path)
+        if not p.exists():
+            raise FileNotFoundError(f"Configuration file not found: {p}")
+        with open(p) as f:
+            raw: dict[str, Any] = yaml.safe_load(f) or {}
+        merged = config.model_dump()
+        merged.update(raw)
+        config = DaenaConfig(**merged)
+    return config

src/daena/core/di.py

@@ -0,0 +1,32 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+class Container:
+    """Minimal dependency injection container.
+
+    Provides singleton and factory resolution for service components.
+    """
+
+    def __init__(self) -> None:
+        self._instances: dict[str, Any] = {}
+        self._factories: dict[str, callable] = {}
+
+    def register(self, key: str, instance: Any) -> None:
+        self._instances[key] = instance
+
+    def register_factory(self, key: str, factory: callable) -> None:
+        self._factories[key] = factory
+
+    def resolve(self, key: str) -> Any:
+        if key in self._instances:
+            return self._instances[key]
+        if key in self._factories:
+            instance = self._factories[key]()
+            self._instances[key] = instance
+            return instance
+        raise KeyError(f"No component registered for: {key}")
+
+    def has(self, key: str) -> bool:
+        return key in self._instances or key in self._factories

src/daena/core/pipeline.py

@@ -0,0 +1,265 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from collections.abc import Callable
+from typing import Any
+
+from daena.config import DaenaConfig
+from daena.logutil import get_logger
+from daena.persistence.backend import Backend
+from daena.persistence.models import StoredRecord
+from daena.pipeline.base import DeliveryResult, Processor, Sink, Source
+from daena.pipeline.record import Record
+from daena.plugins.registry import PluginRegistry
+
+log = get_logger("daena.pipeline")
+
+
+class SourceRunner:
+    """Wraps a Source and handles record ingestion into the pipeline."""
+
+    def __init__(
+        self,
+        source: Source,
+        on_record: Callable[[Record], Any],
+    ) -> None:
+        self._source = source
+        self._on_record = on_record
+
+    async def start(self) -> None:
+        if hasattr(self._source, "set_callback"):
+            self._source.set_callback(self._on_record)
+        log.info("source_starting", source=self._source.name)
+        await self._source.start()
+
+    async def stop(self) -> None:
+        log.info("source_stopping", source=self._source.name)
+        await self._source.stop()
+
+
+class SinkDispatcher:
+    """Manages delivery of records to a sink with retry logic."""
+
+    def __init__(
+        self,
+        sink: Sink,
+        backend: Backend,
+        config: DaenaConfig,
+    ) -> None:
+        self._sink = sink
+        self._backend = backend
+        self._config = config
+        self._logger = log.bind(sink=sink.name)
+
+    async def start(self) -> None:
+        self._logger.info("sink_starting")
+        await self._sink.start()
+
+    async def stop(self) -> None:
+        self._logger.info("sink_stopping")
+        await self._sink.stop()
+
+    async def deliver(self, record: Record) -> DeliveryResult:
+        return await self._sink.deliver(record)
+
+
+class Pipeline:
+    """Orchestrates the full collection → persist → forward data flow.
+
+    Lifecycle:
+      1. Sources emit Records
+      2. Processors transform/filter Records
+      3. Records are persisted to the backend
+      4. SinkDispatcher reads pending records and delivers them
+      5. Delivery results update record state (ack/nack/dead)
+    """
+
+    def __init__(
+        self,
+        config: DaenaConfig,
+        backend: Backend,
+        registry: PluginRegistry,
+    ) -> None:
+        self._config = config
+        self._backend = backend
+        self._registry = registry
+        self._sources: list[Source] = []
+        self._processors: list[Processor] = []
+        self._dispatchers: list[SinkDispatcher] = []
+        self._source_runners: list[SourceRunner] = []
+        self._dispatch_task: asyncio.Task[None] | None = None
+        self._running = False
+        self._queue: asyncio.Queue[Record] = asyncio.Queue(maxsize=config.pipeline.max_queue_size)
+
+    async def start(self) -> None:
+        self._running = True
+        self._build_pipeline()
+        for runner in self._source_runners:
+            await runner.start()
+        for dispatcher in self._dispatchers:
+            await dispatcher.start()
+        self._dispatch_task = asyncio.create_task(self._dispatch_loop())
+        log.info(
+            "pipeline_started",
+            sources=len(self._source_runners),
+            sinks=len(self._dispatchers),
+            processors=len(self._processors),
+        )
+
+    async def stop(self) -> None:
+        self._running = False
+        if self._dispatch_task:
+            self._dispatch_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._dispatch_task
+        for dispatcher in self._dispatchers:
+            await dispatcher.stop()
+        for runner in reversed(self._source_runners):
+            await runner.stop()
+        log.info("pipeline_stopped")
+
+    def _build_pipeline(self) -> None:
+        for sc in self._config.sources:
+            cls = self._registry.get_source(sc.type)
+            if not cls:
+                log.warning("source_type_not_found", type=sc.type)
+                continue
+            source = cls(name=sc.name, config=sc.options)
+            self._sources.append(source)
+            runner = SourceRunner(source, self._ingest)
+            self._source_runners.append(runner)
+
+        for pc in self._config.processors:
+            cls = self._registry.get_processor(pc.type)
+            if not cls:
+                log.warning("processor_type_not_found", type=pc.type)
+                continue
+            proc = cls(name=pc.name, config=pc.options)
+            self._processors.append(proc)
+
+        for sc in self._config.sinks:
+            cls = self._registry.get_sink(sc.type)
+            if not cls:
+                log.warning("sink_type_not_found", type=sc.type)
+                continue
+            sink = cls(name=sc.name, config=sc.options)
+            self._dispatchers.append(SinkDispatcher(sink, self._backend, self._config))
+
+    async def _ingest(self, record: Record) -> None:
+        """Called by sources. Runs processors and persists."""
+        for proc in self._processors:
+            result = await proc.process(record)
+            if result is None:
+                return
+            record = result
+
+        stored = StoredRecord(
+            record_id=record.record_id,
+            source=record.source,
+            record_type=record.record_type,
+            body=record.body,
+            metadata=record.metadata,
+            destination=record.destination,
+            created_at=record.timestamp,
+        )
+        self._backend.put(stored)
+        log.debug("record_persisted", record_id=record.record_id)
+
+    async def _dispatch_loop(self) -> None:
+        """Background loop: pull pending records from backend and deliver."""
+        while self._running:
+            try:
+                await self._dispatch_batch()
+            except asyncio.CancelledError:
+                break
+            except Exception:
+                log.exception("dispatch_loop_error")
+                await asyncio.sleep(1)
+
+    async def _dispatch_batch(self) -> None:
+        pending = self._backend.pending(limit=self._config.pipeline.batch_size)
+        retry = self._backend.ready_for_retry(limit=self._config.pipeline.batch_size)
+        records = pending + retry
+
+        if not records:
+            await asyncio.sleep(self._config.pipeline.flush_interval)
+            return
+
+        for stored in records:
+            if not self._running:
+                break
+            record = Record(
+                source=stored.source,
+                record_type=stored.record_type,
+                body=stored.body,
+                metadata=stored.metadata,
+                destination=stored.destination,
+                record_id=stored.record_id,
+                timestamp=stored.created_at,
+            )
+            for dispatcher in self._dispatchers:
+                await self._deliver_with_retry(dispatcher, record)
+
+        if len(records) < self._config.pipeline.batch_size:
+            await asyncio.sleep(self._config.pipeline.flush_interval)
+
+    async def _deliver_with_retry(self, dispatcher: SinkDispatcher, record: Record) -> None:
+        retry_cfg = None
+        for sc in self._config.sinks:
+            if sc.name == dispatcher._sink.name:
+                retry_cfg = sc.retry
+                break
+
+        max_attempts = retry_cfg.max_attempts if retry_cfg else 3
+        min_delay = retry_cfg.min_delay if retry_cfg else 1.0
+        max_delay = retry_cfg.max_delay if retry_cfg else 120.0
+
+        attempt = 0
+        while attempt < max_attempts:
+            attempt += 1
+            try:
+                result = await dispatcher.deliver(record)
+            except Exception as exc:
+                result = DeliveryResult(
+                    success=False,
+                    record_id=record.record_id,
+                    error=str(exc),
+                )
+
+            if result.success:
+                self._backend.ack(record.record_id)
+                return
+
+            if result.permanent:
+                self._backend.dead(record.record_id, result.error or "permanent failure")
+                return
+
+            backoff = min(min_delay * (2 ** (attempt - 1)), max_delay)
+            stored = self._backend.get(record.record_id)
+            if stored:
+                self._backend.nack(
+                    record.record_id,
+                    error=result.error or "unknown",
+                    retry_delay=backoff,
+                )
+
+            if attempt < max_attempts:
+                await asyncio.sleep(backoff)
+
+        self._backend.dead(
+            record.record_id,
+            f"max retries ({max_attempts}) exceeded: {result.error}",
+        )
+
+    @property
+    def sources(self) -> list[Source]:
+        return list(self._sources)
+
+    @property
+    def sinks(self) -> list[Sink]:
+        return [d._sink for d in self._dispatchers]
+
+    @property
+    def queue_size(self) -> int:
+        return self._queue.qsize()

src/daena/core/runtime.py

@@ -0,0 +1,90 @@
+from __future__ import annotations
+
+from daena.config import DaenaConfig
+from daena.core.pipeline import Pipeline
+from daena.logutil import get_logger
+from daena.persistence.backend import Backend
+from daena.persistence.lmdb import LMDBBackend
+from daena.plugins.loader import discover_plugins
+from daena.plugins.registry import PluginRegistry, get_registry
+
+log = get_logger("daena.runtime")
+
+
+class DaenaRuntime:
+    """Top-level service runtime.
+
+    Owns configuration, persistence backend, pipeline, and lifecycle.
+    Designed to run under asyncio with clean signal-based shutdown.
+    """
+
+    def __init__(self, config: DaenaConfig) -> None:
+        self._config = config
+        self._registry = get_registry()
+        self._backend: Backend | None = None
+        self._pipeline: Pipeline | None = None
+        self._running = False
+
+    @property
+    def config(self) -> DaenaConfig:
+        return self._config
+
+    @property
+    def registry(self) -> PluginRegistry:
+        return self._registry
+
+    @property
+    def backend(self) -> Backend | None:
+        return self._backend
+
+    @property
+    def pipeline(self) -> Pipeline | None:
+        return self._pipeline
+
+    async def start(self) -> None:
+        log.info("runtime_starting")
+
+        discover_plugins()
+        log.info(
+            "plugins_discovered",
+            sources=list(self._registry.list_sources().keys()),
+            processors=list(self._registry.list_processors().keys()),
+            sinks=list(self._registry.list_sinks().keys()),
+        )
+
+        persistence_cfg = self._config.persistence
+        if persistence_cfg.backend == "lmdb":
+            self._backend = LMDBBackend(
+                path=persistence_cfg.lmdb.path,
+                map_size=persistence_cfg.lmdb.map_size,
+                max_databases=persistence_cfg.lmdb.max_databases,
+            )
+        else:
+            raise ValueError(f"Unknown backend: {persistence_cfg.backend}")
+
+        self._backend.open()
+        log.info("backend_opened", backend=persistence_cfg.backend)
+
+        self._pipeline = Pipeline(self._config, self._backend, self._registry)
+        await self._pipeline.start()
+        self._running = True
+        log.info("runtime_started")
+
+    async def stop(self) -> None:
+        log.info("runtime_stopping")
+        self._running = False
+
+        if self._pipeline:
+            await self._pipeline.stop()
+
+        if self._backend:
+            self._backend.close()
+            log.info("backend_closed")
+
+        log.info("runtime_stopped")
+
+    async def reload(self) -> None:
+        log.info("runtime_reloading")
+        await self.stop()
+        await self.start()
+        log.info("runtime_reloaded")

src/daena/exceptions.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+
+class DaenaError(Exception):
+    """Base exception for all daena errors."""
+
+
+class ConfigurationError(DaenaError):
+    """Configuration loading or validation failed."""
+
+
+class BackendError(DaenaError):
+    """Persistence backend operation failed."""
+
+
+class PipelineError(DaenaError):
+    """Pipeline processing error."""
+
+
+class SinkError(DaenaError):
+    """Sink delivery failed."""
+
+
+class SinkPermanentError(SinkError):
+    """Non-retryable sink failure."""
+
+
+class SourceError(DaenaError):
+    """Source collection error."""
+
+
+class PluginError(DaenaError):
+    """Plugin loading or registration error."""
+
+
+class RecordInvalidError(DaenaError):
+    """Record validation failed."""

src/daena/logutil.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import logging
+import sys
+from pathlib import Path
+
+import structlog
+
+
+def setup_logging(
+    level: str = "info",
+    fmt: str = "json",
+    log_file: str | None = None,
+) -> None:
+    level_map: dict[str, int] = {
+        "debug": logging.DEBUG,
+        "info": logging.INFO,
+        "warning": logging.WARNING,
+        "error": logging.ERROR,
+        "critical": logging.CRITICAL,
+    }
+    log_level = level_map.get(level.lower(), logging.INFO)
+
+    processors: list[structlog.typing.Processor] = [
+        structlog.stdlib.add_log_level,
+        structlog.stdlib.add_logger_name,
+        structlog.processors.TimeStamper(fmt="iso"),
+        structlog.processors.StackInfoRenderer(),
+        structlog.dev.set_exc_info,
+        structlog.processors.format_exc_info,
+    ]
+
+    if fmt == "json":
+        processors.append(structlog.processors.JSONRenderer())
+    else:
+        processors.append(structlog.dev.ConsoleRenderer())
+
+    structlog.configure(
+        processors=processors,
+        wrapper_class=structlog.stdlib.BoundLogger,
+        context_class=dict,
+        logger_factory=structlog.stdlib.LoggerFactory(),
+        cache_logger_on_first_use=True,
+    )
+
+    logging.basicConfig(
+        format="%(message)s",
+        stream=sys.stdout,
+        level=log_level,
+        force=True,
+    )
+
+    if log_file:
+        p = Path(log_file)
+        p.parent.mkdir(parents=True, exist_ok=True)
+        handler = logging.FileHandler(str(p))
+        handler.setLevel(log_level)
+        handler.setFormatter(logging.Formatter("%(message)s"))
+        logging.getLogger().addHandler(handler)
+
+
+def get_logger(name: str | None = None) -> structlog.stdlib.BoundLogger:
+    return structlog.get_logger(name or __name__)

src/daena/persistence/backend.py

@@ -0,0 +1,88 @@
+from __future__ import annotations
+
+from collections.abc import Iterator
+from typing import Protocol
+
+from daena.persistence.models import RecordState, StoredRecord
+
+
+class Backend(Protocol):
+    """Abstract interface for persistence backends.
+
+    This Protocol defines the contract that all storage backends must fulfill.
+    The MVP implementation uses LMDB; future implementations may target
+    NATS JetStream, filesystem-backed stores, or cloud queues.
+    """
+
+    def open(self) -> None:
+        """Open or initialize the backend."""
+
+    def close(self) -> None:
+        """Close the backend and release resources."""
+
+    def put(self, record: StoredRecord) -> None:
+        """Store a new record."""
+
+    def get(self, record_id: str) -> StoredRecord | None:
+        """Retrieve a record by ID."""
+
+    def update_state(
+        self,
+        record_id: str,
+        state: RecordState,
+        last_error: str | None = None,
+        next_retry_at: str | None = None,
+    ) -> None:
+        """Update a record's delivery state."""
+
+    def pending(
+        self,
+        limit: int = 100,
+        older_than: str | None = None,
+    ) -> list[StoredRecord]:
+        """Return pending records for delivery, oldest first."""
+
+    def ready_for_retry(
+        self,
+        limit: int = 100,
+        now: str | None = None,
+    ) -> list[StoredRecord]:
+        """Return failed records whose retry time has arrived."""
+
+    def ack(self, record_id: str) -> None:
+        """Mark a record as successfully delivered."""
+
+    def nack(
+        self,
+        record_id: str,
+        error: str,
+        retry_delay: float = 10.0,
+    ) -> None:
+        """Mark delivery failed and schedule retry."""
+
+    def dead(self, record_id: str, error: str) -> None:
+        """Move a record to the dead-letter queue."""
+
+    def count(self, state: RecordState | None = None) -> int:
+        """Count records, optionally filtered by state."""
+
+    def list_by_state(
+        self,
+        state: RecordState,
+        limit: int = 100,
+        offset: int = 0,
+    ) -> list[StoredRecord]:
+        """List records by state with pagination."""
+
+    def iter_pending(self) -> Iterator[StoredRecord]:
+        """Iterate over all pending records (for recovery)."""
+
+    def delete_old(
+        self,
+        before: str,
+        state: RecordState | None = None,
+    ) -> int:
+        """Delete records older than a timestamp. Returns count."""
+
+    def cleanup(self) -> None:
+        """Perform backend-specific maintenance."""

src/daena/persistence/lmdb.py

@@ -0,0 +1,304 @@
+from __future__ import annotations
+
+import json
+from collections.abc import Iterator
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+import lmdb
+
+from daena.exceptions import BackendError
+from daena.persistence.models import RecordState, StoredRecord
+
+_REC_PREFIX = b"rec:"
+_IDX_PENDING = b"idx:pending:"
+_IDX_DELIVERED = b"idx:delivered:"
+_IDX_FAILED = b"idx:failed:"
+_IDX_DEAD = b"idx:dead:"
+_IDX_RETRY = b"idx:retry:"
+
+_PREFIX_MAP: dict[RecordState, bytes] = {
+    RecordState.pending: _IDX_PENDING,
+    RecordState.delivered: _IDX_DELIVERED,
+    RecordState.failed: _IDX_FAILED,
+    RecordState.dead: _IDX_DEAD,
+}
+
+
+class LMDBBackend:
+    """LMDB-based persistence backend using key-prefix partitioning.
+
+    Uses a single LMDB database with key prefixes to emulate named
+    sub-databases.
+
+    Key design:
+      - rec:{record_id}          -> serialized record JSON
+      - idx:pending:{ts}\x00{id}  -> "" (ordered FIFO queue)
+      - idx:delivered:{ts}\x00{id} -> ""
+      - idx:failed:{ts}\x00{id}   -> ""
+      - idx:dead:{ts}\x00{id}     -> ""
+      - idx:retry:{next}\x00{id}  -> ""
+    """
+
+    def __init__(
+        self,
+        path: str | Path = "/var/lib/daena/data",
+        map_size: int = 104_857_600,
+        max_databases: int = 16,
+    ) -> None:
+        self._path = Path(path)
+        self._map_size = map_size
+        self._max_databases = max_databases
+        self._env: lmdb.Environment | None = None
+
+    def _ensure_env(self) -> lmdb.Environment:
+        if self._env is None:
+            raise BackendError("Backend not opened")
+        return self._env
+
+    def open(self) -> None:
+        self._path.mkdir(parents=True, exist_ok=True)
+        self._env = lmdb.open(str(self._path), self._map_size, self._max_databases)
+
+    def close(self) -> None:
+        if self._env is not None:
+            self._env.close()
+            self._env = None
+
+    def _rec_key(self, record_id: str) -> bytes:
+        return _REC_PREFIX + record_id.encode()
+
+    def _idx_key(self, prefix: bytes, record_id: str) -> bytes:
+        ts = datetime.now(UTC).isoformat()
+        return prefix + f"{ts}\x00{record_id}".encode()
+
+    def _record_id_from_idx(self, key: bytes, prefix: bytes) -> str:
+        suffix = key[len(prefix) :].decode()
+        return suffix.split("\x00", 1)[1]
+
+    def _read_record(self, txn: lmdb.Transaction, record_id: str) -> StoredRecord | None:
+        raw = txn.get(self._rec_key(record_id))
+        if raw is None:
+            return None
+        return StoredRecord.from_dict(json.loads(raw))
+
+    def put(self, record: StoredRecord) -> None:
+        env = self._ensure_env()
+        rec_key = self._rec_key(record.record_id)
+        value = json.dumps(record.to_dict()).encode()
+        idx_key = self._idx_key(_IDX_PENDING, record.record_id)
+
+        with env.begin(write=True) as txn:
+            if txn.get(rec_key) is not None:
+                return
+            txn.put(rec_key, value)
+            txn.put(idx_key, b"")
+
+    def get(self, record_id: str) -> StoredRecord | None:
+        env = self._ensure_env()
+        key = self._rec_key(record_id)
+        with env.begin() as txn:
+            value = txn.get(key)
+        if value is None:
+            return None
+        return StoredRecord.from_dict(json.loads(value))
+
+    def update_state(
+        self,
+        record_id: str,
+        state: RecordState,
+        last_error: str | None = None,
+        next_retry_at: str | None = None,
+    ) -> None:
+        env = self._ensure_env()
+        rec_key = self._rec_key(record_id)
+        now = datetime.now(UTC).isoformat()
+
+        with env.begin(write=True) as txn:
+            raw = txn.get(rec_key)
+            if raw is None:
+                return
+            record_dict: dict[str, Any] = json.loads(raw)
+            record_dict["state"] = state.value
+            record_dict["updated_at"] = now
+            if last_error is not None:
+                record_dict["last_error"] = last_error
+            if next_retry_at is not None:
+                record_dict["next_retry_at"] = next_retry_at
+            txn.put(rec_key, json.dumps(record_dict).encode())
+
+            target_prefix = _PREFIX_MAP.get(state)
+            if target_prefix and state != RecordState.pending:
+                new_idx = self._idx_key(target_prefix, record_id)
+                txn.put(new_idx, b"")
+
+            if state == RecordState.failed and next_retry_at:
+                retry_key = _IDX_RETRY + f"{next_retry_at}\x00{record_id}".encode()
+                txn.put(retry_key, b"")
+
+    def pending(self, limit: int = 100, older_than: str | None = None) -> list[StoredRecord]:
+        _ = older_than
+        return self._list_by_state_prefix(_IDX_PENDING, RecordState.pending, limit)
+
+    def ready_for_retry(self, limit: int = 100, now: str | None = None) -> list[StoredRecord]:
+        env = self._ensure_env()
+        now_str = now or datetime.now(UTC).isoformat()
+        results: list[StoredRecord] = []
+        seen: set[str] = set()
+        with env.begin() as txn:
+            cursor = txn.cursor()
+            if not cursor.set_range(_IDX_RETRY):
+                return results
+            for key, _ in cursor:
+                if not key.startswith(_IDX_RETRY):
+                    break
+                record_id = self._record_id_from_idx(key, _IDX_RETRY)
+                retry_ts = key[len(_IDX_RETRY) :].decode().split("\x00", 1)[0]
+                if retry_ts > now_str:
+                    break
+                if record_id in seen:
+                    continue
+                seen.add(record_id)
+                record = self._read_record(txn, record_id)
+                if record and record.state == RecordState.failed:
+                    results.append(record)
+                    if len(results) >= limit:
+                        break
+        return results
+
+    def ack(self, record_id: str) -> None:
+        self.update_state(record_id, RecordState.delivered)
+
+    def nack(self, record_id: str, error: str, retry_delay: float = 10.0) -> None:
+        from datetime import timedelta
+
+        next_retry = (datetime.now(UTC) + timedelta(seconds=retry_delay)).isoformat()
+        self.update_state(
+            record_id,
+            RecordState.failed,
+            last_error=error,
+            next_retry_at=next_retry,
+        )
+
+    def dead(self, record_id: str, error: str) -> None:
+        self.update_state(record_id, RecordState.dead, last_error=error)
+
+    def count(self, state: RecordState | None = None) -> int:
+        env = self._ensure_env()
+        with env.begin() as txn:
+            if state is None:
+                cursor = txn.cursor()
+                count = 0
+                if cursor.set_range(_REC_PREFIX):
+                    for key, _ in cursor:
+                        if not key.startswith(_REC_PREFIX):
+                            break
+                        count += 1
+                return count
+            prefix = _PREFIX_MAP.get(state)
+            if prefix is None:
+                return 0
+            cursor = txn.cursor()
+            count = 0
+            if cursor.set_range(prefix):
+                for key, _ in cursor:
+                    if not key.startswith(prefix):
+                        break
+                    record_id = self._record_id_from_idx(key, prefix)
+                    record = self._read_record(txn, record_id)
+                    if record and record.state == state:
+                        count += 1
+            return count
+
+    def list_by_state(
+        self, state: RecordState, limit: int = 100, offset: int = 0
+    ) -> list[StoredRecord]:
+        prefix = _PREFIX_MAP.get(state)
+        if prefix is None:
+            return []
+        return self._list_by_state_prefix(prefix, state, limit, offset)
+
+    def iter_pending(self) -> Iterator[StoredRecord]:
+        seen: set[str] = set()
+        env = self._ensure_env()
+        with env.begin() as txn:
+            cursor = txn.cursor()
+            if cursor.set_range(_IDX_PENDING):
+                for key, _ in cursor:
+                    if not key.startswith(_IDX_PENDING):
+                        break
+                    record_id = self._record_id_from_idx(key, _IDX_PENDING)
+                    if record_id in seen:
+                        continue
+                    seen.add(record_id)
+                    record = self._read_record(txn, record_id)
+                    if record and record.state == RecordState.pending:
+                        yield record
+
+    def delete_old(self, before: str, state: RecordState | None = None) -> int:
+        env = self._ensure_env()
+        removed = 0
+
+        prefixes = [_IDX_DELIVERED, _IDX_DEAD]
+        if state:
+            p = _PREFIX_MAP.get(state)
+            prefixes = [p] if p else []
+
+        with env.begin(write=True) as txn:
+            for prefix in prefixes:
+                cursor = txn.cursor()
+                if not cursor.set_range(prefix):
+                    continue
+                seen_del: set[str] = set()
+                for key, _ in cursor:
+                    if not key.startswith(prefix):
+                        break
+                    suffix = key[len(prefix) :].decode()
+                    ts_part = suffix.split("\x00", 1)[0]
+                    if ts_part > before:
+                        break
+                    record_id = suffix.split("\x00", 1)[1]
+                    if record_id in seen_del:
+                        continue
+                    seen_del.add(record_id)
+                    txn.delete(self._rec_key(record_id))
+                    txn.delete(key)
+                    removed += 1
+        return removed
+
+    def cleanup(self) -> None:
+        pass
+
+    def _list_by_state_prefix(
+        self,
+        prefix: bytes,
+        state: RecordState,
+        limit: int = 100,
+        offset: int = 0,
+    ) -> list[StoredRecord]:
+        seen: set[str] = set()
+        env = self._ensure_env()
+        results: list[StoredRecord] = []
+        with env.begin() as txn:
+            cursor = txn.cursor()
+            if not cursor.set_range(prefix):
+                return results
+            skipped = 0
+            for key, _ in cursor:
+                if not key.startswith(prefix):
+                    break
+                record_id = self._record_id_from_idx(key, prefix)
+                if record_id in seen:
+                    continue
+                seen.add(record_id)
+                record = self._read_record(txn, record_id)
+                if record is None or record.state != state:
+                    continue
+                if skipped < offset:
+                    skipped += 1
+                    continue
+                results.append(record)
+                if len(results) >= limit:
+                    break
+        return results

src/daena/persistence/models.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import uuid
+from datetime import UTC, datetime
+from enum import StrEnum
+from typing import Any
+
+
+class RecordState(StrEnum):
+    pending = "pending"
+    delivering = "delivering"
+    delivered = "delivered"
+    failed = "failed"
+    dead = "dead"
+
+
+class StoredRecord:
+    """A record as stored in the persistence backend."""
+
+    def __init__(
+        self,
+        record_id: str | None = None,
+        source: str = "",
+        record_type: str = "",
+        body: dict[str, Any] | None = None,
+        metadata: dict[str, Any] | None = None,
+        state: RecordState = RecordState.pending,
+        created_at: str | None = None,
+        updated_at: str | None = None,
+        retry_count: int = 0,
+        next_retry_at: str | None = None,
+        last_error: str | None = None,
+        destination: str | None = None,
+    ) -> None:
+        self.record_id = record_id or str(uuid.uuid4())
+        self.source = source
+        self.record_type = record_type
+        self.body = body or {}
+        self.metadata = metadata or {}
+        self.state = state
+        self.created_at = created_at or datetime.now(UTC).isoformat()
+        self.updated_at = updated_at or self.created_at
+        self.retry_count = retry_count
+        self.next_retry_at = next_retry_at
+        self.last_error = last_error
+        self.destination = destination
+
+    def to_dict(self) -> dict[str, Any]:
+        return {
+            "record_id": self.record_id,
+            "source": self.source,
+            "record_type": self.record_type,
+            "body": self.body,
+            "metadata": self.metadata,
+            "state": self.state.value,
+            "created_at": self.created_at,
+            "updated_at": self.updated_at,
+            "retry_count": self.retry_count,
+            "next_retry_at": self.next_retry_at,
+            "last_error": self.last_error,
+            "destination": self.destination,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> StoredRecord:
+        return cls(
+            record_id=data.get("record_id"),
+            source=data.get("source", ""),
+            record_type=data.get("record_type", ""),
+            body=data.get("body", {}),
+            metadata=data.get("metadata", {}),
+            state=RecordState(data.get("state", "pending")),
+            created_at=data.get("created_at"),
+            updated_at=data.get("updated_at"),
+            retry_count=data.get("retry_count", 0),
+            next_retry_at=data.get("next_retry_at"),
+            last_error=data.get("last_error"),
+            destination=data.get("destination"),
+        )

src/daena/pipeline/base.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
+from typing import Any
+
+from daena.pipeline.record import Record
+
+
+@dataclass
+class DeliveryResult:
+    success: bool
+    record_id: str
+    error: str | None = None
+    permanent: bool = False
+
+
+class Source(ABC):
+    """Base class for all record sources.
+
+    Sources collect or receive data and push Records into the pipeline.
+    """
+
+    name: str
+
+    def __init__(self, name: str, config: dict[str, Any] | None = None) -> None:
+        self.name = name
+        self.config = config or {}
+
+    @abstractmethod
+    async def start(self) -> None:
+        """Start collecting records. Called during service startup."""
+
+    @abstractmethod
+    async def stop(self) -> None:
+        """Stop collecting. Called during shutdown."""
+
+
+class Processor(ABC):
+    """Base class for record processors.
+
+    Processors transform, filter, or enrich Records passing through the pipeline.
+    Return None to drop the record.
+    """
+
+    name: str
+
+    def __init__(self, name: str, config: dict[str, Any] | None = None) -> None:
+        self.name = name
+        self.config = config or {}
+
+    @abstractmethod
+    async def process(self, record: Record) -> Record | None: ...
+
+
+class Sink(ABC):
+    """Base class for record sinks.
+
+    Sinks deliver Records to external destinations.
+    """
+
+    name: str
+
+    def __init__(self, name: str, config: dict[str, Any] | None = None) -> None:
+        self.name = name
+        self.config = config or {}
+
+    @abstractmethod
+    async def deliver(self, record: Record) -> DeliveryResult: ...
+
+    async def start(self) -> None:
+        return
+
+    async def stop(self) -> None:
+        return

src/daena/pipeline/processors/passthrough.py

@@ -0,0 +1,11 @@
+from __future__ import annotations
+
+from daena.pipeline.base import Processor
+from daena.pipeline.record import Record
+
+
+class PassthroughProcessor(Processor):
+    """Passes records through without modification."""
+
+    async def process(self, record: Record) -> Record | None:
+        return record

src/daena/pipeline/record.py

@@ -0,0 +1,19 @@
+from __future__ import annotations
+
+import uuid
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from typing import Any
+
+
+@dataclass
+class Record:
+    """In-memory representation of a collected event or log."""
+
+    source: str
+    record_type: str
+    body: dict[str, Any]
+    metadata: dict[str, Any] = field(default_factory=dict)
+    destination: str | None = None
+    record_id: str = field(default_factory=lambda: str(uuid.uuid4()))
+    timestamp: str = field(default_factory=lambda: datetime.now(UTC).isoformat())

src/daena/pipeline/sinks/dummy.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from daena.pipeline.base import DeliveryResult, Sink
+from daena.pipeline.record import Record
+
+
+class DummySink(Sink):
+    """Logs delivered records without sending anywhere.
+
+    Useful for testing and development.
+    """
+
+    async def deliver(self, record: Record) -> DeliveryResult:
+        return DeliveryResult(success=True, record_id=record.record_id)

src/daena/pipeline/sinks/http.py

@@ -0,0 +1,85 @@
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+
+from daena.pipeline.base import DeliveryResult, Sink
+from daena.pipeline.record import Record
+
+
+class HTTPSink(Sink):
+    """Delivers records to a remote HTTP endpoint.
+
+    Supports configurable URL, method, headers, timeouts, and batching.
+    """
+
+    def __init__(self, name: str, config: dict[str, Any] | None = None) -> None:
+        super().__init__(name, config)
+        cfg = config or {}
+        self._url: str = cfg.get("url", "http://localhost:9999/ingest")
+        self._method: str = cfg.get("method", "POST").upper()
+        self._headers: dict[str, str] = cfg.get("headers", {})
+        self._timeout: float = float(cfg.get("timeout", 30.0))
+        self._client: httpx.AsyncClient | None = None
+
+    async def start(self) -> None:
+        self._client = httpx.AsyncClient(
+            timeout=httpx.Timeout(self._timeout),
+            headers=self._headers,
+        )
+
+    async def stop(self) -> None:
+        if self._client:
+            await self._client.aclose()
+            self._client = None
+
+    async def deliver(self, record: Record) -> DeliveryResult:
+        if self._client is None:
+            return DeliveryResult(
+                success=False,
+                record_id=record.record_id,
+                error="HTTP sink not initialized",
+            )
+
+        payload = {
+            "id": record.record_id,
+            "source": record.source,
+            "type": record.record_type,
+            "timestamp": record.timestamp,
+            "body": record.body,
+            "metadata": record.metadata,
+        }
+
+        try:
+            response = await self._client.request(
+                method=self._method,
+                url=self._url,
+                json=payload,
+            )
+            if response.is_success:
+                return DeliveryResult(success=True, record_id=record.record_id)
+            if 400 <= response.status_code < 500 and response.status_code not in (429,):
+                return DeliveryResult(
+                    success=False,
+                    record_id=record.record_id,
+                    error=f"HTTP {response.status_code}: {response.text[:200]}",
+                    permanent=True,
+                )
+            return DeliveryResult(
+                success=False,
+                record_id=record.record_id,
+                error=f"HTTP {response.status_code}: {response.text[:200]}",
+            )
+        except httpx.TimeoutException as exc:
+            return DeliveryResult(
+                success=False,
+                record_id=record.record_id,
+                error=f"Timeout: {exc}",
+            )
+        except httpx.RequestError as exc:
+            return DeliveryResult(
+                success=False,
+                record_id=record.record_id,
+                error=str(exc),
+            )

src/daena/pipeline/sources/dummy.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from typing import Any
+
+from daena.pipeline.base import Source
+from daena.pipeline.record import Record
+
+
+class DummySource(Source):
+    """Generates test records at a configurable interval.
+
+    Useful for development, testing, and demo purposes.
+    """
+
+    def __init__(self, name: str, config: dict[str, Any] | None = None) -> None:
+        super().__init__(name, config)
+        self._interval: float = float(config.get("interval", 5.0) if config else 5.0)
+        self._message: str = (
+            config.get("message", "daena test event") if config else "daena test event"
+        )
+        self._task: asyncio.Task[None] | None = None
+        self._running = False
+        self._on_record: callable | None = None
+
+    def set_callback(self, cb: callable) -> None:
+        self._on_record = cb
+
+    async def start(self) -> None:
+        self._running = True
+        self._task = asyncio.create_task(self._emit_loop())
+
+    async def stop(self) -> None:
+        self._running = False
+        if self._task:
+            self._task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._task
+
+    async def _emit_loop(self) -> None:
+        counter = 0
+        while self._running:
+            counter += 1
+            record = Record(
+                source=self.name,
+                record_type="test",
+                body={
+                    "message": f"{self._message} #{counter}",
+                    "counter": counter,
+                },
+                metadata={"dummy": True},
+            )
+            if self._on_record:
+                await self._on_record(record)
+            await asyncio.sleep(self._interval)

src/daena/plugins/loader.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from importlib.metadata import entry_points
+
+from daena.exceptions import PluginError
+from daena.plugins.registry import get_registry
+
+SOURCE_EP = "daena.sources"
+PROCESSOR_EP = "daena.processors"
+SINK_EP = "daena.sinks"
+
+
+def discover_plugins() -> None:
+    """Discover plugins via package entry points and register them."""
+    registry = get_registry()
+    errors: list[str] = []
+
+    for ep in entry_points(group=SOURCE_EP):
+        try:
+            cls = ep.load()
+            registry.register_source(ep.name, cls)
+        except Exception as exc:
+            errors.append(f"source {ep.name}: {exc}")
+
+    for ep in entry_points(group=PROCESSOR_EP):
+        try:
+            cls = ep.load()
+            registry.register_processor(ep.name, cls)
+        except Exception as exc:
+            errors.append(f"processor {ep.name}: {exc}")
+
+    for ep in entry_points(group=SINK_EP):
+        try:
+            cls = ep.load()
+            registry.register_sink(ep.name, cls)
+        except Exception as exc:
+            errors.append(f"sink {ep.name}: {exc}")
+
+    if errors:
+        raise PluginError("Plugin loading encountered errors:\n  " + "\n  ".join(errors))

src/daena/plugins/registry.py

@@ -0,0 +1,55 @@
+from __future__ import annotations
+
+
+class PluginRegistry:
+    """Registry for plugin factories.
+
+    Maintains separate maps for sources, processors, and sinks.
+    Each entry maps plugin type name → factory callable.
+    """
+
+    def __init__(self) -> None:
+        self._sources: dict[str, type] = {}
+        self._processors: dict[str, type] = {}
+        self._sinks: dict[str, type] = {}
+
+    def register_source(self, name: str, cls: type) -> None:
+        self._sources[name] = cls
+
+    def register_processor(self, name: str, cls: type) -> None:
+        self._processors[name] = cls
+
+    def register_sink(self, name: str, cls: type) -> None:
+        self._sinks[name] = cls
+
+    def get_source(self, name: str) -> type | None:
+        return self._sources.get(name)
+
+    def get_processor(self, name: str) -> type | None:
+        return self._processors.get(name)
+
+    def get_sink(self, name: str) -> type | None:
+        return self._sinks.get(name)
+
+    def list_sources(self) -> dict[str, type]:
+        return dict(self._sources)
+
+    def list_processors(self) -> dict[str, type]:
+        return dict(self._processors)
+
+    def list_sinks(self) -> dict[str, type]:
+        return dict(self._sinks)
+
+    def all_plugins(self) -> dict[str, dict[str, type]]:
+        return {
+            "sources": self.list_sources(),
+            "processors": self.list_processors(),
+            "sinks": self.list_sinks(),
+        }
+
+
+_registry = PluginRegistry()
+
+
+def get_registry() -> PluginRegistry:
+    return _registry

tests/conftest.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+from collections.abc import Iterator
+from pathlib import Path
+
+import pytest
+import yaml
+
+from daena.config import DaenaConfig
+from daena.persistence.lmdb import LMDBBackend
+from daena.persistence.models import StoredRecord
+
+
+@pytest.fixture
+def tmp_db_path(tmp_path: Path) -> Path:
+    return tmp_path / "lmdb"
+
+
+@pytest.fixture
+def lmdb_backend(tmp_db_path: Path) -> Iterator[LMDBBackend]:
+    backend = LMDBBackend(
+        path=str(tmp_db_path),
+        map_size=10_485_760,
+    )
+    backend.open()
+    yield backend
+    backend.close()
+
+
+@pytest.fixture
+def sample_record() -> StoredRecord:
+    return StoredRecord(
+        source="test_source",
+        record_type="test",
+        body={"message": "hello"},
+        metadata={"env": "test"},
+    )
+
+
+@pytest.fixture
+def minimal_config() -> DaenaConfig:
+    return DaenaConfig()
+
+
+@pytest.fixture
+def config_file(tmp_path: Path) -> Path:
+    path = tmp_path / "daena.yaml"
+    data = {
+        "data_dir": "/tmp/daena_data",
+        "logging": {"level": "debug", "format": "json"},
+        "sources": [
+            {
+                "name": "test",
+                "type": "dummy",
+                "enabled": True,
+                "options": {"interval": 1.0},
+            }
+        ],
+        "sinks": [
+            {
+                "name": "output",
+                "type": "dummy",
+                "enabled": True,
+                "options": {},
+            }
+        ],
+    }
+    with open(path, "w") as f:
+        yaml.dump(data, f)
+    return path

tests/test_config.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from daena.config import DaenaConfig, load_config
+
+
+def test_minimal_config_has_defaults() -> None:
+    config = DaenaConfig()
+    assert config.data_dir == "/var/lib/daena"
+    assert config.logging.level == "info"
+    assert config.api.http.port == 8642
+    assert config.pipeline.batch_size == 100
+
+
+def test_load_config_from_file(config_file: Path) -> None:
+    config = load_config(str(config_file))
+    assert config.data_dir == "/tmp/daena_data"
+    assert config.logging.level == "debug"
+    assert len(config.sources) == 1
+    assert config.sources[0].name == "test"
+    assert config.sources[0].type == "dummy"
+
+
+def test_load_config_nonexistent_file() -> None:
+    try:
+        load_config("/nonexistent/daena.yaml")
+        raise AssertionError("Expected FileNotFoundError")  # noqa: B011
+    except FileNotFoundError:
+        pass
+
+
+def test_config_can_set_persistence_backend() -> None:
+    config = DaenaConfig()
+    assert config.persistence.backend == "lmdb"
+    assert config.persistence.lmdb.map_size == 104_857_600
+
+
+def test_config_source_options(minimal_config: DaenaConfig) -> None:
+    cfg = minimal_config
+    assert cfg.sources == []
+    assert cfg.processors == []
+    assert cfg.sinks == []

tests/test_persistence.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+from daena.persistence.lmdb import LMDBBackend
+from daena.persistence.models import RecordState, StoredRecord
+
+
+def test_put_and_get(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None:
+    lmdb_backend.put(sample_record)
+    retrieved = lmdb_backend.get(sample_record.record_id)
+    assert retrieved is not None
+    assert retrieved.record_id == sample_record.record_id
+    assert retrieved.source == "test_source"
+    assert retrieved.body == {"message": "hello"}
+
+
+def test_pending_returns_new_records(
+    lmdb_backend: LMDBBackend, sample_record: StoredRecord
+) -> None:
+    lmdb_backend.put(sample_record)
+    pending = lmdb_backend.pending(limit=10)
+    assert len(pending) == 1
+    assert pending[0].record_id == sample_record.record_id
+
+
+def test_ack(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None:
+    lmdb_backend.put(sample_record)
+    lmdb_backend.ack(sample_record.record_id)
+    retrieved = lmdb_backend.get(sample_record.record_id)
+    assert retrieved is not None
+    assert retrieved.state == RecordState.delivered
+    pending = lmdb_backend.pending()
+    assert all(r.record_id != sample_record.record_id for r in pending)
+
+
+def test_nack(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None:
+    lmdb_backend.put(sample_record)
+    lmdb_backend.nack(sample_record.record_id, error="test error", retry_delay=1.0)
+    retrieved = lmdb_backend.get(sample_record.record_id)
+    assert retrieved is not None
+    assert retrieved.state == RecordState.failed
+    assert retrieved.last_error == "test error"
+    assert retrieved.next_retry_at is not None
+
+
+def test_dead(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None:
+    lmdb_backend.put(sample_record)
+    lmdb_backend.dead(sample_record.record_id, error="permanent failure")
+    retrieved = lmdb_backend.get(sample_record.record_id)
+    assert retrieved is not None
+    assert retrieved.state == RecordState.dead
+
+
+def test_count(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None:
+    assert lmdb_backend.count() == 0
+    lmdb_backend.put(sample_record)
+    assert lmdb_backend.count() == 1
+    assert lmdb_backend.count(RecordState.pending) == 1
+
+
+def test_list_by_state(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None:
+    lmdb_backend.put(sample_record)
+    pending = lmdb_backend.list_by_state(RecordState.pending)
+    assert len(pending) == 1
+    lmdb_backend.ack(sample_record.record_id)
+    delivered = lmdb_backend.list_by_state(RecordState.delivered)
+    assert len(delivered) == 1
+
+
+def test_delete_old(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None:
+    lmdb_backend.put(sample_record)
+    lmdb_backend.ack(sample_record.record_id)
+    before = "2100-01-01T00:00:00"
+    removed = lmdb_backend.delete_old(before, state=RecordState.delivered)
+    assert removed == 1
+    assert lmdb_backend.get(sample_record.record_id) is None
+
+
+def test_iter_pending(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None:
+    lmdb_backend.put(sample_record)
+    records = list(lmdb_backend.iter_pending())
+    assert len(records) == 1
+    assert records[0].record_id == sample_record.record_id
+
+
+def test_update_state(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None:
+    lmdb_backend.put(sample_record)
+    lmdb_backend.update_state(sample_record.record_id, RecordState.failed, last_error="oops")
+    retrieved = lmdb_backend.get(sample_record.record_id)
+    assert retrieved is not None
+    assert retrieved.state == RecordState.failed
+    assert retrieved.last_error == "oops"
+
+
+def test_stored_record_to_from_dict(sample_record: StoredRecord) -> None:
+    d = sample_record.to_dict()
+    restored = StoredRecord.from_dict(d)
+    assert restored.record_id == sample_record.record_id
+    assert restored.source == sample_record.source
+    assert restored.body == sample_record.body
+    assert restored.state == sample_record.state
+
+
+def test_ready_for_retry(lmdb_backend: LMDBBackend, sample_record: StoredRecord) -> None:
+    lmdb_backend.put(sample_record)
+    lmdb_backend.nack(sample_record.record_id, "retry later", retry_delay=0)
+    ready = lmdb_backend.ready_for_retry(limit=10)
+    assert len(ready) == 1

tests/test_pipeline.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import pytest
+
+from daena.pipeline.base import DeliveryResult, Processor
+from daena.pipeline.processors.passthrough import PassthroughProcessor
+from daena.pipeline.record import Record
+from daena.pipeline.sinks.dummy import DummySink
+from daena.pipeline.sinks.http import HTTPSink
+from daena.pipeline.sources.dummy import DummySource
+
+
+def test_dummy_source_config() -> None:
+    source = DummySource(
+        name="test",
+        config={"interval": 1.0, "message": "hello"},
+    )
+    assert source.name == "test"
+    assert source._interval == 1.0
+    assert source._message == "hello"
+
+
+@pytest.mark.asyncio
+async def test_dummy_sink_delivery() -> None:
+    sink = DummySink(name="test")
+    record = Record(source="test", record_type="test", body={"key": "value"})
+    result = await sink.deliver(record)
+    assert isinstance(result, DeliveryResult)
+    assert result.success is True
+    assert result.record_id == record.record_id
+
+
+@pytest.mark.asyncio
+async def test_passthrough_processor() -> None:
+    proc = PassthroughProcessor(name="test")
+    record = Record(source="test", record_type="test", body={"key": "value"})
+    result = await proc.process(record)
+    assert result is not None
+    assert result.record_id == record.record_id
+    assert result.body == {"key": "value"}
+
+
+def test_http_sink_requires_url() -> None:
+    sink = HTTPSink(name="test", config={"url": "http://example.com/ingest"})
+    assert sink._url == "http://example.com/ingest"
+    assert sink._method == "POST"
+
+
+def test_record_has_id() -> None:
+    r1 = Record(source="s1", record_type="t1", body={})
+    r2 = Record(source="s2", record_type="t2", body={})
+    assert r1.record_id != r2.record_id
+    assert r1.timestamp is not None
+
+
+@pytest.mark.asyncio
+async def test_processor_drop_record() -> None:
+    class DropProcessor(Processor):
+        async def process(self, record: Record) -> Record | None:  # noqa: ARG002
+            return None
+
+    proc = DropProcessor(name="dropper")
+    record = Record(source="test", record_type="test", body={})
+    result = await proc.process(record)
+    assert result is None