diff --git a/CHANGELOG.md b/CHANGELOG.md index 8dd2d39..aae7bcc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,21 @@ Versions follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). --- +## [0.22.0] — 2026-07-10 + +### Added + +**`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 ### Fixed / Enhanced diff --git a/README.md b/README.md index d06acd5..13c7c59 100644 --- a/README.md +++ b/README.md @@ -44,6 +44,7 @@ pip install circuitforge-core[video-service] # Video captioning service ( pip install circuitforge-core[mqtt] # MQTT broker client 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 @@ -82,6 +83,7 @@ pip install circuitforge-core[dev] # All dev dependencies | `cloud_session` | Implemented | Cloud session management primitives | | `input` | Implemented | Input handling — MediaPipe gesture recognition | | `job_quality` | Implemented | Job listing quality scoring and signal extraction | +| `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 | diff --git a/circuitforge_core/task_bridge/__init__.py b/circuitforge_core/task_bridge/__init__.py new file mode 100644 index 0000000..dc31d78 --- /dev/null +++ b/circuitforge_core/task_bridge/__init__.py @@ -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", +] diff --git a/circuitforge_core/task_bridge/client.py b/circuitforge_core/task_bridge/client.py new file mode 100644 index 0000000..d79207c --- /dev/null +++ b/circuitforge_core/task_bridge/client.py @@ -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 diff --git a/circuitforge_core/task_bridge/models.py b/circuitforge_core/task_bridge/models.py new file mode 100644 index 0000000..63c06ad --- /dev/null +++ b/circuitforge_core/task_bridge/models.py @@ -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) diff --git a/docs/modules/task_bridge.md b/docs/modules/task_bridge.md new file mode 100644 index 0000000..f605a07 --- /dev/null +++ b/docs/modules/task_bridge.md @@ -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 `. 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`). diff --git a/mkdocs.yml b/mkdocs.yml index 4766f51..278a132 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -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: diff --git a/pyproject.toml b/pyproject.toml index 6713b7f..50586ae 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 = [ @@ -20,6 +20,9 @@ memory = [ sync = [ "fastapi>=0.110", ] +task-bridge = [ + "httpx>=0.27", +] community = [ "psycopg2>=2.9", ] diff --git a/tests/test_task_bridge/__init__.py b/tests/test_task_bridge/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_task_bridge/test_client.py b/tests/test_task_bridge/test_client.py new file mode 100644 index 0000000..83de114 --- /dev/null +++ b/tests/test_task_bridge/test_client.py @@ -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() diff --git a/tests/test_task_bridge/test_models.py b/tests/test_task_bridge/test_models.py new file mode 100644 index 0000000..efccbf2 --- /dev/null +++ b/tests/test_task_bridge/test_models.py @@ -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