feat(tasks): implement dispatch_task/get_task_status generic caller/args dispatch

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: #67

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

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) |

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",
 ]

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

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.

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 = [

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"