Tasks

The task mixin and manager, the spool, executor, scheduler and stores.

Task capability: the task backbone as a mixin over the base server (◆D22, D16).

TaskMixin is a capability composed over BaseServer AFTER StorageMixin (it needs server.storage) and BEFORE BaseServer (the loop hook must sit inside the D16 chain so it wraps Lifespan). Its cooperative __init__ peels tasks= and forwards everything else down the chain (mirroring StorageMixin.__init__); a composition WITHOUT the mixin has no tasks attribute at all.

tasks= is the on/off switch AND the tuning carrier: absent or truthy → the TaskManager is armed and the worker loop runs at startup; tasks=False → no manager, and the lifespan passes straight through (no loop); a dict — {"enabled": ..., "tick_seconds": ..., "mount": ...} — peels enabled and stashes the rest as tasks_config for the manager to apply. Storage/session are on when their mixin is present, so tasks matches: a composed AsgiServer runs the worker loop out of the box.

TaskConfigElements is the config_grammar companion (D16: the element is declared by the class that peels the kwarg). The site recipe composes it explicitly (config/elements.py) so server(...).tasks(enabled=, tick_seconds=, mount=) lifts to this mixin’s tasks= kwarg through element_kwargs — which enforces strict-unknown children (ConfigError).

The manager is built LAZILY on first tasks access, never in __init__: the cooperative chain runs TaskMixin.__init__ (composed after StorageMixin) from INSIDE StorageMixin.__init__, before that mixin has assigned server.storage — so the manager, which opens its spool over server.storage, cannot be built until the whole chain has completed. Lazy construction defers it to the first use (the lifespan hook, or a caller reaching server.tasks), when the server is fully live.

The loop is server-owned, hooked in __call__ — lifespan.py is NEVER touched (ratified). __call__ intercepts the lifespan scope exactly like CommunicationMixin.__call__: pre-receive lifespan.startup, manager.start(), replay the startup down super().__call__ (so the base Lifespan still runs the app hooks and acks the protocol), and await manager.stop() in finally when the protocol completes at shutdown. Every other scope — and the disabled case — passes straight through.

class genro_asgi.tasks.mixin.TaskMixin(**kwargs)

Bases: object

Task capability mixin, composed after StorageMixin and before the base.

Constructor kwargs peeled here: tasks — the on/off switch (default on) or the {enabled, tick_seconds, mount} tuning dict lifted from the tasks() config element. When enabled, builds the TaskManager and starts/stops its worker loop around the ASGI lifespan protocol.

Parameters:

kwargs (Any)

config_grammar

alias of TaskConfigElements

property tasks: TaskManager

The task manager this server owns (built on first access); disabled is an error.

Lazy: the manager opens its spool over server.storage, which is not yet set while the cooperative __init__ chain is still running — so the first access after the server is live builds it.

property tasks_enabled: bool

Whether the task backbone is armed (tasks not False at init).

property tasks_config: dict[str, Any]

The tuning peeled from a tasks= dict (tick_seconds, mount).

Empty when tasks= was a plain switch; the TaskManager applies it at build time.

class genro_asgi.tasks.mixin.TaskConfigElements

Bases: object

Config elements owned by TaskMixin (its config_grammar companion).

Recipes compose this mixin explicitly (config/elements.py) — no auto-discovery. element_kwargs validates children against it (strict-unknown → ConfigError).

tasks = <genro_builders.builder._decorators._DeclarativeMarker object>

TaskManager — the server-owned task backbone (spool + executor + hub, ◆D22).

The spool and the executor are leaf pieces: the spool is a folder model on storage, the executor runs one resolved task. The manager is what makes them LIVE — a single object owned by the server (dual relationship: self.server) that drives the fire-and-forget worker loop for the whole process. It owns:

• spool — the LocalTaskExecutor’s own spool over server.storage (the stateless seam: the same storage yields an equivalent spool, so the manager reuses the executor’s rather than opening a second);

• executor — the LocalTaskExecutor bound to the live server;

• hub — the in-memory EventHub (the live progress courier for Phase 6);

