chrono-daemon

chrono-daemon is a small Python library for long-running async components whose time can be replayed deterministically. It wraps anyio with four primitives:

• Channel[T]: typed single-producer / single-consumer queues.
• Clock: real time with WallClock, virtual time with SimClock.
• Daemon: a lifecycle unit with on_start, run, and on_stop.
• Supervisor: a structured-concurrency root for hosting daemons.

The practical payoff is simple: production daemons sleep on ctx.clock; tests swap in SimClock and advance seconds of work without waiting for wall time.

Quick Example

from chrono_daemon import Channel, Context, SimClock, Supervisor, daemon, open_channel

@daemon
async def producer(ctx: Context, out: Channel[int]) -> None:
    for i in range(10):
        await ctx.clock.sleep(0.1)
        await out.send.send(i)
    await out.send.aclose()

@daemon
async def consumer(ctx: Context, src: Channel[int]) -> None:
    async for item in src.recv:
        ctx.logger.info("got %d", item)

async def main() -> None:
    ch: Channel[int] = open_channel(maxsize=4)
    clock = SimClock()
    async with Supervisor(clock=clock) as sup:
        sup.add(producer(ch))
        sup.add(consumer(ch))
        await clock.advance(1.5)  # replay 1.5 s of work immediately

Why Use It

• Time-dependent async code is testable without sleeps, polling, or fake task schedulers.
• Wiring is explicit. Every edge is a named SPSC channel, so ownership and backpressure stay visible.
• Lifecycle behavior is structured. Daemons get startup, shutdown, logging, cancellation, and error policy in one place.
• The core runtime is small: pure Python, anyio underneath, and no runtime dependency beyond anyio.

Core API

See docs/concepts.md for details.

Scope

chrono-daemon is for in-process async systems where explicit wiring and deterministic time matter: evaluation loops, agent pipelines, robotics-style control mocks, and testable service internals.

It is intentionally not a topic broker, service registry, RPC framework, parameter server, CLI launcher, network runtime, or continuous-time numerical simulator.

Install / dev

uv sync --dev --extra zmq
make check        # ruff + pyrefly
make test         # pytest on asyncio + trio
make all          # format + check + test

Python 3.11+. Only runtime dependency is anyio>=4; --extra zmq installs the optional transport dependencies needed for the full test suite. Remote ZMQ channels are optional:

pip install "chrono-daemon[zmq]"

Where to look next

• docs/concepts.md: what each primitive is and the invariants the test suite pins.
• docs/adr/: why each decision looks the way it does (Topic-less, on-error-shutdown by default, anyio-only).
• docs/recipes.md: patterns kept off the core surface but importable under chrono_daemon.recipes.*, grouped as routing, coordination, state, buffering, and hosting helpers under src/chrono_daemon/recipes/.
• examples/: runnable System 2/1/0 demos and notes.

Status

Early. In-process channels are the default and keep zero runtime dependencies beyond anyio. ZMQ remote channels are available as an optional extra (chrono-daemon[zmq]) for asyncio-backed deployments.