feat(pipeline-run): fall back to file stem for a nameless submit spec

packages/tangle-cli/src/tangle_cli/pipeline_run_manager.py

@@ -926,6 +926,30 @@ def prepare_pipeline_spec_for_submit(
             hydrate=hydrate,
         )
 
+    @staticmethod
+    def _resolve_fallback_pipeline_name(
+        pipeline_spec: dict[str, Any],
+        fallback_name: str | None,
+    ) -> dict[str, Any]:
+        """Fill a missing pipeline ``name`` from a caller-resolved fallback.
+
+        A file-based submit whose spec omits a usable ``name`` still validates
+        and runs under a fallback (historically the source-file stem), matching
+        the long-standing downstream behavior. A declared non-blank ``name``
+        always wins; a missing, non-string, or whitespace-only ``name`` is
+        treated as genuinely nameless and takes the fallback, so other
+        structural errors still surface at validation. A no-op when the spec is
+        not a mapping or no fallback name is available.
+        """
+        if not isinstance(pipeline_spec, dict):
+            return pipeline_spec
+        name = pipeline_spec.get("name")
+        if isinstance(name, str) and name.strip():
+            return pipeline_spec
+        if not (isinstance(fallback_name, str) and fallback_name):
+            return pipeline_spec
+        return {**pipeline_spec, "name": fallback_name}
+
     def prepare_submit_payload_from_spec(
         self,
         pipeline_spec: dict[str, Any],
@@ -954,6 +978,10 @@ def prepare_submit_payload_from_spec(
         )
         prepared_run_args = self.hooks.prepare_run_arguments(prepared_spec, run_args)
         prepared_spec = self.apply_run_name_template(prepared_spec, prepared_run_args)
+        prepared_spec = self._resolve_fallback_pipeline_name(
+            prepared_spec,
+            Path(str(pipeline_path)).stem if pipeline_path is not None else None,
+        )
         if not skip_validation:
             validation_errors = self.hooks.validate_pipeline_for_run(
                 prepared_spec,

packages/tangle-cli/src/tangle_cli/pipeline_runner.py

@@ -391,6 +391,14 @@ def prepare_pipeline_for_run(
                 if isinstance(spec_name, str) and spec_name:
                     pipeline_name = spec_name
 
+            # Resolve the fallback run name into the spec before validation.
+            # ``pipeline_name`` has already fallen back to ``initial_pipeline_name``
+            # (the path stem) whenever the spec omits a usable ``name``; mirror
+            # that into the spec so a file-based submit with no declared name
+            # still validates and runs under the fallback. A declared name has
+            # already won above, so only the genuinely-nameless case is filled.
+            pipeline_spec = self._resolve_fallback_pipeline_name(pipeline_spec, pipeline_name)
+
             validation_errors = hooks.validate_pipeline_for_run(
                 pipeline_spec,
                 pipeline_path=pipeline_path,

tests/test_pipeline_runs_cli.py

@@ -203,6 +203,73 @@ def test_pipeline_runs_submit_builds_create_payload(monkeypatch, tmp_path: Path,
     assert root_task["arguments"] == {"query": "default", "required": "value"}
 
 
+def test_pipeline_runs_submit_falls_back_to_file_stem_when_spec_has_no_name(
+    monkeypatch, tmp_path: Path, capsys
+):
+    # A file-based submit whose spec omits ``name`` still validates and runs,
+    # taking the source-file stem as the run name (the long-standing downstream
+    # fallback, now honored natively in OSS rather than rejected at submit).
+    pipeline_path = tmp_path / "my_pipeline.yaml"
+    pipeline_path.write_text(
+        yaml.safe_dump({"implementation": {"graph": {"tasks": {}}}}, sort_keys=False),
+        encoding="utf-8",
+    )
+    fake_client = FakeClient()
+    monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client)
+    app = cli.build_app()
+
+    run_app(app, ["sdk", "pipeline-runs", "submit", str(pipeline_path), "--no-hydrate"])
+
+    result = json.loads(capsys.readouterr().out)
+    assert result == {"id": "run-1", "root_execution_id": "exec-1"}
+    assert fake_client.created[0]["root_task"]["componentRef"]["spec"]["name"] == "my_pipeline"
+
+
+def test_pipeline_runs_submit_prefers_declared_name_over_file_stem(
+    monkeypatch, tmp_path: Path, capsys
+):
+    # A declared non-empty ``name`` always wins over the file-stem fallback,
+    # even when the file stem (here ``config``) differs from the declared name.
+    pipeline_path = tmp_path / "config.yaml"
+    pipeline_path.write_text(
+        yaml.safe_dump(
+            {"name": "Demo Pipeline", "implementation": {"graph": {"tasks": {}}}},
+            sort_keys=False,
+        ),
+        encoding="utf-8",
+    )
+    fake_client = FakeClient()
+    monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client)
+    app = cli.build_app()
+
+    run_app(app, ["sdk", "pipeline-runs", "submit", str(pipeline_path), "--no-hydrate"])
+
+    assert fake_client.created[0]["root_task"]["componentRef"]["spec"]["name"] == "Demo Pipeline"
+
+
+def test_pipeline_runs_submit_falls_back_to_file_stem_for_whitespace_only_name(
+    monkeypatch, tmp_path: Path, capsys
+):
+    # A whitespace-only ``name`` is not a usable declared name, so it is treated
+    # as nameless and takes the file-stem fallback rather than being submitted
+    # verbatim as a blank run name.
+    pipeline_path = tmp_path / "my_pipeline.yaml"
+    pipeline_path.write_text(
+        yaml.safe_dump(
+            {"name": "   ", "implementation": {"graph": {"tasks": {}}}},
+            sort_keys=False,
+        ),
+        encoding="utf-8",
+    )
+    fake_client = FakeClient()
+    monkeypatch.setattr(pipeline_runs_cli, "LazyTangleApiClient", lambda **kwargs: fake_client)
+    app = cli.build_app()
+
+    run_app(app, ["sdk", "pipeline-runs", "submit", str(pipeline_path), "--no-hydrate"])
+
+    assert fake_client.created[0]["root_task"]["componentRef"]["spec"]["name"] == "my_pipeline"
+
+
 def test_pipeline_runs_submit_rejects_authoring_validation_errors(monkeypatch, tmp_path: Path):
     pipeline_path = _write_invalid_authoring_pipeline(tmp_path / "pipeline.yaml")
     fake_client = FakeClient()