• scheduler — the TaskScheduler (the recurring loop, over task_store);

• task_store — the FileTaskStore (persistent schedules, secure when keys);

• worker_id — the single logical worker of the mono-process core ("local").

start()/stop() are the lifecycle the server’s lifespan hook calls (TaskMixin.__call__): start launches three tasks on the running loop — the fire-and-forget _worker_loop, the scheduler tick loop, and the built-in _purge_loop (session GC) — and stop cancels/awaits them all (in-flight executions are their own tasks and are left to finish). Each loop mirrors the same shape: a failing pass is logged and never kills the loop.

The worker loop is FIRE-AND-FORGET on the event loop: it polls list_pending, assign``s each task to ``worker_id, and launches executor.execute as its own task. The D2 thread pool stays reserved for the blocking handler BODY inside execute (via server.run_sync) — the loop itself never blocks the pool. The session-purge loop is a plain internal job (NOT a scheduler store record — it is no @route(task=...), so it has no registry callable): it calls server.session_store.purge_expired() through run_sync every PURGE_SECONDS. Distributed dispatch (worker processes, a batch commander) is out of scope (D22).

class genro_asgi.tasks.manager.TaskManager(server)

Bases: object

Owns the spool/executor/hub and drives the fire-and-forget worker loop.

Note

Bound to the live server (dual relationship: self.server). The loop task and its event loop are held on the instance; running reports whether the loop task is live.

Parameters:

server (BaseServer)

__init__(server)

Bind to the live server and build the executor, hub, store and scheduler.

server.tasks_config (the tuning dict the mixin peeled from a tasks= dict, empty otherwise) is applied here: mount overrides the store’s by-keys choice, tick_seconds retunes the scheduler.

Parameters:

server (BaseServer)

Return type:

None

property spool: Any

The task spool (the executor’s own, over server.storage).

property running: bool

Whether the worker loop task is currently live.

start()

Launch the worker, scheduler and purge loops (lifespan startup).

Return type:

None

async stop()

Cancel the worker, purge and scheduler loops (lifespan shutdown).

In-flight executions are their own tasks and are left to finish.

Return type:

None

publish_progress(task_id, data)

Write a progress snapshot AND publish it live, in one paired call.

The pairing rule of the push channel: every progress.json write also fans out on the hub, keyed by the descriptor’s launching session_id (None = no push channel, the spool write still happens). The task must be ACTIVE on this manager’s worker — writing progress anywhere else would create a stray spool folder.

Raises:

LookupError – if the task is unknown or not in the active state.

Return type:

None

Parameters:

• task_id (str)

• data (dict[str, Any])

TaskSpool — the file spool of batch tasks (folder-move model, ◆D22).

A batch task is a FOLDER on storage; its STATE is its POSITION in the tree. The sender (an app handler, or the scheduler) creates the folder under pending/; the manager MOVES it into the assigned worker’s folder; the worker runs it and MOVES it to terminated/ or aborted/. A move is the only state transition — atomic on one mount (LocalStorageNode.move_to, a plain Path.rename).

Layout (all on the plain site mount — these are service data, never secrets, so NO encryption; distinct from the task STORE, which goes secure when keys are installed):

batches/
  pending/<task_id>/            sender writes here; the manager polls it
  active/<worker_id>/<task_id>/ the manager moves it here to assign it
  terminated/<task_id>/         completed
  aborted/<task_id>/            interrupted (user cancel / error / crash)

Inside a task folder:

descriptor.json   the TaskDescriptor (identity + node_path + status + outcome)
params.pkl        the call kwargs (pickled: may carry Python objects)
progress.json     the worker's latest progress snapshot (single-writer: the worker)
cancel            marker file: present == the user requested a stop
result            the batch result (written by the worker on completion)

A batch_id is terminal (§5.7): it never resumes or relaunches itself. An orphan (a task left under active/ whose worker died) is settled aborted at boot; relaunch = a NEW id. There are no locks: safety is structural — one single writer per state directory (sender on pending, one worker per active subfolder), and a claim is an atomic rename.

