From 43f9e9a54c3d81caca1ff57c8ea2a2a61d73ca36 Mon Sep 17 00:00:00 2001 From: pyr0ball Date: Fri, 10 Jul 2026 18:08:48 -0700 Subject: [PATCH] feat(tasks): implement dispatch_task/get_task_status generic caller/args dispatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Pagepiper (and possibly other products copying its pattern) imported dispatch_task(caller, args) -> task_id / get_task_status(task_id) -> dict from circuitforge_core.tasks, expecting a "product/task_name" + kwargs-dict interface. Neither function existed, so every call silently hit an except-Exception fallback with no visible error. This is a new module, not a TaskScheduler wrapper — TaskScheduler is keyed by task_id/job_id/params against a specific SQLite background_tasks table (VRAM-budgeted LLM queue), a different shape from the generic named-runnable dispatch pagepiper actually needed. Scope: free-tier, in-process, single-node only. Routing through the circuitforge-orch coordinator would need a new generic task-dispatch endpoint on that separate BSL package (CFOrchClient only exposes model/ service allocation today) — tracked as follow-up, not attempted here. Consuming products additionally need to call register_task_runner() at startup to benefit; that's product-side work in a separate repo. Bump to 0.22.0. Closes: https://git.opensourcesolarpunk.com/Circuit-Forge/circuitforge-core/issues/67 --- CHANGELOG.md | 17 ++++ README.md | 2 +- circuitforge_core/tasks/__init__.py | 12 +++ circuitforge_core/tasks/dispatch.py | 138 ++++++++++++++++++++++++++++ docs/modules/tasks.md | 19 ++++ pyproject.toml | 2 +- tests/test_tasks/test_dispatch.py | 116 +++++++++++++++++++++++ 7 files changed, 304 insertions(+), 2 deletions(-) create mode 100644 circuitforge_core/tasks/dispatch.py create mode 100644 tests/test_tasks/test_dispatch.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 8dd2d39..9e02bd3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,23 @@ Versions follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). --- +## [0.22.0] — 2026-07-10 + +### Added + +**`circuitforge_core.tasks.dispatch_task` / `get_task_status`** — generic caller/args task dispatch (closes #67) + +Pagepiper (and potentially other products copying its pattern) imported `dispatch_task(caller, args) -> task_id` / `get_task_status(task_id) -> dict` from `circuitforge_core.tasks`, expecting a `"product/task_name"` + kwargs-dict interface — neither function existed anywhere, so every call silently hit an `except Exception` fallback to local `BackgroundTasks`, with no visible error. This is a different API shape from the existing VRAM-budgeted `TaskScheduler` (keyed by `task_id`/`job_id`/`params` against a specific SQLite `background_tasks` table), so it's a new module rather than a `TaskScheduler` wrapper. + +- `register_task_runner(caller, fn)` — register a runnable under a name (e.g. `"pagepiper/ingest_pdf"`) once at product startup. +- `dispatch_task(caller, args)` — runs the registered runnable as `fn(**args)` on a background thread, returns a `task_id` immediately. Raises `LookupError` if `caller` isn't registered — products with an `except Exception: ...` fallback (like pagepiper's `_dispatch_ingest`) keep working unchanged. +- `get_task_status(task_id)` — returns `{"status": "queued"|"running"|"complete"|"error", "progress": int|None, "error": str|None}`. Raises `KeyError` for an unknown `task_id`. +- `reset_dispatch_registry()` — test teardown only. +- **Scope note:** this is the free-tier, in-process, single-node implementation — no cross-node distribution. Routing through the `circuitforge-orch` coordinator (BSL, separate package) would need a new generic task-dispatch endpoint on that coordinator, which doesn't exist today (`circuitforge_orch.client.CFOrchClient` only exposes model/service allocation, not a generic caller/args job queue); that's out of scope for this cf-core-only PR and tracked as follow-up. Consuming products (pagepiper) additionally need to call `register_task_runner()` at startup to benefit — not done here, since that's product-side work in a separate repo. +- 10 tests: unregistered lookup, task_id uniqueness, args passed as kwargs, success/error status transitions, unregister, unknown task_id, status snapshot immutability. + +--- + ## [0.20.0] — 2026-05-05 ### Fixed / Enhanced diff --git a/README.md b/README.md index d06acd5..a120312 100644 --- a/README.md +++ b/README.md @@ -63,7 +63,7 @@ pip install circuitforge-core[dev] # All dev dependencies | `documents` | Implemented | PDF, DOCX, and image OCR ingestion into `StructuredDocument` | | `affiliates` | Implemented | Affiliate URL wrapping with per-user opt-out and env-var fallback | | `preferences` | Implemented | User preference store — local YAML with pluggable backend; dot-path get/set | -| `tasks` | Implemented | VRAM-aware LLM task scheduler; shared slot manager across services | +| `tasks` | Implemented | VRAM-aware LLM task scheduler; shared slot manager across services; generic caller/args `dispatch_task`/`get_task_status` | | `manage` | Implemented | Cross-platform product process manager (Docker and native modes) | | `resources` | Implemented | VRAM allocation, eviction engine, GPU profile registry | | `text` | Implemented | Text utilities (normalize, chunk, truncate) + local LLM inference service (GGUF/transformers/VLM/classifier backends, multimodal content-block API) | diff --git a/circuitforge_core/tasks/__init__.py b/circuitforge_core/tasks/__init__.py index d9ee25b..e2bfadd 100644 --- a/circuitforge_core/tasks/__init__.py +++ b/circuitforge_core/tasks/__init__.py @@ -1,3 +1,10 @@ +from circuitforge_core.tasks.dispatch import ( + dispatch_task, + get_task_status, + register_task_runner, + reset_dispatch_registry, + unregister_task_runner, +) from circuitforge_core.tasks.scheduler import ( TaskScheduler, LocalScheduler, @@ -12,4 +19,9 @@ __all__ = [ "detect_available_vram_gb", "get_scheduler", "reset_scheduler", + "dispatch_task", + "get_task_status", + "register_task_runner", + "unregister_task_runner", + "reset_dispatch_registry", ] diff --git a/circuitforge_core/tasks/dispatch.py b/circuitforge_core/tasks/dispatch.py new file mode 100644 index 0000000..b43b099 --- /dev/null +++ b/circuitforge_core/tasks/dispatch.py @@ -0,0 +1,138 @@ +# circuitforge_core/tasks/dispatch.py +""" +Generic caller/args task dispatch — cf-core #67. + +Products (e.g. pagepiper) were already importing `dispatch_task`/ +`get_task_status` from `circuitforge_core.tasks` expecting a +"product/task_name" + kwargs-dict interface, distinct from the VRAM-budgeted +`TaskScheduler` in `scheduler.py` (which is keyed by task_id/job_id/params +against a specific SQLite `background_tasks` table). Neither function +existed, so every call silently hit the `except Exception` fallback. + +This is the free-tier (in-process, single-node) implementation: a runnable +is registered under a name once at product startup, then dispatched by that +name any number of times. There is no coordinator here — cross-node +distribution via the separate circuitforge-orch package would need a new +generic task-dispatch endpoint on that coordinator; this module doesn't +attempt that, and is written so a future coordinator-backed implementation +can slot in behind the same two function signatures. + +Usage:: + + from circuitforge_core.tasks import register_task_runner, dispatch_task, get_task_status + + register_task_runner("pagepiper/ingest_pdf", run_ingest_pdf) + + task_id = dispatch_task("pagepiper/ingest_pdf", {"doc_id": "...", "file_path": "..."}) + get_task_status(task_id) # {"status": "running", "progress": 0, "error": None} +""" +from __future__ import annotations + +import logging +import threading +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor +from typing import Any +from uuid import uuid4 + +logger = logging.getLogger(__name__) + +_registry_lock = threading.Lock() +_task_runners: dict[str, Callable[..., None]] = {} + +_status_lock = threading.Lock() +_task_status: dict[str, dict[str, Any]] = {} + +_executor_lock = threading.Lock() +_executor: ThreadPoolExecutor | None = None + + +def _get_executor() -> ThreadPoolExecutor: + global _executor + with _executor_lock: + if _executor is None: + _executor = ThreadPoolExecutor(thread_name_prefix="cf-task-dispatch") + return _executor + + +def register_task_runner(caller: str, fn: Callable[..., None]) -> None: + """ + Register a runnable under `caller` (e.g. "pagepiper/ingest_pdf"). + + Must be called once at product startup before dispatch_task() is used + with the same `caller` string. `fn` is called as `fn(**args)`. + """ + with _registry_lock: + _task_runners[caller] = fn + + +def unregister_task_runner(caller: str) -> None: + """Remove a registration. TEST TEARDOWN ONLY.""" + with _registry_lock: + _task_runners.pop(caller, None) + + +def dispatch_task(caller: str, args: dict[str, Any]) -> str: + """ + Dispatch the runnable registered under `caller` with `args` as kwargs, + running it on a background thread. Returns immediately with a task_id. + + Raises LookupError if no runnable is registered for `caller` — callers + that assume this always routes through a coordinator (the pattern + pagepiper's `_dispatch_ingest` uses) should catch and fall back, exactly + as they already do for the ImportError this used to raise. + """ + with _registry_lock: + fn = _task_runners.get(caller) + if fn is None: + raise LookupError( + f"No task runner registered for {caller!r}. " + "Call register_task_runner() before dispatch_task()." + ) + + task_id = str(uuid4()) + with _status_lock: + _task_status[task_id] = {"status": "queued", "progress": 0, "error": None} + + def _run() -> None: + with _status_lock: + _task_status[task_id] = {"status": "running", "progress": 0, "error": None} + try: + fn(**args) + except Exception as exc: + logger.exception("Task %s (%s) failed", task_id, caller) + with _status_lock: + _task_status[task_id] = {"status": "error", "progress": None, "error": str(exc)} + else: + with _status_lock: + _task_status[task_id] = {"status": "complete", "progress": 100, "error": None} + + _get_executor().submit(_run) + return task_id + + +def get_task_status(task_id: str) -> dict[str, Any]: + """ + Return `{"status", "progress", "error"}` for `task_id`. + + Raises KeyError if `task_id` is unknown — never dispatched, or the + process restarted since (status is in-memory only, not persisted). + """ + with _status_lock: + status = _task_status.get(task_id) + if status is None: + raise KeyError(f"Unknown task_id: {task_id!r}") + return dict(status) + + +def reset_dispatch_registry() -> None: + """Clear all registrations and status, and shut down the executor. TEST TEARDOWN ONLY.""" + global _executor + with _registry_lock: + _task_runners.clear() + with _status_lock: + _task_status.clear() + with _executor_lock: + if _executor is not None: + _executor.shutdown(wait=True) + _executor = None diff --git a/docs/modules/tasks.md b/docs/modules/tasks.md index 4f341cf..1070b53 100644 --- a/docs/modules/tasks.md +++ b/docs/modules/tasks.md @@ -76,3 +76,22 @@ def get_scheduler(): ``` Always re-export `reset_scheduler` from the shim so the FastAPI lifespan can import it from one place. + +## Generic caller/args dispatch + +Separate from the VRAM-budgeted scheduler above: `dispatch_task()`/`get_task_status()` are a generic, product-agnostic pair for running a named callable in the background and polling its status, keyed by a `"product/task_name"` string rather than the scheduler's `task_id`/`job_id`/SQLite-table shape. + +```python +from circuitforge_core.tasks import register_task_runner, dispatch_task, get_task_status + +# Once, at product startup: +register_task_runner("pagepiper/ingest_pdf", run_ingest_pdf) + +# Anywhere a task needs dispatching: +task_id = dispatch_task("pagepiper/ingest_pdf", {"doc_id": "...", "file_path": "..."}) +get_task_status(task_id) # {"status": "running", "progress": 0, "error": None} +``` + +Free-tier, in-process, single-node — runs on a background thread pool, no cross-node distribution. `dispatch_task()` raises `LookupError` for an unregistered `caller`, so products with an `except Exception: ...` fallback to local execution keep working unchanged if they forget to register a runner. + +`register_task_runner`/`dispatch_task`/`get_task_status` route through `circuitforge_core.tasks.dispatch` internally — nothing shared with `scheduler.py`'s VRAM-aware queue. diff --git a/pyproject.toml b/pyproject.toml index 6713b7f..266cb24 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "circuitforge-core" -version = "0.21.0" +version = "0.22.0" description = "Shared scaffold for CircuitForge products (MIT)" requires-python = ">=3.11" dependencies = [ diff --git a/tests/test_tasks/test_dispatch.py b/tests/test_tasks/test_dispatch.py new file mode 100644 index 0000000..36bd45a --- /dev/null +++ b/tests/test_tasks/test_dispatch.py @@ -0,0 +1,116 @@ +"""Tests for circuitforge_core.tasks.dispatch (cf-core #67).""" +from __future__ import annotations + +import threading +import time + +import pytest + +from circuitforge_core.tasks.dispatch import ( + dispatch_task, + get_task_status, + register_task_runner, + reset_dispatch_registry, + unregister_task_runner, +) + + +@pytest.fixture(autouse=True) +def _clean_registry(): + reset_dispatch_registry() + yield + reset_dispatch_registry() + + +def _wait_for_status(task_id, target_statuses, timeout=2.0): + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + status = get_task_status(task_id) + if status["status"] in target_statuses: + return status + time.sleep(0.01) + raise AssertionError(f"task {task_id} did not reach {target_statuses} in time") + + +class TestDispatchTask: + def test_raises_lookup_error_when_unregistered(self): + with pytest.raises(LookupError): + dispatch_task("pagepiper/ingest_pdf", {"doc_id": "1"}) + + def test_returns_a_task_id_string(self): + register_task_runner("test/noop", lambda **kwargs: None) + task_id = dispatch_task("test/noop", {}) + assert isinstance(task_id, str) and task_id + + def test_two_dispatches_get_different_task_ids(self): + register_task_runner("test/noop", lambda **kwargs: None) + id1 = dispatch_task("test/noop", {}) + id2 = dispatch_task("test/noop", {}) + assert id1 != id2 + + def test_task_runs_registered_fn_with_args_as_kwargs(self): + received = {} + event = threading.Event() + + def runner(*, doc_id, file_path): + received["doc_id"] = doc_id + received["file_path"] = file_path + event.set() + + register_task_runner("pagepiper/ingest_pdf", runner) + task_id = dispatch_task( + "pagepiper/ingest_pdf", {"doc_id": "abc", "file_path": "/tmp/x.pdf"} + ) + assert event.wait(timeout=2.0) + assert received == {"doc_id": "abc", "file_path": "/tmp/x.pdf"} + _wait_for_status(task_id, {"complete"}) + + def test_task_status_reaches_complete_on_success(self): + register_task_runner("test/ok", lambda **kwargs: None) + task_id = dispatch_task("test/ok", {}) + status = _wait_for_status(task_id, {"complete", "error"}) + assert status == {"status": "complete", "progress": 100, "error": None} + + def test_task_status_reaches_error_on_exception(self): + def failing(**kwargs): + raise ValueError("boom") + + register_task_runner("test/fail", failing) + task_id = dispatch_task("test/fail", {}) + status = _wait_for_status(task_id, {"complete", "error"}) + assert status["status"] == "error" + assert status["error"] == "boom" + + def test_unregister_removes_runner(self): + register_task_runner("test/temp", lambda **kwargs: None) + unregister_task_runner("test/temp") + with pytest.raises(LookupError): + dispatch_task("test/temp", {}) + + +class TestGetTaskStatus: + def test_raises_key_error_for_unknown_task_id(self): + with pytest.raises(KeyError): + get_task_status("not-a-real-task-id") + + def test_status_immediately_after_dispatch_is_queued_or_running(self): + gate = threading.Event() + + def slow(**kwargs): + gate.wait(timeout=2.0) + + register_task_runner("test/slow", slow) + try: + task_id = dispatch_task("test/slow", {}) + status = get_task_status(task_id) + assert status["status"] in ("queued", "running") + finally: + gate.set() + + def test_returns_a_copy_not_the_internal_dict(self): + register_task_runner("test/ok", lambda **kwargs: None) + task_id = dispatch_task("test/ok", {}) + _wait_for_status(task_id, {"complete"}) + status = get_task_status(task_id) + status["status"] = "mutated" + assert get_task_status(task_id)["status"] == "complete"