Add support for streaming

.github/workflows/ci.yml

@@ -67,6 +67,10 @@ jobs:
           uvx ruff format --check .
           uvx ruff check .
 
+      - name: Run adapter tests
+        working-directory: adapters/python
+        run: uv run --with pytest pytest tests/
+
       - name: Check version consistency
         shell: python3 {0}
         run: |

README.md

@@ -30,14 +30,17 @@ A **workflow** is the entry point for a run. It can call **tasks**, which are th
 ```python
 import coflux as cf
 
+
 @cf.task(retries=cf.Retries(3, when=ConnectionError))
 def fetch_data(url: str) -> dict:
     return requests.get(url).json()
 
+
 @cf.task(cache=True)
 def transform(data: dict) -> list:
     return sorted(data["items"], key=lambda x: x["score"], reverse=True)
 
+
 @cf.workflow()
 def my_pipeline(url: str):
     return transform(fetch_data(url))
@@ -56,10 +59,12 @@ Reuse results across runs, with TTLs, parameter filtering, and cross-workspace c
 @cf.task(cache=True)
 def get_user(user_id): ...
 
+
 # cache for 10 minutes
 @cf.task(cache=600)
 def fetch_prices(): ...
 
+
 # cache by specific params only
 @cf.task(cache=cf.Cache(params=["product_id"]))
 def get_product(product_id, include_reviews=False): ...
@@ -71,8 +76,7 @@ Retry on specific exceptions with configurable backoff:
 
 ```python
 @cf.task(retries=cf.Retries(5, backoff=(1, 60), when=TransientError))
-def call_api():
-    ...
+def call_api(): ...
 ```
 
 ### Parallel execution
@@ -82,9 +86,9 @@ Submit tasks concurrently and collect results:
 ```python
 @cf.workflow()
 def process_order(user_id, product_id):
-    user = load_user.submit(user_id)              # starts immediately
-    product = load_product.submit(product_id)     # starts immediately
-    create_order(user.result(), product.result()) # waits for results before calling
+    user = load_user.submit(user_id)  # starts immediately
+    product = load_product.submit(product_id)  # starts immediately
+    create_order(user.result(), product.result())  # waits for results before calling
 ```
 
 ### Assets
@@ -97,6 +101,7 @@ def generate_report() -> cf.Asset:
     Path("report.csv").write_text(build_csv())
     return cf.asset(match="*.csv")
 
+
 @cf.workflow()
 def my_workflow():
     report = generate_report()
@@ -112,6 +117,7 @@ Memoise task calls within a single run (unlike caching, which works across runs)
 def send_email(recipient):
     mailer.send(recipient.email, ...)
 
+
 @cf.workflow()
 def notify(campaign_id):
     for r in get_recipients(campaign_id):
@@ -125,6 +131,7 @@ Record numeric values from tasks and visualise them in Studio:
 ```python
 loss = cf.Metric("loss", group="training")
 
+
 @cf.task()
 def train(epochs):
     for epoch in range(epochs):
@@ -191,10 +198,12 @@ Or [run it with Docker](https://docs.coflux.com/getting_started/server).
 
 import coflux as cf
 
+
 @cf.task()
 def greet(name: str) -> str:
     return f"Hello, {name}!"
 
+
 @cf.workflow()
 def hello(name: str):
     print(greet(name))

adapters/python/README.md

@@ -22,11 +22,13 @@ poetry install
 ```python
 from coflux import task, workflow, log_info
 
+
 @task()
 def process_item(item_id: int) -> dict:
     log_info("Processing item {id}", id=item_id)
     return {"id": item_id, "status": "done"}
 
+
 @workflow()
 def batch_process(items: list[int]) -> list[dict]:
     results = []

adapters/python/coflux/__init__.py

@@ -16,18 +16,31 @@
 from .errors import (
     ExecutionAbandoned,
     ExecutionCancelled,
+    ExecutionCrashed,
     ExecutionError,
     ExecutionTerminated,
     ExecutionTimeout,
     InputDismissed,
+    StreamSuperseded,
 )
 from .metric import Metric, MetricGroup, MetricScale, progress
-from .models import Asset, AssetEntry, AssetMetadata, Execution, Input, ModelSchema
+from .models import (
+    Asset,
+    AssetEntry,
+    AssetMetadata,
+    AsyncStreamIterator,
+    Execution,
+    Input,
+    Stream,
+    StreamIterator,
+)
 from .prompt import Prompt
 from .state import get_context
-from .target import Cache, Defer, Retries
+from .streams import stream
+from .target import Cache, Defer, Retries, Streams
 
-__all__ = [
+# Grouped by category rather than sorted alphabetically.
+__all__ = [  # noqa: RUF022
     # Version
     "__version__",
     # Decorators
@@ -41,19 +54,26 @@
     "ExecutionCancelled",
     "ExecutionTimeout",
     "ExecutionAbandoned",
+    "ExecutionCrashed",
+    "StreamSuperseded",
     "InputDismissed",
     "Input",
-    "ModelSchema",
     "Metric",
     "MetricGroup",
     "MetricScale",
     "Prompt",
     "Cache",
     "Defer",
     "Retries",
+    "Streams",
     "Asset",
     "AssetEntry",
     "AssetMetadata",
+    "Stream",
+    "StreamIterator",
+    "AsyncStreamIterator",
+    # Producer-side stream helper
+    "stream",
     # Context functions
     "group",
     "suspense",

adapters/python/coflux/_version.py

@@ -1,6 +1,7 @@
 """Package version, isolated to avoid circular imports."""
 
-from importlib.metadata import PackageNotFoundError, version as _pkg_version
+from importlib.metadata import PackageNotFoundError
+from importlib.metadata import version as _pkg_version
 
 try:
     __version__ = _pkg_version("coflux")