The spool is the shared object the sender, the manager and the worker all use, each with the methods that concern it. Storage is synchronous by construction (core 1b): async callers dispatch blocking spool calls via server.run_sync.

class genro_asgi.tasks.spool.TaskSpool(storage)

Bases: object

The file spool: create in pending, move between states, read progress/cancel.

Note

Holds the shared LocalStorage (dual relationship: self.storage), never raw paths. All I/O is plain synchronous storage calls.

Parameters:

storage (LocalStorage)

__init__(storage)

Bind the spool to the server’s storage service (the site mount).

Parameters:

storage (LocalStorage)

Return type:

None

create(descriptor, params)

Create a task folder under pending/ with its descriptor and params.

Writes descriptor.json and the pickled params.pkl. Returns the task_id. This is a plain storage fact done by whoever launches the batch; the manager only ever MOVES the folder afterwards.

Return type:

str

Parameters:

• descriptor (dict[str, Any])

• params (dict[str, Any])

read_params(task_id)

Unpickle a task’s params (the worker reads them to run the task).

Return type:

dict[str, Any]

Parameters:

task_id (str)

list_pending()

The descriptors of every task under pending/ (the manager’s queue).

Return type:

list[dict[str, Any]]

list_active(worker_id)

The descriptors of worker_id’s active tasks (that worker’s queue).

Return type:

list[dict[str, Any]]

Parameters:

worker_id (str)

assign(task_id, worker_id)

Move pending/<id> -> active/<worker_id>/<id> (the manager’s act).

Stamps status=active, worker_id and started_ts on the descriptor, then moves the folder. The move is atomic on the mount.

Raises:

LookupError – if the task is not in pending.

Return type:

None

Parameters:

• task_id (str)

• worker_id (str)

settle(task_id, worker_id, outcome, error=None)

Move active/<worker_id>/<id> -> terminated/ or aborted/.

outcome == “ok” settles under terminated/; anything else (error, aborted, orphan) settles under aborted/. Stamps the descriptor with ended_ts/outcome/error. Terminal: a settled task never moves again — re-settling raises because the folder is no longer active.

Raises:

LookupError – if the task is not active on worker_id.

Return type:

None

Parameters:

• task_id (str)

• worker_id (str)

• outcome (str)

• error (str | None)

write_progress(task_id, worker_id, data)

Write the worker’s latest progress snapshot (single-writer: the worker).

Return type:

None

Parameters:

• task_id (str)

• worker_id (str)

• data (dict[str, Any])

read_progress(task_id)

Read a task’s latest progress snapshot, or None (monitor / SSE baseline).

Return type:

dict[str, Any] | None

Parameters:

task_id (str)

request_cancel(task_id)

Drop the cancel marker in the task folder (the user’s stop request).

Return type:

None

Parameters:

task_id (str)

is_cancelled(task_id)

True if a cancel marker is present (the worker checks at each tick).

Return type:

bool

Parameters:

task_id (str)

write_result(task_id, worker_id, data)

Persist the batch result in the task folder (pickled).

Return type:

None

Parameters:

• task_id (str)

• worker_id (str)

• data (Any)

read_result(task_id)

Unpickle a task’s result, or None if not written yet.

Return type:

Any

Parameters:

task_id (str)

get(task_id)

The descriptor of task_id in whatever state it sits, or None.

Return type:

dict[str, Any] | None

Parameters:

task_id (str)

list_by_owner(owner)

Every task of owner across ALL states (pending/active/terminated/aborted).

The user’s full picture: queued, running, finished, aborted. Reads all state directories; the monitor uses the raw states, the page filters by owner here.

Return type:

list[dict[str, Any]]

Parameters:

owner (str)

list_by_status(status)

Every task in one state (the monitor’s pending / per-worker views).

For active this spans all worker subfolders; each descriptor carries its worker_id, so the monitor can group per worker.

Return type:

list[dict[str, Any]]

Parameters:

status (str)

purge(task_id)

Remove a settled task’s folder entirely (tree-removal). True if removed.

