merge: feat/66-task-bridge into freeze/0.22.0
# Conflicts: # CHANGELOG.md # README.md # pyproject.toml
This commit is contained in:
commit
2f6d5e5d87
11 changed files with 436 additions and 0 deletions
|
|
@ -45,6 +45,15 @@ Standardizes retry/backoff behavior for cf-core modules and products that make e
|
|||
- New core dependency: `backon>=4.0`.
|
||||
- Scope note: only the standalone wrapper module ships in this PR. Wiring it into `llm`, `affiliates`, `reranker`, and `activitypub` (the call sites identified in #65) is left as follow-up work per module, rather than retrofitting `LLMRouter`'s existing fallback-chain error handling in the same change.
|
||||
|
||||
**`circuitforge_core.task_bridge`** — shared data contract for pushing tasks into an external scheduler (closes #66)
|
||||
|
||||
Kiwi is the pilot consumer, pushing pantry expiry-alert tasks into Ashley Venn's Focus Flow scheduler (AGPL-3.0, external project). To keep AGPL and BSL code from ever sharing a process or artifact, the task data contract lives here in cf-core (MIT); each product implements its own exporter/receiver. Design spec: `circuitforge-plans/shared/superpowers/specs/2026-07-05-focus-flow-task-bridge-design.md`.
|
||||
|
||||
- `models.py` — `ExternalTask` frozen dataclass: `schema_version`, `source_product`, `external_id`, `title`, `notes`, `due_at` (ISO 8601 UTC), `kind` (always `"flexible"` in v1 — external sources can never inject urgency into another product's UX), `status` (`"active"` / `"cancelled"`). Validates required fields and `kind`/`status` values in `__post_init__`.
|
||||
- `client.py` — `push_tasks(endpoint, token, tasks)`: thin `httpx` wrapper POSTing a batch as `{"tasks": [...]}` with a bearer token. Raises `TaskBridgeError` on transport failure or non-2xx response. No transport server, no auth/token generation — that lives on the receiving side.
|
||||
- New `task-bridge` extra (`httpx>=0.27`).
|
||||
- 19 tests: schema/validation/serialization (no network) plus a contract test against a real local `http.server` instance standing in for Focus Flow's importer, verifying the client emits a conformant payload.
|
||||
|
||||
---
|
||||
|
||||
## [0.20.0] — 2026-05-05
|
||||
|
|
|
|||
|
|
@ -45,6 +45,7 @@ pip install circuitforge-core[mqtt] # MQTT broker client
|
|||
pip install circuitforge-core[signal-bus] # SSE event publisher for real-time signal streams
|
||||
pip install circuitforge-core[meshtastic-service] # Meshtastic mesh radio + MQTT + FastAPI
|
||||
pip install circuitforge-core[memory] # Knowledge graph via mnemo sidecar
|
||||
pip install circuitforge-core[task-bridge] # Push tasks into an external scheduler (e.g. Focus Flow)
|
||||
pip install circuitforge-core[community] # PostgreSQL-backed community store
|
||||
pip install circuitforge-core[manage] # cf-manage CLI (Typer)
|
||||
pip install circuitforge-core[dev] # All dev dependencies
|
||||
|
|
@ -85,6 +86,7 @@ pip install circuitforge-core[dev] # All dev dependencies
|
|||
| `job_quality` | Implemented | Job listing quality scoring and signal extraction |
|
||||
| `signal_bus` | Implemented | Generic SSE event publisher for real-time signal streams |
|
||||
| `retry` | Implemented | Standard retry/backoff wrapper over `backon` for external-call modules |
|
||||
| `task_bridge` | Implemented | Shared data contract + push client for external task schedulers (e.g. Focus Flow) |
|
||||
| `vision` | Stub | Vision router (moondream2 / SigLIP dispatch — planned) |
|
||||
| `wizard` | Stub | First-run wizard base class — products subclass `BaseWizard` |
|
||||
| `pipeline` | Stub | Staging queue base — products provide concrete schema |
|
||||
|
|
|
|||
33
circuitforge_core/task_bridge/__init__.py
Normal file
33
circuitforge_core/task_bridge/__init__.py
Normal file
|
|
@ -0,0 +1,33 @@
|
|||
# circuitforge_core/task_bridge/__init__.py
|
||||
"""
|
||||
task_bridge — shared MIT data contract for pushing tasks from a CF product
|
||||
into an external scheduler (pilot consumer: Kiwi -> Focus Flow, AGPL-3.0).
|
||||
|
||||
Pure data contract + reusable push helper. No transport server, no auth/token
|
||||
generation logic (that lives on the receiving side), no product-specific
|
||||
behavior — keeps AGPL and BSL code from ever sharing a process or artifact.
|
||||
|
||||
Typical usage::
|
||||
|
||||
from circuitforge_core.task_bridge import ExternalTask, push_tasks
|
||||
|
||||
task = ExternalTask(
|
||||
source_product="kiwi",
|
||||
external_id="kiwi:item:1234",
|
||||
title="Use up milk",
|
||||
due_at="2026-07-07T00:00:00Z",
|
||||
notes="Opened 2026-07-01",
|
||||
)
|
||||
push_tasks("http://127.0.0.1:8512/import/tasks", token, [task])
|
||||
|
||||
Design spec: circuitforge-plans/shared/superpowers/specs/2026-07-05-focus-flow-task-bridge-design.md
|
||||
"""
|
||||
from .client import TaskBridgeError, push_tasks
|
||||
from .models import SCHEMA_VERSION, ExternalTask
|
||||
|
||||
__all__ = [
|
||||
"ExternalTask",
|
||||
"SCHEMA_VERSION",
|
||||
"TaskBridgeError",
|
||||
"push_tasks",
|
||||
]
|
||||
53
circuitforge_core/task_bridge/client.py
Normal file
53
circuitforge_core/task_bridge/client.py
Normal file
|
|
@ -0,0 +1,53 @@
|
|||
"""Push client for the cf-core task_bridge module.
|
||||
|
||||
MIT licensed. Thin httpx wrapper any CF product reuses to push a batch of
|
||||
ExternalTask records to a configured local HTTP endpoint. No transport
|
||||
server, no auth/token generation logic — that lives on the receiving side
|
||||
(e.g. Focus Flow's importer). See:
|
||||
circuitforge-plans/shared/superpowers/specs/2026-07-05-focus-flow-task-bridge-design.md
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
import httpx
|
||||
|
||||
from circuitforge_core.task_bridge.models import ExternalTask
|
||||
|
||||
|
||||
class TaskBridgeError(RuntimeError):
|
||||
"""Raised when pushing tasks to the configured endpoint fails."""
|
||||
|
||||
|
||||
def push_tasks(
|
||||
endpoint: str,
|
||||
token: str,
|
||||
tasks: Sequence[ExternalTask],
|
||||
*,
|
||||
timeout: float = 10.0,
|
||||
) -> httpx.Response:
|
||||
"""
|
||||
POST a batch of ExternalTask records to `endpoint` as `{"tasks": [...]}`,
|
||||
authenticated with a bearer token.
|
||||
|
||||
Raises:
|
||||
TaskBridgeError: the request failed to send, or the endpoint returned
|
||||
a non-2xx status.
|
||||
"""
|
||||
payload = {"tasks": [t.to_dict() for t in tasks]}
|
||||
try:
|
||||
resp = httpx.post(
|
||||
endpoint,
|
||||
json=payload,
|
||||
headers={"Authorization": f"Bearer {token}"},
|
||||
timeout=timeout,
|
||||
)
|
||||
except httpx.HTTPError as exc:
|
||||
raise TaskBridgeError(f"task_bridge push to {endpoint!r} failed: {exc}") from exc
|
||||
|
||||
if resp.status_code >= 300:
|
||||
raise TaskBridgeError(
|
||||
f"task_bridge push to {endpoint!r} returned {resp.status_code}: {resp.text}"
|
||||
)
|
||||
|
||||
return resp
|
||||
59
circuitforge_core/task_bridge/models.py
Normal file
59
circuitforge_core/task_bridge/models.py
Normal file
|
|
@ -0,0 +1,59 @@
|
|||
"""Data models for the cf-core task_bridge module.
|
||||
|
||||
MIT licensed. Ticket: cf-core #66. Design spec:
|
||||
circuitforge-plans/shared/superpowers/specs/2026-07-05-focus-flow-task-bridge-design.md
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import asdict, dataclass
|
||||
from typing import Any, Literal
|
||||
|
||||
SCHEMA_VERSION = 1
|
||||
|
||||
# v1 external tasks are always "flexible" — Focus Flow's calm task kind.
|
||||
# External sources never set "inflexible" / "critical" / "locked" / "surprise",
|
||||
# so no CF product can inject urgency into another product's UX.
|
||||
TaskKind = Literal["flexible"]
|
||||
TaskStatus = Literal["active", "cancelled"]
|
||||
|
||||
_VALID_STATUSES: frozenset[str] = frozenset({"active", "cancelled"})
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class ExternalTask:
|
||||
"""
|
||||
A task pushed from a CF product into an external scheduler (e.g. Focus Flow).
|
||||
|
||||
Pure data contract — no transport, no auth, no product-specific behavior.
|
||||
`external_id` must be stable and idempotent per source item (e.g.
|
||||
"kiwi:item:1234") so repeated pushes upsert rather than duplicate.
|
||||
"""
|
||||
|
||||
source_product: str
|
||||
external_id: str
|
||||
title: str
|
||||
due_at: str # ISO 8601 UTC, e.g. "2026-07-07T00:00:00Z"
|
||||
notes: str | None = None
|
||||
kind: TaskKind = "flexible"
|
||||
status: TaskStatus = "active"
|
||||
schema_version: int = SCHEMA_VERSION
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if not self.source_product:
|
||||
raise ValueError("source_product must not be empty")
|
||||
if not self.external_id:
|
||||
raise ValueError("external_id must not be empty")
|
||||
if not self.title:
|
||||
raise ValueError("title must not be empty")
|
||||
if self.kind != "flexible":
|
||||
raise ValueError(
|
||||
f"kind must be 'flexible' in schema v{SCHEMA_VERSION}, got {self.kind!r}"
|
||||
)
|
||||
if self.status not in _VALID_STATUSES:
|
||||
raise ValueError(
|
||||
f"status must be one of {sorted(_VALID_STATUSES)}, got {self.status!r}"
|
||||
)
|
||||
|
||||
def to_dict(self) -> dict[str, Any]:
|
||||
"""Serialize to the wire format expected by an importer's HTTP API."""
|
||||
return asdict(self)
|
||||
44
docs/modules/task_bridge.md
Normal file
44
docs/modules/task_bridge.md
Normal file
|
|
@ -0,0 +1,44 @@
|
|||
# task_bridge
|
||||
|
||||
Shared MIT data contract for pushing tasks from a CF product into an external scheduler. Pilot consumer: Kiwi's pantry expiry alerts pushed into [Focus Flow](https://git.opensourcesolarpunk.com/Circuit-Forge/focus-flow), Ashley Venn's ADHD-first scheduler (AGPL-3.0, external project).
|
||||
|
||||
## Why this lives in cf-core
|
||||
|
||||
Focus Flow is AGPL-3.0; CF's AI features are BSL 1.1. To avoid any question of AGPL reach extending into CF's licensed code, the two sides never share a process or artifact:
|
||||
|
||||
- The **data contract** (this module) is MIT, a pure data definition with no product-specific logic.
|
||||
- Each **product's exporter/receiver** lives in that product's own repo, under its own licensing.
|
||||
- The two sides talk only over local HTTP, never a shared library or process.
|
||||
|
||||
Design spec: `circuitforge-plans/shared/superpowers/specs/2026-07-05-focus-flow-task-bridge-design.md`.
|
||||
|
||||
## Usage
|
||||
|
||||
```python
|
||||
from circuitforge_core.task_bridge import ExternalTask, push_tasks
|
||||
|
||||
task = ExternalTask(
|
||||
source_product="kiwi",
|
||||
external_id="kiwi:item:1234", # stable, idempotent per source item — upserts, not duplicates
|
||||
title="Use up milk",
|
||||
due_at="2026-07-07T00:00:00Z", # ISO 8601 UTC
|
||||
notes="Opened 2026-07-01",
|
||||
)
|
||||
|
||||
push_tasks("http://127.0.0.1:8512/import/tasks", token, [task])
|
||||
```
|
||||
|
||||
`kind` is always `"flexible"` in schema v1 — Focus Flow's calm task kind. External sources can never set `"inflexible"`, `"critical"`, `"locked"`, or `"surprise"`, so no CF product can inject urgency into another product's UX.
|
||||
|
||||
`status` is `"active"` (default) or `"cancelled"` — push a `"cancelled"` record when the source item is deleted or edited away before the external scheduler marks it complete.
|
||||
|
||||
## API
|
||||
|
||||
- `ExternalTask(source_product, external_id, title, due_at, notes=None, kind="flexible", status="active")` — frozen dataclass; validates required fields and `kind`/`status` in `__post_init__`. `.to_dict()` serializes to the wire format.
|
||||
- `push_tasks(endpoint, token, tasks, *, timeout=10.0)` — POSTs `{"tasks": [...]}` with `Authorization: Bearer <token>`. Raises `TaskBridgeError` on transport failure or non-2xx response.
|
||||
|
||||
## What this module does *not* do
|
||||
|
||||
No transport server, no auth/token generation or pairing logic, no product-specific behavior (diffing, callback handling, inventory updates). Those live entirely on the consuming product's side — see Kiwi's `focus_flow` service for the reference implementation.
|
||||
|
||||
Install: `pip install circuitforge-core[task-bridge]` (pulls in `httpx`).
|
||||
|
|
@ -71,6 +71,7 @@ nav:
|
|||
- stt: modules/stt.md
|
||||
- tts: modules/tts.md
|
||||
- pipeline: modules/pipeline.md
|
||||
- task_bridge: modules/task_bridge.md
|
||||
- vision: modules/vision.md
|
||||
- wizard: modules/wizard.md
|
||||
- Developer Guide:
|
||||
|
|
|
|||
|
|
@ -24,6 +24,9 @@ sync = [
|
|||
signal-bus = [
|
||||
"fastapi>=0.110",
|
||||
]
|
||||
task-bridge = [
|
||||
"httpx>=0.27",
|
||||
]
|
||||
community = [
|
||||
"psycopg2>=2.9",
|
||||
]
|
||||
|
|
|
|||
0
tests/test_task_bridge/__init__.py
Normal file
0
tests/test_task_bridge/__init__.py
Normal file
149
tests/test_task_bridge/test_client.py
Normal file
149
tests/test_task_bridge/test_client.py
Normal file
|
|
@ -0,0 +1,149 @@
|
|||
"""Tests for circuitforge_core.task_bridge.client (cf-core #66)."""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import threading
|
||||
from contextlib import contextmanager
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from circuitforge_core.task_bridge.client import TaskBridgeError, push_tasks
|
||||
from circuitforge_core.task_bridge.models import ExternalTask
|
||||
|
||||
|
||||
def _task(**overrides) -> ExternalTask:
|
||||
fields = {
|
||||
"source_product": "kiwi",
|
||||
"external_id": "kiwi:item:1234",
|
||||
"title": "Use up milk",
|
||||
"due_at": "2026-07-07T00:00:00Z",
|
||||
}
|
||||
fields.update(overrides)
|
||||
return ExternalTask(**fields)
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _mocked_httpx_post(handler):
|
||||
"""Patch httpx.post to route through an httpx.MockTransport handler."""
|
||||
transport = httpx.MockTransport(handler)
|
||||
|
||||
def fake_post(url, *, json=None, headers=None, timeout=None):
|
||||
with httpx.Client(transport=transport) as client:
|
||||
return client.post(url, json=json, headers=headers, timeout=timeout)
|
||||
|
||||
import circuitforge_core.task_bridge.client as client_module
|
||||
|
||||
original = client_module.httpx.post
|
||||
client_module.httpx.post = fake_post
|
||||
try:
|
||||
yield
|
||||
finally:
|
||||
client_module.httpx.post = original
|
||||
|
||||
|
||||
class TestPushTasksUnit:
|
||||
def test_sends_conformant_payload(self):
|
||||
captured = {}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
captured["body"] = json.loads(request.content)
|
||||
captured["auth"] = request.headers.get("authorization")
|
||||
return httpx.Response(200, json={"accepted": 1})
|
||||
|
||||
with _mocked_httpx_post(handler):
|
||||
resp = push_tasks(
|
||||
"http://127.0.0.1:9999/import/tasks", "tok123", [_task()]
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert captured["auth"] == "Bearer tok123"
|
||||
assert captured["body"] == {"tasks": [_task().to_dict()]}
|
||||
|
||||
def test_pushes_multiple_tasks_in_one_batch(self):
|
||||
captured = {}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
captured["body"] = json.loads(request.content)
|
||||
return httpx.Response(200, json={"accepted": 2})
|
||||
|
||||
tasks = [_task(external_id="kiwi:item:1"), _task(external_id="kiwi:item:2")]
|
||||
with _mocked_httpx_post(handler):
|
||||
push_tasks("http://127.0.0.1:9999/import/tasks", "tok", tasks)
|
||||
|
||||
assert len(captured["body"]["tasks"]) == 2
|
||||
|
||||
def test_raises_on_non_2xx_response(self):
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(401, text="bad token")
|
||||
|
||||
with _mocked_httpx_post(handler):
|
||||
with pytest.raises(TaskBridgeError):
|
||||
push_tasks("http://127.0.0.1:9999/import/tasks", "bad", [_task()])
|
||||
|
||||
def test_raises_on_transport_error(self):
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
raise httpx.ConnectError("connection refused")
|
||||
|
||||
with _mocked_httpx_post(handler):
|
||||
with pytest.raises(TaskBridgeError):
|
||||
push_tasks("http://127.0.0.1:9999/import/tasks", "tok", [_task()])
|
||||
|
||||
def test_empty_task_list_sends_empty_batch(self):
|
||||
captured = {}
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
captured["body"] = json.loads(request.content)
|
||||
return httpx.Response(200, json={"accepted": 0})
|
||||
|
||||
with _mocked_httpx_post(handler):
|
||||
push_tasks("http://127.0.0.1:9999/import/tasks", "tok", [])
|
||||
|
||||
assert captured["body"] == {"tasks": []}
|
||||
|
||||
|
||||
class _FakeImporterHandler(BaseHTTPRequestHandler):
|
||||
"""A minimal stand-in for Focus Flow's import API."""
|
||||
|
||||
received: list[dict] = []
|
||||
|
||||
def do_POST(self):
|
||||
length = int(self.headers["Content-Length"])
|
||||
body = json.loads(self.rfile.read(length))
|
||||
_FakeImporterHandler.received.append(
|
||||
{"body": body, "auth": self.headers.get("Authorization")}
|
||||
)
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.end_headers()
|
||||
self.wfile.write(json.dumps({"accepted": len(body["tasks"])}).encode())
|
||||
|
||||
def log_message(self, format, *args):
|
||||
pass # silence test output
|
||||
|
||||
|
||||
class TestPushTasksContract:
|
||||
"""Real local HTTP server standing in for Focus Flow's importer."""
|
||||
|
||||
def test_client_emits_conformant_payload_over_real_socket(self):
|
||||
_FakeImporterHandler.received = []
|
||||
server = HTTPServer(("127.0.0.1", 0), _FakeImporterHandler)
|
||||
port = server.server_address[1]
|
||||
thread = threading.Thread(target=server.handle_request, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
task = _task(notes="Opened 2026-07-01")
|
||||
resp = push_tasks(
|
||||
f"http://127.0.0.1:{port}/import/tasks", "pairing-token-abc", [task]
|
||||
)
|
||||
thread.join(timeout=5)
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json() == {"accepted": 1}
|
||||
assert len(_FakeImporterHandler.received) == 1
|
||||
received = _FakeImporterHandler.received[0]
|
||||
assert received["auth"] == "Bearer pairing-token-abc"
|
||||
assert received["body"] == {"tasks": [task.to_dict()]}
|
||||
finally:
|
||||
server.server_close()
|
||||
83
tests/test_task_bridge/test_models.py
Normal file
83
tests/test_task_bridge/test_models.py
Normal file
|
|
@ -0,0 +1,83 @@
|
|||
"""Tests for circuitforge_core.task_bridge.models (cf-core #66). No network."""
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from circuitforge_core.task_bridge.models import SCHEMA_VERSION, ExternalTask
|
||||
|
||||
|
||||
def _task(**overrides) -> ExternalTask:
|
||||
fields = {
|
||||
"source_product": "kiwi",
|
||||
"external_id": "kiwi:item:1234",
|
||||
"title": "Use up milk",
|
||||
"due_at": "2026-07-07T00:00:00Z",
|
||||
}
|
||||
fields.update(overrides)
|
||||
return ExternalTask(**fields)
|
||||
|
||||
|
||||
class TestExternalTaskDefaults:
|
||||
def test_default_kind_is_flexible(self):
|
||||
assert _task().kind == "flexible"
|
||||
|
||||
def test_default_status_is_active(self):
|
||||
assert _task().status == "active"
|
||||
|
||||
def test_default_schema_version(self):
|
||||
assert _task().schema_version == SCHEMA_VERSION == 1
|
||||
|
||||
def test_notes_defaults_to_none(self):
|
||||
assert _task().notes is None
|
||||
|
||||
def test_frozen(self):
|
||||
task = _task()
|
||||
with pytest.raises((AttributeError, TypeError)):
|
||||
task.title = "other" # type: ignore
|
||||
|
||||
|
||||
class TestExternalTaskValidation:
|
||||
def test_rejects_empty_source_product(self):
|
||||
with pytest.raises(ValueError):
|
||||
_task(source_product="")
|
||||
|
||||
def test_rejects_empty_external_id(self):
|
||||
with pytest.raises(ValueError):
|
||||
_task(external_id="")
|
||||
|
||||
def test_rejects_empty_title(self):
|
||||
with pytest.raises(ValueError):
|
||||
_task(title="")
|
||||
|
||||
def test_rejects_non_flexible_kind(self):
|
||||
with pytest.raises(ValueError):
|
||||
_task(kind="critical")
|
||||
|
||||
def test_rejects_unknown_status(self):
|
||||
with pytest.raises(ValueError):
|
||||
_task(status="done")
|
||||
|
||||
def test_accepts_cancelled_status(self):
|
||||
assert _task(status="cancelled").status == "cancelled"
|
||||
|
||||
|
||||
class TestExternalTaskToDict:
|
||||
def test_to_dict_contains_all_fields(self):
|
||||
task = _task(notes="Opened 2026-07-01")
|
||||
d = task.to_dict()
|
||||
assert d == {
|
||||
"source_product": "kiwi",
|
||||
"external_id": "kiwi:item:1234",
|
||||
"title": "Use up milk",
|
||||
"due_at": "2026-07-07T00:00:00Z",
|
||||
"notes": "Opened 2026-07-01",
|
||||
"kind": "flexible",
|
||||
"status": "active",
|
||||
"schema_version": 1,
|
||||
}
|
||||
|
||||
def test_to_dict_is_json_serializable(self):
|
||||
import json
|
||||
|
||||
task = _task()
|
||||
json.dumps(task.to_dict()) # must not raise
|
||||
Loading…
Reference in a new issue