fix: authenticate cross-worker image fetches to FlowMesh origin only

[timzsu]

Purpose

Fix a cross-worker image-fetch failure in VLM/image workflows on multi-node deployments (reported as PIL.UnidentifiedImageError). When an image artifact is produced on one worker and consumed on a different worker without a local copy, the artifact resolves to a FlowMesh results URL that the image branch in DataMixin fetched without auth — so under enforced auth the endpoint returns a 401 non-image body and the fetch/decode fails. Co-located runs never hit this (the artifact resolves to a local path). The naive fix (auth on every image URL) would leak the worker's bearer token to arbitrary hosts, so this PR authenticates only the worker's own FlowMesh origin.

Changes

• artifacts.py — add is_flowmesh_origin_url(url): true only when the URL's scheme+host match the worker's FLOWMESH_BASE_URL.
• data.py — split the http(s) image branch: FlowMesh-origin URLs go through the auth-aware _load_image_from_artifact; all other URLs keep the original unauthenticated fetch with a bounded 15s timeout.
• test_data_mixin_lineage.py — two regression tests: origin URL gets Authorization: Bearer; external URL gets no credential (even with FLOWMESH_API_KEY set) and the 15s timeout.

Design

Only the cross-worker results-URL fallback in data.py was not already auth-aware. It now routes through the same auth-aware helper, gated to the worker's own origin via FLOWMESH_BASE_URL (the existing trust boundary, already used by ssh_executor.py and stamped on uploaded artifacts). Origin-only (scheme+host+port) keeps external/public URLs unauthenticated and bounds their timeout to 15s, avoiding both token leakage and the 1800s default stall.

Test Plan

uv sync --all-packages --group ci
uv run pytest tests/worker/test_data_mixin_lineage.py tests/worker/test_artifact_utils.py -q
uv run pre-commit run --files src/worker/executors/mixins/data.py src/worker/executors/utils/artifacts.py tests/worker/test_data_mixin_lineage.py

End-to-end multi-node validation was run on a real two-node deployment (below), reproducing the producer/consumer-on-separate-results-storage condition on a single host.

Test Result

• Targeted tests: 15 passed. ruff / black / isort / mypy on the changed files: clean.
• Full tests/worker/ suite not run locally (GPU/model-download heavy, >20 min) — deferred to CI.

End-to-end (multi-node, enforced auth)

Validated on a real two-node FlowMesh deployment on a single host, adapted from the 35_multi_node_deployment env template: root node (redis_control + redis_telemetry + server :8020) plus a worker node (server only, Redis → root), with distinct stack suffixes / ports / results volumes so producer and consumer do not share results storage — the condition that forces the HTTP-URL fallback. Auth was genuinely enforced via the simple_plugin IdentityProvider (confirmed on /api/v1/nodes: 401 no token, 401 bad token, 200 with token).

A producer artifact was stored on root via the authenticated POST /api/v1/results/{task_id}/files; the shipped string-URL image branch of DataMixin._collect_prompts_for_spec was then exercised inside the real worker container (host-networked, real FLOWMESH_API_KEY / FLOWMESH_BASE_URL) against the resolved results URL:

Worker image	Branch taken	Result
v0.1.7 (unpatched)	bare requests.get(url, timeout=15)	FAIL — HTTPError 401 Unauthorized
v0.1.7 + this fix	is_flowmesh_origin_url → _load_image_from_artifact → resolve_artifact (auth)	PASS — image decoded, RGB 64×64

Note: against current shipped code the pre-fix failure surfaces as requests.HTTPError 401 — the response.raise_for_status() on the bare-fetch line raises before Image.open, i.e. just ahead of the UnidentifiedImageError in the original report. The user-visible effect (the visual-embedding task hard-fails on a cross-node image ref) and the fix are unchanged.

---

Pre-submission Checklist

• I have read the contribution guidelines.
• I have run pre-commit run --all-files and fixed any issues.
• I have added or updated tests covering my changes (if applicable).
• I have verified that uv run pytest tests/ passes locally.
• If I changed shared schemas or proto definitions, I have checked downstream compatibility across Server and Worker.
• If I changed the SDK or CLI, I have verified the affected packages work (uv sync --all-packages --group ci --frozen).
• If this is a breaking change, I have prefixed the PR title with [BREAKING] and described migration steps above.
• I have updated documentation or config examples if user-facing behavior changed.

[kaiitunnz]

LGTM.