Return type:

bool

Parameters:

task_id (str)

genro_asgi.tasks.spool.new_descriptor(task_id, owner, mount, node_path, session_id=None)

Build a fresh TaskDescriptor dict in the pending state.

The autosufficient shape a worker needs to run the task without the sender’s context: identity, the owner (for list_by_owner and progress), and how to resolve the code — mount (the app) + node_path (resolved by the worker with Router.node(node_path), callable, no HTTP). session_id is the launching MCP session, the key the executor publishes progress events under (None when the sender has no push channel). params travel in a separate pickled file, not inline here.

Return type:

dict[str, Any]

Parameters:

• task_id (str)

• owner (str)

• mount (str)

• node_path (str)

• session_id (str | None)

LocalTaskExecutor — run one spool task in-process on the live server (◆D22).

A batch task names its code by mount + node_path (the app it lives in, and the route path inside that app’s router). Running it needs the app instance — which needs the whole server (storage, dbs, sibling apps). In this mono-process core the server is ALREADY live: the executor binds to it directly and reaches the target app through server.mounts / server.primary. It never serves HTTP; it only resolves and runs handlers. (Distributed execution — a worker process that rebuilds the server from a config path — belongs to the orchestration package, D22, and is out of scope here.)

Running a handler needs NO request context: a @route handler is a bound method, so self.server / self.db are already reachable from the app instance. The executor resolves the callable with app.route.node(node_path) (a RouterNode, callable, no HTTP) and invokes it with the task’s params — the same async/sync split the dispatcher uses: an async handler stays on the loop, a sync handler goes through server.run_sync (the Macro 1 pool protocol; routed_application.py/applications/mcp.py are the mirrors).

The manager has already MOVED the task into active/<worker>/; the executor reads it there, runs it, and settles it once: any outcome (ok / error) moves the folder to terminated / aborted and a batch_id never moves again (§5.7 terminal-by-position). There is a single logical worker in the core, WORKER_ID == "local"; the per-worker active/<worker>/ structure stays for D22 forward-compat.

The A<->C bridge (core 1e Phase 6): a descriptor carrying the launching MCP session_id gets its lifecycle published on server.tasks.hub — started before the run, settled (with outcome/error) after — so a subscribed SSE stream follows the task live. session_id None = no push channel, no-op. The spool stays the source of truth; the hub is only the live courier.

class genro_asgi.tasks.executor.LocalTaskExecutor(server)

Bases: object

Resolve a spool task by mount/node_path and run it on the live server.

Note

Bound to the live server (dual relationship: self.server); owns the TaskSpool over the server’s site storage. Both attributes are the seam the TaskManager reuses (the spool is stateless — the same storage yields an equivalent spool).

Parameters:

server (BaseServer)

__init__(server)

Bind to the live server and open the spool over its storage.

Parameters:

server (BaseServer)

Return type:

None

resolve(descriptor)

Return the callable RouterNode for a task descriptor.

Looks the app up by mount in server.mounts (the primary answers an empty/unmatched mount, mirroring the request demux) and resolves node_path in that app’s router. The node is callable — invoking it runs the handler, no HTTP.

Raises:

LookupError – if mount names neither a secondary mount nor the primary.

Return type:

Any

Parameters:

descriptor (dict[str, Any])

async execute(task_id, worker_id)

Run the task active on worker_id and settle it; return the outcome.

Reads the descriptor and params from the spool, resolves the handler, runs it (async on the loop, sync on the pool via server.run_sync), writes the result, and settles the folder to terminated (ok) or aborted (error). The outcome is "ok" or "error"; on error the exception text is stamped on the descriptor via settle.

Raises:

LookupError – if no task task_id exists in the spool.

Return type:

str

Parameters:

• task_id (str)

• worker_id (str)

TaskScheduler — the loop that runs due schedules (◆D22).

One asyncio task owned by the TaskManager (started with the manager on lifespan startup, dies in shutdown), tick ~30s (cron granularity is the minute). Each tick:

