diff --git a/agents/completed.md b/agents/completed.md index c8cbf20..1ddee7a 100644 --- a/agents/completed.md +++ b/agents/completed.md @@ -8,6 +8,46 @@ --- +# #375: `yield ProducedFlavor` conversion contract is gone — from-scratch example (and training-endpoints quant endpoints) silently publish nothing + +**Completed:** yes +**Status:** DONE (2026-07-04, Claude) — settled the ONE contract, option (a): producer endpoints (conversion/dataset/training) write files locally, call `cozy_convert.publish_flavors(ctx, flavors)` — one Tensorhub /commits commit per ProducedFlavor (path = file or dir) — and return a result struct; the executor stays contract-free. The dead yield-shape is deleted AND enforced: @endpoint rejects generator handlers for producer kinds at decoration time (TypeError pointing at publish_flavors), so the old shape fails loudly at import instead of silently streaming. examples/from-scratch rewritten to the contract (returns FromScratchResult w/ revision_ids) + discovery smoke test (kind=conversion, output_mode=single, result struct) + publish_flavors e2e against the fake /commits server. Docs: @endpoint docstring, endpoint-authoring.md (worked example), local-dev.md, produced.py/publish.py docstrings. training-endpoints quant/convert endpoints rewrite against this contract (their #37); live publish-then-consume e2e (tensorhub #542 scenario) deferred to the e2e suite. + +## Metadata +- Category: bug / api-contract +- Status: planned +- Passes: false + +## Tasks +- [x] Decide the ONE contract: (a) endpoints call `cozy_convert.publish_flavors(ctx, flavors)` explicitly and return a result struct, or (b) the executor collects `ProducedFlavor` yields from conversion-kind handlers and publishes them. Pick, document on @endpoint, delete the other shape. +- [x] Fix examples/from-scratch to the chosen contract and cover it (discovery+lock smoke at minimum; ideally the e2e publish-then-consume scenario of tensorhub #542). + +## Acceptance +from-scratch runs against a live stack and its commit lands in the CAS; the dead shape no longer imports. + +--- + +# #374: cozy_convert publish/clone robustness — retries, resume, ingest junk in published trees, silent empty publish + +**Completed:** yes +**Status:** DONE (2026-07-04, Claude) — hub.py: bounded retries w/ backoff + Retry-After on commit POST, part PUTs, complete, and finalize polling (429/5xx/network; abort-DELETE on terminal failure unchanged). run_clone: persistent workdir keyed by sha256(provider|source|destination) under $COZY_CONVERT_WORKDIR (default /cozy-convert) — retained on failure so a retry resumes the HF snapshot (hf local-dir metadata kept for exactly that reason), deleted on success; partial flavor trees are wiped per-retry. Junk filtering: files_from_tree AND _copy_non_weights skip `.cache/huggingface/**` so mirrors are byte-faithful (root dotfiles like a real `.gitignore` still publish). `if not result.published: raise` — publishing nothing can never read as success (empty `outputs` was already defaulted to bf16 by normalize_outputs; the guard now covers every path). Legacy `publish_repo_revision` (~450 LOC, legacy /publish route) deleted from _PublisherMixin along with its local-context stub, finalize-poll constants, and orphaned _helpers; checkpoint publishing is ONLY cozy_convert.publish_flavors (/commits). New tests: retry/junk in test_hub.py + run_clone lifecycle in test_publish.py (fake /commits server, real HTTP). Deliberate: publish_dataset_revision (datasets subsystem) untouched; civitai download resume is #373's scope. + +## Metadata +- Category: bug / cozy_convert +- Status: planned +- Passes: false + +## Tasks +- [x] Live probe: the published repo contains `.cache/huggingface/**` lock/metadata files and `.gitignore` — `ingest_huggingface` snapshots into the tree and `files_from_tree` (hub.py) uploads EVERYTHING. Filter HF-cache internals (or snapshot to a clean copy) so mirrors are byte-faithful to the source repo, not to huggingface_hub's cache layout. +- [x] HubClient has no retry/backoff/429 handling on part PUTs, commit POST, or complete (hub.py:113-156) — one transient S3 hiccup after a multi-GB download+convert fails the whole clone; run_clone `shutil.rmtree`s the workdir in `finally` (clone.py:552) so a retry is a full re-download. Add bounded retries + a keyed persistent workdir for resume. +- [x] `run_clone` with an empty `outputs` list publishes NOTHING and returns success (the specs loop simply doesn't run; the no-publish error only fires when failed_flavors is non-empty). Default to publish-as-is or make empty outputs an error. +- [x] Kill the second publish path: `gen_worker.publish_repo_revision` (request_context/__init__.py:995) still posts to the legacy `/repos/:tenant/:name/publish` route while cozy_convert uses `/commits` (#515 "the ONE publish path"). Convert remaining callers or delete. + +## Acceptance +A mirrored repo's tree equals the source repo's tree; transient network errors don't restart multi-GB work; publishing nothing cannot read as success. + +--- + # #372: transport hardening — auth-failure gating, HelloAck deadline, redirect hop reset, TLS on redirects **Completed:** yes diff --git a/agents/progress.md b/agents/progress.md index 32ab7a0..73fbc76 100644 --- a/agents/progress.md +++ b/agents/progress.md @@ -379,41 +379,3 @@ Tasks: Alternating requests across two endpoints on one GPU show demote/promote transitions (no full reloads) and both run unoffloaded when resident alone. --- - -# #374: cozy_convert publish/clone robustness — retries, resume, ingest junk in published trees, silent empty publish - -**Completed:** no -**Status:** OPEN (2026-07-04, mirror-flow audit + live probe) - -## Metadata -- Category: bug / cozy_convert -- Status: planned -- Passes: false - -## Tasks -- [ ] Live probe: the published repo contains `.cache/huggingface/**` lock/metadata files and `.gitignore` — `ingest_huggingface` snapshots into the tree and `files_from_tree` (hub.py) uploads EVERYTHING. Filter HF-cache internals (or snapshot to a clean copy) so mirrors are byte-faithful to the source repo, not to huggingface_hub's cache layout. -- [ ] HubClient has no retry/backoff/429 handling on part PUTs, commit POST, or complete (hub.py:113-156) — one transient S3 hiccup after a multi-GB download+convert fails the whole clone; run_clone `shutil.rmtree`s the workdir in `finally` (clone.py:552) so a retry is a full re-download. Add bounded retries + a keyed persistent workdir for resume. -- [ ] `run_clone` with an empty `outputs` list publishes NOTHING and returns success (the specs loop simply doesn't run; the no-publish error only fires when failed_flavors is non-empty). Default to publish-as-is or make empty outputs an error. -- [ ] Kill the second publish path: `gen_worker.publish_repo_revision` (request_context/__init__.py:995) still posts to the legacy `/repos/:tenant/:name/publish` route while cozy_convert uses `/commits` (#515 "the ONE publish path"). Convert remaining callers or delete. - -## Acceptance -A mirrored repo's tree equals the source repo's tree; transient network errors don't restart multi-GB work; publishing nothing cannot read as success. - ---- - -# #375: `yield ProducedFlavor` conversion contract is gone — from-scratch example (and training-endpoints quant endpoints) silently publish nothing - -**Completed:** no -**Status:** OPEN (2026-07-04) — examples/from-scratch yields `ProducedFlavor` from a `kind="conversion"` generator, but post-#367/#368 gen_worker contains ZERO ProducedFlavor handling: the executor treats the generator as a streaming handler and nothing publishes. training-endpoints' quant/convert endpoints are written against the same dead contract (their #37 depends on this decision). - -## Metadata -- Category: bug / api-contract -- Status: planned -- Passes: false - -## Tasks -- [ ] Decide the ONE contract: (a) endpoints call `cozy_convert.publish_flavors(ctx, flavors)` explicitly and return a result struct, or (b) the executor collects `ProducedFlavor` yields from conversion-kind handlers and publishes them. Pick, document on @endpoint, delete the other shape. -- [ ] Fix examples/from-scratch to the chosen contract and cover it (discovery+lock smoke at minimum; ideally the e2e publish-then-consume scenario of tensorhub #542). - -## Acceptance -from-scratch runs against a live stack and its commit lands in the CAS; the dead shape no longer imports. diff --git a/docs/endpoint-authoring.md b/docs/endpoint-authoring.md index 198eb10..9b5d9d1 100644 --- a/docs/endpoint-authoring.md +++ b/docs/endpoint-authoring.md @@ -100,14 +100,35 @@ Resources(gpu=True, vram_gb=24, compute_capability=8.0, libraries=("nunchaku",)) ## Kinds `@endpoint(kind="conversion" | "training" | "dataset")` selects the context -subclass the handler receives: `ConversionContext` adds -`publish_repo_revision` / `save_checkpoint` / `mktemp` / `source` / -`destination`; `DatasetContext` adds `publish_dataset_revision` / -`resolve_dataset`; `TrainingContext` adds the repo-metadata RPCs. +subclass the handler receives: `ConversionContext` adds `save_checkpoint` / +`mktemp` / `source` / `destination`; `DatasetContext` adds +`publish_dataset_revision` / `resolve_dataset`; `TrainingContext` adds the +repo-metadata RPCs. + +Producer endpoints publish **explicitly**: write files locally, call +`cozy_convert.publish_flavors(ctx, flavors)` — one Tensorhub commit per +`ProducedFlavor` (path = file or directory) — and return a result struct: + +```python +@endpoint(kind="conversion") +class Convert: + def run(self, ctx: ConversionContext, p: In) -> Out: + out_dir = ctx.mktemp() + ... # write model files under out_dir + commits = publish_flavors( + ctx, [ProducedFlavor(path=out_dir, flavor="bf16")], + destination_repo=p.destination_repo, + ) + return Out(revision_ids=[c.revision_id for c in commits]) +``` + +Generator handlers are rejected for producer kinds — yielding streams +chunks, it never publishes. ## Streaming -An async-generator handler streams; each yielded struct is one chunk: +An async-generator handler streams (inference kinds only); each yielded +struct is one chunk: ```python async def stream(self, ctx, p: In) -> AsyncIterator[Out]: diff --git a/docs/local-dev.md b/docs/local-dev.md index b9a5dd0..f64f23c 100644 --- a/docs/local-dev.md +++ b/docs/local-dev.md @@ -173,10 +173,13 @@ The `.gen-worker-run/` directory is throwaway. Add it to `.gitignore` / ## Conversion / dataset endpoints -`ConversionContext.publish_repo_revision` and `materialize_blob` are -stubbed by default — they print the would-be call to stderr and return -a fake response. Pass `--allow-publish` to call the real tensorhub APIs -(useful for round-tripping against a dev tensorhub). +Checkpoint publishing goes through `cozy_convert.publish_flavors(ctx, +flavors)`, which talks to tensorhub directly using the worker capability +token — with no token configured (plain local runs) it fails loudly +instead of pretending to publish. `ConversionContext.materialize_blob` +is stubbed against the local CAS by default; pass `--allow-publish` to +call the real tensorhub API (useful for round-tripping against a dev +tensorhub). ```bash gen-worker run --payload '{"source":{"ref":"..."},"specs":[...]}' --allow-publish diff --git a/examples/from-scratch/README.md b/examples/from-scratch/README.md index 95042cb..829272e 100644 --- a/examples/from-scratch/README.md +++ b/examples/from-scratch/README.md @@ -1,11 +1,10 @@ # from-scratch -A `@conversion` that emits an **orphan checkpoint** — a brand-new set of weights with no source repo, no parent lineage. The SDK accepts an empty lineage array and lands the checkpoint as a root node in the lineage DAG. +An `@endpoint(kind="conversion")` that publishes an **orphan checkpoint** — a brand-new set of weights with no source repo, no parent lineage. ## What it demonstrates -- The `from-scratch` training kind: `@conversion(kind="from-scratch", concurrency="sequential")` — declares to the runtime that this job genuinely has no upstream model to materialize. -- **`ProducedFlavor`** as the return contract — the function generates weights, writes them to a path, returns `[ProducedFlavor(path=..., flavor=...)]`; the library handles upload + finalize + tag application. -- Tenant code never touches tensorhub's upload API directly — the SDK owns the session lifecycle. +- **The producer publish contract**: write files locally, call `cozy_convert.publish_flavors(ctx, flavors)` — one Tensorhub commit per `ProducedFlavor` — and return a result struct. Nothing publishes implicitly; generator handlers are rejected for producer kinds. +- Tenant code never touches tensorhub's upload API directly — `publish_flavors` owns hashing, presigned part PUTs, dedup, and finalize. ## When to copy it - Generating random-init weights for a new architecture. @@ -13,5 +12,5 @@ A `@conversion` that emits an **orphan checkpoint** — a brand-new set of weigh - Any job that produces checkpoints from nothing (synthesis, distillation from a non-Cozy source, etc.). ## Files -- `from_scratch.py` — the function; uses `torch.manual_seed` for deterministic output. +- `from_scratch.py` — the endpoint; uses `torch.manual_seed` for deterministic output. - `endpoint.toml` — declares CPU-only resources (this example doesn't need GPU). diff --git a/examples/from-scratch/from_scratch.py b/examples/from-scratch/from_scratch.py index a87e5e9..be33ef5 100644 --- a/examples/from-scratch/from_scratch.py +++ b/examples/from-scratch/from_scratch.py @@ -1,18 +1,17 @@ """from-scratch — example ``@endpoint(kind="conversion")`` that generates random weights and publishes them as an orphan checkpoint (no parent lineage). -Tenant code never touches tensorhub's upload contract: it writes files -locally, yields ``ProducedFlavor``s, and ``cozy_convert.publish_flavors`` -turns each into one Tensorhub commit. +The publish contract: a conversion handler writes files locally, calls +``cozy_convert.publish_flavors(ctx, flavors)`` — one Tensorhub commit per +``ProducedFlavor`` — and returns a result struct. Nothing publishes +implicitly; generator handlers are rejected for producer kinds. """ from __future__ import annotations -from typing import Iterator - import msgspec -from cozy_convert import ProducedFlavor +from cozy_convert import ProducedFlavor, publish_flavors from gen_worker import ConversionContext, endpoint @@ -22,15 +21,21 @@ class FromScratchInput(msgspec.Struct, forbid_unknown_fields=True): hidden_dim: int = 64 +class FromScratchResult(msgspec.Struct): + destination_repo: str + revision_ids: list[str] + published_files: int + + @endpoint(kind="conversion") class FromScratch: - """Generate random weights and emit them as an orphan checkpoint.""" + """Generate random weights and publish them as an orphan checkpoint.""" def generate( self, ctx: ConversionContext, payload: FromScratchInput, - ) -> Iterator[ProducedFlavor]: + ) -> FromScratchResult: import torch from safetensors.torch import save_file @@ -43,7 +48,17 @@ def generate( } weights_path = ctx.mktemp() / "weights.safetensors" save_file(weights, str(weights_path)) - yield ProducedFlavor(path=weights_path, flavor="fp32") + + commits = publish_flavors( + ctx, + [ProducedFlavor(path=weights_path, flavor="fp32")], + destination_repo=payload.destination_repo, + ) + return FromScratchResult( + destination_repo=payload.destination_repo, + revision_ids=[c.revision_id for c in commits], + published_files=sum(c.uploaded + c.deduped for c in commits), + ) -__all__ = ["FromScratchInput", "FromScratch"] +__all__ = ["FromScratchInput", "FromScratchResult", "FromScratch"] diff --git a/packages/cozy_convert/src/cozy_convert/clone.py b/packages/cozy_convert/src/cozy_convert/clone.py index 35500e9..23f993b 100644 --- a/packages/cozy_convert/src/cozy_convert/clone.py +++ b/packages/cozy_convert/src/cozy_convert/clone.py @@ -11,6 +11,8 @@ from __future__ import annotations +import hashlib +import logging import os import re import shutil @@ -23,6 +25,8 @@ from .ingest import IngestedSource, ingest_civitai, ingest_huggingface from .writer import MAX_SAFETENSORS_SHARD_BYTES, shard_safetensors_by_offset +logger = logging.getLogger(__name__) + _PUBLIC_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9.-]{0,127}$") _PUBLIC_TAG_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,62}$") @@ -177,6 +181,8 @@ def _copy_non_weights(source_dir: Path, out_dir: Path, *, skip_components: set[s if not f.is_file(): continue rel = f.relative_to(source_dir) + if rel.parts[:2] == (".cache", "huggingface"): + continue # hf local-dir download metadata, not repo content comp = rel.parts[0] if len(rel.parts) > 1 else "" name = f.name is_weightish = _is_weight(f) or name.endswith(".safetensors.index.json") @@ -326,6 +332,20 @@ def build_flavor_tree( # run_clone — ingest, convert, ONE finalize path # --------------------------------------------------------------------------- +def _clone_workdir(provider: str, source_key: str, destination: str) -> Path: + """Persistent workdir keyed by (provider, source, destination): a failed + clone keeps its downloaded snapshot so a retry resumes instead of + re-downloading. Deleted on success. Base dir: ``$COZY_CONVERT_WORKDIR`` + or ``/cozy-convert``.""" + base = Path(os.environ.get("COZY_CONVERT_WORKDIR", "").strip() + or Path(tempfile.gettempdir()) / "cozy-convert") + digest = hashlib.sha256( + f"{provider}|{source_key}|{destination}".encode("utf-8")).hexdigest()[:16] + workdir = base / f"clone-{digest}" + workdir.mkdir(parents=True, exist_ok=True) + return workdir + + def run_clone( ctx: Any, *, @@ -359,7 +379,11 @@ def _progress(p: float, stage: str) -> None: except Exception: pass - workdir = Path(tempfile.mkdtemp(prefix=f"clone-{getattr(ctx, 'request_id', 'x')}-")) + source_key = source_ref if provider == "huggingface" else str(civitai_model_version_id or 0) + if source_revision: + source_key = f"{source_key}@{source_revision}" + workdir = _clone_workdir(provider, source_key, destination) + succeeded = False try: _progress(0.05, "clone.ingest") @@ -423,8 +447,14 @@ def _dl_progress(done: int, total: Optional[int]) -> None: attrs = dict(source.attrs) flavor_label = source_dtype or spec.dtype else: + # Wipe any partial flavor tree from a prior failed run — + # only the downloaded source is resumable. + flavor_dir = workdir / f"flavor-{spec.label}" + shutil.rmtree(flavor_dir, ignore_errors=True) + shutil.rmtree(workdir / f"flavor-{spec.label}.__repack__", + ignore_errors=True) tree, attrs = build_flavor_tree( - source, spec, workdir / f"flavor-{spec.label}", + source, spec, flavor_dir, quantize_components=quantize_components, ) except InlineConversionNotPossible as exc: @@ -490,8 +520,10 @@ def _dl_progress(done: int, total: Optional[int]) -> None: "total_bytes": commit.total_bytes, }) - if not result.published and result.failed_flavors: - reasons = "; ".join(str(f.get("reason") or "") for f in result.failed_flavors) + if not result.published: + reasons = "; ".join( + str(f.get("reason") or "") for f in result.failed_flavors + ) or "no output spec produced anything" raise RuntimeError(f"clone produced no publishable flavor: {reasons}") result.metadata["destination_repo"] = destination @@ -499,9 +531,13 @@ def _dl_progress(done: int, total: Optional[int]) -> None: if result.failed_flavors: result.metadata["failed_flavor_count"] = str(len(result.failed_flavors)) _progress(1.0, "clone.completed") + succeeded = True return result finally: - shutil.rmtree(workdir, ignore_errors=True) + if succeeded: + shutil.rmtree(workdir, ignore_errors=True) + else: + logger.warning("clone failed; workdir retained for resume: %s", workdir) def from_huggingface(ctx: Any, payload: Any, *, hf_token: str | None = None) -> CloneResult: diff --git a/packages/cozy_convert/src/cozy_convert/hub.py b/packages/cozy_convert/src/cozy_convert/hub.py index 63c858f..0b5142c 100644 --- a/packages/cozy_convert/src/cozy_convert/hub.py +++ b/packages/cozy_convert/src/cozy_convert/hub.py @@ -17,19 +17,61 @@ from __future__ import annotations import json +import logging +import random import time import urllib.parse from dataclasses import dataclass, field from pathlib import Path -from typing import Any, Mapping, Optional +from typing import Any, Callable, Mapping, Optional import requests +logger = logging.getLogger(__name__) + +_RETRY_ATTEMPTS = 5 +_RETRY_BASE_DELAY_S = 1.0 +_RETRY_MAX_DELAY_S = 30.0 + class HubPublishError(RuntimeError): """Terminal failure talking to tensorhub's commit API.""" +def _retry_after_s(resp: requests.Response) -> Optional[float]: + try: + value = float(str(resp.headers.get("Retry-After") or "").strip()) + except Exception: + return None + return min(value, _RETRY_MAX_DELAY_S) if value > 0 else None + + +def _send_with_retries(what: str, send: Callable[[], requests.Response]) -> requests.Response: + """Bounded retries on network errors, 429, and 5xx (honors Retry-After). + + Returns the last response for non-retryable statuses — callers keep their + own status handling. + """ + delay = _RETRY_BASE_DELAY_S + for attempt in range(1, _RETRY_ATTEMPTS + 1): + try: + resp = send() + except requests.RequestException as exc: + if attempt == _RETRY_ATTEMPTS: + raise HubPublishError(f"{what} failed (network): {exc}") from exc + else: + code = int(resp.status_code) + if code != 429 and code < 500: + return resp + if attempt == _RETRY_ATTEMPTS: + return resp + delay = _retry_after_s(resp) or delay + logger.warning("%s retrying (attempt %d/%d)", what, attempt, _RETRY_ATTEMPTS) + time.sleep(delay + random.uniform(0, delay * 0.1)) + delay = min(delay * 2, _RETRY_MAX_DELAY_S) + raise HubPublishError(f"{what} failed after {_RETRY_ATTEMPTS} attempts") + + def blake3_file(path: Path, *, chunk: int = 8 * 1024 * 1024) -> str: from blake3 import blake3 @@ -113,12 +155,12 @@ def _repo_path(self, destination_repo: str) -> str: ) def _post(self, path: str, payload: Optional[dict] = None, *, timeout: float | None = None) -> requests.Response: - return requests.post( + return _send_with_retries(f"POST {path}", lambda: requests.post( f"{self.base_url}{path}", headers=self._headers(), data=json.dumps(payload) if payload is not None else None, timeout=timeout or self.timeout_s, - ) + )) @staticmethod def _json(resp: requests.Response) -> dict[str, Any]: @@ -141,7 +183,10 @@ def _upload_one(self, repo_path: str, revision_id: str, entry: Mapping[str, Any] buf = f.read(part_size) if not buf and i > 0: break - resp = requests.put(url, data=buf, timeout=self.timeout_s * 5) + def _put(u: str = url, b: bytes = buf) -> requests.Response: + return requests.put(u, data=b, timeout=self.timeout_s * 5) + + resp = _send_with_retries(f"part PUT {entry.get('path')!r} #{i + 1}", _put) if resp.status_code < 200 or resp.status_code >= 300: raise HubPublishError( f"part PUT failed ({resp.status_code}) for {entry.get('path')!r}") @@ -287,12 +332,18 @@ def commit( def files_from_tree(tree: Path, *, prefix: str = "") -> list[CommitFile]: - """Build CommitFile entries for every regular file under ``tree``.""" + """Build CommitFile entries for every regular file under ``tree``. + + ``.cache/huggingface/**`` is skipped: huggingface_hub's local-dir download + metadata is cache-layout junk, never repo content.""" tree = Path(tree) out: list[CommitFile] = [] for f in sorted(tree.rglob("*")): if not f.is_file(): continue + rel_parts = f.relative_to(tree).parts + if rel_parts[:2] == (".cache", "huggingface"): + continue rel = f.relative_to(tree).as_posix() if prefix: rel = f"{prefix.rstrip('/')}/{rel}" diff --git a/packages/cozy_convert/src/cozy_convert/produced.py b/packages/cozy_convert/src/cozy_convert/produced.py index a90d3e4..7df1a6d 100644 --- a/packages/cozy_convert/src/cozy_convert/produced.py +++ b/packages/cozy_convert/src/cozy_convert/produced.py @@ -1,9 +1,10 @@ -"""ProducedFlavor — what a tenant transform returns per output. +"""ProducedFlavor — what a tenant transform hands to ``publish_flavors``. -A tenant's ``@conversion`` returns ``list[ProducedFlavor]`` — one -entry per flavor the job produces into the destination checkpoint. The -library uploads each flavor's ``path`` (file OR directory) and attaches -the declared ``attributes`` to the final checkpoint flavor publish payload. +A producer endpoint builds ``list[ProducedFlavor]`` — one entry per flavor +the job produces into the destination checkpoint — and calls +``cozy_convert.publish_flavors(ctx, flavors)``. The library uploads each +flavor's ``path`` (file OR directory) as one Tensorhub commit and attaches +the declared ``attributes`` to the commit payload. Attribute-bag ownership (issue #22 — server-authoritative metadata): - Tenant declares ONLY tenant-specific attributes (technique config, diff --git a/packages/cozy_convert/src/cozy_convert/publish.py b/packages/cozy_convert/src/cozy_convert/publish.py index 6739bd4..f9d5a76 100644 --- a/packages/cozy_convert/src/cozy_convert/publish.py +++ b/packages/cozy_convert/src/cozy_convert/publish.py @@ -1,8 +1,10 @@ -"""Publish ProducedFlavor outputs to Tensorhub (replaces conversion/dispatch.py). +"""Publish ProducedFlavor outputs to Tensorhub — THE producer publish contract. -A conversion / dataset tenant returns ``list[ProducedFlavor]``; each flavor's +A conversion / dataset / training endpoint writes files locally, calls +``publish_flavors(ctx, flavors)``, and returns a result struct. Each flavor's ``path`` (file or directory) becomes ONE Tensorhub commit against the -destination repo declared on the job payload. +destination repo (explicit ``destination_repo=`` or the job payload's +reserved ``destination.repo`` field). Nothing publishes implicitly. """ from __future__ import annotations diff --git a/packages/cozy_convert/tests/conftest.py b/packages/cozy_convert/tests/conftest.py new file mode 100644 index 0000000..5d641e8 --- /dev/null +++ b/packages/cozy_convert/tests/conftest.py @@ -0,0 +1,117 @@ +"""Shared fake of tensorhub's /commits API (threaded HTTP server).""" + +from __future__ import annotations + +import json +import threading +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from typing import Any + +import pytest + +from cozy_convert.hub import HubClient + + +class _FakeHub(BaseHTTPRequestHandler): + server_version = "FakeTensorhub/1.0" + state: dict[str, Any] = {} + + def log_message(self, *args: Any) -> None: # silence + pass + + def _read_json(self) -> dict: + n = int(self.headers.get("Content-Length") or 0) + body = self.rfile.read(n) if n else b"" + return json.loads(body) if body else {} + + def _send(self, code: int, payload: dict | None = None) -> None: + body = json.dumps(payload or {}).encode() + self.send_response(code) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_POST(self) -> None: # noqa: N802 + st = _FakeHub.state + if self.path.endswith("/commits"): + if st.get("fail_commit_posts", 0) > 0: + st["fail_commit_posts"] -= 1 + self._send(503, {"error": "unavailable"}) + return + req = self._read_json() + st["commit_request"] = req + st.setdefault("commit_requests", []).append(req) + st["auth"] = self.headers.get("Authorization", "") + uploads = [] + for i, op in enumerate(req.get("operations", [])): + if op["type"] != "add": + continue + if op["blake3"] in st.get("existing_blobs", set()): + uploads.append({"path": op["path"], "blake3": op["blake3"], "exists": True}) + continue + uid = f"up-{i}" + base = f"http://127.0.0.1:{self.server.server_port}" + uploads.append({ + "path": op["path"], "blake3": op["blake3"], "exists": False, + "upload_id": uid, + "part_urls": [f"{base}/put/{uid}/1"], + "part_size": max(int(op["size_bytes"]), 1), + "total_parts": 1, + }) + self._send(201, {"revision_id": "rev-1", "uploads": uploads, + "deletions": [], "copies": [], "tags": req.get("tags") or [], + "mode": req.get("mode") or "merge"}) + return + if "/uploads/" in self.path and self.path.endswith("/complete"): + if st.get("fail_completes", 0) > 0: + st["fail_completes"] -= 1 + self._send(500, {"error": "boom"}) + return + st.setdefault("completed", []).append(self.path) + body = self._read_json() + st.setdefault("complete_bodies", []).append(body) + self._send(200, {"ok": True}) + return + if self.path.endswith("/finalize"): + n = st.get("finalize_calls", 0) + 1 + st["finalize_calls"] = n + if n == 1: + self._send(202, {"status": "running"}) # first call -> poll + else: + self._send(200, {"ok": True, "checkpoint_id": "blake3:abc"}) + return + self._send(404, {"error": "not_found"}) + + def do_PUT(self) -> None: # noqa: N802 + n = int(self.headers.get("Content-Length") or 0) + data = self.rfile.read(n) if n else b"" + st = _FakeHub.state + if st.get("fail_puts", 0) > 0: + st["fail_puts"] -= 1 + self.send_response(500) + self.send_header("Content-Length", "0") + self.end_headers() + return + _FakeHub.state.setdefault("put_bytes", {})[self.path] = data + self.send_response(200) + self.send_header("ETag", '"etag-1"') + self.send_header("Content-Length", "0") + self.end_headers() + + +@pytest.fixture() +def fake_hub(): + _FakeHub.state = {"existing_blobs": set()} + server = ThreadingHTTPServer(("127.0.0.1", 0), _FakeHub) + t = threading.Thread(target=server.serve_forever, daemon=True) + t.start() + yield server + server.shutdown() + + +def _client(server) -> HubClient: + return HubClient( + base_url=f"http://127.0.0.1:{server.server_port}", + token="cap-token", owner="acme", + ) diff --git a/packages/cozy_convert/tests/test_hub.py b/packages/cozy_convert/tests/test_hub.py index 06f7fba..1c6a101 100644 --- a/packages/cozy_convert/tests/test_hub.py +++ b/packages/cozy_convert/tests/test_hub.py @@ -1,109 +1,19 @@ """HubClient against a threaded fake of tensorhub's /commits API. Exercises the real HTTP code path: POST /commits (presign), part PUTs, -per-upload complete, finalize (202 poll -> 200), and blake3 dedup skips. +per-upload complete, finalize (202 poll -> 200), blake3 dedup skips, and +bounded retries on transient failures. """ from __future__ import annotations -import json -import threading -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path -from typing import Any import pytest -from cozy_convert.hub import CommitFile, HubClient, HubPublishError, blake3_file, files_from_tree - - -class _FakeHub(BaseHTTPRequestHandler): - server_version = "FakeTensorhub/1.0" - state: dict[str, Any] = {} - - def log_message(self, *args: Any) -> None: # silence - pass - - def _read_json(self) -> dict: - n = int(self.headers.get("Content-Length") or 0) - body = self.rfile.read(n) if n else b"" - return json.loads(body) if body else {} - - def _send(self, code: int, payload: dict | None = None) -> None: - body = json.dumps(payload or {}).encode() - self.send_response(code) - self.send_header("Content-Type", "application/json") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - - def do_POST(self) -> None: # noqa: N802 - st = _FakeHub.state - if self.path.endswith("/commits"): - req = self._read_json() - st["commit_request"] = req - st["auth"] = self.headers.get("Authorization", "") - uploads = [] - for i, op in enumerate(req.get("operations", [])): - if op["type"] != "add": - continue - if op["blake3"] in st.get("existing_blobs", set()): - uploads.append({"path": op["path"], "blake3": op["blake3"], "exists": True}) - continue - uid = f"up-{i}" - base = f"http://127.0.0.1:{self.server.server_port}" - uploads.append({ - "path": op["path"], "blake3": op["blake3"], "exists": False, - "upload_id": uid, - "part_urls": [f"{base}/put/{uid}/1"], - "part_size": max(int(op["size_bytes"]), 1), - "total_parts": 1, - }) - self._send(201, {"revision_id": "rev-1", "uploads": uploads, - "deletions": [], "copies": [], "tags": req.get("tags") or [], - "mode": req.get("mode") or "merge"}) - return - if "/uploads/" in self.path and self.path.endswith("/complete"): - st.setdefault("completed", []).append(self.path) - body = self._read_json() - st.setdefault("complete_bodies", []).append(body) - self._send(200, {"ok": True}) - return - if self.path.endswith("/finalize"): - n = st.get("finalize_calls", 0) + 1 - st["finalize_calls"] = n - if n == 1: - self._send(202, {"status": "running"}) # first call -> poll - else: - self._send(200, {"ok": True, "checkpoint_id": "blake3:abc"}) - return - self._send(404, {"error": "not_found"}) - - def do_PUT(self) -> None: # noqa: N802 - n = int(self.headers.get("Content-Length") or 0) - data = self.rfile.read(n) if n else b"" - _FakeHub.state.setdefault("put_bytes", {})[self.path] = data - self.send_response(200) - self.send_header("ETag", '"etag-1"') - self.send_header("Content-Length", "0") - self.end_headers() - - -@pytest.fixture() -def fake_hub(): - _FakeHub.state = {"existing_blobs": set()} - server = ThreadingHTTPServer(("127.0.0.1", 0), _FakeHub) - t = threading.Thread(target=server.serve_forever, daemon=True) - t.start() - yield server - server.shutdown() - - -def _client(server) -> HubClient: - return HubClient( - base_url=f"http://127.0.0.1:{server.server_port}", - token="cap-token", owner="acme", - ) +from cozy_convert.hub import CommitFile, HubPublishError, blake3_file, files_from_tree + +from conftest import _FakeHub, _client def test_commit_uploads_completes_and_finalizes(fake_hub, tmp_path: Path, monkeypatch) -> None: @@ -165,3 +75,36 @@ def test_destination_must_be_owner_repo(fake_hub) -> None: destination_repo="just-a-name", files=[CommitFile(path="x", local_path=Path("/nonexistent"))], ) + + +def test_transient_failures_are_retried(fake_hub, tmp_path: Path, monkeypatch) -> None: + """One 503 on the commit POST, one 500 on a part PUT, one 500 on complete: + the commit still lands (bounded retries, no restart of the whole job).""" + monkeypatch.setattr("time.sleep", lambda *_: None) + f = tmp_path / "model.safetensors" + f.write_bytes(b"\x02" * 48) + _FakeHub.state.update( + {"fail_commit_posts": 1, "fail_puts": 1, "fail_completes": 1, + "finalize_calls": 1}) + + result = _client(fake_hub).commit( + destination_repo="acme/my-model", + files=[CommitFile(path="model.safetensors", local_path=f)], + ) + assert result.revision_id == "rev-1" + assert result.uploaded == 1 + # the successful PUT actually carried the bytes + assert list(_FakeHub.state["put_bytes"].values()) == [b"\x02" * 48] + + +def test_files_from_tree_skips_hf_cache_junk(tmp_path: Path) -> None: + (tmp_path / "config.json").write_text("{}") + junk = tmp_path / ".cache" / "huggingface" / "download" + junk.mkdir(parents=True) + (junk / "config.json.metadata").write_text("etag") + (tmp_path / ".cache" / "huggingface" / ".gitignore").write_text("*") + # a real dotfile at repo root is content, not junk + (tmp_path / ".gitignore").write_text("logs/") + + paths = [f.path for f in files_from_tree(tmp_path)] + assert paths == [".gitignore", "config.json"] diff --git a/packages/cozy_convert/tests/test_publish.py b/packages/cozy_convert/tests/test_publish.py new file mode 100644 index 0000000..dbee68c --- /dev/null +++ b/packages/cozy_convert/tests/test_publish.py @@ -0,0 +1,174 @@ +"""publish_flavors + run_clone against the fake /commits server. + +Covers the producer publish contract (#375) and clone robustness (#374): +explicit publish, HF-cache junk never uploaded, keyed workdir retained on +failure / removed on success, and no-publish-cannot-read-as-success. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from cozy_convert import ProducedFlavor, publish_flavors +from cozy_convert.clone import run_clone +from cozy_convert.ingest import IngestedSource + +from conftest import _FakeHub + + +class _Ctx: + def __init__(self, server) -> None: + self._file_api_base_url = f"http://127.0.0.1:{server.server_port}" + self._worker_capability_token = "cap-token" + self.owner = "acme" + self.request_id = "req-1" + self.destination = {"repo": "acme/fallback"} + + +def test_publish_flavors_file_and_dir(fake_hub, tmp_path: Path) -> None: + _FakeHub.state["finalize_calls"] = 1 + weights = tmp_path / "weights.safetensors" + weights.write_bytes(b"\x00" * 32) + tree = tmp_path / "tree" + (tree / ".cache" / "huggingface").mkdir(parents=True) + (tree / ".cache" / "huggingface" / "x.lock").write_text("junk") + (tree / "config.json").write_text("{}") + + results = publish_flavors( + _Ctx(fake_hub), + [ + ProducedFlavor(path=weights, flavor="fp32"), + ProducedFlavor(path=tree, flavor="bf16"), + ], + destination_repo="acme/dest", + ) + assert [r.revision_id for r in results] == ["rev-1", "rev-1"] + reqs = _FakeHub.state["commit_requests"] + assert len(reqs) == 2 + assert reqs[0]["flavor"] == "fp32" + assert [op["path"] for op in reqs[0]["operations"]] == ["weights.safetensors"] + # directory flavor: HF-cache junk never reaches the commit + assert [op["path"] for op in reqs[1]["operations"]] == ["config.json"] + + +def test_publish_flavors_destination_falls_back_to_ctx(fake_hub, tmp_path: Path) -> None: + _FakeHub.state["finalize_calls"] = 1 + f = tmp_path / "weights.safetensors" + f.write_bytes(b"\x01" * 16) + publish_flavors(_Ctx(fake_hub), [ProducedFlavor(path=f, flavor="fp32")]) + assert _FakeHub.state["auth"] == "Bearer cap-token" + + +def test_publish_flavors_requires_destination(fake_hub, tmp_path: Path) -> None: + ctx = _Ctx(fake_hub) + ctx.destination = {} + f = tmp_path / "w.safetensors" + f.write_bytes(b"\x01") + with pytest.raises(ValueError, match="destination_repo"): + publish_flavors(ctx, [ProducedFlavor(path=f)]) + + +# --------------------------------------------------------------------------- +# run_clone: junk filtering, workdir lifecycle, empty publish +# --------------------------------------------------------------------------- + + +def _fake_source(dest_dir: Path) -> IngestedSource: + return IngestedSource( + provider="huggingface", + source_ref="org/tiny", + source_revision="sha-1", + dir=dest_dir, + layout="diffusers", + model_family="", + model_family_variant="", + classification=None, + attrs={"dtype": "bf16"}, + metadata={"source_provider": "huggingface"}, + repo_spec={"kind": "model", "library_name": "diffusers"}, + ) + + +def _install_fake_ingest(monkeypatch, *, fail_first: bool = False) -> dict: + calls = {"n": 0} + + def fake_ingest(source_ref, dest_dir, **kwargs): + calls["n"] += 1 + dest_dir = Path(dest_dir) + dest_dir.mkdir(parents=True, exist_ok=True) + (dest_dir / "partial.bin").write_bytes(b"partial") + if fail_first and calls["n"] == 1: + raise RuntimeError("network died mid-download") + (dest_dir / "config.json").write_text("{}") + junk = dest_dir / ".cache" / "huggingface" + junk.mkdir(parents=True, exist_ok=True) + (junk / "config.json.metadata").write_text("etag") + (junk / ".gitignore").write_text("*") + return _fake_source(dest_dir) + + monkeypatch.setattr("cozy_convert.clone.ingest_huggingface", fake_ingest) + return calls + + +def test_run_clone_publishes_clean_tree_and_removes_workdir( + fake_hub, tmp_path: Path, monkeypatch +) -> None: + _FakeHub.state["finalize_calls"] = 1 + monkeypatch.setenv("COZY_CONVERT_WORKDIR", str(tmp_path / "work")) + _install_fake_ingest(monkeypatch) + + result = run_clone( + _Ctx(fake_hub), provider="huggingface", source_ref="org/tiny", + destination_repo="acme/dest", + ) + assert len(result.published) == 1 + ops = [op["path"] for op in _FakeHub.state["commit_request"]["operations"]] + assert "config.json" in ops + assert not any(o.startswith(".cache/") for o in ops) + # success: keyed workdir is gone + assert list((tmp_path / "work").glob("clone-*")) == [] + + +def test_run_clone_failure_retains_workdir_for_resume( + fake_hub, tmp_path: Path, monkeypatch +) -> None: + _FakeHub.state["finalize_calls"] = 1 + monkeypatch.setenv("COZY_CONVERT_WORKDIR", str(tmp_path / "work")) + calls = _install_fake_ingest(monkeypatch, fail_first=True) + + with pytest.raises(RuntimeError, match="network died"): + run_clone( + _Ctx(fake_hub), provider="huggingface", source_ref="org/tiny", + destination_repo="acme/dest", + ) + kept = list((tmp_path / "work").glob("clone-*")) + assert len(kept) == 1 + assert (kept[0] / "source" / "partial.bin").exists() + + # retry lands in the SAME keyed workdir and succeeds; workdir removed + result = run_clone( + _Ctx(fake_hub), provider="huggingface", source_ref="org/tiny", + destination_repo="acme/dest", + ) + assert calls["n"] == 2 + assert len(result.published) == 1 + assert list((tmp_path / "work").glob("clone-*")) == [] + + +def test_run_clone_publishing_nothing_is_an_error( + fake_hub, tmp_path: Path, monkeypatch +) -> None: + monkeypatch.setenv("COZY_CONVERT_WORKDIR", str(tmp_path / "work")) + + def empty_ingest(source_ref, dest_dir, **kwargs): + Path(dest_dir).mkdir(parents=True, exist_ok=True) + return _fake_source(Path(dest_dir)) + + monkeypatch.setattr("cozy_convert.clone.ingest_huggingface", empty_ingest) + with pytest.raises(RuntimeError, match="no publishable flavor"): + run_clone( + _Ctx(fake_hub), provider="huggingface", source_ref="org/tiny", + destination_repo="acme/dest", + ) diff --git a/src/gen_worker/api/decorators.py b/src/gen_worker/api/decorators.py index bbf1f85..e5734e0 100644 --- a/src/gen_worker/api/decorators.py +++ b/src/gen_worker/api/decorators.py @@ -15,7 +15,12 @@ def setup(self, model: FluxPipeline): self.model = model def generate(self, ctx, p: Input) -> Output: ... * ``kind="conversion" | "training" | "dataset"`` selects the context subclass. -* An async-generator handler streams; there is no separate streaming decorator. + Producer kinds publish explicitly — write files locally, call + ``cozy_convert.publish_flavors(ctx, flavors)`` (one Tensorhub commit per + ``ProducedFlavor``), and return a result struct. Generator handlers are + rejected for producer kinds: nothing is ever published by yielding. +* An async-generator handler streams (inference only); there is no separate + streaming decorator. * ``runtime="vllm"`` boots an engine-hosting server subprocess before setup. * ``variants={name: (binding, Resources)}`` stamps one separately-routable, separately-placeable function per variant from a single handler body. @@ -112,6 +117,21 @@ def _validate_handler_shape(owner: str, fn: Callable[..., Any], *, is_method: bo ) +def _reject_producer_generator(owner: str, fn: Callable[..., Any], kind: str) -> None: + """Producer kinds (conversion/training/dataset) publish explicitly and + return a result; a generator handler would stream chunks and publish + NOTHING. Fail at decoration time instead of silently at runtime.""" + if kind == "inference": + return + if inspect.isgeneratorfunction(fn) or inspect.isasyncgenfunction(fn): + raise TypeError( + f"@endpoint(kind={kind!r}): {owner} must not be a generator. " + "Producer endpoints write files locally, publish explicitly " + "(cozy_convert.publish_flavors(ctx, flavors)), and return a " + "result struct; streaming generators are inference-only." + ) + + def _normalize_models( model: Optional[Binding], models: Optional[Mapping[str, Binding]] ) -> dict[str, Binding]: @@ -301,6 +321,8 @@ def _decorate_class( runtime: Optional[str], ) -> type: handlers = _find_handler_methods(cls) + for attr, member in handlers: + _reject_producer_generator(f"{cls.__name__}.{attr}", member, kind) models = _resolve_single_slot(cls, models, handlers) _validate_class_models(cls, models) @@ -345,6 +367,7 @@ def _decorate_function( name: Optional[str], ) -> Callable[..., Any]: _validate_handler_shape(fn.__name__, fn, is_method=False) + _reject_producer_generator(fn.__name__, fn, kind) if "" in models: binding = models.pop("") injected = [p.name for p in _handler_params(fn, is_method=False)[2:]] diff --git a/src/gen_worker/cli/local_context.py b/src/gen_worker/cli/local_context.py index 11f27f3..575c97b 100644 --- a/src/gen_worker/cli/local_context.py +++ b/src/gen_worker/cli/local_context.py @@ -10,9 +10,10 @@ - ``save_bytes`` / ``save_file`` materialize files under ``./.gen-worker-run/outputs/`` and return an Asset with ``local_path`` set (no tensorhub upload). -- ``publish_repo_revision`` and ``materialize_blob`` on ConversionContext - print the would-be payload to stderr and return a fake response, gated - by ``--allow-publish``. +- ``materialize_blob`` on ConversionContext falls back to the local CAS + unless ``--allow-publish`` delegates to the real implementation. + (Checkpoint publishing goes through ``cozy_convert.publish_flavors``, + which talks to tensorhub directly and fails loudly without credentials.) - ``_canceled`` is toggled by the installed SIGINT handler in ``run.py``. Construction is shaped so the only producer of a LocalRequestContext is the @@ -133,25 +134,11 @@ class LocalRequestContext(LocalRequestContextMixin, RequestContext): class LocalConversionContext(LocalRequestContextMixin, ConversionContext): """Conversion-kind local context. - ``publish_repo_revision`` / ``materialize_blob`` are stubbed: they print - the would-be payload to stderr and return a fake response unless + ``materialize_blob`` is stubbed against the local CAS unless ``--allow-publish`` was passed (in which case we delegate to the real implementation — useful for round-tripping against a dev tensorhub). """ - def publish_repo_revision(self, **kwargs: Any) -> Dict[str, Any]: # type: ignore[override] - if self._local_allow_publish: - return super().publish_repo_revision(**kwargs) - _stderr_emitter({ - "kind": "publish_repo_revision.stubbed", - "would_publish": kwargs, - }) - return { - "stubbed": True, - "destination_repo": kwargs.get("destination_repo"), - "revision_id": f"local-{uuid.uuid4().hex[:8]}", - } - def materialize_blob(self, digest: str, dest: "str | os.PathLike[str]") -> Path: # type: ignore[override] if self._local_allow_publish: return super().materialize_blob(digest, dest) diff --git a/src/gen_worker/cli/run.py b/src/gen_worker/cli/run.py index 3b249dc..34a60e4 100644 --- a/src/gen_worker/cli/run.py +++ b/src/gen_worker/cli/run.py @@ -94,9 +94,8 @@ def add_subparser(sub: argparse._SubParsersAction[Any]) -> None: p.add_argument( "--allow-publish", action="store_true", help=( - "Allow ConversionContext.publish_repo_revision / materialize_blob " - "to call the real tensorhub APIs. Default: stubbed (print payload " - "to stderr, return a fake response)." + "Allow ConversionContext.materialize_blob to call the real " + "tensorhub APIs. Default: stubbed against the local CAS." ), ) p.add_argument( diff --git a/src/gen_worker/request_context/__init__.py b/src/gen_worker/request_context/__init__.py index ab4f2f5..71d5425 100644 --- a/src/gen_worker/request_context/__init__.py +++ b/src/gen_worker/request_context/__init__.py @@ -5,7 +5,6 @@ import logging import os import base64 -import random import re import shutil import tempfile @@ -57,10 +56,6 @@ def _as_asset(asset: Asset, cls: type) -> Any: logger = logging.getLogger(__name__) -_REPO_REVISION_FINALIZE_REQUEST_TIMEOUT_S = 30 -_REPO_REVISION_FINALIZE_POLL_MAX_S = 30 * 60 -_REPO_REVISION_FINALIZE_POLL_INITIAL_S = 1.0 -_REPO_REVISION_FINALIZE_POLL_MAX_DELAY_S = 10.0 # Helpers, constants, and JWT/SSRF utilities live in _helpers.py. They are # re-exported here so existing `from gen_worker.request_context import _foo` @@ -75,7 +70,6 @@ def _as_asset(asset: Asset, cls: type) -> Any: _FILE_API_STREAM_CHUNK_TIMEOUT_S, _FILE_API_STREAM_FINALIZE_TIMEOUT_S, _FILE_API_STREAM_REPLAY_TIMEOUT_S, - _PUBLIC_TAG_RE, _canonicalize_model_ref_string, _decode_unverified_jwt_claims, _encode_ref_for_url, @@ -84,11 +78,9 @@ def _as_asset(asset: Asset, cls: type) -> Any: _infer_mime_type, _infer_tensors_format, _is_private_ip_str, - _normalize_destination_repo_tags, _normalize_output_ref, _normalize_repo_name, _parse_owner_repo, - _parse_owner_repo_with_optional_tag, _require_worker_capability_token, _resolve_hint_first_string, _sha256_file, @@ -734,13 +726,13 @@ def _save_file_create(self, ref: str, local_path: str | os.PathLike[str]) -> Ass # # RequestContext is the per-inference base. Conversion, dataset-producing, # and trainer endpoints get richer subclasses that carry the -# producer-contract RPCs (publish_repo_revision, publish_dataset_revision, -# resolve_dataset, read/write_repo_metadata, materialize_blob). +# producer-contract RPCs (publish_dataset_revision, resolve_dataset, +# read/write_repo_metadata, materialize_blob). # # ConversionContext / DatasetContext / TrainingContext share `_PublisherMixin` # for the producer-contract HTTP helpers (repo metadata read/write, blob -# fetch + materialization by digest). publish_repo_revision lives on the base -# RequestContext — producer-style @inference handlers (clone) call it too. +# fetch + materialization by digest). Checkpoint publishing is NOT here: +# producer endpoints call cozy_convert.publish_flavors (the /commits path). # --------------------------------------------------------------------------- @@ -992,455 +984,6 @@ def open_checkpoint_stream( attributes=attributes, ) - def publish_repo_revision( - self, - *, - destination_repo: str, - artifact_refs: Optional[List[Any]] = None, - metadata: Optional[Dict[str, Any]] = None, - create_if_missing: bool = True, - destination_repo_tags: Optional[List[str]] = None, - source_repo: Optional[str] = None, - source_version_id: Optional[str] = None, - target_version_id: Optional[str] = None, - snapshot_manifest: Optional[Dict[str, Any]] = None, - # HARD-CUT issue #14: the /publish endpoint takes checkpoint_flavors - # plus per-checkpoint lineage and a tags map. - relationship_kind: str = "import", - auto_create_external_parent: bool = True, - ) -> dict[str, Any]: - """Publish checkpoints + lineage to Tensorhub via the worker-cap /publish endpoint. - - Every entry in metadata['checkpoint_flavors'] becomes one concrete - checkpoint attached to the destination tag group in Tensorhub. - tensorhub.repo_checkpoints (checkpoint_id IS the snapshot digest). - `source_repo` + `source_version_id` (if set) produce one lineage - edge per checkpoint with relationship_kind. Tags in - `destination_repo_tags` all get pointed at the first checkpoint in - the list (typical case: one output_spec with tag='prod'). - - For imports (clone_huggingface), pass source_repo as - 'external-sources/upstream' and source_version_id as the external - reference. The external-source identifier is an internal tensorhub - schema encoding (separate from the model-ref wire format) that - prefixes the upstream provider, e.g. 'hf:black-forest-labs/FLUX.2-klein-4B' - for huggingface or 'civitai:' for civitai. Set - `relationship_kind='import'` + `auto_create_external_parent=True`. - """ - import requests - owner, repo = _parse_owner_repo(destination_repo) - normalized_source_version_id = str(source_version_id or "").strip().lower() - normalized_tags = _normalize_destination_repo_tags(destination_repo_tags) - - md = dict(metadata or {}) - base = (self._file_api_base_url or "").strip().rstrip("/") - token = (self._worker_capability_token or "").strip() - logger.info( - "worker_finalize_attempt request_id=%s job_id=%s owner=%s repo=%s", - self.request_id, - self._job_id or "", - owner, - repo, - ) - if not base or not token: - # Local/dev mode: no remote publish channel configured. - return {"ok": False, "skipped": True, "reason": "missing_worker_capability_channel", "owner": owner, "name": repo} - - headers = { - "Authorization": f"Bearer {token}", - "Content-Type": "application/json", - "X-Cozy-Owner": owner, - } - - def _parse_retry_after(raw: Any) -> Optional[float]: - try: - value = float(str(raw or "").strip()) - except Exception: - return None - if value <= 0: - return None - return min(value, _REPO_REVISION_FINALIZE_POLL_MAX_DELAY_S) - - def _response_json(resp: requests.Response) -> Dict[str, Any]: - try: - parsed = resp.json() - if isinstance(parsed, dict): - return parsed - except Exception: - pass - return {"ok": True} if not resp.text else {"ok": False, "raw": resp.text} - - def _raise_publish_http_error(code: int, path: str, text: str) -> None: - if code in (401, 403): - detail = "" - try: - detail = str((text or "").strip()) - except Exception: - detail = "" - if detail != "": - raise AuthError( - f"repo publish unauthorized ({code}): check worker_capability_token validity; response={detail[:256]}" - ) - raise AuthError(f"repo publish unauthorized ({code}): check worker_capability_token validity") - raise RuntimeError(f"repo publish request failed ({code}) {path}: {text[:512]}") - - def _poll_repo_revision_finalize(status_path: str, initial_delay_s: Optional[float] = None) -> Dict[str, Any]: - delay_s = initial_delay_s or _REPO_REVISION_FINALIZE_POLL_INITIAL_S - deadline = time.monotonic() + _REPO_REVISION_FINALIZE_POLL_MAX_S - while True: - if self._cancel_event.is_set(): - raise RuntimeError("repo finalize canceled while polling") - sleep_s = max(0.0, min(delay_s, _REPO_REVISION_FINALIZE_POLL_MAX_DELAY_S)) - if sleep_s > 0: - time.sleep(sleep_s + random.uniform(0, min(0.25, sleep_s * 0.1))) - url = f"{base}{status_path}" - resp = requests.get(url, headers=headers, timeout=_REPO_REVISION_FINALIZE_REQUEST_TIMEOUT_S) - code = int(resp.status_code) - if code in (401, 403): - _raise_publish_http_error(code, status_path, resp.text or "") - if code < 200 or code >= 300: - raise RuntimeError(f"repo finalize status poll failed ({code}) {status_path}: {resp.text[:512]}") - parsed = _response_json(resp) - state = str(parsed.get("state") or "").strip().lower() - finalize_status = str(parsed.get("finalize_status") or "").strip().lower() - if state == "finalized" or finalize_status == "succeeded": - result = parsed.get("finalize_result") - if isinstance(result, dict): - return result - return parsed - if state == "failed" or finalize_status == "failed": - err = parsed.get("finalize_error") - if isinstance(err, dict): - err_msg = json.dumps(err, sort_keys=True) - else: - err_msg = str(err or parsed) - raise RuntimeError(f"repo finalize failed: {err_msg[:512]}") - retry_after = _parse_retry_after(parsed.get("retry_after_seconds")) - if retry_after is not None: - delay_s = retry_after - else: - delay_s = min(delay_s * 1.6, _REPO_REVISION_FINALIZE_POLL_MAX_DELAY_S) - if time.monotonic() >= deadline: - raise RuntimeError( - f"repo finalize timed out after {_REPO_REVISION_FINALIZE_POLL_MAX_S}s polling {status_path}" - ) - - def _status_path_from_location(path: str, location: str) -> str: - loc = str(location or "").strip() - if not loc: - return path[:-len("/finalize")] if path.endswith("/finalize") else path - parsed = urllib.parse.urlparse(loc) - if parsed.scheme or parsed.netloc: - status_path = parsed.path or "/" - if parsed.query: - status_path = f"{status_path}?{parsed.query}" - return status_path - return loc - - def _request_repo_revision_finalize(path: str, payload: Dict[str, Any]) -> Dict[str, Any]: - deadline = time.monotonic() + _REPO_REVISION_FINALIZE_POLL_MAX_S - body = json.dumps(payload) - delay_s = _REPO_REVISION_FINALIZE_POLL_INITIAL_S - while True: - url = f"{base}{path}" - try: - resp = requests.post( - url, - headers=headers, - data=body, - timeout=_REPO_REVISION_FINALIZE_REQUEST_TIMEOUT_S, - ) - except requests.RequestException as exc: - if time.monotonic() >= deadline: - raise RuntimeError(f"repo finalize failed (network): {exc}") from exc - logger.warning( - "worker_finalize_post_retry request_id=%s job_id=%s path=%s error=%r", - self.request_id, - self._job_id or "", - path, - exc, - ) - time.sleep(delay_s + random.uniform(0, min(0.25, delay_s * 0.1))) - delay_s = min(delay_s * 1.6, _REPO_REVISION_FINALIZE_POLL_MAX_DELAY_S) - continue - - code = int(resp.status_code) - if code == 202: - status_path = _status_path_from_location(path, resp.headers.get("Location", "")) - retry_after = _parse_retry_after(resp.headers.get("Retry-After")) - if retry_after is None: - try: - retry_after = _parse_retry_after(_response_json(resp).get("retry_after_seconds")) - except Exception: - retry_after = None - return _poll_repo_revision_finalize(status_path, retry_after) - if code in (401, 403): - _raise_publish_http_error(code, path, resp.text or "") - if code >= 500 and time.monotonic() < deadline: - logger.warning( - "worker_finalize_post_retry request_id=%s job_id=%s path=%s status=%s response=%s", - self.request_id, - self._job_id or "", - path, - code, - (resp.text or "")[:256], - ) - time.sleep(delay_s + random.uniform(0, min(0.25, delay_s * 0.1))) - delay_s = min(delay_s * 1.6, _REPO_REVISION_FINALIZE_POLL_MAX_DELAY_S) - continue - if code < 200 or code >= 300: - _raise_publish_http_error(code, path, resp.text or "") - return _response_json(resp) - - def _request_json(method: str, path: str, payload: Dict[str, Any], *, allow_404: bool = False) -> Dict[str, Any]: - if method.upper() == "POST" and path.endswith("/finalize"): - return _request_repo_revision_finalize(path, payload) - url = f"{base}{path}" - resp = requests.request( - method=method, - url=url, - headers=headers, - data=json.dumps(payload), - timeout=30, - ) - code = int(resp.status_code) - if code in (401, 403): - _raise_publish_http_error(code, path, resp.text or "") - if allow_404 and code == 404: - return {"ok": False, "not_found": True} - if code < 200 or code >= 300: - # Create may be idempotent and already exists. - if path == "/api/v1/repos" and code in (400, 409): - return {"ok": True, "already_exists": True} - _raise_publish_http_error(code, path, resp.text or "") - return _response_json(resp) - - # TensorHub auto-creates the repo and lineage record on the upload path, - # so no explicit repo creation or conversion-jobs/start call is needed. - # The job_id comes from the capability token (via execution_hints). - scope = self._repo_job_upload_scope() - if scope is not None: - _, _, job_id = scope - else: - job_id = "" - if job_id == "": - raise RuntimeError("repo publish failed: no job_id in execution_hints (missing destination_repo or job scope)") - - commit_checkpoint_flavors = [] - raw_checkpoint_flavors = md.get("checkpoint_flavors") - if isinstance(raw_checkpoint_flavors, list): - for item in raw_checkpoint_flavors: - if isinstance(item, dict): - commit_checkpoint_flavors.append(dict(item)) - - # Wire artifact_refs into checkpoint_flavors if not already present. - parsed_artifacts = [] - for ref in (artifact_refs or []): - if isinstance(ref, dict): - art: Dict[str, Any] = {} - digest = str(ref.get("digest") or ref.get("blob_digest") or "").strip() - if digest: - art["digest"] = digest - path_val = str(ref.get("path") or ref.get("blob_path") or "").strip() - if path_val: - art["path"] = path_val - size_val = ref.get("size_bytes") - if size_val is not None: - art["size_bytes"] = int(size_val) - domain_val = str(ref.get("domain") or ref.get("blob_domain") or "private").strip() - art["domain"] = domain_val - if art.get("digest"): - parsed_artifacts.append(art) - elif isinstance(ref, str) and ref.strip(): - parsed_artifacts.append({"path": ref.strip()}) - - if parsed_artifacts and commit_checkpoint_flavors: - for flavor_item in commit_checkpoint_flavors: - if "artifacts" not in flavor_item: - flavor_item["artifacts"] = parsed_artifacts - - # ----- Build the /publish request body (issue #14 shape). ----- - manifest_entries: List[Dict[str, Any]] = [] - if isinstance(snapshot_manifest, dict): - raw_entries = snapshot_manifest.get("entries") - if isinstance(raw_entries, list): - manifest_entries = [e for e in raw_entries if isinstance(e, dict)] - - # Each commit_checkpoint_flavors entry becomes one checkpoint row. The - # server computes checkpoint_id from the per-flavor snapshot_manifest. - publish_checkpoints: List[Dict[str, Any]] = [] - parent_repo_ref = "" - if source_repo and str(source_repo).strip(): - parent_repo_ref = str(source_repo).strip() - parent_checkpoint_id = normalized_source_version_id - - for v in commit_checkpoint_flavors: - # The server computes checkpoint_id from the per-flavor - # snapshot_manifest. Checkpoint rows carry only concrete fields and - # flavor pointers; no arbitrary checkpoint attributes are sent. - flavor = str(v.get("flavor") or "").strip() - raw_flavors = v.get("flavors") - flavors: List[str] = [] - if isinstance(raw_flavors, list): - for raw_flavor in raw_flavors: - item = str(raw_flavor or "").strip() - if item and item not in flavors: - flavors.append(item) - if flavor and flavor not in flavors: - flavors.insert(0, flavor) - label = str(v.get("display_label") or "").strip() - lineage: List[Dict[str, Any]] = [] - if parent_repo_ref and parent_checkpoint_id: - edge: Dict[str, Any] = { - "parent_repo": parent_repo_ref, - "parent_checkpoint_id": parent_checkpoint_id, - "relationship_kind": relationship_kind or "import", - } - # Issue #258 task 3: forward per-edge metadata when the - # caller attached it (e.g. quantization_method + - # quantization_library for quantization edges). The - # publish handler in tensorhub validates the shape via - # ValidateQuantizationLineageMetadata for relationship_kind - # == "quantization" and rejects with 400 if required - # fields are missing. - raw_meta = v.get("lineage_metadata") - if isinstance(raw_meta, dict) and raw_meta: - edge["metadata"] = raw_meta - lineage.append(edge) - # Per-flavor snapshot_manifest is taken from the caller's - # v["snapshot_manifest"] when present; otherwise falls back to - # the shared top-level manifest_entries (clone path). - v_manifest = v.get("snapshot_manifest") - if isinstance(v_manifest, list): - per_flavor_entries = [e for e in v_manifest if isinstance(e, dict)] - else: - per_flavor_entries = manifest_entries - # Explicit deletions list — paths from the prior :latest checkpoint - # that the caller wants removed from this revision. The server - # always merges with :latest, so this is the only way to express - # overwrite semantics (e.g. clone's overwrite_repo path enumerates - # the prior manifest and lists every path here). - v_deletions_raw = v.get("deletions") - per_flavor_deletions: List[str] = [] - if isinstance(v_deletions_raw, list): - for d in v_deletions_raw: - if isinstance(d, str): - s = d.strip() - if s: - per_flavor_deletions.append(s) - ck_entry: Dict[str, Any] = { - "snapshot_manifest": per_flavor_entries, - "flavor": flavor, - "flavors": flavors, - "lineage": lineage, - } - if label: - ck_entry["display_label"] = label - for axis_key in ("dtype", "file_layout", "file_type"): - axis_value = str(v.get(axis_key) or "").strip() - if axis_value: - ck_entry[axis_key] = axis_value - if per_flavor_deletions: - ck_entry["deletions"] = per_flavor_deletions - # Per-checkpoint metadata — tensorhub stores this as - # `checkpoints.metadata` jsonb. Carries intrinsic facts like - # `size_facts` used by the orchestrator's VRAM-aware placement. - raw_metadata = v.get("metadata") - if isinstance(raw_metadata, dict) and raw_metadata: - ck_entry["metadata"] = dict(raw_metadata) - publish_checkpoints.append(ck_entry) - - primary_default_flavor = "" - if publish_checkpoints: - primary_default_flavor = str(publish_checkpoints[0].get("flavor") or "").strip() - if not primary_default_flavor: - raw_primary_flavors = publish_checkpoints[0].get("flavors") - if isinstance(raw_primary_flavors, list) and raw_primary_flavors: - primary_default_flavor = str(raw_primary_flavors[0] or "").strip() - # Top-level tags list (renamed from tag_groups). Each entry points - # at a flavor; the server selects which checkpoint that flavor - # resolves to. `default_checkpoint_id` is no longer sent — - # `default_flavor` carries the same information. - tags: List[Dict[str, Any]] = [] - for tag in normalized_tags: - clean_tag = str(tag or "").strip() - if not clean_tag or clean_tag.lower() == "latest": - continue - entry: Dict[str, Any] = {"tag": clean_tag} - if primary_default_flavor: - entry["default_flavor"] = primary_default_flavor - tags.append(entry) - - if not publish_checkpoints: - logger.warning( - "worker_finalize_skipped request_id=%s job_id=%s owner=%s repo=%s reason=no_checkpoints", - self.request_id, self._job_id or "", owner, repo, - ) - else: - # `merge_with_existing` is no longer sent: the server always - # merges with :latest. Overwrite semantics are expressed by - # the caller populating per-checkpoint `deletions` lists; this - # method passes them through unchanged. - publish_req_body = { - "tags": tags, - "checkpoint_flavors": publish_checkpoints, - "auto_create_external_parent": bool(auto_create_external_parent), - } - # Issue #20: finalize hits the session-scoped URL. The session - # was opened lazily on the first save_* call to this repo and - # cached in the ctx-level manager. - finalize_resp: Dict[str, Any] = {} - try: - revision_id = self._checkpoint_revision_id(owner, repo) - finalize_resp = _request_json( - "POST", - f"/api/v1/repos/{urllib.parse.quote(owner, safe='')}/{urllib.parse.quote(repo, safe='')}/revisions/{urllib.parse.quote(revision_id, safe='')}/finalize", - publish_req_body, - ) or {} - # Finalize succeeded — drop the session from the manager's - # cache so close_all at request end doesn't try to abort it. - mgr = self._upload_session_manager() - sess = mgr.get("checkpoint", {"repo_owner": owner, "repo_name": repo}) - if sess is not None: - # Finalize endpoint already returned; just forget cache. - mgr._sessions.pop(sess.scope_key, None) # type: ignore[attr-defined] - except Exception as exc: - logger.warning( - "worker_finalize_catalog_failed request_id=%s job_id=%s owner=%s repo=%s error=%r", - self.request_id, self._job_id or "", owner, repo, exc, - ) - raise - - # extract server-computed checkpoint_ids from the - # finalize response so callers (and the post-finalize tag - # promotion below) can reference them. - resp_checkpoints = finalize_resp.get("checkpoints") - if isinstance(resp_checkpoints, list): - for idx, rc in enumerate(resp_checkpoints): - if idx < len(publish_checkpoints) and isinstance(rc, dict): - cid = str(rc.get("checkpoint_id") or "").strip() - if cid: - publish_checkpoints[idx]["checkpoint_id"] = cid - - published_ids = [str(c.get("checkpoint_id") or "") for c in publish_checkpoints] - logger.info( - "worker_finalize_succeeded request_id=%s job_id=%s owner=%s repo=%s checkpoints=%d primary=%s", - self.request_id, self._job_id or "", owner, repo, - len(published_ids), published_ids[0] if published_ids else "", - ) - - out: Dict[str, Any] = { - "ok": True, - "owner": owner, - "name": repo, - "job_id": job_id, - "checkpoint_ids": published_ids, - "output_versions": published_ids, - } - if normalized_tags: - out["destination_repo_tags"] = normalized_tags - return out - def _download_blob_by_digest(self, digest: str, dest: Path) -> None: """Fetch a blob by ``:`` digest to ``dest``. @@ -1661,7 +1204,7 @@ def publish_dataset_revision( ) -> dict[str, Any]: """Publish a dataset revision into ``tensorhub.datasets``. - Parallel to ``publish_repo_revision`` but writes to the datasets + Writes to the datasets subsystem instead of ``repo_checkpoints``. The flow: 1. Resolve ``destination_dataset`` (owner/name) against tensorhub. diff --git a/src/gen_worker/request_context/_helpers.py b/src/gen_worker/request_context/_helpers.py index 619708e..1ea8462 100644 --- a/src/gen_worker/request_context/_helpers.py +++ b/src/gen_worker/request_context/_helpers.py @@ -16,7 +16,7 @@ class files only hold class bodies. import socket import urllib.parse import urllib.request -from typing import Any, Dict, List, Optional +from typing import Any, Dict, Optional from ..api.errors import OutputTooLargeError from ..models.refs import parse_model_ref @@ -166,35 +166,6 @@ def _parse_owner_repo(value: str) -> tuple[str, str]: return owner, repo -def _parse_owner_repo_with_optional_tag(value: str) -> tuple[str, str, str]: - raw = str(value or "").strip().strip("/") - tag = "" - if ":" in raw: - raw, tag = raw.rsplit(":", 1) - tag = str(tag or "").strip().lower() - owner, repo = _parse_owner_repo(raw) - return owner, repo, tag - - -def _normalize_destination_repo_tags(values: Optional[List[str]]) -> List[str]: - out: List[str] = [] - seen: set[str] = set() - for item in list(values or []): - tag = str(item or "").strip().lower() - if not tag: - continue - if not _PUBLIC_TAG_RE.match(tag): - raise ValueError("destination_repo_tags contains an invalid tag") - if tag == "latest": - raise ValueError("destination_repo_tags must not include latest") - if tag in seen: - continue - seen.add(tag) - out.append(tag) - out.sort() - return out - - def _decode_unverified_jwt_claims(token: str) -> Dict[str, Any]: raw = str(token or "").strip() if raw.count(".") < 2: diff --git a/src/gen_worker/request_context/_stream.py b/src/gen_worker/request_context/_stream.py index 9331603..f737b2f 100644 --- a/src/gen_worker/request_context/_stream.py +++ b/src/gen_worker/request_context/_stream.py @@ -65,8 +65,7 @@ def __init__( self._expected_size_bytes = int(expected_size_bytes or 0) # Lineage/attributes are retained for local fallback compatibility. # Repo-CAS `/complete` is deliberately parts-only; checkpoint-level - # metadata is sent once in publish_repo_revision's final - # checkpoint_flavors[] payload. + # metadata rides the commit payload (cozy_convert.publish_flavors). self._lineage_produced_by_kind = (str(produced_by_kind or "").strip() or None) self._lineage_step_number = step_number if isinstance(step_number, int) else None self._lineage_epoch_number = epoch_number if isinstance(epoch_number, int) else None diff --git a/tests/test_discovery_and_decorators.py b/tests/test_discovery_and_decorators.py index 92b2625..6cc4512 100644 --- a/tests/test_discovery_and_decorators.py +++ b/tests/test_discovery_and_decorators.py @@ -376,3 +376,49 @@ def test_discovery_walks_layouts_dedups_and_skips_third_party(tmp_pkg: Path, cap assert any( "outside the walked package" in r.message for r in caplog.records ) + + +# --------------------------------------------------------------------------- # +# 9. producer kinds publish explicitly — generator handlers are rejected # +# --------------------------------------------------------------------------- # + + +def test_producer_kind_rejects_sync_generator_class_handler() -> None: + with pytest.raises(TypeError, match="publish_flavors"): + @endpoint(kind="conversion") + class BadConversion: + def run(self, ctx: RequestContext, data: _In): + yield _Out(result="x") + + +def test_producer_kind_rejects_async_generator_class_handler() -> None: + with pytest.raises(TypeError, match="inference-only"): + @endpoint(kind="training") + class BadTraining: + async def run(self, ctx: RequestContext, data: _In): + yield _Out(result="x") + + +def test_producer_kind_rejects_generator_function() -> None: + with pytest.raises(TypeError, match="must not be a generator"): + @endpoint(kind="dataset") + def bad_dataset(ctx: RequestContext, data: _In): + yield _Out(result="x") + + +def test_from_scratch_example_uses_publish_contract() -> None: + """Discovery smoke for examples/from-scratch: imports, one conversion-kind + non-generator handler, result struct output (the explicit-publish shape).""" + import importlib.util + + path = Path(__file__).resolve().parents[1] / "examples" / "from-scratch" / "from_scratch.py" + spec = importlib.util.spec_from_file_location("from_scratch_example", path) + assert spec is not None and spec.loader is not None + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + + (s,) = extract_specs(mod.FromScratch) + assert s.kind == "conversion" + assert s.output_mode == "single" + assert not s.is_async_gen + assert s.output_type is mod.FromScratchResult diff --git a/tests/test_request_context.py b/tests/test_request_context.py index 3e14a38..ffbe1a8 100644 --- a/tests/test_request_context.py +++ b/tests/test_request_context.py @@ -100,14 +100,15 @@ def test_producer_contexts_are_real_subclasses() -> None: assert issubclass(cls, RequestContext) ctx = cls(request_id="r1") # producer surface lives on the subclass, not the base - assert hasattr(ctx, "publish_repo_revision") assert hasattr(ctx, "save_checkpoint") assert hasattr(ctx, "set_repo_spec") assert hasattr(ctx, "hf_token") + # checkpoint publishing is cozy_convert.publish_flavors, not a ctx RPC + assert not hasattr(ctx, "publish_repo_revision") assert hasattr(ConversionContext(request_id="r1"), "mktemp") assert hasattr(DatasetContext(request_id="r1"), "resolve_dataset") base = RequestContext(request_id="r1") - for producer_only in ("publish_repo_revision", "save_checkpoint", "mktemp"): + for producer_only in ("save_checkpoint", "mktemp"): assert not hasattr(base, producer_only)