feat(tasks): implement dispatch_task/get_task_status used by product cf-orch integrations #67

New issue

Closed

opened 2026-07-10 16:14:02 -07:00 by pyr0ball · 0 comments

pyr0ball commented

2026-07-10 16:14:02 -07:00

Owner

Context

While working on pagepiper (PR Circuit-Forge/pagepiper#12), found that app/api/library.py imports and calls:

from circuitforge_core.tasks import dispatch_task  # dispatch_task(caller=task_name, args=args) -> task_id
from circuitforge_core.tasks import get_task_status  # get_task_status(task_id) -> dict

Neither function exists anywhere in circuitforge_core — grepped the main source tree and every worktree, zero matches. The only thing that exists in circuitforge_core.tasks today is the VRAM-aware TaskScheduler (get_scheduler().submit(task_type, payload, vram_gb) / .result(task_id)), which is a different API shape (task_type/payload/vram_gb, not caller/args).

Impact

Pagepiper's cf-orch dispatch path (_dispatch_shelve in app/api/library.py) always hits the ImportError fallback and silently runs shelving via local BackgroundTasks instead of routing through cf-orch. This is currently invisible — it just works via the fallback — but it means:

No actual cross-node task distribution for pagepiper's shelve pipeline, despite the code appearing to support it
Any other product written against this same caller=/args= interface (copied from pagepiper's pattern, e.g. if Turnstone or a future product assumes it exists) will hit the same silent fallback

Scope

Implement in circuitforge_core/tasks/:

dispatch_task(caller: str, args: dict) -> str — submits a task identified by a "product/task_name" string (e.g. "pagepiper/shelve_pdf") with a plain kwargs dict, routes through the existing coordinator/agent scheduling infra, returns a task_id
get_task_status(task_id: str) -> dict — polls status, shape compatible with what app/api/ingest.py (pagepiper) / app/api/shelve.py already expect: {"status": ..., "progress": ..., "error": ...}

Open question: should this be a new thin wrapper over the existing TaskScheduler/get_scheduler() API (translating caller/args into task_type/payload), or a genuinely new lower-level primitive? Whoever picks this up should check how _dispatch_shelve's fallback-on-ImportError pattern is used across products before deciding, since other products may have copied the same assumed interface.

Testing

Unit tests for dispatch_task/get_task_status, plus (once implemented) an integration check that pagepiper's _dispatch_shelve actually routes through cf-orch instead of falling back, to close the silent-fallback gap this ticket describes.

## Context While working on `pagepiper` (PR https://git.opensourcesolarpunk.com/Circuit-Forge/pagepiper/pulls/12), found that `app/api/library.py` imports and calls: ```python from circuitforge_core.tasks import dispatch_task # dispatch_task(caller=task_name, args=args) -> task_id from circuitforge_core.tasks import get_task_status # get_task_status(task_id) -> dict ``` Neither function exists anywhere in `circuitforge_core` — grepped the main source tree and every worktree, zero matches. The only thing that exists in `circuitforge_core.tasks` today is the VRAM-aware `TaskScheduler` (`get_scheduler().submit(task_type, payload, vram_gb)` / `.result(task_id)`), which is a different API shape (task_type/payload/vram_gb, not caller/args). ## Impact Pagepiper's cf-orch dispatch path (`_dispatch_shelve` in `app/api/library.py`) always hits the `ImportError` fallback and silently runs shelving via local `BackgroundTasks` instead of routing through cf-orch. This is currently invisible — it just works via the fallback — but it means: - No actual cross-node task distribution for pagepiper's shelve pipeline, despite the code appearing to support it - Any other product written against this same `caller=/args=` interface (copied from pagepiper's pattern, e.g. if Turnstone or a future product assumes it exists) will hit the same silent fallback ## Scope Implement in `circuitforge_core/tasks/`: - `dispatch_task(caller: str, args: dict) -> str` — submits a task identified by a `"product/task_name"` string (e.g. `"pagepiper/shelve_pdf"`) with a plain kwargs dict, routes through the existing coordinator/agent scheduling infra, returns a task_id - `get_task_status(task_id: str) -> dict` — polls status, shape compatible with what `app/api/ingest.py` (pagepiper) / `app/api/shelve.py` already expect: `{"status": ..., "progress": ..., "error": ...}` Open question: should this be a new thin wrapper over the existing `TaskScheduler`/`get_scheduler()` API (translating `caller`/`args` into `task_type`/`payload`), or a genuinely new lower-level primitive? Whoever picks this up should check how `_dispatch_shelve`'s fallback-on-`ImportError` pattern is used across products before deciding, since other products may have copied the same assumed interface. ## Testing Unit tests for `dispatch_task`/`get_task_status`, plus (once implemented) an integration check that pagepiper's `_dispatch_shelve` actually routes through cf-orch instead of falling back, to close the silent-fallback gap this ticket describes.