• scan: walk every mounted app’s routing tree (nodes(lazy=True, forbidden=True) — a gated route is still a declared task) collecting entries whose metadata carries task. The tree IS the live registry: no register() API. A task_name declared twice is an ERROR — both are excluded, no silent pick. A record whose task_name nobody declares is an ORPHAN: never executed, never deleted (a UI concern).

• sync defaults: a declared task_every/task_cron auto-creates the missing code-default record (code = task_name); an existing record always wins (store override) — deleting it resets to the code default at the next scan.

• run: for each due record resolve the live callable and spawn it — async handlers on the loop, sync ones through server.run_sync (the D2 pool). No overlap: a schedule whose previous run is still in flight is skipped. Runs are SYSTEM calls: no middleware chain, no auth filters.

On completion the record’s last_*/next_run_ts update and a JSONL line is appended to the task’s capped log. run_now fires a schedule immediately (from another thread when the loop is up, inline otherwise) with the same no-overlap guard and the same outcome path.

The store lives on the TaskManager (manager.task_store), not the server; the scheduler reaches the live server through manager.server. Store I/O and sync task bodies are synchronous by construction (core 1b) and dispatched via server.run_sync (a zero-arg closure — run_sync takes no kwargs).

class genro_asgi.tasks.scheduler.TaskScheduler(manager)

Bases: object

The scheduling loop bound to its manager (dual relationship).

Parameters:

manager (TaskManager)

__init__(manager)

Bind the scheduler to the manager owning the store, server and loop.

Parameters:

manager (TaskManager) – The TaskManager; the store is manager.task_store, the live server is manager.server, the live registry is the mounted apps’ routing trees.

Return type:

None

property server: Any

The live server (via the manager).

property store: Any

The manager’s TaskStore.

property running: set[str]

The codes with a run currently in flight (a copy).

start()

Start the tick loop on the running event loop (lifespan startup).

Return type:

None

async stop()

Cancel the tick loop (lifespan shutdown); in-flight runs finish.

Return type:

None

scan()

task_name -> {“callable”, “metadata”} from every mounted app’s tree.

Duplicates (the same task_name declared by two routes) are an explicit error: logged and EXCLUDED — no silent pick.

Return type:

dict[str, dict[str, Any]]

sync_defaults(registry, now)

Auto-create the code-default record for declared default schedules.

Return type:

None

Parameters:

• registry (dict[str, dict[str, Any]])

• now (float)

async tick()

One pass: scan, sync defaults, spawn every due schedule.

Store I/O (sync_defaults writes, due_rows reads every record) runs on the server pool via run_sync, never on the loop.

Return type:

None

run_now(code)

Fire a schedule immediately (the UI’s Run now). Thread-safe.

Return type:

str

Returns:

"started" (fired on the live loop), "done" (executed inline — no loop running, e.g. in tests), or "running" (skipped: the previous run is still in flight).

Raises:

LookupError – If the code is unknown or its task_name is an orphan.

Parameters:

code (str)

Schedule parsing and next-run computation — the three task kinds.

• every: "30s" | "15m" | "2h" | "1d" — next = after + interval.

• cron: classic 5-field string (min hour dom month dow) with * , - /, parsed in-house (~60 lines; croniter stays out — zero dependencies). System cron semantics: dow accepts 0-7 (0 and 7 = Sunday); when BOTH dom and dow are restricted a date matches if EITHER matches; evaluation is in LOCAL time, like system cron.

• at: a list of ISO timestamps (naive = local time), each fires once; an exhausted list yields no next run (a DERIVED state — the record stays).

All computed instants are POSIX epoch seconds (the store’s *_ts fields). No replay: the next run is always computed forward from after (now), never backfilled for downtime.

genro_asgi.tasks.schedule.parse_every(spec)

"30s" | "15m" | "2h" | "1d" -> the interval in seconds.

Raises:

ValueError – On a malformed spec, an unknown unit or a zero interval.

Return type:

int

Parameters:

spec (str)

genro_asgi.tasks.schedule.parse_at(spec)

A list of ISO timestamps -> sorted epoch seconds (naive = local time).

