release: gen-worker 0.43.0 — v4 private-input manifest hard cut (gw#585)

CHANGELOG.md

@@ -1,5 +1,25 @@
 # Changelog
 
+## 0.43.0 (2026-07-21)
+
+- **gw#585: tensorhub v4 private-input manifests — gRPC protocol v3 -> v4
+  HARD CUT (th#886).** The hub no longer rewrites canonical payload bytes
+  with presigned URLs. `RunJob.input_assets` (field 15) carries the ordered
+  credential-free manifest; the worker resolves fresh transport URLs itself
+  with one strict `POST /worker/input-assets/resolve` under its
+  attempt-scoped capability, verifies exact size/BLAKE3/MIME/kind, preserves
+  opaque `Asset.ref`, sets only `local_path`, and cleans attempt-owned temp
+  files on every terminal path. Endpoint build now rejects Asset-bearing
+  `set`/`frozenset` and non-string-keyed mappings (unordered containers have
+  no manifest order); base `Asset`/`MediaAsset` discover as `kind=media`.
+  A v0.43 worker cannot serve a v3 hub and vice versa — deploy in lockstep
+  with the tensorhub v4 release.
+- **`GEN_WORKER_INTERNAL_OBJECT_HOSTS`**: exact-host allowlist that exempts
+  resolver-minted private-input URLs from the private-IP SSRF gate for
+  deployments whose object store lives on an internal network. Caller public
+  transports always face the full SSRF policy.
+- marco-polo example: `marco-polo-attach` private-input echo probe (e2e).
+
 ## 0.42.0 (2026-07-21)
 
 - **gw#615: disk telemetry can no longer freeze the event loop (0.40.7

examples/marco-polo/src/marco_polo/main.py

@@ -1,10 +1,12 @@
 # source-hash bust: 2026-07-03 (#368: @endpoint API rewrite)
 import asyncio
+import os
 import time
 from typing import AsyncIterator
 
 import msgspec
 from gen_worker import RequestContext, ValidationError, endpoint
+from gen_worker.api.types import Asset
 
 
 class MarcoPoloInput(msgspec.Struct):
@@ -15,6 +17,19 @@ class MarcoPoloOutput(msgspec.Struct):
     response: str
 
 
+class MarcoAttachInput(msgspec.Struct):
+    image: Asset
+    text: str = ""
+    delay_s: float = 0.0
+
+
+class MarcoAttachOutput(msgspec.Struct):
+    response: str
+    blake3: str
+    size_bytes: int
+    local_path: str
+
+
 @endpoint
 class MarcoPolo:
     def marco_polo(self, ctx: RequestContext, data: MarcoPoloInput) -> MarcoPoloOutput:
@@ -61,6 +76,32 @@ async def marco_polo_wedge(
         time.sleep(1800)
         return MarcoPoloOutput(response="unreachable")
 
+    async def marco_polo_attach(
+        self, ctx: RequestContext, data: MarcoAttachInput
+    ) -> MarcoAttachOutput:
+        """th#886 v4 private-input probe: echoes the materialized attachment's
+        BLAKE3/size/attempt-local path so the harness can verify exact bytes
+        and post-terminal temp cleanup. delay_s>0 holds the job in flight for
+        disconnect/retry chaos."""
+        import blake3
+
+        if str(data.text or "").strip().lower() != "marco":
+            raise ValidationError(f"expected 'marco', got {data.text!r}")
+        path = os.fspath(data.image)
+        with open(path, "rb") as f:
+            raw = f.read()
+        waited = 0.0
+        while waited < float(data.delay_s or 0.0):
+            ctx.raise_if_cancelled()
+            await asyncio.sleep(0.15)
+            waited += 0.15
+        return MarcoAttachOutput(
+            response="polo",
+            blake3=blake3.blake3(raw).hexdigest(),
+            size_bytes=len(raw),
+            local_path=path,
+        )
+
     async def marco_polo_stream(
         self, ctx: RequestContext, data: MarcoPoloInput
     ) -> AsyncIterator[MarcoPoloOutput]:

proto/CONTRACT.md

@@ -3,7 +3,7 @@
 Proto: `worker_scheduler.proto`, package `cozy.scheduler`, service
 `WorkerScheduler`, one RPC: `Connect(stream WorkerMessage) returns (stream SchedulerMessage)`.
 This REPLACES the old `scheduler.v1` proto entirely. No compat, no negotiation:
-`ProtocolVersion.PROTOCOL_VERSION_CURRENT = 3` is the only accepted version.
+`ProtocolVersion.PROTOCOL_VERSION_CURRENT = 4` is the only accepted version.
 
 Roles: **W** = gen-worker (Python), **O** = orchestrator (tensorhub Go).
 Every field below names its producer and consumer; anything without both was deleted.
@@ -269,7 +269,7 @@ Reply to Hello; re-sent on config change (full replace).
 |---|---|---|---|
 | `protocol_version` | O constant | W version gate | symmetric validation |
 | `file_base_url` | O config (`FileAPIBaseURL`) | W upload stack / blob-ref PUT flow (§8) | tensorhub HTTP base for capability-token calls |
-| `keep` | legacy v2 release config | none in v3 | retained field number only; v3 W ignores it |
+| `keep` | legacy v2 release config | none in v4 | retained field number only; v4 W ignores it |
 | `resolutions` | O precision resolver | W endpoint bindings | full-replace declared-ref to worker-specific ref/cast picks |
 | `desired_residency` | O scheduler/controller | W model reconciler | full-replace per-worker disk and hot-instance goal (§7) |
 
@@ -355,7 +355,7 @@ not inherit another alias's proof.
 | `request_id` | O | W executor | unique request id |
 | `attempt` | O fencing (§2) | W (echoed on all replies) | dispatch fence |
 | `function_name` | O from client submit | W registry lookup | dispatch key; unknown fn ⇒ immediate `JobResult{INVALID}` |
-| `input_payload` | O assignment payload derived from `/invoke` | W deserializes, validates against the fn payload type | MessagePack bytes; O keeps canonical stored refs in durable request identity but replaces typed Asset refs only in this ephemeral payload with fresh authorized HTTP(S) URLs. W rejects opaque refs and caller `local_path`, downloads distinct URLs once in payload order, assigns that single fresh attempt-local path to every duplicate occurrence, and cleans it on every terminal path before model/handler work can observe a failure |
+| `input_payload` | O copies the canonical `/invoke` payload byte-for-byte | W deserializes, materializes, then validates against the fn payload type | MessagePack bytes. Stored Asset refs remain opaque `source_ref` values; transport URLs never enter canonical bytes. Caller HTTP(S) refs remain in this payload and are absent from `input_assets`. W rejects caller `local_path` values, downloads each distinct input once in deterministic payload order, assigns one fresh attempt-local path to every duplicate occurrence, and cleans it on every terminal path |
 | `timeout_ms` | O from endpoint config | W deadline watchdog | 0 = none; on expiry W aborts and sends `JobResult{FATAL, safe_message:"deadline exceeded"}` |
 | `tenant` | O request state | W upload scoping + structured logs | invoking owner slug |
 | `invoker_id` | O request state | W `ctx.invoker_id` (read-only tenant surface) | optional |
@@ -365,6 +365,46 @@ not inherit another alias's proof.
 | `models` | O binding resolver (endpoint defaults + `_models` envelope) | W model injection (`ensure_local` + typed path/pipeline) + per-request LoRA overlays | slot → ref (+ optional `loras`) |
 | `snapshots` | O resolver | W download stack | presigned snapshots for tensorhub-CAS refs in `models` (including LoRA overlay refs) that O doesn't know to be on this worker's disk; W ignores entries already local (digest match). hf/civitai refs need no snapshot |
 | `required_compile` | O unique target selection | W pre-setup + pre-execution fence | exact target/cell/contract required for this attempt; mandatory for W8A8 and fail-closed whenever present |
+| `input_assets` | O submit-time stored-asset inspector | W private-input materializer | Unique by `(source_ref, kind)` in deterministic schema traversal order: endpoint field declaration, then list index, with `*` map keys sorted; the first traversal occurrence of a duplicate wins. Each entry is exactly `asset_id`, `source_ref`, lowercase BLAKE3 hex, `size_bytes`, `kind`, and lowercase `mime_type`; it contains no URL, path, auth, schema, or count field. |
+
+### Input-asset manifests (RunJob.input_assets)
+
+The endpoint contract must expose Asset-bearing collections through ordered
+`list`/`tuple` shapes. Asset-bearing `set`/`frozenset` shapes and mappings with
+non-string keys are rejected by the worker's schema discovery at build time:
+they have no stable occurrence order, and O does not invent a sorting rule.
+Generic `Asset`/`MediaAsset` fields use `kind = "media"`; typed
+image/video/audio fields use their exact kind. Leading or trailing whitespace
+in a typed Asset `ref` is invalid, so `input_payload` and
+`input_assets[].source_ref` cannot disagree.
+
+W derives the ordered unique private sequence by `(ref, declared kind)` from
+the decoded payload and requires exact count/order/ref/kind plus valid
+immutable metadata against `input_assets` BEFORE any resolver call or GET; a
+mismatch is a non-retryable failure. For non-empty `input_assets`, the
+capability JWT carries `attempt` and the signed manifest digest, and W resolves
+the whole list with exactly one bounded strict request per attempt:
+
+`POST {file_base_url}/api/v1/worker/input-assets/resolve`
+
+`Authorization: Bearer <RunJob.capability_token>`
+
+`{"request_id":"<RunJob.request_id>","attempt":<RunJob.attempt>}`
+
+The response is `{"assets":[...]}` with the same six manifest fields plus
+`url` and `url_expires_at`. W rejects a missing, extra, reordered, changed,
+unknown, or malformed entry, validates every URL before downloading, downloads
+each unique input once, verifies exact byte count, streaming BLAKE3, media
+kind, caller limits, and real image decode where applicable, keeps every
+decoded Asset field (`ref` stays the opaque `source_ref`), and sets only the
+worker-local `local_path`. URLs never replace `ref` and are never persisted. A
+resolver `409` means the attempt is stale ⇒ W stops as CANCELED without
+publishing; `503`/network failures and malformed platform responses are
+RETRYABLE. Capabilities, URLs, opaque refs, and response bodies never appear in
+errors or logs. Materialized inputs are attempt-scoped and transactional:
+success, handler failure, cancellation, and supersession each clean exactly
+their own attempt's directory — attempt N+1 never deletes N's scope and a
+superseded attempt never publishes its result.
 
 ### RequiredCompileExecution (embedded in RunJob)
 
@@ -572,7 +612,7 @@ There is no queued-only/item-level cancel.
 
 ### ModelOp (O → W)
 Producer: compile-cache adoption only. Model download/load/unload lifecycle is
-declarative through `HelloAck.desired_residency`; v3 O MUST NOT send the legacy
+declarative through `HelloAck.desired_residency`; v4 O MUST NOT send the legacy
 DOWNLOAD/LOAD/UNLOAD enum values.
 
 | field | producer | consumer | semantics |

proto/worker_scheduler.proto

@@ -30,7 +30,7 @@ option go_package = "github.com/cozy-creator/tensorhub/internal/orchestrator/grp
 // FAILED_PRECONDITION. No minor versions, no feature gates, no negotiation.
 enum ProtocolVersion {
   PROTOCOL_VERSION_UNSPECIFIED = 0;
-  PROTOCOL_VERSION_CURRENT     = 3;
+  PROTOCOL_VERSION_CURRENT     = 4;
 }
 
 service WorkerScheduler {
@@ -135,7 +135,7 @@ message ModelResidency {
   ResidencyTier tier = 2;      // highest tier the model currently occupies
   int64 vram_bytes = 3;        // measured footprint; set when tier = VRAM, else 0
   // Exact immutable bytes and the DesiredResidency generation that caused
-  // their materialization. Both are additive protocol-v3 evidence: legacy
+  // their materialization. Both are protocol-v4 evidence: legacy
   // workers leave them zero/empty, which means identity is unknown rather
   // than inferred from the mutable ref.
   string snapshot_digest = 4;
@@ -159,14 +159,14 @@ message InFlightJob {
 message HelloAck {
   ProtocolVersion protocol_version = 1; // orchestrator's version, for symmetric validation
   string file_base_url = 2;  // tensorhub base URL for capability-token HTTP calls (uploads, blob refs)
-  repeated string keep = 3;  // legacy v2 field; ignored by protocol v3 workers
+  repeated string keep = 3;  // legacy v2 field; ignored by protocol v4 workers
   // th#697 precision ladder: per-model flavor resolutions for THIS worker's
   // card. The worker applies each entry to bindings whose wire ref equals
   // `ref` (a bare declared ref): download/load `resolved_ref` and apply the
   // `cast` transform at load. Full-replace with the ack. Additive: old
   // workers ignore it and load their declared bindings.
   repeated ModelResolution resolutions = 4;
-  // Full-replace desired model residency for this worker (protocol v3).
+  // Full-replace desired model residency for this worker (protocol v4).
   DesiredResidency desired_residency = 5;
 }
 
@@ -338,7 +338,7 @@ message RunJob {
   string request_id = 1;
   int64  attempt    = 2; // fencing token; bumped by the orchestrator on every (re)queue
   string function_name = 3;
-  bytes  input_payload = 4; // MessagePack-encoded input struct (input file refs pre-materialized to URLs)
+  bytes  input_payload = 4; // canonical MessagePack input; stored Asset refs remain opaque
   int64  timeout_ms = 5;    // execution deadline; 0 = no explicit timeout
   string tenant     = 6;    // invoking owner (org/user slug)
   string invoker_id = 7;    // surfaced read-only as ctx.invoker_id
@@ -362,6 +362,18 @@ message RunJob {
   // instruction (worker's own policy). A worker that cannot serve the named
   // lane answers a typed refusal naming it — never a silent fallback.
   string lane = 14;
+  // Ordered, credential-free stored inputs authorized at submit. The worker
+  // resolves fresh transport URLs for the exact request+attempt capability.
+  repeated InputAsset input_assets = 15;
+}
+
+message InputAsset {
+  string asset_id = 1;
+  string source_ref = 2;
+  string blake3 = 3;
+  int64 size_bytes = 4;
+  string kind = 5;
+  string mime_type = 6;
 }
 
 message RequiredCompileExecution {
@@ -604,7 +616,7 @@ message ModelEvent {
   // Immutable bytes and DesiredResidency generation for this ref transition.
   // A receiver must treat an empty digest or zero generation as identity
   // unknown. Generation fences delayed events after a tag moves or a newer
-  // desired replacement is reconciled without changing protocol version 3.
+  // desired replacement is reconciled without changing protocol version 4.
   string snapshot_digest = 16;
   uint64 residency_generation = 17;
   // Echoes ModelOp.operation_id on ADOPTED/adopt-FAILED and at most one later
@@ -615,14 +627,14 @@ message ModelEvent {
   // one later causal compiled-runtime FAILED; empty on boot-attached cells and
   // ordinary model events.
   string target_incarnation_id = 19;
-  // th#850 managed-tier fill hierarchy (gw#599): bytes actually fetched over
-  // the network (R2 origin) to produce this ON_DISK transition, as opposed
-  // to bytes served from a warm local-disk or endpoint-volume cache hit. Set
-  // on every fresh materialization's ON_DISK event, including an explicit 0
-  // — pair with the DOWNLOADING events' bytes_total for this ref to read
-  // "warm boot ⇒ ~0 R2 bytes". Additive: old workers never set it, which a
-  // receiver must treat as identity unknown (same convention as
-  // snapshot_digest/residency_generation above), never as a verified zero.
+  // th#850: cumulative bytes actually fetched over the network (the public
+  // R2 source, never a local/volume CAS hit) for this DOWNLOADING operation.
+  // Monotonically non-decreasing within one operation, same cadence as
+  // bytes_done; set on progress events and the last one before completion is
+  // authoritative. This is the minimal worker-side signal the "volume-
+  // attached boot with warm blobs ⇒ R2 bytes ≈ 0" runtime assertion needs —
+  // distinct from bytes_done/bytes_total, which count ALL bytes materialized
+  // (including cache hits, both local disk and volume-backed CAS roots).
   int64 network_bytes = 20;
 }
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "gen-worker"
-version = "0.42.0"
+version = "0.43.0"
 description = "A library used to build custom functions in Cozy Creator's serverless function platform."
 readme = "README.md"
 license = "MIT"

src/gen_worker/discovery/discover.py

@@ -48,9 +48,6 @@ def _is_msgspec_struct(t: Any) -> bool:
         return False
 
 
-_MEDIA_ASSET_TYPES = (MediaAsset, ImageAsset, VideoAsset, AudioAsset)
-
-
 def _media_kind(t: type) -> str:
     if issubclass(t, ImageAsset):
         return "image"
@@ -61,6 +58,46 @@ def _media_kind(t: type) -> str:
     return "media"
 
 
+def _annotation_carries_asset(ann: Any, _seen: Optional[Set[type]] = None) -> bool:
+    """True when the annotation subtree can hold an input ``Asset``."""
+    seen = _seen if _seen is not None else set()
+    origin = typing.get_origin(ann)
+    if origin is typing.Annotated:
+        args = typing.get_args(ann)
+        return bool(args) and _annotation_carries_asset(args[0], seen)
+    if origin in (typing.Union, py_types.UnionType):
+        return any(
+            _annotation_carries_asset(arg, seen)
+            for arg in typing.get_args(ann)
+            if arg is not type(None)
+        )
+    if origin in (list, tuple, set, frozenset):
+        args = typing.get_args(ann)
+        return bool(args) and _annotation_carries_asset(args[0], seen)
+    if origin is dict:
+        args = typing.get_args(ann)
+        return len(args) == 2 and _annotation_carries_asset(args[1], seen)
+    if isinstance(ann, type):
+        if issubclass(ann, Asset):
+            return True
+        if issubclass(ann, Tensors):
+            return False
+        if _is_msgspec_struct(ann):
+            if ann in seen:
+                return False
+            seen.add(ann)
+            try:
+                hints = typing.get_type_hints(ann, include_extras=True)
+            except Exception:
+                hints = getattr(ann, "__annotations__", {})
+            return any(
+                _annotation_carries_asset(hints[field], seen)
+                for field in getattr(ann, "__struct_fields__", ()) or ()
+                if field in hints
+            )
+    return False
+
+
 def _collect_payload_moderation_metadata(payload_type: type) -> Dict[str, Any]:
     out: Dict[str, list[Dict[str, str]]] = {"prompts": [], "media": []}
     seen_structs: set[type] = set()
@@ -90,7 +127,20 @@ def walk(ann: Any, path: str) -> None:
                     walk(arg, path)
             return
 
-        if origin in (list, tuple, set, frozenset):
+        if origin in (set, frozenset):
+            # th#886: input-asset manifests are ordered; unordered containers
+            # have no stable occurrence order, so an Asset here is a build error.
+            args = typing.get_args(ann)
+            if args and _annotation_carries_asset(args[0]):
+                raise ValueError(
+                    f"{path}: Asset fields cannot ride unordered set/frozenset "
+                    "containers; use list or tuple"
+                )
+            if args:
+                walk(args[0], f"{path}[]")
+            return
+
+        if origin in (list, tuple):
             args = typing.get_args(ann)
             if args:
                 walk(args[0], f"{path}[]")
@@ -99,14 +149,20 @@ def walk(ann: Any, path: str) -> None:
         if origin is dict:
             args = typing.get_args(ann)
             if len(args) == 2:
+                if args[0] is not str and _annotation_carries_asset(args[1]):
+                    raise ValueError(
+                        f"{path}: Asset-bearing mappings require string keys"
+                    )
                 walk(args[1], f"{path}.*")
             return
 
         if isinstance(ann, type):
-            if issubclass(ann, _MEDIA_ASSET_TYPES):
-                out["media"].append({"field": path, "kind": _media_kind(ann)})
+            if issubclass(ann, Tensors):
                 return
-            if issubclass(ann, (Asset, Tensors)):
+            if issubclass(ann, Asset):
+                # Base Asset/MediaAsset = kind "media"; typed subclasses carry
+                # their exact kind (th#886 manifest vocabulary).
+                out["media"].append({"field": path, "kind": _media_kind(ann)})
                 return
             if _is_msgspec_struct(ann):
                 if ann in seen_structs: