From 7447ca8e57fc47f4acc5e3bb4e4c6569b4a89c64 Mon Sep 17 00:00:00 2001 From: Volv G Date: Wed, 16 Sep 2026 12:27:24 -0700 Subject: [PATCH 1/4] Make component publishing monotonic and surface the compared digest Regular component publishing is now monotonic against the highest non-deprecated, owner-scoped published version: a local version older than the latest published version is a no-op SKIP instead of publishing the older spec and deprecating the newer one. Equal is still a SKIP and newer still PROCEEDs. Ordering comes from the existing compare_versions. Skips now surface the exact digest they were compared against so callers (for example ship-on-publish) can pin it: - ProcessingResult.latest_digest: digest of the selected latest non-deprecated owner-scoped published version, set on PROCEED/SKIP and carried through the SUCCESS/ERROR results that follow a version check. - ProcessingResult.resolved_digest: digest or latest_digest, but None unless the outcome is SUCCESS or SKIP, so a publish request that raised and may still have landed never hands back the stale digest. - ProcessingResult.digest keeps its existing SUCCESS-only meaning. If several non-deprecated owner-scoped components tie at the selected latest version, no exact digest can be chosen, so the check fails closed with an error naming the tied digests rather than guessing from API ordering. Deprecated components are never selected as latest. Deliberate downgrades remain possible via an explicit opt-in: --allow-downgrade on published-components publish, or allow_downgrade=True on the publisher and its wrappers. Default is off. Assisted-By: devx/6aa1a44a-8f9f-4be7-b20c-752af4396eb9 (cherry picked from commit 58593ea3c453ca2429efbe668ad7a8ec48dfd0e8) --- README.md | 23 +- .../src/tangle_cli/component_publisher.py | 134 ++++++++++-- .../tangle_cli/published_components_cli.py | 12 + tests/test_component_publisher.py | 205 ++++++++++++++++++ tests/test_components_cli.py | 3 + 5 files changed, 363 insertions(+), 14 deletions(-) diff --git a/README.md b/README.md index af9d42d..67b0e75 100644 --- a/README.md +++ b/README.md @@ -353,7 +353,28 @@ uv run tangle sdk published-components publish components/my-component.yaml --dr uv run tangle sdk published-components deprecate sha256:old --superseded-by sha256:new ``` -`publish` accepts `--image`, `--name`, `--description`, `--annotations` (JSON), `--dry-run`, `--published-by`, generic git metadata fields, generic API auth fields, `--log-type`, and `--config`. By default it scopes version checks and automatic old-version deprecation to the current authenticated user via `users_me()`; use `--published-by` to supply an explicit owner/publisher filter. Publishing fails closed if no owner can be determined. +`publish` accepts `--image`, `--name`, `--description`, `--annotations` (JSON), `--dry-run`, `--allow-downgrade`, `--published-by`, generic git metadata fields, generic API auth fields, `--log-type`, and `--config`. By default it scopes version checks and automatic old-version deprecation to the current authenticated user via `users_me()`; use `--published-by` to supply an explicit owner/publisher filter. Publishing fails closed if no owner can be determined. + +#### Monotonic publishing and result digests + +Publishing is monotonic against the highest **non-deprecated, owner-scoped** published version of the component (ordering comes from `compare_versions`, which zero-pads shorter versions so `1.0.1 > 1.0`): + +| Local vs latest published | Outcome | Notes | +| --- | --- | --- | +| nothing published / no readable remote version | `proceed` | first publish | +| local strictly newer | `proceed` | publishes, then deprecates older owner-scoped versions | +| local equal | `skip` | no create/deprecate calls | +| local strictly older | `skip` | no-op; never publishes an older version and never deprecates a newer one | + +Every result carries the digest of the version it compared against: + +- `digest` — digest of a **newly created** publication (SUCCESS only, unchanged meaning). +- `latest_digest` — exact digest of the selected latest published version (set on PROCEED/SKIP, and carried through the SUCCESS/ERROR results that follow a version check). JSON output includes it as `latest_digest`. +- `ProcessingResult.resolved_digest` — the digest a caller should pin: `digest or latest_digest`, but deliberately `None` for any outcome other than SUCCESS/SKIP, so a failed publish never hands back a stale-but-plausible digest. + +If two or more non-deprecated owner-scoped components tie at the selected latest version, no exact digest can be chosen, so the check fails closed with an `error` naming the tied digests instead of guessing from API ordering. Deprecated components are never selected as "latest". + +**Contract change:** republishing an older version used to proceed (publishing the older spec and deprecating the newer one). It is now a skip. Deliberate downgrades must opt in with `--allow-downgrade` on the CLI, or `ComponentPublisher(allow_downgrade=True)` / `allow_downgrade=True` on the `publish_component_to_tangle` / `perform_version_check` wrappers. Republishing the same version is still a skip, as before. There is no separate OSS `publish-all` command. To publish multiple components, pass a YAML/JSON config list, or `_defaults` + `configs`, to the same `published-components publish` command; the command aggregates results and exits nonzero if any component errors. A top-level `_select` node can choose between such documents per environment (see [Environment-selected configs](#environment-selected-configs-_select)). diff --git a/packages/tangle-cli/src/tangle_cli/component_publisher.py b/packages/tangle-cli/src/tangle_cli/component_publisher.py index 5db339d..ea15255 100644 --- a/packages/tangle-cli/src/tangle_cli/component_publisher.py +++ b/packages/tangle-cli/src/tangle_cli/component_publisher.py @@ -36,7 +36,17 @@ class ProcessingOutcome(str, Enum): @dataclass class ProcessingResult: - """Result for one component publish/deprecate processing step.""" + """Result for one component publish/deprecate processing step. + + ``digest`` is the digest of a *newly created* publication and is therefore + only populated on :attr:`ProcessingOutcome.SUCCESS`. ``latest_digest`` is + the exact digest of the selected latest non-deprecated owner-scoped + published version that the local component was compared against; it is + populated on PROCEED/SKIP (and carried through SUCCESS/ERROR results that + happen after a version check) whenever it can be resolved unambiguously. + Callers that want "the digest to pin, whatever happened" should use + :attr:`resolved_digest`. + """ outcome: ProcessingOutcome local_version: str | None = None @@ -44,8 +54,23 @@ class ProcessingResult: spec: Any = None reason: str | None = None digest: str | None = None + latest_digest: str | None = None response: Any = None + @property + def resolved_digest(self) -> str | None: + """Digest a caller can pin: newly published digest, else latest published. + + Deliberately ``None`` for any outcome other than SUCCESS/SKIP: after an + ERROR (including a publish request that raised and may still have + landed server-side) the correct digest is unknown, so this must not + hand back the pre-existing one. + """ + + if self.outcome not in (ProcessingOutcome.SUCCESS, ProcessingOutcome.SKIP): + return None + return self.digest or self.latest_digest + def to_dict(self) -> dict[str, Any]: payload: dict[str, Any] = { "status": self.outcome.value, @@ -54,6 +79,7 @@ def to_dict(self) -> dict[str, Any]: "latest_version": self.latest_version, "reason": self.reason, "digest": self.digest, + "latest_digest": self.latest_digest, "response": _to_plain(self.response), } if self.spec is not None: @@ -139,6 +165,7 @@ def __init__( hooks: Sequence[ComponentPublishHook] | None = None, logger: Logger | None = None, base_url: str | None = None, + allow_downgrade: bool = False, ) -> None: """Initialize the ComponentPublisher. @@ -147,6 +174,11 @@ def __init__( ``client_factory`` is a downstream seam for lazily constructing a custom authenticated client; subclasses may also override :meth:`_get_client` for more control. + + ``allow_downgrade`` is an explicit opt-out from monotonic publishing. + By default a local version older than the latest published + owner-scoped version is a no-op SKIP; set it to ``True`` only for a + deliberate manual downgrade/republish. """ super().__init__( @@ -157,6 +189,7 @@ def __init__( base_url=base_url, ) self.published_by = published_by + self.allow_downgrade = allow_downgrade self.hooks = list(hooks or []) self.results: list[tuple[str, ProcessingResult]] = [] @@ -190,6 +223,19 @@ def component_digest(self, component: Any) -> str | None: digest = getattr(component, "digest", None) return str(digest) if digest else None + def component_is_deprecated(self, component: Any) -> bool: + """Return whether a published component is marked deprecated. + + Listing is already non-deprecated by default; this is a defensive + filter so deprecated entries can never be selected as "latest". + """ + + if isinstance(component, Mapping): + value = component.get("deprecated") + else: + value = getattr(component, "deprecated", None) + return bool(value) + def current_user_id(self, client: Any) -> str | None: """Return the current Tangle user id for owner-scoped lookups.""" @@ -206,18 +252,33 @@ def current_user_id(self, client: Any) -> str | None: return str(value) if value else None def perform_version_check(self, spec: Any) -> ProcessingResult: - """Perform owner-scoped version checking for a component. + """Perform owner-scoped, monotonic version checking for a component. If ``published_by`` is omitted, the current authenticated user is resolved via ``client.users_me().id``. Failure to determine an owner is an error so callers do not accidentally compare/deprecate components owned by others. + + Ordering is defined by :func:`tangle_cli.utils.compare_versions` + against the highest non-deprecated owner-scoped published version: + + * nothing published (or no readable remote version) → PROCEED + * local strictly newer → PROCEED + * local equal → SKIP, carrying that version's exact digest + * local strictly older → SKIP (no-op), carrying the newer published + digest; never publishes an older version and never deprecates a + newer one. ``allow_downgrade=True`` opts out of this last rule. + + If several non-deprecated owner-scoped components tie at the selected + latest version, a SKIP cannot name one exact digest, so the check fails + closed with an ERROR rather than guessing from API ordering. """ local_version = spec.version self.log.info(f" Local version: {local_version}") - latest_version = None + latest_version: str | None = None + latest_digests: list[str] = [] if self.dry_run: test_version = os.environ.get("TEST_LATEST_VERSION") @@ -257,16 +318,21 @@ def perform_version_check(self, spec: Any) -> ProcessingResult: digest = self.component_digest(component) if not digest: continue + if self.component_is_deprecated(component): + continue try: full_spec = client.get_component_spec(digest) remote_version = full_spec.version if full_spec else None - if remote_version and ( - not latest_version or utils.compare_versions(remote_version, latest_version) > 0 - ): - latest_version = remote_version except Exception as exc: self.log.warn(f" Warning: Failed to get version for component {digest[:16]}: {exc}") continue + if not remote_version: + continue + if latest_version is None or utils.compare_versions(remote_version, latest_version) > 0: + latest_version = remote_version + latest_digests = [digest] + elif utils.compare_versions(remote_version, latest_version) == 0 and digest not in latest_digests: + latest_digests.append(digest) if latest_version: self.log.info(f" Remote version: {latest_version}") @@ -275,11 +341,31 @@ def perform_version_check(self, spec: Any) -> ProcessingResult: f" ℹ️ Found {len(existing_components)} component(s) but couldn't extract version" ) - should_proceed = not latest_version or utils.compare_versions(local_version, latest_version) != 0 + if len(latest_digests) > 1: + tied = ", ".join(latest_digests) + self.log.error( + f" ❌ Ambiguous latest published version {latest_version}: " + f"{len(latest_digests)} non-deprecated components share it ({tied})" + ) + return ProcessingResult( + outcome=ProcessingOutcome.ERROR, + local_version=local_version, + latest_version=latest_version, + spec=spec, + reason=( + f"Ambiguous latest published version {latest_version}: " + f"{len(latest_digests)} non-deprecated components share it ({tied}); " + "cannot select an exact digest" + ), + ) + + latest_digest = latest_digests[0] if latest_digests else None + + comparison = 0 if latest_version is None else utils.compare_versions(local_version, latest_version) + should_proceed = latest_version is None or comparison > 0 or (comparison < 0 and self.allow_downgrade) if should_proceed: - is_older = latest_version is not None and utils.compare_versions(latest_version, local_version) > 0 - version_suffix = " (older)" if is_older else "" + version_suffix = " (older)" if comparison < 0 else "" self.log.info( " ➡️ Version " + (f"{latest_version}{version_suffix}" if latest_version else "new") @@ -289,17 +375,29 @@ def perform_version_check(self, spec: Any) -> ProcessingResult: outcome=ProcessingOutcome.PROCEED, local_version=local_version, latest_version=latest_version, + latest_digest=latest_digest, spec=spec, ) - self.log.info(f" ⏭️ Skipping: Version {local_version} unchanged") + if comparison < 0: + self.log.info( + f" ⏭️ Skipping: Local version {local_version} is older than published {latest_version}" + ) + reason = ( + f"Version {local_version} is older than published version {latest_version} " + "(no-op; publishing is monotonic)" + ) + else: + self.log.info(f" ⏭️ Skipping: Version {local_version} unchanged") + reason = f"Version {local_version} unchanged (matches remote)" return ProcessingResult( outcome=ProcessingOutcome.SKIP, local_version=local_version, latest_version=latest_version, + latest_digest=latest_digest, spec=spec, - reason=f"Version {local_version} unchanged (matches remote)", + reason=reason, ) def deprecate_old_components( @@ -528,6 +626,7 @@ def publish_component( outcome=ProcessingOutcome.SUCCESS, local_version=version_check_result.local_version, latest_version=version_check_result.latest_version, + latest_digest=version_check_result.latest_digest, spec=spec, reason=f"Dry-run: would publish {spec.name}", response={"name": spec.name, "text": local_yaml_content}, @@ -542,6 +641,7 @@ def publish_component( outcome=ProcessingOutcome.ERROR, local_version=version_check_result.local_version, latest_version=version_check_result.latest_version, + latest_digest=version_check_result.latest_digest, spec=spec, reason="Cannot determine current user for author filtering", ) @@ -562,6 +662,7 @@ def publish_component( spec=spec, reason=f"Successfully published with digest: {new_digest}", digest=str(new_digest), + latest_digest=version_check_result.latest_digest, response=result, ) @@ -570,6 +671,7 @@ def publish_component( outcome=ProcessingOutcome.ERROR, local_version=version_check_result.local_version, latest_version=version_check_result.latest_version, + latest_digest=version_check_result.latest_digest, spec=spec, reason="Component published but no digest returned", response=result, @@ -580,6 +682,7 @@ def publish_component( outcome=ProcessingOutcome.ERROR, local_version=version_check_result.local_version, latest_version=version_check_result.latest_version, + latest_digest=version_check_result.latest_digest, spec=spec, reason=f"Request failed: {exc}", ) @@ -772,14 +875,16 @@ def perform_version_check( client: Any = None, logger: Logger | None = None, published_by: str | None = None, + allow_downgrade: bool = False, ) -> ProcessingResult: - """Perform owner-scoped version checking for a component.""" + """Perform owner-scoped, monotonic version checking for a component.""" return ComponentPublisher( dry_run=dry_run, client=client, logger=logger, published_by=published_by, + allow_downgrade=allow_downgrade, ).perform_version_check(spec) @@ -797,6 +902,7 @@ def publish_component_to_tangle( client: Any = None, client_factory: Callable[[], Any] | None = None, published_by: str | None = None, + allow_downgrade: bool = False, ) -> ProcessingResult: """Publish one component using ``ComponentPublisher.publish_component``.""" @@ -809,6 +915,7 @@ def publish_component_to_tangle( git_remote_url=git_remote_url, git_repo=git_repo, published_by=published_by, + allow_downgrade=allow_downgrade, ) return publisher.publish_component( file_path, @@ -830,6 +937,7 @@ def publish_component(client: Any, component_path: str | Path, **kwargs: Any) -> git_root=kwargs.pop("git_root", None), git_repo=kwargs.pop("git_repo", None), published_by=kwargs.pop("published_by", None), + allow_downgrade=bool(kwargs.pop("allow_downgrade", False)), client=client, client_factory=kwargs.pop("client_factory", None), logger=kwargs.pop("logger", None), diff --git a/packages/tangle-cli/src/tangle_cli/published_components_cli.py b/packages/tangle-cli/src/tangle_cli/published_components_cli.py index 733eb02..0744278 100644 --- a/packages/tangle-cli/src/tangle_cli/published_components_cli.py +++ b/packages/tangle-cli/src/tangle_cli/published_components_cli.py @@ -247,6 +247,16 @@ def published_components_publish( Parameter(help="Custom annotations as a JSON object."), ] = None, dry_run: bool | None = None, + allow_downgrade: Annotated[ + bool | None, + Parameter( + help=( + "Publish even when the local version is older than the latest published " + "owner-scoped version. Off by default: publishing is monotonic and an older " + "local version is a no-op skip." + ) + ), + ] = None, git_remote_sha: str | None = None, git_remote_branch: str | None = None, git_remote_url: str | None = None, @@ -269,6 +279,7 @@ def published_components_publish( description=(description, None), annotations=("annotations", annotations, None, True), dry_run=(dry_run, None), + allow_downgrade=(allow_downgrade, None), git_remote_sha=(git_remote_sha, None), git_remote_branch=(git_remote_branch, None), git_remote_url=(git_remote_url, None), @@ -302,6 +313,7 @@ def published_components_publish( git_remote_url=args.git_remote_url, git_root=args.git_root, published_by=args.published_by, + allow_downgrade=bool(args.allow_downgrade), client=client, logger=logger, ) diff --git a/tests/test_component_publisher.py b/tests/test_component_publisher.py index a6ac801..8541a4c 100644 --- a/tests/test_component_publisher.py +++ b/tests/test_component_publisher.py @@ -30,6 +30,7 @@ class User: class ExistingComponent: digest: str name: str = "demo" + deprecated: bool = False class FakeClient: @@ -215,6 +216,210 @@ def test_version_check_skips_unchanged_owner_scoped_version() -> None: assert "unchanged" in (result.reason or "") +# --------------------------------------------------------------------------- +# Monotonic publishing contract +# --------------------------------------------------------------------------- + + +def _version_check( + local_version: str, + published: dict[str, str], + *, + deprecated: set[str] | None = None, + allow_downgrade: bool = False, +) -> tuple[ProcessingResult, FakeClient]: + spec = ComponentSpec.from_yaml( + f"name: demo\nmetadata:\n annotations:\n version: '{local_version}'\n" + ) + client = FakeClient() + client.existing = [ + ExistingComponent(digest, deprecated=digest in (deprecated or set())) for digest in published + ] + client.component_versions = dict(published) + result = perform_version_check( + spec=spec, + dry_run=False, + client=client, + allow_downgrade=allow_downgrade, + ) + return result, client + + +def test_version_check_proceeds_when_nothing_published() -> None: + result, client = _version_check("1.0", {}) + + assert result.outcome == ProcessingOutcome.PROCEED + assert result.latest_version is None + assert result.latest_digest is None + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_version_check_proceeds_when_local_is_newer() -> None: + result, _ = _version_check("1.1", {"sha256:v10": "1.0"}) + + assert result.outcome == ProcessingOutcome.PROCEED + assert result.latest_version == "1.0" + assert result.latest_digest == "sha256:v10" + + +def test_version_check_equal_version_skips_with_exact_digest() -> None: + result, client = _version_check("1.0", {"sha256:v10": "1.0"}) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_version == "1.0" + assert result.latest_digest == "sha256:v10" + assert result.resolved_digest == "sha256:v10" + assert "unchanged" in (result.reason or "") + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_version_check_older_local_version_skips_with_newer_digest() -> None: + result, client = _version_check("1.0", {"sha256:v20": "2.0"}) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_version == "2.0" + assert result.latest_digest == "sha256:v20" + assert result.resolved_digest == "sha256:v20" + assert "older" in (result.reason or "") + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_version_check_selects_highest_of_multiple_published_versions() -> None: + result, _ = _version_check( + "1.5", + {"sha256:v10": "1.0", "sha256:v201": "2.0.1", "sha256:v20": "2.0"}, + ) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_version == "2.0.1" + assert result.latest_digest == "sha256:v201" + + +def test_version_check_fails_closed_on_ambiguous_latest_digests() -> None: + result, client = _version_check("1.0", {"sha256:a": "2.0", "sha256:b": "2.0"}) + + assert result.outcome == ProcessingOutcome.ERROR + assert result.latest_digest is None + assert result.resolved_digest is None + assert "Ambiguous latest published version 2.0" in (result.reason or "") + assert "sha256:a" in (result.reason or "") and "sha256:b" in (result.reason or "") + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_version_check_ignores_deprecated_components_when_selecting_latest() -> None: + result, _ = _version_check( + "1.0", + {"sha256:v10": "1.0", "sha256:v20": "2.0"}, + deprecated={"sha256:v20"}, + ) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_version == "1.0" + assert result.latest_digest == "sha256:v10" + + +def test_version_check_ignores_deprecated_duplicate_of_latest_version() -> None: + result, _ = _version_check( + "1.0", + {"sha256:live": "2.0", "sha256:dead": "2.0"}, + deprecated={"sha256:dead"}, + ) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_digest == "sha256:live" + + +def test_version_check_allow_downgrade_opts_back_into_publishing_older() -> None: + result, _ = _version_check("1.0", {"sha256:v20": "2.0"}, allow_downgrade=True) + + assert result.outcome == ProcessingOutcome.PROCEED + assert result.latest_version == "2.0" + assert result.latest_digest == "sha256:v20" + + +def test_publish_older_version_is_a_noop_that_pins_newer_digest(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="1.0") + client = FakeClient() + client.existing = [ExistingComponent("sha256:v20")] + client.component_versions = {"sha256:v20": "2.0"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.digest is None + assert result.latest_digest == "sha256:v20" + assert result.resolved_digest == "sha256:v20" + assert client.create_calls == [] + assert client.update_calls == [] + assert result.to_dict()["latest_digest"] == "sha256:v20" + + +def test_publish_newer_version_reports_new_and_previous_digests(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = FakeClient() + client.existing = [ExistingComponent("sha256:v10")] + client.component_versions = {"sha256:v10": "1.0"} + client.publish_response = {"digest": "sha256:v20"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.SUCCESS + assert result.digest == "sha256:v20" + assert result.latest_digest == "sha256:v10" + assert result.resolved_digest == "sha256:v20" + + +def test_publish_ambiguous_latest_digests_publishes_nothing(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="1.0") + client = FakeClient() + client.existing = [ExistingComponent("sha256:a"), ExistingComponent("sha256:b")] + client.component_versions = {"sha256:a": "2.0", "sha256:b": "2.0"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.ERROR + assert result.resolved_digest is None + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_resolved_digest_is_none_for_error_outcomes(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = FakeClient() + client.existing = [ExistingComponent("sha256:v10")] + client.component_versions = {"sha256:v10": "1.0"} + + def failing_create(**kwargs: Any) -> dict[str, Any]: + raise RuntimeError("boom") + + client.published_components_create = failing_create # type: ignore[method-assign] + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.ERROR + assert result.latest_digest == "sha256:v10" + assert result.resolved_digest is None + assert client.update_calls == [] + + +def test_version_check_without_owner_reports_error_and_no_digest() -> None: + spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '1.0'\n") + client = FakeClient() + client.user = None + + result = perform_version_check(spec=spec, dry_run=False, client=client) + + assert result.outcome == ProcessingOutcome.ERROR + assert result.latest_digest is None + assert result.resolved_digest is None + assert client.find_calls == [] + assert client.create_calls == [] + + def test_version_check_progress_uses_logger_not_tangle_verbose(monkeypatch, capsys) -> None: spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '1.0'\n") client = FakeClient() diff --git a/tests/test_components_cli.py b/tests/test_components_cli.py index c52b5d8..d2b23c0 100644 --- a/tests/test_components_cli.py +++ b/tests/test_components_cli.py @@ -88,6 +88,7 @@ def fake_client_from_options(**kwargs: Any) -> object: "git_remote_url": None, "git_root": None, "published_by": None, + "allow_downgrade": False, "client": None, "logger": ANY, } @@ -283,6 +284,7 @@ def fake_client_from_options(**kwargs: Any) -> object: "git_remote_url": None, "git_root": None, "published_by": "first@example.com", + "allow_downgrade": False, "client": fake_client, "logger": ANY, }, @@ -293,6 +295,7 @@ def fake_client_from_options(**kwargs: Any) -> object: "git_remote_url": None, "git_root": None, "published_by": "second@example.com", + "allow_downgrade": False, "client": None, "logger": ANY, }, From ff6339f95072524a38cc1c6cf1bad9b704e48ab1 Mon Sep 17 00:00:00 2001 From: Volv G Date: Wed, 16 Sep 2026 12:43:45 -0700 Subject: [PATCH 2/4] Fail closed on unreadable rows and re-check published state before publish Follow-up to the monotonic publishing change, addressing review findings that monotonicity could still be violated. Preflight previously warned and skipped any candidate whose spec/version could not be read, and the publish path then re-listed and deprecated every returned row. A listed-but-unreadable row (possibly newer) could therefore be deprecated sight-unseen, and a version appearing between the two lookups could be deprecated by a decision made before it existed. - Reading the published state is now a single observation (collect_published_state) returning a verified digest -> version snapshot plus the identifiers of any unreadable non-deprecated candidate. Any unreadable candidate, including one with no digest, is an ERROR: nothing is published and nothing is deprecated. - Policy lives in one place (_evaluate_published_state) and is re-applied to a fresh observation taken immediately before create, so a concurrently published equal/newer/ambiguous row still turns the publish into a SKIP or ERROR with no create or deprecate calls. - Only digests proven strictly older in that final observation are deprecated, in sorted order. A row first seen after the publish decision is never deprecated. allow_downgrade likewise never deprecates a strictly newer row. A race after the final read needs a server-side conditional/CAS operation and is out of scope here. - Latest/tied digest lists are sorted and the latest version is chosen deterministically, so results and diagnostics no longer depend on API response ordering. resolved_digest outcome safety and allow_downgrade semantics are unchanged, and the added regressions cover unreadable newer rows, concurrent newer/equal/ambiguous rows between lookups, refusal without create/deprecate, and order-independent diagnostics. Assisted-By: devx/6aa1a44a-8f9f-4be7-b20c-752af4396eb9 (cherry picked from commit 585a998228c7433c11d8a01a184c66d984cccc53) --- README.md | 11 +- .../src/tangle_cli/component_publisher.py | 319 +++++++++++++----- tests/test_component_publisher.py | 185 ++++++++++ 3 files changed, 428 insertions(+), 87 deletions(-) diff --git a/README.md b/README.md index 67b0e75..0b86894 100644 --- a/README.md +++ b/README.md @@ -372,9 +372,16 @@ Every result carries the digest of the version it compared against: - `latest_digest` — exact digest of the selected latest published version (set on PROCEED/SKIP, and carried through the SUCCESS/ERROR results that follow a version check). JSON output includes it as `latest_digest`. - `ProcessingResult.resolved_digest` — the digest a caller should pin: `digest or latest_digest`, but deliberately `None` for any outcome other than SUCCESS/SKIP, so a failed publish never hands back a stale-but-plausible digest. -If two or more non-deprecated owner-scoped components tie at the selected latest version, no exact digest can be chosen, so the check fails closed with an `error` naming the tied digests instead of guessing from API ordering. Deprecated components are never selected as "latest". +The check **fails closed** (an `error`, with nothing published and nothing deprecated) whenever the published state cannot be read completely: -**Contract change:** republishing an older version used to proceed (publishing the older spec and deprecating the newer one). It is now a skip. Deliberate downgrades must opt in with `--allow-downgrade` on the CLI, or `ComponentPublisher(allow_downgrade=True)` / `allow_downgrade=True` on the `publish_component_to_tangle` / `perform_version_check` wrappers. Republishing the same version is still a skip, as before. +- any non-deprecated owner-scoped candidate whose digest is missing, or whose spec/version cannot be fetched or parsed — an unreadable row could be newer than the local version, and must never be deprecated sight-unseen; +- two or more non-deprecated candidates tied at the selected latest version, where no exact digest can be chosen. The reason names the tied digests instead of guessing from API ordering. + +Deprecated components are never selected as "latest", and all digest lists in results/logs are sorted, so diagnostics do not depend on API response order. + +The published state is re-read immediately before create, and the same policy is re-applied to that fresh observation: a version that appeared concurrently since the first check can still turn the publish into a skip or an error. After a successful create, only digests **proven strictly older** in that final observation are deprecated — a row first seen after the publish decision is never deprecated on the strength of the earlier one. A race after the final read is not preventable client-side and needs a server-side conditional/CAS operation. + +**Contract change:** republishing an older version used to proceed (publishing the older spec and deprecating the newer one). It is now a skip. `--allow-downgrade` publishes the older spec but still never deprecates a strictly newer row; deprecate those explicitly with `published-components deprecate` if that is really intended. Deliberate downgrades must opt in with `--allow-downgrade` on the CLI, or `ComponentPublisher(allow_downgrade=True)` / `allow_downgrade=True` on the `publish_component_to_tangle` / `perform_version_check` wrappers. Republishing the same version is still a skip, as before. There is no separate OSS `publish-all` command. To publish multiple components, pass a YAML/JSON config list, or `_defaults` + `configs`, to the same `published-components publish` command; the command aggregates results and exits nonzero if any component errors. A top-level `_select` node can choose between such documents per environment (see [Environment-selected configs](#environment-selected-configs-_select)). diff --git a/packages/tangle-cli/src/tangle_cli/component_publisher.py b/packages/tangle-cli/src/tangle_cli/component_publisher.py index ea15255..49298ba 100644 --- a/packages/tangle-cli/src/tangle_cli/component_publisher.py +++ b/packages/tangle-cli/src/tangle_cli/component_publisher.py @@ -34,6 +34,59 @@ class ProcessingOutcome(str, Enum): ERROR = "error" +@dataclass(frozen=True) +class _PublishedState: + """One observation of the owner-scoped published state of a component. + + ``versions`` maps digest → verified published version for non-deprecated + candidates. ``unreadable`` holds identifiers of non-deprecated candidates + whose version could not be determined; any entry there means the view is + incomplete and callers must fail closed rather than publish/deprecate. + Derived digest lists are sorted so diagnostics never depend on API order. + """ + + versions: Mapping[str, str] = field(default_factory=dict) + unreadable: tuple[str, ...] = () + deprecated_count: int = 0 + found_count: int = 0 + + @property + def latest_version(self) -> str | None: + """Highest verified published version, deterministic across API orders.""" + + latest: str | None = None + for version in sorted(set(self.versions.values())): + if latest is None or utils.compare_versions(version, latest) > 0: + latest = version + return latest + + @property + def latest_digests(self) -> tuple[str, ...]: + """Sorted digests whose version compares equal to :attr:`latest_version`.""" + + latest = self.latest_version + if latest is None: + return () + return tuple( + sorted( + digest + for digest, version in self.versions.items() + if utils.compare_versions(version, latest) == 0 + ) + ) + + def digests_older_than(self, local_version: str) -> tuple[str, ...]: + """Sorted digests proven strictly older than ``local_version``.""" + + return tuple( + sorted( + digest + for digest, version in self.versions.items() + if utils.compare_versions(version, local_version) < 0 + ) + ) + + @dataclass class ProcessingResult: """Result for one component publish/deprecate processing step. @@ -56,6 +109,9 @@ class ProcessingResult: digest: str | None = None latest_digest: str | None = None response: Any = None + #: Verified digest → version snapshot the decision was made from. In-process + #: detail for callers/tests; deliberately not emitted by :meth:`to_dict`. + published_versions: Mapping[str, str] = field(default_factory=dict) @property def resolved_digest(self) -> str | None: @@ -262,101 +318,178 @@ def perform_version_check(self, spec: Any) -> ProcessingResult: Ordering is defined by :func:`tangle_cli.utils.compare_versions` against the highest non-deprecated owner-scoped published version: - * nothing published (or no readable remote version) → PROCEED + * nothing published → PROCEED * local strictly newer → PROCEED * local equal → SKIP, carrying that version's exact digest * local strictly older → SKIP (no-op), carrying the newer published digest; never publishes an older version and never deprecates a newer one. ``allow_downgrade=True`` opts out of this last rule. - If several non-deprecated owner-scoped components tie at the selected - latest version, a SKIP cannot name one exact digest, so the check fails - closed with an ERROR rather than guessing from API ordering. + The check fails closed — ERROR, and callers must not publish or + deprecate anything — when the published state cannot be read + completely: + + * any non-deprecated owner-scoped candidate whose digest is missing or + whose spec/version cannot be fetched or parsed (an unreadable row + could be newer than the local version, and would later be deprecated + on a blind "deprecate everything" pass); + * several non-deprecated candidates tied at the selected latest + version, where no exact digest can be chosen. + + Diagnostics list digests in sorted order so results never depend on + API response ordering. """ local_version = spec.version self.log.info(f" Local version: {local_version}") - latest_version: str | None = None - latest_digests: list[str] = [] - if self.dry_run: test_version = os.environ.get("TEST_LATEST_VERSION") + state = _PublishedState() if test_version: - latest_version = test_version - self.log.info(f" Remote version (test): {latest_version}") - else: - client = self._get_client() - if client is None: - return ProcessingResult( - outcome=ProcessingOutcome.ERROR, - local_version=str(local_version), - latest_version=None, - reason="Failed to create API client", - ) + state = _PublishedState(latest_version=test_version) + self.log.info(f" Remote version (test): {test_version}") + return self._evaluate_published_state(spec, state) - filter_by = self.published_by or self.current_user_id(client) - if not filter_by: - self.log.error( - "❌ Cannot determine current user — aborting to avoid deprecating components owned by others" - ) - return ProcessingResult( - outcome=ProcessingOutcome.ERROR, - local_version=str(local_version), - latest_version=None, - reason="Cannot determine current user for author filtering", - ) + client = self._get_client() + if client is None: + return ProcessingResult( + outcome=ProcessingOutcome.ERROR, + local_version=str(local_version), + latest_version=None, + reason="Failed to create API client", + ) + + filter_by = self.owner_filter(client) + if not filter_by: + return ProcessingResult( + outcome=ProcessingOutcome.ERROR, + local_version=str(local_version), + latest_version=None, + reason="Cannot determine current user for author filtering", + ) + + state = self.collect_published_state(client, spec, filter_by, verbose=False) + return self._evaluate_published_state(spec, state) + + def owner_filter(self, client: Any) -> str | None: + """Resolve the owner scope for lookups/deprecation, logging on failure.""" - existing_components = client.find_existing_components( - spec.search_names, - verbose=False, - published_by=filter_by, + filter_by = self.published_by or self.current_user_id(client) + if not filter_by: + self.log.error( + "❌ Cannot determine current user — aborting to avoid deprecating components owned by others" ) + return filter_by - if existing_components: - for component in existing_components: - digest = self.component_digest(component) - if not digest: - continue - if self.component_is_deprecated(component): - continue - try: - full_spec = client.get_component_spec(digest) - remote_version = full_spec.version if full_spec else None - except Exception as exc: - self.log.warn(f" Warning: Failed to get version for component {digest[:16]}: {exc}") - continue - if not remote_version: - continue - if latest_version is None or utils.compare_versions(remote_version, latest_version) > 0: - latest_version = remote_version - latest_digests = [digest] - elif utils.compare_versions(remote_version, latest_version) == 0 and digest not in latest_digests: - latest_digests.append(digest) - - if latest_version: - self.log.info(f" Remote version: {latest_version}") + def collect_published_state( + self, + client: Any, + spec: Any, + filter_by: str, + *, + verbose: bool = False, + ) -> "_PublishedState": + """Read the owner-scoped published state for ``spec`` exactly once. + + Returns a verified digest → version snapshot plus the identifiers of + any non-deprecated candidate that could not be read. Callers decide + policy; this method only observes. + """ + + existing_components = client.find_existing_components( + spec.search_names, + verbose=verbose, + published_by=filter_by, + ) + + versions: dict[str, str] = {} + unreadable: list[str] = [] + deprecated_count = 0 + + for component in existing_components or []: + if self.component_is_deprecated(component): + deprecated_count += 1 + continue + digest = self.component_digest(component) + if not digest: + name = None + if isinstance(component, Mapping): + name = component.get("name") else: - self.log.info( - f" ℹ️ Found {len(existing_components)} component(s) but couldn't extract version" - ) + name = getattr(component, "name", None) + unreadable.append(f"") + continue + try: + full_spec = client.get_component_spec(digest) + remote_version = full_spec.version if full_spec else None + except Exception as exc: + self.log.warn(f" Warning: Failed to get version for component {digest[:16]}: {exc}") + unreadable.append(digest) + continue + if not remote_version: + unreadable.append(digest) + continue + versions[digest] = str(remote_version) + + return _PublishedState( + versions=versions, + unreadable=tuple(sorted(unreadable)), + deprecated_count=deprecated_count, + found_count=len(existing_components or []), + ) + + def _evaluate_published_state( + self, + spec: Any, + state: "_PublishedState", + *, + stage: str = "version check", + ) -> ProcessingResult: + """Apply the monotonic publish policy to an observed published state.""" + + local_version = spec.version + prefix = "" if stage == "version check" else f"({stage}) " + + if state.unreadable: + listed = ", ".join(state.unreadable) + reason = ( + f"Cannot read published version for {len(state.unreadable)} non-deprecated " + f"component(s) ({listed}); refusing to publish or deprecate without a complete view" + ) + self.log.error(f" ❌ {prefix}{reason}") + return ProcessingResult( + outcome=ProcessingOutcome.ERROR, + local_version=local_version, + latest_version=None, + spec=spec, + reason=reason, + published_versions=state.versions, + ) + + latest_version = state.latest_version + latest_digests = state.latest_digests + + if state.versions: + self.log.info(f" Remote version: {latest_version}") + elif state.found_count and state.deprecated_count == state.found_count: + self.log.info(f" ℹ️ Found {state.found_count} component(s), all deprecated") if len(latest_digests) > 1: tied = ", ".join(latest_digests) - self.log.error( - f" ❌ Ambiguous latest published version {latest_version}: " - f"{len(latest_digests)} non-deprecated components share it ({tied})" + reason = ( + f"Ambiguous latest published version {latest_version}: " + f"{len(latest_digests)} non-deprecated components share it ({tied}); " + "cannot select an exact digest" ) + self.log.error(f" ❌ {prefix}{reason}") return ProcessingResult( outcome=ProcessingOutcome.ERROR, local_version=local_version, latest_version=latest_version, spec=spec, - reason=( - f"Ambiguous latest published version {latest_version}: " - f"{len(latest_digests)} non-deprecated components share it ({tied}); " - "cannot select an exact digest" - ), + reason=reason, + published_versions=state.versions, ) latest_digest = latest_digests[0] if latest_digests else None @@ -367,7 +500,7 @@ def perform_version_check(self, spec: Any) -> ProcessingResult: if should_proceed: version_suffix = " (older)" if comparison < 0 else "" self.log.info( - " ➡️ Version " + f" ➡️ {prefix}Version " + (f"{latest_version}{version_suffix}" if latest_version else "new") + f" → {local_version}" ) @@ -377,18 +510,19 @@ def perform_version_check(self, spec: Any) -> ProcessingResult: latest_version=latest_version, latest_digest=latest_digest, spec=spec, + published_versions=state.versions, ) if comparison < 0: self.log.info( - f" ⏭️ Skipping: Local version {local_version} is older than published {latest_version}" + f" ⏭️ {prefix}Skipping: Local version {local_version} is older than published {latest_version}" ) reason = ( f"Version {local_version} is older than published version {latest_version} " "(no-op; publishing is monotonic)" ) else: - self.log.info(f" ⏭️ Skipping: Version {local_version} unchanged") + self.log.info(f" ⏭️ {prefix}Skipping: Version {local_version} unchanged") reason = f"Version {local_version} unchanged (matches remote)" return ProcessingResult( @@ -398,6 +532,7 @@ def perform_version_check(self, spec: Any) -> ProcessingResult: latest_digest=latest_digest, spec=spec, reason=reason, + published_versions=state.versions, ) def deprecate_old_components( @@ -632,11 +767,8 @@ def publish_component( response={"name": spec.name, "text": local_yaml_content}, ) - filter_by = self.published_by or self.current_user_id(client) + filter_by = self.owner_filter(client) if not filter_by: - self.log.error( - "❌ Cannot determine current user — aborting to avoid deprecating components owned by others" - ) return ProcessingResult( outcome=ProcessingOutcome.ERROR, local_version=version_check_result.local_version, @@ -645,7 +777,23 @@ def publish_component( spec=spec, reason="Cannot determine current user for author filtering", ) - existing_components = client.find_existing_components(spec.search_names, verbose=True, published_by=filter_by) + + # Re-observe the published state immediately before create/deprecate and + # re-apply the same policy: a row that appeared after the first check + # must be able to stop the publish, and must never be deprecated on the + # strength of the earlier decision. + final_state = self.collect_published_state(client, spec, filter_by, verbose=True) + final_check = self._evaluate_published_state(spec, final_state, stage="refreshed check") + if final_check.outcome != ProcessingOutcome.PROCEED: + if final_check.outcome == ProcessingOutcome.SKIP: + self.log.info(f" ⏭️ Skipping API publish: {final_check.reason}") + else: + self.log.error(f" ❌ Cannot proceed due to error: {final_check.reason}") + return final_check + + deprecation_candidates = [ + {"digest": digest} for digest in final_state.digests_older_than(str(spec.version)) + ] try: result = client.published_components_create(name=spec.name, text=local_yaml_content) @@ -654,24 +802,25 @@ def publish_component( if new_digest: self.log.info(f"✅ Published: {spec.name} (digest: {str(new_digest)[:16]}...)") - self.deprecate_old_components(existing_components, str(new_digest)) + self.deprecate_old_components(deprecation_candidates, str(new_digest)) return ProcessingResult( outcome=ProcessingOutcome.SUCCESS, - local_version=version_check_result.local_version, - latest_version=version_check_result.latest_version, + local_version=final_check.local_version, + latest_version=final_check.latest_version, spec=spec, reason=f"Successfully published with digest: {new_digest}", digest=str(new_digest), - latest_digest=version_check_result.latest_digest, + latest_digest=final_check.latest_digest, response=result, + published_versions=final_state.versions, ) self.log.warn("⚠️ Component published but no digest returned") return ProcessingResult( outcome=ProcessingOutcome.ERROR, - local_version=version_check_result.local_version, - latest_version=version_check_result.latest_version, - latest_digest=version_check_result.latest_digest, + local_version=final_check.local_version, + latest_version=final_check.latest_version, + latest_digest=final_check.latest_digest, spec=spec, reason="Component published but no digest returned", response=result, @@ -680,9 +829,9 @@ def publish_component( self.log.error(f"❌ Request failed: {exc}") return ProcessingResult( outcome=ProcessingOutcome.ERROR, - local_version=version_check_result.local_version, - latest_version=version_check_result.latest_version, - latest_digest=version_check_result.latest_digest, + local_version=final_check.local_version, + latest_version=final_check.latest_version, + latest_digest=final_check.latest_digest, spec=spec, reason=f"Request failed: {exc}", ) diff --git a/tests/test_component_publisher.py b/tests/test_component_publisher.py index 8541a4c..62c9971 100644 --- a/tests/test_component_publisher.py +++ b/tests/test_component_publisher.py @@ -406,6 +406,191 @@ def failing_create(**kwargs: Any) -> dict[str, Any]: assert client.update_calls == [] +# --------------------------------------------------------------------------- +# Fail-closed reads, lookup races, deterministic diagnostics +# --------------------------------------------------------------------------- + + +class RaceClient(FakeClient): + """Client whose published state changes between the two owner-scoped lookups.""" + + def __init__(self, first: list[ExistingComponent], second: list[ExistingComponent]) -> None: + super().__init__() + self._sequence = [first, second] + self.existing = first + + def find_existing_components(self, components: Any, **kwargs: Any) -> list[ExistingComponent]: + self.existing = self._sequence[min(len(self.find_calls), len(self._sequence) - 1)] + return super().find_existing_components(components, **kwargs) + + +def test_version_check_fails_closed_on_unreadable_candidate(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = FakeClient() + client.existing = [ExistingComponent("sha256:v10"), ExistingComponent("sha256:unknown")] + # "sha256:unknown" is absent from component_versions, so get_component_spec raises. + client.component_versions = {"sha256:v10": "1.0"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.ERROR + assert "Cannot read published version" in (result.reason or "") + assert "sha256:unknown" in (result.reason or "") + assert result.resolved_digest is None + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_version_check_fails_closed_on_candidate_without_digest() -> None: + spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '2.0'\n") + client = FakeClient() + client.existing = [ExistingComponent("", name="demo")] + + result = perform_version_check(spec=spec, dry_run=False, client=client) + + assert result.outcome == ProcessingOutcome.ERROR + assert "no digest" in (result.reason or "") + assert client.create_calls == [] + + +def test_unreadable_row_appearing_before_publish_blocks_create_and_deprecate(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = RaceClient( + [ExistingComponent("sha256:v10")], + [ExistingComponent("sha256:v10"), ExistingComponent("sha256:unknown")], + ) + client.component_versions = {"sha256:v10": "1.0"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.ERROR + assert "sha256:unknown" in (result.reason or "") + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_concurrent_newer_row_between_lookups_skips_without_publishing(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = RaceClient( + [ExistingComponent("sha256:v10")], + [ExistingComponent("sha256:v10"), ExistingComponent("sha256:v30")], + ) + client.component_versions = {"sha256:v10": "1.0", "sha256:v30": "3.0"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_version == "3.0" + assert result.latest_digest == "sha256:v30" + assert result.resolved_digest == "sha256:v30" + assert "older" in (result.reason or "") + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_concurrent_equal_row_between_lookups_skips_without_publishing(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = RaceClient([], [ExistingComponent("sha256:v20")]) + client.component_versions = {"sha256:v20": "2.0"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_digest == "sha256:v20" + assert "unchanged" in (result.reason or "") + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_concurrent_ambiguous_rows_between_lookups_fail_closed(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = RaceClient( + [ExistingComponent("sha256:v10")], + [ExistingComponent("sha256:b"), ExistingComponent("sha256:a")], + ) + client.component_versions = {"sha256:v10": "1.0", "sha256:a": "3.0", "sha256:b": "3.0"} + + result = publish_component_to_tangle(component_path, client=client) + + assert result.outcome == ProcessingOutcome.ERROR + assert "Ambiguous latest published version 3.0" in (result.reason or "") + assert client.create_calls == [] + assert client.update_calls == [] + + +def test_row_appearing_after_version_check_is_never_deprecated(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="2.0") + client = RaceClient( + [ExistingComponent("sha256:v10")], + [ExistingComponent("sha256:v10"), ExistingComponent("sha256:v11")], + ) + client.component_versions = {"sha256:v10": "1.0", "sha256:v11": "1.1"} + client.publish_response = {"digest": "sha256:v20"} + + result = publish_component_to_tangle(component_path, client=client) + + # Both late rows are proven older than 2.0, so both may be deprecated, in + # sorted order; nothing unverified is ever touched. + assert result.outcome == ProcessingOutcome.SUCCESS + assert [call["digest"] for call in client.update_calls] == ["sha256:v10", "sha256:v11"] + + +def test_allow_downgrade_never_deprecates_newer_rows(tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="1.0") + client = FakeClient() + client.existing = [ExistingComponent("sha256:v20"), ExistingComponent("sha256:v005")] + client.component_versions = {"sha256:v20": "2.0", "sha256:v005": "0.5"} + client.publish_response = {"digest": "sha256:v10"} + + result = publish_component_to_tangle(component_path, client=client, allow_downgrade=True) + + assert result.outcome == ProcessingOutcome.SUCCESS + assert [call["digest"] for call in client.update_calls] == ["sha256:v005"] + + +def test_ambiguity_diagnostics_are_sorted_independently_of_api_order() -> None: + spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '1.0'\n") + reasons = [] + for digests in (["sha256:a", "sha256:b"], ["sha256:b", "sha256:a"]): + client = FakeClient() + client.existing = [ExistingComponent(digest) for digest in digests] + client.component_versions = {"sha256:a": "2.0", "sha256:b": "2.0"} + result = perform_version_check(spec=spec, dry_run=False, client=client) + assert result.outcome == ProcessingOutcome.ERROR + reasons.append(result.reason) + + assert reasons[0] == reasons[1] + assert "(sha256:a, sha256:b)" in (reasons[0] or "") + + +def test_latest_digest_selection_is_independent_of_api_order() -> None: + spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '1.0'\n") + selected = [] + for digests in (["sha256:v10", "sha256:v30", "sha256:v20"], ["sha256:v30", "sha256:v20", "sha256:v10"]): + client = FakeClient() + client.existing = [ExistingComponent(digest) for digest in digests] + client.component_versions = {"sha256:v10": "1.0", "sha256:v20": "2.0", "sha256:v30": "3.0"} + result = perform_version_check(spec=spec, dry_run=False, client=client) + selected.append((result.latest_version, result.latest_digest)) + + assert selected[0] == selected[1] == ("3.0", "sha256:v30") + + +def test_compare_equal_raw_versions_pick_deterministic_representative() -> None: + spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '0.9'\n") + picked = [] + for digests in (["sha256:a", "sha256:b"], ["sha256:b", "sha256:a"]): + client = FakeClient() + client.existing = [ExistingComponent(digest) for digest in digests] + # "1.0" and "1.0.0" compare equal, so this is an ambiguous tie either way. + client.component_versions = {"sha256:a": "1.0", "sha256:b": "1.0.0"} + result = perform_version_check(spec=spec, dry_run=False, client=client) + assert result.outcome == ProcessingOutcome.ERROR + picked.append((result.latest_version, result.reason)) + + assert picked[0] == picked[1] + + def test_version_check_without_owner_reports_error_and_no_digest() -> None: spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '1.0'\n") client = FakeClient() From 402a7a9c586d8a9832158f381a49be6baad028bf Mon Sep 17 00:00:00 2001 From: Volv G Date: Wed, 16 Sep 2026 12:48:47 -0700 Subject: [PATCH 3/4] Fix dry-run TEST_LATEST_VERSION state and correct the publish matrix docs The dry-run override built _PublishedState(latest_version=...), but latest_version is a derived property, so any dry-run with TEST_LATEST_VERSION set raised TypeError before reaching the policy. The synthetic override now has its own field. It has no backing published component, so it yields no digest: latest_version reports it, while latest_digest and resolved_digest stay None, and the normal older/equal/newer rules apply. The README outcome table also still claimed that an unreadable remote version proceeds, which contradicted the fail-closed behavior described directly below it. Unreadable and ambiguous states are now their own error row. Assisted-By: devx/6aa1a44a-8f9f-4be7-b20c-752af4396eb9 (cherry picked from commit 11e45bc3bea322150b20f4af7640e944b33f87ae) --- README.md | 5 +- .../src/tangle_cli/component_publisher.py | 8 ++- tests/test_component_publisher.py | 50 +++++++++++++++++++ 3 files changed, 60 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 0b86894..64d7f94 100644 --- a/README.md +++ b/README.md @@ -361,10 +361,11 @@ Publishing is monotonic against the highest **non-deprecated, owner-scoped** pub | Local vs latest published | Outcome | Notes | | --- | --- | --- | -| nothing published / no readable remote version | `proceed` | first publish | -| local strictly newer | `proceed` | publishes, then deprecates older owner-scoped versions | +| nothing published (no non-deprecated owner-scoped version) | `proceed` | first publish | +| local strictly newer | `proceed` | publishes, then deprecates owner-scoped versions proven older | | local equal | `skip` | no create/deprecate calls | | local strictly older | `skip` | no-op; never publishes an older version and never deprecates a newer one | +| published version unreadable, or ambiguous tie at the latest version | `error` | fails closed; no create/deprecate calls | Every result carries the digest of the version it compared against: diff --git a/packages/tangle-cli/src/tangle_cli/component_publisher.py b/packages/tangle-cli/src/tangle_cli/component_publisher.py index 49298ba..7d7d507 100644 --- a/packages/tangle-cli/src/tangle_cli/component_publisher.py +++ b/packages/tangle-cli/src/tangle_cli/component_publisher.py @@ -49,11 +49,17 @@ class _PublishedState: unreadable: tuple[str, ...] = () deprecated_count: int = 0 found_count: int = 0 + #: Dry-run ``TEST_LATEST_VERSION`` override. Synthetic: it has no backing + #: published component, so it never contributes a digest. + synthetic_latest_version: str | None = None @property def latest_version(self) -> str | None: """Highest verified published version, deterministic across API orders.""" + if not self.versions: + return self.synthetic_latest_version + latest: str | None = None for version in sorted(set(self.versions.values())): if latest is None or utils.compare_versions(version, latest) > 0: @@ -347,7 +353,7 @@ def perform_version_check(self, spec: Any) -> ProcessingResult: test_version = os.environ.get("TEST_LATEST_VERSION") state = _PublishedState() if test_version: - state = _PublishedState(latest_version=test_version) + state = _PublishedState(synthetic_latest_version=test_version) self.log.info(f" Remote version (test): {test_version}") return self._evaluate_published_state(spec, state) diff --git a/tests/test_component_publisher.py b/tests/test_component_publisher.py index 62c9971..a1f3838 100644 --- a/tests/test_component_publisher.py +++ b/tests/test_component_publisher.py @@ -591,6 +591,56 @@ def test_compare_equal_raw_versions_pick_deterministic_representative() -> None: assert picked[0] == picked[1] +def test_dry_run_test_latest_version_applies_monotonic_rules(monkeypatch) -> None: + spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '2.0'\n") + + monkeypatch.setenv("TEST_LATEST_VERSION", "1.0") + newer = perform_version_check(spec=spec, dry_run=True) + assert newer.outcome == ProcessingOutcome.PROCEED + assert newer.latest_version == "1.0" + assert newer.latest_digest is None + + monkeypatch.setenv("TEST_LATEST_VERSION", "2.0") + equal = perform_version_check(spec=spec, dry_run=True) + assert equal.outcome == ProcessingOutcome.SKIP + assert equal.latest_version == "2.0" + assert equal.latest_digest is None + assert equal.resolved_digest is None + assert "unchanged" in (equal.reason or "") + + monkeypatch.setenv("TEST_LATEST_VERSION", "3.0") + older = perform_version_check(spec=spec, dry_run=True) + assert older.outcome == ProcessingOutcome.SKIP + assert older.latest_version == "3.0" + assert older.latest_digest is None + assert older.resolved_digest is None + assert "older" in (older.reason or "") + + +def test_dry_run_without_test_latest_version_proceeds(monkeypatch) -> None: + spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '2.0'\n") + monkeypatch.delenv("TEST_LATEST_VERSION", raising=False) + + result = perform_version_check(spec=spec, dry_run=True) + + assert result.outcome == ProcessingOutcome.PROCEED + assert result.latest_version is None + assert result.latest_digest is None + + +def test_dry_run_publish_skips_when_test_latest_version_is_newer(monkeypatch, tmp_path: Path) -> None: + component_path = write_component(tmp_path / "component.yaml", version="1.0") + client = FakeClient() + monkeypatch.setenv("TEST_LATEST_VERSION", "2.0") + + result = publish_component_to_tangle(component_path, dry_run=True, client=client) + + assert result.outcome == ProcessingOutcome.SKIP + assert result.latest_version == "2.0" + assert client.create_calls == [] + assert client.update_calls == [] + + def test_version_check_without_owner_reports_error_and_no_digest() -> None: spec = ComponentSpec.from_yaml("name: demo\nmetadata:\n annotations:\n version: '1.0'\n") client = FakeClient() From 9e771972746478b428c9aa75e6580f3e282e6b78 Mon Sep 17 00:00:00 2001 From: Volv G Date: Thu, 17 Sep 2026 14:36:16 -0700 Subject: [PATCH 4/4] Add @Publish with owner-scoped component resolution (v0.1.15) Introduce a `@Publish` decorator that records a component's registry name and version as real `CallableRef` fields, and emit that declaration into the compiled resolver sidecar so separate tooling can act on it: orders-loader: name: orders-loader version: "1.0" publisher: me local_from_python: {...} publish: true The declaration uses the entry's OWN generic name/version fields, so ordinary hydration looks the published component up and falls back to the local block only when no such component exists. A registry error is not a fallback: resolution fails closed rather than quietly building from local source. `publish` is a separate literal `true` and is never inferred from the other fields, so a hand-written entry of that shape is not silently published. `publisher` is the symbolic `me`, resolved at hydration time to whoever is authenticated. Compilation stays offline and records no account id, so an artifact is not bound to one author's account. Without this scoping the lookup is global, and a component published by someone else under the same name and version is a valid candidate -- a name is not an identity control. The symbol is interpreted only alongside `publish: true`, which the compiler always writes together, so `me` is not a reserved account id and a hand-authored publisher is still used verbatim and still resolves without authentication. Owner filtering is made exact. The API owner parameter is a substring match, so an exact `published_by` request is post-filtered on the returned rows: a superset id (`alice` vs `alice2`) and a row with no owner are both refused. Explicit `published_by_substring` callers keep partial matching. This also tightens the publisher's owner-scoped version check, which must not treat another account's components as its own. Identity parsing is shared by the publisher and the hydrator so the two cannot disagree about who the current user is. Publication is deliberately NOT part of component dedup identity, so two declarations that agree are still one component. Assisted-By: devx/6257e672-aaa7-443c-a4e1-a0150a485d9d --- .../tangle-cli/src/tangle_cli/__init__.py | 2 +- .../src/tangle_cli/authenticated_identity.py | 77 +++ packages/tangle-cli/src/tangle_cli/client.py | 10 + .../src/tangle_cli/component_from_func.py | 9 +- .../src/tangle_cli/component_publisher.py | 13 +- .../src/tangle_cli/pipeline_compiler.py | 92 ++- .../src/tangle_cli/pipeline_hydrator.py | 14 + .../tangle_cli/python_pipeline/__init__.py | 2 + .../src/tangle_cli/python_pipeline/publish.py | 142 +++++ .../src/tangle_cli/python_pipeline/ref.py | 10 + pyproject.toml | 2 +- tests/test_component_publisher.py | 50 ++ tests/test_packaging.py | 2 +- tests/test_python_pipeline_dsl.py | 1 + tests/test_python_pipeline_publish.py | 565 ++++++++++++++++++ tests/test_static_client.py | 94 +++ tests/test_symbolic_publisher_hydration.py | 189 ++++++ uv.lock | 2 +- 18 files changed, 1254 insertions(+), 22 deletions(-) create mode 100644 packages/tangle-cli/src/tangle_cli/authenticated_identity.py create mode 100644 packages/tangle-cli/src/tangle_cli/python_pipeline/publish.py create mode 100644 tests/test_python_pipeline_publish.py create mode 100644 tests/test_symbolic_publisher_hydration.py diff --git a/packages/tangle-cli/src/tangle_cli/__init__.py b/packages/tangle-cli/src/tangle_cli/__init__.py index a4243cf..f087eea 100644 --- a/packages/tangle-cli/src/tangle_cli/__init__.py +++ b/packages/tangle-cli/src/tangle_cli/__init__.py @@ -14,6 +14,6 @@ try: __version__ = metadata_version("tangle-cli") except PackageNotFoundError: - __version__ = "0.1.14" + __version__ = "0.1.15" __all__ = ["TangleDynamicDiscoveryClient", "__version__"] diff --git a/packages/tangle-cli/src/tangle_cli/authenticated_identity.py b/packages/tangle-cli/src/tangle_cli/authenticated_identity.py new file mode 100644 index 0000000..9e65edd --- /dev/null +++ b/packages/tangle-cli/src/tangle_cli/authenticated_identity.py @@ -0,0 +1,77 @@ +"""Resolution of the authenticated account, and the symbolic ``me`` publisher. + +Compilation is offline, so a compiler cannot write the author's account id into +a component entry. It writes the symbolic publisher :data:`ME` instead, and the +account is resolved at hydration time by whoever is actually authenticated. + +Two callers need the account: the publisher, which scopes its version check and +deprecation to the owner, and hydration of entries whose publisher is ``me``. +They must agree, so the parsing lives here once. + +They differ only in what an *unknown* account means, so that choice is left to +the caller: the publisher degrades, while resolution must fail closed -- see +:func:`require_authenticated_user_id`. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import Any + +#: Symbolic publisher meaning "whoever is authenticated at hydration time". +#: Matched exactly and case-sensitively, so a literal account id that happens to +#: differ in case is never mistaken for the sentinel. +ME = "me" + +__all__ = [ + "ME", + "IdentityUnavailableError", + "authenticated_user_id", + "is_symbolic_me", + "require_authenticated_user_id", +] + + +class IdentityUnavailableError(RuntimeError): + """The authenticated account could not be determined.""" + + +def is_symbolic_me(publisher: Any) -> bool: + """Whether ``publisher`` is the symbolic self-reference rather than an id.""" + return publisher == ME + + +def authenticated_user_id(client: Any) -> str | None: + """Return the current user id, or ``None`` if it cannot be read. + + An empty or missing id is as unusable as no answer at all. + """ + try: + user_info = client.users_me() + except Exception: + return None + if user_info is None: + return None + if isinstance(user_info, Mapping): + value = user_info.get("id") + else: + value = getattr(user_info, "id", None) + return str(value) if value else None + + +def require_authenticated_user_id(client: Any) -> str: + """Return the current user id, raising when it cannot be determined. + + For owner-scoped *resolution* an unknown account must never widen the + search: continuing unscoped is what would let a component published by + someone else under the same name be resolved as the author's own code. + """ + user_id = authenticated_user_id(client) + if not user_id: + raise IdentityUnavailableError( + "Cannot determine the authenticated account, so a component " + f"published by '{ME}' cannot be resolved. Refusing to fall back to " + "an unscoped lookup, which could resolve a component owned by " + "someone else. Re-authenticate and retry." + ) + return user_id diff --git a/packages/tangle-cli/src/tangle_cli/client.py b/packages/tangle-cli/src/tangle_cli/client.py index 2c15a9b..5f07831 100644 --- a/packages/tangle-cli/src/tangle_cli/client.py +++ b/packages/tangle-cli/src/tangle_cli/client.py @@ -952,12 +952,22 @@ def find_existing_components( search_digests.add(str(data["digest"])) publisher_filter = published_by_substring or published_by + # The API parameter is a SUBSTRING match. When the caller asked for an + # exact owner it is therefore only a prefilter, and accepting its + # results verbatim would let a superset owner id (``alice`` matching + # ``alice2``) satisfy an exact request. An owner-scoped lookup is an + # identity control, so exactness is enforced here on the returned rows + # and a row with no owner is never accepted. Explicit + # ``published_by_substring`` callers keep substring semantics. + exact_owner = published_by if published_by and not published_by_substring else None found: dict[str, ComponentInfo] = {} def add(info: ComponentInfo) -> None: key = info.digest or info.name if not key: return + if exact_owner is not None and info.published_by != exact_owner: + return found[key] = info if verbose: self.logger.info(f" Found existing component: {info.name} ({key[:16]}...)") diff --git a/packages/tangle-cli/src/tangle_cli/component_from_func.py b/packages/tangle-cli/src/tangle_cli/component_from_func.py index df246b8..7ea20f3 100644 --- a/packages/tangle-cli/src/tangle_cli/component_from_func.py +++ b/packages/tangle-cli/src/tangle_cli/component_from_func.py @@ -986,8 +986,13 @@ def _is_main_str(n: ast.expr) -> bool: # _strip_authoring_constructs). ``registered`` marks an op published separately # via its own gen_config.yaml; when that same op is baked (through its # local_from_python entry) the decorator + its authoring import must be stripped -# too, exactly like @task. -_AUTHORING_DECORATOR_NAMES = frozenset({"task", "pipeline", "subpipeline", "registered"}) +# too, exactly like @task. ``Publish`` likewise only records a publication +# declaration for the compiler; leaving it in the baked program would raise +# ``NameError`` at container startup, since its import is stripped with the +# rest of the authoring surface. +_AUTHORING_DECORATOR_NAMES = frozenset( + {"task", "pipeline", "subpipeline", "registered", "Publish"} +) # The python-pipeline authoring modules. ONLY imports of these modules (and # their submodules) are authoring-only and stripped from the baked source. We diff --git a/packages/tangle-cli/src/tangle_cli/component_publisher.py b/packages/tangle-cli/src/tangle_cli/component_publisher.py index 7d7d507..24317f2 100644 --- a/packages/tangle-cli/src/tangle_cli/component_publisher.py +++ b/packages/tangle-cli/src/tangle_cli/component_publisher.py @@ -18,6 +18,7 @@ import tangle_cli.utils as utils +from .authenticated_identity import authenticated_user_id from .handler import TangleCliHandler from .logger import Logger @@ -301,17 +302,7 @@ def component_is_deprecated(self, component: Any) -> bool: def current_user_id(self, client: Any) -> str | None: """Return the current Tangle user id for owner-scoped lookups.""" - try: - user_info = client.users_me() - except Exception: - return None - if user_info is None: - return None - if isinstance(user_info, Mapping): - value = user_info.get("id") - else: - value = getattr(user_info, "id", None) - return str(value) if value else None + return authenticated_user_id(client) def perform_version_check(self, spec: Any) -> ProcessingResult: """Perform owner-scoped, monotonic version checking for a component. diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py b/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py index c850d4d..b7365d4 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_compiler.py @@ -52,6 +52,7 @@ import yaml +from .authenticated_identity import ME from .component_from_func import build_unwrapped_inputs_schema from .handler import TangleCliHandler from .python_pipeline.cfg import Cfg, _coerce_override, load_cfg @@ -1814,6 +1815,57 @@ def _task_component_identity( } +def _task_publication(ref: CallableRef) -> dict[str, str] | None: + """The ``@Publish`` declaration on one traced ``@task`` call, if any. + + Returns: + The neutral ``publish`` marker mapping, or ``None`` when the task + declares no publication. + """ + if ref._task_publish_name is None: + return None + return { + "component_name": ref._task_publish_name, + "version": ref._task_publish_version or "", + } + + +def _require_agreeing_publication( + existing: dict[str, str] | None, + incoming: dict[str, str] | None, + *, + identity_payload: Mapping[str, Any], +) -> None: + """Refuse when refs sharing ONE component disagree about publication. + + Publication is intentionally not part of the dedup identity, so two call + sites that generate the same component collapse into a single sidecar + entry which can carry only one marker. Every ref folded into that entry + must therefore make the same declaration. + + By construction they do -- ``@Publish`` decorates the function, and every + derived ref copies the fields -- so this is a guard against a future + caller constructing refs directly, not the mechanism that makes the common + case work. + + Raises: + CompileError: If the two declarations differ. + """ + if existing == incoming: + return + + def _render(value: dict[str, str] | None) -> str: + if value is None: + return "no @Publish" + return f"@Publish({value['component_name']!r}, version={value['version']!r})" + + raise CompileError( + f"@Publish disagreement for {_task_identity_label(identity_payload)}: " + f"the same generated component is declared as {_render(existing)} and " + f"{_render(incoming)}. One component can carry one publication." + ) + + def _task_identity_label(identity_payload: Mapping[str, Any]) -> str: """Human-readable ``::`` label for an identity payload. @@ -1946,14 +1998,23 @@ def _plan_task_sidecar( identity = _stable_payload_hash(identity_payload) by_identity = variants.setdefault(base, {}) previous = by_identity.get(identity) + # Publication intent rides ALONGSIDE the identity, never inside it: + # it does not change the generated component, so folding it into the + # identity would split one component into two byte-identical ones. + # Every ref deduped into one component must therefore agree about it. + publication = _task_publication(ref) + if previous is not None: + _require_agreeing_publication( + previous[2], publication, identity_payload=identity_payload + ) if previous is None: - by_identity[identity] = (emitted, identity_payload) + by_identity[identity] = (emitted, identity_payload, publication) elif _emitted_form_rank(emitted) < _emitted_form_rank(previous[0]): # Same component, different SPELLING (e.g. one call site omits # ``mode`` while another writes ``mode="inline"``). Pick the # canonical representative rather than whichever was traced first, # so the sidecar text does not depend on call order. - by_identity[identity] = (emitted, previous[1]) + by_identity[identity] = (emitted, previous[1], previous[2]) legacy_fragments.setdefault((base, identity), _fragment_for_task(ref, unwrapped_schema)) task_identities.append((task_id, base, identity)) @@ -1962,7 +2023,7 @@ def _plan_task_sidecar( identity_labels: dict[str, str] = {} for base, by_identity in variants.items(): if len(by_identity) == 1: - identity, (emitted, _identity_payload) = next(iter(by_identity.items())) + identity, (emitted, _identity_payload, _publication) = next(iter(by_identity.items())) named = [(legacy_fragments[(base, identity)], identity, emitted)] else: # EVERY colliding variant is suffixed — the trace-order "first" one @@ -1970,7 +2031,7 @@ def _plan_task_sidecar( # sorted order so the sidecar text is call-order independent. named = sorted( (f"{base}--{identity}", identity, emitted) - for identity, (emitted, _identity_payload) in by_identity.items() + for identity, (emitted, _identity_payload, _publication) in by_identity.items() ) for fragment, identity, emitted in named: if fragment in entries: # defensive — digests are content-addressed @@ -1981,7 +2042,28 @@ def _plan_task_sidecar( ) fragment_by_identity[(base, identity)] = fragment identity_labels[fragment] = _task_identity_label(by_identity[identity][1]) - entries[fragment] = {"local_from_python": emitted} + entry: dict[str, Any] = {} + publication = by_identity[identity][2] + if publication is not None: + # The declaration is emitted as the resolver's OWN generic + # name/version/publisher fields, so ordinary hydration can look + # the published component up, with ``local_from_python`` as the + # candidate used when none is found. ``publish: true`` is a + # separate explicit marker: publication is never inferred from + # those fields alone, so a hand-written or fallback entry of + # that shape is not silently published. + # + # The publisher is the SYMBOLIC ``me``: compiling is offline and + # must stay so, so it cannot resolve an account id. Hydration + # resolves it to whoever is authenticated, which keeps the + # lookup owner-scoped instead of global. + entry["name"] = publication["component_name"] + entry["version"] = publication["version"] + entry["publisher"] = ME + entry["local_from_python"] = emitted + if publication is not None: + entry["publish"] = True + entries[fragment] = entry fragment_by_task = { task_id: fragment_by_identity[(base, identity)] for task_id, base, identity in task_identities diff --git a/packages/tangle-cli/src/tangle_cli/pipeline_hydrator.py b/packages/tangle-cli/src/tangle_cli/pipeline_hydrator.py index 485aeeb..4646ed5 100644 --- a/packages/tangle-cli/src/tangle_cli/pipeline_hydrator.py +++ b/packages/tangle-cli/src/tangle_cli/pipeline_hydrator.py @@ -23,6 +23,7 @@ import yaml from . import utils +from .authenticated_identity import is_symbolic_me, require_authenticated_user_id from .api_transport import DEFAULT_TIMEOUT_SECONDS from .component_generator import ComponentGenerator from .handler import TangleCliHandler @@ -1149,6 +1150,19 @@ def _resolve_by_name_with_filters( """Resolve a component by name with optional filters.""" component_name = entry["name"] publisher = entry.get("publisher") + if is_symbolic_me(publisher) and entry.get("publish") is True: + # Compilation is offline, so the compiler wrote a symbolic owner + # rather than an account id. Resolve it to whoever is actually + # authenticated: without an owner the lookup is global, and a + # component published by anyone else under this name and version + # would be a valid candidate. Unresolvable identity fails closed + # instead of widening the search. + # + # Gated on the marker as well as the symbol, because the compiler + # always writes both. An entry WITHOUT the marker is hand-authored, + # where ``me`` has always meant a literal account id; reinterpreting + # it would silently retarget or break such a config. + publisher = require_authenticated_user_id(self._api_client()) version_constraint = entry.get("version") required_annotations = entry.get("annotations") diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py index 18460b0..56bf560 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/__init__.py @@ -24,6 +24,7 @@ from .pipeline import pipeline from .raw import raw from .ref import ref +from .publish import Publish from .registered import registered from .subpipeline import subpipeline from .task import task @@ -33,6 +34,7 @@ __all__ = [ "pipeline", "task", + "Publish", "registered", "ref", "raw", diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/publish.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/publish.py new file mode 100644 index 0000000..b1b3c8a --- /dev/null +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/publish.py @@ -0,0 +1,142 @@ +"""``@Publish`` decorator -- an inert metadata stub. + +This package deliberately contains no publication implementation. ``@Publish`` +records a component name and version for a local ``@task``, and the compiler +writes the neutral marker shown below into the generated +``.components.yaml`` resolver sidecar, beside that component's +``local_from_python`` block; neither action publishes, contacts a registry, or +expresses publication policy, and decorating changes nothing about how the +component is generated. Some current consumers are closed source, though their +implementations are expected to be opened in the future. The marker schema is +the integration contract and is kept deliberately neutral so any +implementation can consume it:: + + orders-loader: + name: orders-loader + version: "1.0" + publisher: me + local_from_python: {...} + publish: true + +Separate tooling reads ``publish`` and decides whether, when, and under what +policy to publish. The declaration is emitted as the entry's OWN generic +``name``/``version`` fields, so ordinary hydration looks the published +component up at the exact declared version and uses ``local_from_python`` when +no such component exists yet. A registry error is NOT a fallback: resolution +fails closed rather than quietly building from local source. + +``publish`` is deliberately a separate literal ``true`` and is never inferred +from ``name`` + ``version`` + ``local_from_python``, so a hand-written or +fallback entry of that shape is not silently published. + +``publisher`` is the SYMBOLIC value ``me``, matched exactly and +case-sensitively, and is resolved at hydration time to whoever is +authenticated; compiling stays offline and never learns or records an account +id. This keeps the lookup owner-scoped: an unscoped name search would let a +component published by someone else under the same name and version be +resolved in its place, and a name is not an identity control. If the account +cannot be determined, resolution fails closed rather than widening. + +The symbol is interpreted only on an entry that also carries ``publish: true``, +which the compiler always writes together. ``me`` is therefore NOT a reserved +account id: in a hand-authored entry, and in any entry without the marker, a +publisher is used verbatim -- including one that happens to be ``me``. An entry +with no publisher keeps the ordinary cross-publisher behaviour. + +Ordering: ``@Publish`` goes directly ABOVE ``@task``, so it receives the +``CallableRef`` that ``@task`` produced:: + + from tangle_cli.python_pipeline import Publish, task + + @Publish(component_name="orders-loader", version="1.0") + @task(image="python:3.11") + def load_orders(source: str = "orders") -> str: + '''Load orders.''' + ... + +The declaration is stored as REAL ``CallableRef`` fields, so it survives +fluent composition (``.bind()``, ``.named()``) and is carried by refs imported +from an already-loaded module. This decorator does not consult the registry, +read the filesystem, or perform any I/O. +""" +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from .errors import CompileError +from .ref import CallableRef + +_USAGE = ( + "@Publish(component_name='my-component', version='1.0') directly above " + "the @task it publishes" +) + + +def _refuse(message: str) -> CompileError: + return CompileError(f"@Publish {message}") + + +def _require_text(value: Any, *, field: str) -> str: + """Validate one author-supplied string. + + Both values are echoed into diagnostics and written into a YAML document, + so control characters are rejected at the authoring boundary rather than + escaped at each point of use. + """ + if not isinstance(value, str) or not value.strip(): + raise _refuse(f"requires a non-empty string {field!r}. Write {_USAGE}.") + text = value.strip() + if any(ord(character) < 0x20 or ord(character) == 0x7F for character in text): + raise _refuse(f"{field!r} must not contain control characters.") + return text + + +@dataclass(frozen=True) +class Publish: + """Declare that the decorated ``@task`` component should be published. + + Args: + component_name: Registry name to publish under. Required, non-empty, + and never derived from the function name -- the registry name is a + deliberate, stable identifier that must not change because someone + renamed a Python function. A distinctive name is the author's + responsibility. + version: Version to publish. Required and non-empty; there is no + default and none is inferred. + + Returns: + The same :class:`CallableRef`, carrying the publication declaration. + + Raises: + CompileError: If either value is missing, blank, or contains control + characters, or if the decorated object is not a ``@task`` ref + (a plain function, a ``ref()``/``@registered`` ref, or ``@task`` + applied in the wrong order). + """ + + component_name: str + version: str + + def __post_init__(self) -> None: + object.__setattr__( + self, "component_name", _require_text(self.component_name, field="component_name") + ) + object.__setattr__(self, "version", _require_text(self.version, field="version")) + + def __call__(self, target: Any) -> CallableRef: + if not isinstance(target, CallableRef) or target._task_source_path is None: + raise _refuse( + f"can only publish a @task component, but it was applied to " + f"{type(target).__name__}. Write {_USAGE}." + ) + if target._task_publish_name is not None: + raise _refuse( + f"is already declared on this task as " + f"{target._task_publish_name!r} {target._task_publish_version}. " + f"One publication per task." + ) + return target._replace( + _task_publish_name=self.component_name, + _task_publish_version=self.version, + ) diff --git a/packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py b/packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py index fa0891f..e86b55c 100644 --- a/packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py +++ b/packages/tangle-cli/src/tangle_cli/python_pipeline/ref.py @@ -142,6 +142,16 @@ class CallableRef: _task_resolve_root: Path | None = None _task_custom_annotations: dict[str, str] | None = None _task_unwrap: tuple[str, ...] = () + # ``@Publish`` metadata. ``None`` unless the author declared a publication + # for this ``@task``. REAL dataclass fields rather than attributes stamped + # on the object: ``_replace`` re-copies only a dunder allowlist, so a + # dynamic attribute would be lost by ``.bind()``/``.named()``, whereas + # ``dataclasses.replace`` carries every field. That is what lets the marker + # survive fluent composition and imported/cached refs. Publication is NOT + # generation-affecting -- it never changes the generated component bytes -- + # so it is deliberately absent from the dedup identity. + _task_publish_name: str | None = None + _task_publish_version: str | None = None # ``@registered`` metadata. ``None`` for ``ref()``/``@task`` refs; # populated by the ``@registered`` decorator. Drives the compile-time # rewrite of the ``registered://pending`` sentinel URL to a diff --git a/pyproject.toml b/pyproject.toml index 30c1dd5..ddfb5bf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "tangle-cli" -version = "0.1.14" +version = "0.1.15" description = "CLI for Tangle, the open-source ML pipeline orchestration platform" readme = "README.md" authors = [ diff --git a/tests/test_component_publisher.py b/tests/test_component_publisher.py index a1f3838..93466b4 100644 --- a/tests/test_component_publisher.py +++ b/tests/test_component_publisher.py @@ -878,3 +878,53 @@ def test_publish_components_returns_nonzero_for_errors(tmp_path: Path) -> None: ProcessingOutcome.SUCCESS, ProcessingOutcome.ERROR, ] + + +def test_the_owner_argument_a_version_check_passes_is_matched_exactly() -> None: + """Pin the semantics of the owner argument ``perform_version_check`` sends. + + This is a CONTRACT test, not an end-to-end publisher run: it calls + ``find_existing_components`` directly, which is the call the version check + makes. The companion test above proves the publisher passes ``published_by``; + this proves what that argument then means. Together they stop the + publisher's owner scope loosening silently. + + It matters because the server query is a substring match: without the + client's exact filter, a component owned by ``alice2`` would be read as part + of ``alice``'s published state -- constraining alice's next version, or + being considered for deprecation. + """ + from tangle_cli.client import TangleApiClient + from tangle_cli.models import ComponentInfo + + rows = [ + ComponentInfo(name="orders-loader", digest="sha256:theirs", version="9.9", published_by="alice2"), + ComponentInfo(name="orders-loader", digest="sha256:mine", version="1.0", published_by="alice"), + ] + + class _Session: + def request(self, *a: Any, **kw: Any) -> Any: # pragma: no cover - never called + raise AssertionError("no HTTP in this test") + + class _Client(TangleApiClient): + def list_published_component_infos( # type: ignore[override] + self, + include_deprecated: bool = False, + name_substring: str | None = None, + published_by_substring: str | None = None, + digest: str | None = None, + *, + fetch_specs: bool = False, + ) -> list[Any]: + out = rows + if published_by_substring: + out = [i for i in out if published_by_substring in (i.published_by or "")] + if name_substring: + out = [i for i in out if name_substring.lower() in i.name.lower()] + return out + + client = _Client("https://api.test", session=_Session()) + + found = client.find_existing_components(["orders-loader"], published_by="alice") + + assert [i.digest for i in found] == ["sha256:mine"] diff --git a/tests/test_packaging.py b/tests/test_packaging.py index 21f20bd..68d5980 100644 --- a/tests/test_packaging.py +++ b/tests/test_packaging.py @@ -183,7 +183,7 @@ def test_tangle_cli_wheel_supports_expert_no_deps_import_path_without_tangle_api requires_dist = [line for line in metadata.splitlines() if line.startswith("Requires-Dist: ")] assert not any(name.startswith("tangle_api/") for name in names) assert "tangle_cli/openapi/openapi.json" not in names - assert "Version: 0.1.14" in metadata + assert "Version: 0.1.15" in metadata assert "Requires-Dist: tangle-api==0.1.1" in requires_dist assert not any("extra == 'native'" in line for line in requires_dist) assert "Provides-Extra: native" in metadata diff --git a/tests/test_python_pipeline_dsl.py b/tests/test_python_pipeline_dsl.py index 3a876e9..f4211db 100644 --- a/tests/test_python_pipeline_dsl.py +++ b/tests/test_python_pipeline_dsl.py @@ -53,6 +53,7 @@ def test_all_names_are_exported(self): assert set(pp.__all__) == { "pipeline", "task", + "Publish", "registered", "ref", "raw", diff --git a/tests/test_python_pipeline_publish.py b/tests/test_python_pipeline_publish.py new file mode 100644 index 0000000..b4ccbd3 --- /dev/null +++ b/tests/test_python_pipeline_publish.py @@ -0,0 +1,565 @@ +"""``@Publish`` marker: decoration contract, ref propagation, sidecar emission. + +``@Publish`` is a pure marker. Compiling a marked pipeline records the +declaration in the resolver sidecar beside ``local_from_python`` and publishes +NOTHING; a downstream tool reads the marker and owns the policy. These tests +cover the authoring refusals, the propagation properties that make the marker +trustworthy (fluent composition, imported/cached refs, repeated compiles), and +the emission rules (dedup agreement, subgraph children, hydration unaffected). +""" +from __future__ import annotations + +import importlib +import sys +from pathlib import Path +from typing import Any + +import pytest +import yaml + +from tangle_cli.pipeline_compiler import compile_pipeline +from tangle_cli.python_pipeline import Publish, ref, registered, task +from tangle_cli.python_pipeline.errors import CompileError + +FIXTURES = Path(__file__).parent / "fixtures" / "python_pipeline" + + +def _assert_marked(entry: dict, name: str, version: str) -> None: + """Assert the emitted shape for a declared component. + + The declaration is emitted as the resolver's OWN + ``name``/``version``/``publisher`` fields, so ordinary hydration looks the + published component up; ``local_from_python`` remains as the candidate used + when none is found. ``publish`` is a separate literal ``True`` so + publication is never inferred from those fields alone. + + ``publisher`` is the SYMBOLIC ``me``, never a resolved account id: the + compiler is offline and must not contact the API to learn who is running it. + """ + assert entry["name"] == name + assert entry["version"] == version + assert entry["publisher"] == "me" + assert entry["publish"] is True + assert "local_from_python" in entry + + +def _task_ref(**kwargs): + """A minimal ``@task`` ref to apply ``@Publish`` to.""" + + def load_orders(source: str = "orders") -> str: + """Load orders. + + Metadata: + Name: Load Orders + """ + return source + + return task(image="registry.example/loader:1", **kwargs)(load_orders) + + +# --------------------------------------------------------------------------- +# Authoring contract: both values required, and the target must be a @task. + + +@pytest.mark.parametrize( + "kwargs", + [ + pytest.param({"component_name": "", "version": "1.0"}, id="empty-name"), + pytest.param({"component_name": " ", "version": "1.0"}, id="blank-name"), + pytest.param({"component_name": "orders", "version": ""}, id="empty-version"), + pytest.param({"component_name": None, "version": "1.0"}, id="none-name"), + pytest.param({"component_name": "orders", "version": 1.0}, id="non-string-version"), + pytest.param({"component_name": "ord\ners", "version": "1.0"}, id="newline-in-name"), + pytest.param({"component_name": "orders", "version": "1.0\x7f"}, id="delete-in-version"), + ], +) +def test_publish_refuses_unusable_values(kwargs): + """Both values are required, non-blank, and control-character free: they + are echoed into diagnostics and written into a YAML document.""" + with pytest.raises(CompileError, match="@Publish"): + Publish(**kwargs) + + +def test_publish_requires_both_values(): + with pytest.raises(TypeError): + Publish(component_name="orders") # type: ignore[call-arg] + + +def test_publish_strips_surrounding_whitespace(): + declared = Publish(component_name=" orders-loader ", version=" 1.0 ") + assert (declared.component_name, declared.version) == ("orders-loader", "1.0") + + +def test_publish_refuses_a_plain_function(): + """Applied BELOW @task (wrong order) the decorator sees a function.""" + with pytest.raises(CompileError, match="can only publish a @task component"): + + @Publish(component_name="orders-loader", version="1.0") + def load_orders() -> str: + return "orders" + + +def test_publish_refuses_a_plain_ref(): + """``ref()`` names an existing component; there is no local source to + generate and publish from.""" + with pytest.raises(CompileError, match="can only publish a @task component"): + Publish(component_name="orders-loader", version="1.0")( + ref("file://component.yaml") + ) + + +def test_publish_refuses_a_registered_ref(): + """``@registered`` points at an ALREADY published component.""" + + @registered(fragment="run-query", gen_config="gs://bucket/gen_config.yaml") + def run_query(sql: str = "SELECT 1") -> str: + """Run a query.""" + return sql + + with pytest.raises(CompileError, match="can only publish a @task component"): + Publish(component_name="run-query", version="1.0")(run_query) + + +def test_publish_refuses_a_second_declaration(): + published = Publish(component_name="orders-loader", version="1.0")(_task_ref()) + with pytest.raises(CompileError, match="already declared"): + Publish(component_name="other", version="2.0")(published) + + +# --------------------------------------------------------------------------- +# Propagation: the marker must survive everything that produces a NEW ref. + + +def test_the_marker_is_a_real_field_carried_by_fluent_composition(): + """``_replace`` re-copies only a dunder allowlist, so a dynamically + stamped attribute would be dropped here. Real dataclass fields are copied + by ``dataclasses.replace`` itself.""" + published = Publish(component_name="orders-loader", version="1.0")(_task_ref()) + + derived = published.named("first").bind(source="a") + + assert derived._task_publish_name == "orders-loader" + assert derived._task_publish_version == "1.0" + + +def test_the_marker_does_not_leak_onto_an_undeclared_task(): + plain = _task_ref() + Publish(component_name="orders-loader", version="1.0")(plain) + + assert plain._task_publish_name is None + assert plain._task_publish_version is None + + +# --------------------------------------------------------------------------- +# Sidecar emission. + + +def _keys_named(data: Any, key: str) -> bool: + """Whether ``key`` appears as a mapping key anywhere in ``data``.""" + if isinstance(data, dict): + return key in data or any(_keys_named(value, key) for value in data.values()) + if isinstance(data, list): + return any(_keys_named(item, key) for item in data) + return False + + +def _sidecar(pipeline_path: Path, out: Path): + result = compile_pipeline(pipeline_path, out) + return yaml.safe_load(result.components_path.read_text()) + + +def _project_with(tmp_path: Path, body: str) -> Path: + """Write a one-module pipeline project and return its pipeline path.""" + src = tmp_path / "project" / "src" + src.mkdir(parents=True) + pipeline_path = src / "pipeline.py" + pipeline_path.write_text(body, encoding="utf-8") + return pipeline_path + + +_PUBLISHING_PIPELINE = ( + "from tangle_cli.python_pipeline import Out, Publish, pipeline, task\n\n" + "@Publish(component_name='orders-loader', version='1.0')\n" + "@task(image='registry.example/loader:1')\n" + "def load_orders(source: str = 'orders') -> str:\n" + ' """Load orders.\n\n' + " Metadata:\n" + " Name: Load Orders\n" + ' """\n' + " return source\n\n" + "@pipeline('Publishing Pipeline')\n" + "def publishing_pipeline() -> Out[str]:\n" + " loaded = load_orders(source='orders')\n" + " return loaded\n" +) + + +def test_a_declared_task_emits_a_publish_marker_beside_local_from_python(tmp_path): + pipeline_path = _project_with(tmp_path, _PUBLISHING_PIPELINE) + + sidecar = _sidecar(pipeline_path, pipeline_path.parent / "compiled.yaml") + + entry = sidecar["load-orders"] + _assert_marked(entry, "orders-loader", "1.0") + # The local resolver is untouched and stays the fallback candidate. + assert entry["local_from_python"]["function"] == "load_orders" + + +def test_an_undeclared_task_emits_no_publish_key(tmp_path): + """Absence is expressed by omission, never by a null or empty mapping.""" + pipeline_path = _project_with( + tmp_path, _PUBLISHING_PIPELINE.replace( + "@Publish(component_name='orders-loader', version='1.0')\n", "" + ) + ) + + sidecar = _sidecar(pipeline_path, pipeline_path.parent / "compiled.yaml") + + assert "publish" not in sidecar["load-orders"] + + +def test_compiling_a_declared_pipeline_publishes_nothing(tmp_path, monkeypatch): + """Compile records intent and performs no registry call. The marker is + inert: publication is a separate, downstream decision.""" + import tangle_cli.component_publisher as component_publisher + + def _fail(*args, **kwargs): # pragma: no cover - must never run + raise AssertionError("compile must not publish") + + monkeypatch.setattr( + component_publisher.ComponentPublisher, "publish_components", _fail, raising=False + ) + pipeline_path = _project_with(tmp_path, _PUBLISHING_PIPELINE) + + sidecar = _sidecar(pipeline_path, pipeline_path.parent / "compiled.yaml") + + assert sidecar["load-orders"]["name"] == "orders-loader" + + +def test_two_call_sites_of_one_declared_task_share_one_entry_and_one_marker(tmp_path): + """Dedup is unchanged by publication: one generated component, one entry, + one marker.""" + pipeline_path = _project_with( + tmp_path, + "from tangle_cli.python_pipeline import Out, Publish, pipeline, task\n\n" + "@Publish(component_name='orders-loader', version='1.0')\n" + "@task(image='registry.example/loader:1')\n" + "def load_orders(source: str = 'orders') -> str:\n" + ' """Load orders.\n\n' + " Metadata:\n" + " Name: Load Orders\n" + ' """\n' + " return source\n\n" + "@pipeline('Twice Pipeline')\n" + "def twice_pipeline() -> Out[str]:\n" + " load_orders.named('first')(source='a')\n" + " return load_orders.named('second')(source='b')\n", + ) + + sidecar = _sidecar(pipeline_path, pipeline_path.parent / "compiled.yaml") + + assert list(sidecar) == ["load-orders"] + _assert_marked(sidecar["load-orders"], "orders-loader", "1.0") + + +def test_two_declarations_folded_into_one_component_are_refused(tmp_path): + """Publication is deliberately NOT part of the dedup identity, so two + refs that generate the same component collapse into one entry that can + carry only one marker. Reachable by applying @Publish functionally to one + of two otherwise identical refs — so this refusal is not merely defensive. + """ + pipeline_path = _project_with( + tmp_path, + "from tangle_cli.python_pipeline import Out, Publish, pipeline, task\n\n" + "def load_orders(source: str = 'orders') -> str:\n" + ' """Load orders.\n\n' + " Metadata:\n" + " Name: Load Orders\n" + ' """\n' + " return source\n\n" + "base = task(image='registry.example/loader:1')(load_orders)\n" + "declared = Publish(component_name='orders-loader', version='1.0')(base)\n\n" + "@pipeline('Disagreeing Pipeline')\n" + "def disagreeing_pipeline() -> Out[str]:\n" + " base.named('plain')(source='a')\n" + " return declared.named('declared')(source='b')\n", + ) + + with pytest.raises(CompileError, match="@Publish disagreement"): + compile_pipeline(pipeline_path, pipeline_path.parent / "compiled.yaml") + + +@pytest.mark.parametrize( + ("first", "second"), + [ + pytest.param("", ", mode='inline'", id="lean-then-explicit"), + pytest.param(", mode='inline'", "", id="explicit-then-lean"), + ], +) +def test_the_marker_survives_the_canonical_spelling_choice(tmp_path, first, second): + """``@task()`` and ``@task(mode='inline')`` are one component spelled two + ways, and the compiler keeps the leanest spelling regardless of call + order. Whichever emitted form wins must still carry the declaration -- + in the order where the SECOND call site supplies the canonical form, the + representative is replaced after the marker was first recorded. + """ + src = tmp_path / "project" / "src" + src.mkdir(parents=True) + pipeline_path = src / "pipeline.py" + pipeline_path.write_text( + "from tangle_cli.python_pipeline import Out, Publish, pipeline, task\n\n" + "def load_orders(source: str = 'orders') -> str:\n" + ' """Load orders.\n\n Metadata:\n Name: Load Orders\n """\n' + " return source\n\n" + "publish = Publish(component_name='orders-loader', version='1.0')\n" + f"first = publish(task(image='registry.example/loader:1'{first})(load_orders))\n" + f"second = publish(task(image='registry.example/loader:1'{second})(load_orders))\n\n" + "@pipeline('Spelling Pipeline')\n" + "def spelling_pipeline() -> Out[str]:\n" + " a = first.named('a')(source='a')\n" + " b = second.named('b')(source='b')\n" + " return b\n", + encoding="utf-8", + ) + + sidecar = _sidecar(pipeline_path, src / "compiled.yaml") + + assert list(sidecar) == ["load-orders"] + _assert_marked(sidecar["load-orders"], "orders-loader", "1.0") + + +def test_two_different_components_each_carry_their_own_marker(tmp_path): + """Multiple distinct publications per compile are representable upstream: + the one-per-ship rule is a downstream policy, not a compiler rule.""" + pipeline_path = _project_with( + tmp_path, + "from tangle_cli.python_pipeline import Out, Publish, pipeline, task\n\n" + "@Publish(component_name='orders-loader', version='1.0')\n" + "@task(image='registry.example/loader:1')\n" + "def load_orders(source: str = 'orders') -> str:\n" + ' """Load orders.\n\n Metadata:\n Name: Load Orders\n """\n' + " return source\n\n" + "@Publish(component_name='orders-shipper', version='2.0')\n" + "@task(image='registry.example/shipper:1')\n" + "def ship_orders(source: str = 'orders') -> str:\n" + ' """Ship orders.\n\n Metadata:\n Name: Ship Orders\n """\n' + " return source\n\n" + "@pipeline('Two Publications Pipeline')\n" + "def two_publications_pipeline() -> Out[str]:\n" + " loaded = load_orders(source='orders')\n" + " shipped = ship_orders(source=loaded)\n" + " return shipped\n", + ) + + sidecar = _sidecar(pipeline_path, pipeline_path.parent / "compiled.yaml") + + assert sidecar["load-orders"]["name"] == "orders-loader" + assert sidecar["ship-orders"]["name"] == "orders-shipper" + assert sidecar["ship-orders"]["version"] == "2.0" + + +# --------------------------------------------------------------------------- +# The case that defeated a capture-based design: the declaration lives in a +# module that is ALREADY imported, so its decorators do not run again. + + +def _shared_task_project(tmp_path: Path, module: str) -> Path: + """A project whose pipeline imports its declared @task from ``module``.""" + src = tmp_path / "project" / "src" + src.mkdir(parents=True) + (src / f"{module}.py").write_text( + "from tangle_cli.python_pipeline import Publish, task\n\n" + "@Publish(component_name='orders-loader', version='1.0')\n" + "@task(image='registry.example/loader:1')\n" + "def load_orders(source: str = 'orders') -> str:\n" + ' """Load orders.\n\n' + " Metadata:\n" + " Name: Load Orders\n" + ' """\n' + " return source\n", + encoding="utf-8", + ) + pipeline_path = src / "pipeline.py" + pipeline_path.write_text( + "from tangle_cli.python_pipeline import Out, pipeline\n" + f"from {module} import load_orders\n\n" + "@pipeline('Imported Pipeline')\n" + "def imported_pipeline() -> Out[str]:\n" + " loaded = load_orders(source='orders')\n" + " return loaded\n", + encoding="utf-8", + ) + return pipeline_path + + +def test_a_declaration_in_an_imported_module_is_emitted(tmp_path): + """The marker travels on the ref, so it does not matter which module the + declaration was written in.""" + module = f"pubshared_{abs(hash(str(tmp_path))):x}" + pipeline_path = _shared_task_project(tmp_path, module) + + sidecar = _sidecar(pipeline_path, pipeline_path.parent / "compiled.yaml") + + assert sidecar["load-orders"]["name"] == "orders-loader" + + +def test_a_preloaded_declaring_module_changes_nothing(tmp_path): + """The decisive property, and the exact case that defeats a capture-based + design: when the declaring module is ALREADY imported its ``@Publish`` + does not execute again, so a mechanism that observed decoration would see + nothing. The marker rides on the ref, so a cold compile and a compile with + the module preloaded produce the same sidecar. + """ + module = f"pubcached_{abs(hash(str(tmp_path))):x}" + pipeline_path = _shared_task_project(tmp_path, module) + source_dir = pipeline_path.parent + + cold = _sidecar(pipeline_path, source_dir / "cold.yaml") + + sys.path.insert(0, str(source_dir)) + try: + preloaded = importlib.import_module(module) + # Verify the preload resolved to the module just written, rather than + # some same-named module left behind by another test. + assert Path(preloaded.__file__ or "") == source_dir / f"{module}.py" + assert preloaded.load_orders._task_publish_name == "orders-loader" + + warm = _sidecar(pipeline_path, source_dir / "warm.yaml") + finally: + sys.path.remove(str(source_dir)) + sys.modules.pop(module, None) + + _assert_marked(cold["load-orders"], "orders-loader", "1.0") + assert warm == cold + + +# --------------------------------------------------------------------------- +# Subgraph children and hydration. + + +def test_a_declaration_inside_a_subpipeline_child_is_emitted(tmp_path): + """A child graph gets its own ``.components.yaml``; the marker must + ride into the child's sidecar, not the root's.""" + src = tmp_path / "project" / "src" + src.mkdir(parents=True) + pipeline_path = src / "pipeline.py" + pipeline_path.write_text( + "from tangle_cli.python_pipeline import Out, Publish, pipeline, subpipeline, task\n\n" + "@Publish(component_name='orders-loader', version='1.0')\n" + "@task(image='registry.example/loader:1')\n" + "def load_orders(source: str = 'orders') -> str:\n" + ' """Load orders.\n\n Metadata:\n Name: Load Orders\n """\n' + " return source\n\n" + "@pipeline('Child Pipeline')\n" + "def child_pipeline() -> Out[str]:\n" + " loaded = load_orders(source='orders')\n" + " return loaded\n\n" + "@pipeline('Parent Pipeline')\n" + "def parent_pipeline() -> Out[str]:\n" + " child = subpipeline(child_pipeline)()\n" + " return child\n", + encoding="utf-8", + ) + out = src / "compiled.yaml" + + result = compile_pipeline(pipeline_path, out, pipeline_name="Parent Pipeline") + + assert len(result.subgraph_paths) == 1 + child_graph = result.subgraph_paths[0] + child_sidecar_path = child_graph.with_name(f"{child_graph.stem}.components.yaml") + child_sidecar = yaml.safe_load(child_sidecar_path.read_text()) + _assert_marked(child_sidecar["load-orders"], "orders-loader", "1.0") + + +def test_the_marker_is_stripped_from_the_baked_operation_program(): + """``@Publish`` is authoring-only. The baked program drops the authoring + import, so a surviving ``@Publish`` line would raise ``NameError`` at + container startup for every marked task. + """ + from tangle_cli.component_from_func import _strip_authoring_constructs + + baked = _strip_authoring_constructs( + "from tangle_cli.python_pipeline import Publish, task\n\n" + '@Publish(component_name="orders-loader", version="1.0")\n' + '@task(image="registry.example/loader:1")\n' + 'def load_orders(source: str = "orders") -> str:\n' + " return source\n" + ) + + assert "Publish" not in baked + assert "tangle_cli.python_pipeline" not in baked + # The decisive check: it must RUN with no authoring names in scope. + namespace: dict[str, Any] = {} + exec(compile(baked, "", "exec"), namespace) + assert namespace["load_orders"]("orders") == "orders" + + +def test_the_generated_component_command_carries_no_authoring_construct(tmp_path): + """End-to-end counterpart: the command actually baked into the resolved + component must be free of the decorator and its import.""" + from unittest.mock import MagicMock + + from tangle_cli.pipeline_hydrator import PipelineHydrator + + pipeline_path = _project_with(tmp_path, _PUBLISHING_PIPELINE) + out = pipeline_path.parent / "compiled.yaml" + compile_pipeline(pipeline_path, out) + + hydrated = PipelineHydrator(client=MagicMock()).hydrate_file(out) + + tasks = hydrated.data["implementation"]["graph"]["tasks"] + spec = next(iter(tasks.values()))["componentRef"]["spec"] + program = "\n".join( + part for part in spec["implementation"]["container"]["command"] if isinstance(part, str) + ) + assert "@Publish" not in program + assert "from tangle_cli.python_pipeline import" not in program + assert "def load_orders" in program + + +def test_a_marked_component_still_hydrates_from_local_source(tmp_path): + """Hydration reads ``local_from_python`` and ignores the marker, so a + declared component resolves exactly as an undeclared one does.""" + from unittest.mock import MagicMock + + from tangle_cli.pipeline_hydrator import PipelineHydrator + + declared_path = _project_with(tmp_path, _PUBLISHING_PIPELINE) + declared_out = declared_path.parent / "compiled.yaml" + compile_pipeline(declared_path, declared_out) + + hydrated = PipelineHydrator(client=MagicMock()).hydrate_file(declared_out) + + tasks = hydrated.data["implementation"]["graph"]["tasks"] + task_spec = next(iter(tasks.values())) + # The component resolved from local source, exactly as an unmarked one. + assert "spec" in task_spec["componentRef"] + assert task_spec["componentRef"]["spec"]["name"] == "Load Orders" + # The marker did not travel into the hydrated graph. Checked STRUCTURALLY: + # the rendered YAML embeds the pipeline source, whose own identifiers + # contain the substring "publish". + assert not _keys_named(hydrated.data, "publish") + + +def test_compilation_emits_a_symbolic_publisher_without_contacting_the_api(tmp_path): + """The emitted owner must be symbolic, not a resolved account. + + Resolving an id at compile time would make compilation require + authentication and network access, and would bake one author's account into + an artifact that another author may legitimately rebuild. + """ + pipeline_path = _project_with(tmp_path, _PUBLISHING_PIPELINE) + + sidecar = _sidecar(pipeline_path, pipeline_path.parent / "compiled.yaml") + + entry = sidecar["load-orders"] + assert entry["publisher"] == "me" + + +def test_the_symbolic_publisher_is_matched_exactly_and_case_sensitively() -> None: + """A literal account id that merely looks like the sentinel stays literal.""" + from tangle_cli.authenticated_identity import is_symbolic_me + + assert is_symbolic_me("me") + for other in ("ME", "Me", " me", "me ", "me@example.com", "", None, True): + assert not is_symbolic_me(other) diff --git a/tests/test_static_client.py b/tests/test_static_client.py index 9668b13..418930c 100644 --- a/tests/test_static_client.py +++ b/tests/test_static_client.py @@ -830,3 +830,97 @@ def _refresh_auth(self) -> None: assert client.refreshes == 2 assert len(session.calls) == 2 assert session.calls[-1]["headers"]["Authorization"] == "Bearer refreshed-2" + + +# --- exact vs substring owner scoping --------------------------------------- +# +# ``published_by`` and ``published_by_substring`` are distinct public +# arguments. The server query is a SUBSTRING match, so for an exact request the +# server result is only a prefilter and exactness must be enforced locally. +# An owner scope is used as an identity control by callers (owner-scoped +# publication/deprecation, and resolution of publish-marked components), so +# these are contract tests, not cosmetics. + + +def _owner_client(rows: list[dict[str, Any]], *, server_filters: bool = True) -> TangleApiClient: + """A client whose listing layer behaves like the server: substring owner. + + ``server_filters=False`` models a server that returns a row the owner query + should have excluded -- absent publication metadata, or a backend that does + not apply the filter. The client must enforce the identity control itself + rather than trusting the response. + """ + from tangle_cli.models import ComponentInfo + + infos = [ComponentInfo(**row) for row in rows] + + class _Client(TangleApiClient): + def list_published_component_infos( # type: ignore[override] + self, + include_deprecated: bool = False, + name_substring: str | None = None, + published_by_substring: str | None = None, + digest: str | None = None, + *, + fetch_specs: bool = False, + ) -> list[Any]: + out = infos + if published_by_substring and server_filters: + out = [i for i in out if published_by_substring in (i.published_by or "")] + if name_substring: + out = [i for i in out if name_substring.lower() in i.name.lower()] + return out + + return _Client("https://api.test", session=FakeSession([])) + + +def test_exact_published_by_rejects_a_superset_owner_id() -> None: + """``alice`` must not match ``alice2``: the server substring query returns + it, so only the local exact filter can refuse it.""" + client = _owner_client( + [{"name": "orders-loader", "digest": "sha256:lookalike", "published_by": "alice2"}] + ) + + found = client.find_existing_components(["orders-loader"], published_by="alice") + + assert found == [] + + +def test_exact_published_by_accepts_only_the_exact_owner() -> None: + client = _owner_client( + [ + {"name": "orders-loader", "digest": "sha256:theirs", "published_by": "alice2"}, + {"name": "orders-loader", "digest": "sha256:mine", "published_by": "alice"}, + ] + ) + + found = client.find_existing_components(["orders-loader"], published_by="alice") + + assert [i.digest for i in found] == ["sha256:mine"] + + +def test_exact_published_by_rejects_a_row_with_no_owner() -> None: + """A missing owner field is not evidence of ownership, and the client must + not assume the server applied the filter it was given.""" + client = _owner_client( + [{"name": "orders-loader", "digest": "sha256:ownerless", "published_by": None}], + server_filters=False, + ) + + found = client.find_existing_components(["orders-loader"], published_by="alice") + + assert found == [] + + +def test_substring_published_by_keeps_partial_matching() -> None: + """The exactness applies to exact requests only; substring callers rely on + partial matching.""" + client = _owner_client( + [{"name": "orders-loader", "digest": "sha256:bot", "published_by": "team-alpha-bot"}] + ) + + found = client.find_existing_components( + ["orders-loader"], published_by_substring="team-alpha" + ) + + assert [i.digest for i in found] == ["sha256:bot"] diff --git a/tests/test_symbolic_publisher_hydration.py b/tests/test_symbolic_publisher_hydration.py new file mode 100644 index 0000000..fca903a --- /dev/null +++ b/tests/test_symbolic_publisher_hydration.py @@ -0,0 +1,189 @@ +"""Owner scoping of entries whose publisher is the symbolic ``me``. + +The compiler emits ``publisher: me`` because compilation is offline. Hydration +resolves it to the authenticated account, which is what keeps the lookup from +being a global name search in which anyone's same-named component is a +candidate. These tests pin that identity control at the layer every consumer +inherits, not just in one downstream CLI. +""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from tangle_cli.authenticated_identity import IdentityUnavailableError +from tangle_cli.client import TangleApiClient +from tangle_cli.models import ComponentInfo +from tangle_cli.pipeline_hydrator import PipelineHydrator + +_ME = "user-me" +_GENERATION = {"file": "daily_pulse.py", "function": "load_orders", "image": "img:1"} + + +def _rows(*specs: tuple[str, str | None]) -> list[ComponentInfo]: + return [ + ComponentInfo(name="orders-loader", digest=digest, version="1.0", published_by=owner) + for digest, owner in specs + ] + + +class _Client: + """Client whose listing layer behaves like the server: substring owner.""" + + base_url = "https://tangle.example" + logger = None + + def __init__(self, rows: list[ComponentInfo], *, identity: Any = _ME) -> None: + self._rows = rows + self._identity = identity + self.owner_filters: list[str | None] = [] + + def users_me(self) -> Any: + if isinstance(self._identity, Exception): + raise self._identity + return {"id": self._identity} if self._identity is not None else None + + def list_published_component_infos( + self, + include_deprecated: bool = False, + name_substring: str | None = None, + published_by_substring: str | None = None, + digest: str | None = None, + *, + fetch_specs: bool = False, + ) -> list[ComponentInfo]: + out = self._rows + if published_by_substring: + out = [i for i in out if published_by_substring in (i.published_by or "")] + if name_substring: + out = [i for i in out if name_substring.lower() in i.name.lower()] + return out + + def find_existing_components(self, *args: Any, **kwargs: Any) -> list[ComponentInfo]: + self.owner_filters.append(kwargs.get("published_by")) + return TangleApiClient.find_existing_components(self, *args, **kwargs) + + def get_component_spec(self, digest: str) -> dict[str, Any]: + return {"name": "orders-loader", "digest": digest} + + +def _entry(**over: Any) -> dict[str, Any]: + base = { + "name": "orders-loader", + "version": "1.0", + "publisher": "me", + "local_from_python": dict(_GENERATION), + "publish": True, + } + base.update(over) + return base + + +def test_symbolic_me_resolves_to_the_authenticated_account() -> None: + client = _Client(_rows(("sha256:mine", _ME))) + + resolved = PipelineHydrator(client=client)._resolve_by_name_with_filters(_entry()) + + assert resolved is not None + assert resolved[0] == "sha256:mine" + assert client.owner_filters == [_ME] # the symbol never reaches the API + + +def test_a_foreign_component_is_not_selected_even_when_returned_first() -> None: + client = _Client(_rows(("sha256:attacker", "attacker"), ("sha256:mine", _ME))) + + resolved = PipelineHydrator(client=client)._resolve_by_name_with_filters(_entry()) + + assert resolved is not None + assert resolved[0] == "sha256:mine" + + +def test_a_foreign_only_collision_yields_no_candidate() -> None: + """Someone else holding the name must not supply the code; the caller then + falls back to the local candidate.""" + client = _Client(_rows(("sha256:attacker", "attacker"))) + + assert PipelineHydrator(client=client)._resolve_by_name_with_filters(_entry()) is None + + +@pytest.mark.parametrize( + "identity", + [ + pytest.param(RuntimeError("auth expired"), id="lookup-raises"), + pytest.param(None, id="no-user"), + pytest.param("", id="empty-id"), + ], +) +def test_an_unresolvable_identity_fails_closed(identity: Any) -> None: + """Never widen to a global search: that is the vulnerability.""" + client = _Client(_rows(("sha256:mine", _ME)), identity=identity) + + with pytest.raises(IdentityUnavailableError, match="unscoped"): + PipelineHydrator(client=client)._resolve_by_name_with_filters(_entry()) + assert client.owner_filters == [] # no request was issued at all + + +def test_a_literal_publisher_id_is_used_verbatim() -> None: + client = _Client(_rows(("sha256:theirs", "other-team"))) + + resolved = PipelineHydrator(client=client)._resolve_by_name_with_filters( + _entry(publisher="other-team") + ) + + assert resolved is not None + assert client.owner_filters == ["other-team"] + + +def test_an_entry_without_a_publisher_keeps_the_existing_global_behaviour() -> None: + """Hand-authored cross-publisher resolution is unchanged.""" + client = _Client(_rows(("sha256:someone-elses", "someone-else"))) + entry = {"name": "orders-loader", "version": "1.0"} + + resolved = PipelineHydrator(client=client)._resolve_by_name_with_filters(entry) + + assert resolved is not None + assert client.owner_filters == [None] + + +def test_a_publisher_that_merely_resembles_the_sentinel_stays_literal() -> None: + """``ME`` is not ``me``: only the exact symbol is self-referential, so a + real account id is never swapped for the current user.""" + client = _Client(_rows(("sha256:odd", "ME"))) + + resolved = PipelineHydrator(client=client)._resolve_by_name_with_filters( + _entry(publisher="ME") + ) + + assert resolved is not None + assert client.owner_filters == ["ME"] + + +def test_an_unmarked_entry_treats_me_as_a_literal_account_id() -> None: + """``me`` is only self-referential on a compiler-authored marked entry. + + Hand-authored configs predate the sentinel, where ``publisher: me`` meant a + registry account literally named ``me`` -- the API does not reserve the + string. Substituting there would silently retarget an existing config, or + fail it on an auth error, so the marker is required too. + """ + client = _Client(_rows(("sha256:literal-me", "me"))) + entry = {"name": "orders-loader", "version": "1.0", "publisher": "me"} + + resolved = PipelineHydrator(client=client)._resolve_by_name_with_filters(entry) + + assert resolved is not None + assert resolved[0] == "sha256:literal-me" + assert client.owner_filters == ["me"] # passed through, not substituted + + +def test_an_unmarked_entry_with_me_does_not_require_authentication() -> None: + """The compatibility guarantee has teeth: an unmarked config must keep + resolving even when no account can be determined.""" + client = _Client(_rows(("sha256:literal-me", "me")), identity=RuntimeError("no auth")) + entry = {"name": "orders-loader", "version": "1.0", "publisher": "me"} + + resolved = PipelineHydrator(client=client)._resolve_by_name_with_filters(entry) + + assert resolved is not None diff --git a/uv.lock b/uv.lock index 8c0b313..ac7fbb8 100644 --- a/uv.lock +++ b/uv.lock @@ -2083,7 +2083,7 @@ requires-dist = [{ name = "pydantic", specifier = ">=2.0" }] [[package]] name = "tangle-cli" -version = "0.1.14" +version = "0.1.15" source = { editable = "." } dependencies = [ { name = "cloud-pipelines" },