Raises:

ValueError – If spec is not a list or an entry is not ISO-parseable.

Return type:

list[float]

Parameters:

spec (Any)

class genro_asgi.tasks.schedule.CronSpec(spec)

Bases: object

A parsed 5-field cron string; next_after walks to the next match.

Parameters:

spec (str)

__init__(spec)

Parse "min hour dom month dow" into value sets.

Raises:

ValueError – On a wrong field count or a malformed/out-of-range field.

Parameters:

spec (str)

Return type:

None

next_after(after_ts)

The first matching instant strictly after after_ts (epoch seconds).

Raises:

ValueError – If nothing matches within ~4 years (an impossible date, e.g. "0 0 31 2 *").

Return type:

float

Parameters:

after_ts (float)

genro_asgi.tasks.schedule.next_run(kind, spec, after_ts)

The next due instant for a schedule, or None (an exhausted at list).

Parameters:

• kind (str) – "every" | "cron" | "at".

• spec (Any) – The kind’s spec — interval string, cron string, or ISO list.

• after_ts (float) – Compute the first occurrence strictly after this instant.

Raises:

ValueError – On an unknown kind or a malformed spec.

Return type:

float | None

TaskStore — the persistent half of the task backbone (NO SQLite).

The routing tree declares tasks (the live registry); the store remembers the SCHEDULES: when each one runs, whether it is enabled, what happened last time. Filesystem on the storage layer behind a small contract — one JSON per schedule record at tasks/<code>.json, one capped JSONL log per task at tasks/logs/<task_name>.jsonl. The mount is chosen exactly as the other server stores’ data of record: the encrypted secure mount when key material is installed, else the plain site mount with a warning (distinct from the task SPOOL, which is always site — spool data are transient service state, task RECORDS may carry schedule kwargs worth encrypting). A future Db-backed store swaps behind the same contract.

The record:

{
  "code": "shop_cleanup",          # PK; the default row's code IS the task name
  "task_name": "shop_cleanup",     # joins the live registry (the tree)
  "target_kind": "task",           # "path" reserved for a future privileged mode
  "kwargs": {},                    # passed to the callable
  "kind": "every",                 # every | cron | at
  "spec": "15m",                   # interval | cron string | ISO list
  "enabled": true,
  "next_run_ts": 1783900000.0,     # epoch; null = nothing due (exhausted "at")
  "last_run_ts": null, "last_outcome": null,
  "last_error": null, "last_duration": null
}

The store never computes schedules (that is tasks.schedule) and never resolves callables (that is the scheduler): it persists, lists and filters. upsert_default is the code-default rule: created when absent, an existing record ALWAYS wins (the config/preferences pattern). Storage is synchronous by construction (core 1b): async callers dispatch store calls via server.run_sync.

class genro_asgi.tasks.store.TaskStore

Bases: object

Contract for the schedule store (see module docstring).

load_all()

Return every schedule record.

Return type:

list[dict[str, Any]]

get(code)

Return the record for code or None.

Return type:

dict[str, Any] | None

Parameters:

code (str)

save(record)

Persist record (create or update), atomically.

Return type:

None

Parameters:

record (dict[str, Any])

delete(code)

Remove code. True if a record was removed, False if absent.

Return type:

bool

Parameters:

code (str)

append_log(task_name, entry)

Append one JSONL line to the task’s log, keeping the last LOG_CAP.

Return type:

None

Parameters:

• task_name (str)

• entry (dict[str, Any])

read_log(task_name, limit=200)

The task’s most recent log entries, oldest first.

Return type:

list[dict[str, Any]]

Parameters:

• task_name (str)

• limit (int)

due_rows(now)

The enabled records whose next_run_ts has expired.

Return type:

list[dict[str, Any]]

Parameters:

now (float)

upsert_default(record)

Create the code-default record when absent; an existing one wins.

Return type:

None

Parameters:

record (dict[str, Any])

update_run(code, **fields)

Merge run-outcome fields (last_*, next_run_ts) into a record.

Return type:

None

Parameters:

