feat(task-bridge): add shared data contract for external task schedulers

New circuitforge_core.task_bridge module: ExternalTask (frozen dataclass,
schema v1) plus push_tasks() httpx wrapper. Pilot consumer is Kiwi, pushing
pantry expiry alerts into Focus Flow (AGPL-3.0, external project). The
contract lives in cf-core (MIT) so AGPL and BSL code never share a process
or artifact — each product implements its own exporter/receiver.

No transport server, no auth/token generation, no product-specific
behavior — pure data contract + reusable push helper, same spirit as the
existing sync and tasks modules.

Bump to 0.22.0.

Closes: #66

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

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 |

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

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

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)

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

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:

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

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

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