• code (str)

• fields (Any)

set_enabled(code, value)

Flip the enabled flag; the updated record, or None if absent.

Return type:

dict[str, Any] | None

Parameters:

• code (str)

• value (bool)

class genro_asgi.tasks.store.FileTaskStore(storage, mount=None)

Bases: TaskStore

One JSON per schedule + one JSONL log per task, on the storage layer.

Note

The store holds the shared LocalStorage instance (dual relationship: self.storage), never raw paths — the mount decides encryption at rest.

Parameters:

• storage (LocalStorage)

• mount (str | None)

__init__(storage, mount=None)

Bind the store to the server’s storage service.

Parameters:

• storage (LocalStorage) – The server’s LocalStorage; task files live on its secure mount when keys are installed, else site.

• mount (str | None) – Explicit mount override (the tasks(mount=...) config element); None keeps the automatic by-keys choice.

Return type:

None

property mount: str

the explicit override, else by installed keys.

Type:

The task-files mount

load_all()

Read every *.json record under <mount>:tasks/.

Return type:

list[dict[str, Any]]

get(code)

Read one schedule’s record, or None if the file does not exist.

Return type:

dict[str, Any] | None

Parameters:

code (str)

save(record)

Persist record through the storage node, warning if unencrypted.

code is the file identity. created_at is kept from the existing record on update; updated_at is refreshed on every write.

Return type:

None

Parameters:

record (dict[str, Any])

delete(code)

Remove one schedule’s file. True if it existed, False otherwise.

Return type:

bool

Parameters:

code (str)

append_log(task_name, entry)

Append one JSONL line, rewriting with only the last LOG_CAP lines.

Return type:

None

Parameters:

• task_name (str)

• entry (dict[str, Any])

read_log(task_name, limit=200)

The task’s most recent limit entries, oldest first.

Return type:

list[dict[str, Any]]

Parameters:

• task_name (str)

• limit (int)

EventHub — in-memory per-session event fan-out (the live courier, ◆D22).

The spool on storage is the SOURCE OF TRUTH for task progress (the worker writes progress.json at each tick); the hub is the LIVE COURIER that carries the same event to whoever is watching NOW. A subscriber is a bounded asyncio.Queue keyed by session_id (the launching MCP session). The executor pairs each spool.write_progress(...) with a hub.publish(session_id, event) (wired in Phase 6); the MCP push channel (Phase 6) subscribes on a GET and drains the queue into an SSE stream.

Fire-and-forget, shaped like ChannelClient.send (channel/client.py) but pure in-memory — no transport, no frames, nothing to import. A publish to a session with no subscriber is a no-op: progress is not lost, it lives on the spool; the hub only serves live watchers. The queue is bounded and DROPS THE OLDEST event when full: progress is idempotent (each event is a snapshot superseding the previous), so a slow reader loses intermediate frames, never the meaning.

No durable replay log lives here (that reads as orchestration, D22): resumability is snapshot-baseline — a late subscriber replays the current progress.json snapshot (Phase 5/6), then follows the live queue.

class genro_asgi.tasks.hub.EventHub

Bases: object

Per-session_id fan-out of progress events over bounded queues.

Note

Pure in-memory state on the instance (self._subscribers): a session maps to the set of its live queues (one server may serve several SSE streams for the same session). No locks — every method runs on the one server event loop.

__init__()

Start with no subscribers.

Return type:

None

subscribe(session_id)

Register a fresh bounded queue for session_id and return it.

The caller (the SSE handler) drains the queue until it unsubscribes.

Return type:

Queue[Any]

Parameters:

session_id (str)

unsubscribe(session_id, queue)

Drop queue from session_id (the SSE stream closed); tidy up empties.

Return type:

None

Parameters:

• session_id (str)

• queue (Queue[Any])

publish(session_id, event)

Deliver event to every live queue of session_id (no-op if none).

Full queue → drop the oldest event and enqueue the new one: progress is a snapshot, so the freshest event is the one that matters.

Return type:

None

Parameters:

• session_id (str)

• event (Any)