chore: Add seam for incremental opt-out of legacy CLI commands

packages/nemo_platform_plugin/src/nemo_platform_plugin/commands.py

@@ -328,6 +328,12 @@ def _register_job_subgroup(
     cli: NemoCLI | None = None,
 ) -> None:
     """Register a ``<job-name>`` sub-group with run / submit / explain verbs."""
+    if not job_cls.generate_legacy_verbs:
+        _add_submit_command(cli_app, job_cls, scheduler, cli=cli, command_name=job_cls.name, rich_help_panel="Jobs")
+        if cli is not None:
+            cli.update_job_cli(job_cls, cli_app)
+        return
+
     job_group = typer.Typer(
         name=job_cls.name,
         help=job_cls.description or f"Manage the {job_cls.name} job.",
@@ -441,7 +447,7 @@ def _do_run() -> Any:
             renderer.on_complete(ctx=rctx)
 
     help_text = "Run locally, in-process."
-    _run.__signature__ = _build_job_run_signature(leaves)  # type: ignore[attr-defined]
+    setattr(_run, "__signature__", _build_job_run_signature(leaves))
     group.command(name="run", help=help_text)(_run)
 
 
@@ -530,6 +536,8 @@ def _add_submit_command(
     scheduler: NemoJobScheduler,
     *,
     cli: NemoCLI | None = None,
+    command_name: str = "submit",
+    rich_help_panel: str | None = None,
 ) -> None:
     """Register the ``submit`` verb. Generates per-field flags + static submit flags.
 
@@ -618,8 +626,8 @@ def _do_submit() -> Any:
             renderer.on_complete(ctx=rctx)
 
     help_text = "Submit to a cluster."
-    _submit.__signature__ = _build_job_submit_signature(leaves)  # type: ignore[attr-defined]
-    group.command(name="submit", help=help_text)(_submit)
+    setattr(_submit, "__signature__", _build_job_submit_signature(leaves))
+    group.command(name=command_name, help=help_text, rich_help_panel=rich_help_panel)(_submit)
 
 
 def _build_job_submit_signature(leaves: list[SpecLeafField]) -> inspect.Signature:
@@ -866,6 +874,12 @@ def _register_function_subgroup(
     cli: NemoCLI | None = None,
 ) -> None:
     """Register a ``<fn-name>`` sub-group with run / submit verbs."""
+    if not fn_cls.generate_legacy_verbs:
+        _add_function_submit_command(cli_app, fn_cls, cli=cli, command_name=fn_cls.name, rich_help_panel="Functions")
+        if cli is not None:
+            cli.update_function_cli(fn_cls, cli_app)
+        return
+
     fn_group = typer.Typer(
         name=fn_cls.name,
         help=fn_cls.description or f"Manage the {fn_cls.name} function.",
@@ -960,7 +974,7 @@ def _run(typer_ctx: typer.Context, **kwargs: object) -> None:
 
     help_text = f"Run {fn_cls.name} locally, in-process."
     epilog = build_epilog(schema=fn_cls.spec_schema, leaves=leaves, kind="Function")
-    _run.__signature__ = _build_function_run_signature(leaves)  # type: ignore[attr-defined]
+    setattr(_run, "__signature__", _build_function_run_signature(leaves))
     group.command(name="run", help=help_text, epilog=epilog)(_run)
 
 
@@ -1102,6 +1116,8 @@ def _add_function_submit_command(
     fn_cls: type[NemoFunction],
     *,
     cli: NemoCLI | None = None,
+    command_name: str = "submit",
+    rich_help_panel: str | None = None,
 ) -> None:
     """Register the ``submit`` verb. Generates per-field flags + static submit flags.
 
@@ -1113,12 +1129,12 @@ def _add_function_submit_command(
 
     def _submit(typer_ctx: typer.Context, **kwargs: object) -> None:
         original_kwargs = dict(kwargs)
-        spec_str: str = kwargs.pop("spec", "{}")  # type: ignore[assignment]
-        spec_file: Path | None = kwargs.pop("spec_file", None)  # type: ignore[assignment]
-        cluster: str | None = kwargs.pop("cluster", None)  # type: ignore[assignment]
-        base_url: str | None = kwargs.pop("base_url", None)  # type: ignore[assignment]
-        workspace: str = kwargs.pop("workspace", "default")  # type: ignore[assignment]
-        request_id: str | None = kwargs.pop("request_id", None)  # type: ignore[assignment]
+        spec_str: str = cast(str, kwargs.pop("spec", "{}"))
+        spec_file: Path | None = cast("Path | None", kwargs.pop("spec_file", None))
+        cluster: str | None = cast("str | None", kwargs.pop("cluster", None))
+        base_url: str | None = cast("str | None", kwargs.pop("base_url", None))
+        workspace: str = cast(str, kwargs.pop("workspace", "default"))
+        request_id: str | None = cast("str | None", kwargs.pop("request_id", None))
 
         base = _load_spec(spec_str, spec_file)
         overlay = build_overlay(leaves, kwargs, unset_sentinel=UNSET)
@@ -1164,8 +1180,8 @@ def _submit(typer_ctx: typer.Context, **kwargs: object) -> None:
 
     help_text = f"Submit {fn_cls.name} over HTTP."
     epilog = build_epilog(schema=fn_cls.spec_schema, leaves=leaves, kind="Function")
-    _submit.__signature__ = _build_function_submit_signature(leaves)  # type: ignore[attr-defined]
-    group.command(name="submit", help=help_text, epilog=epilog)(_submit)
+    setattr(_submit, "__signature__", _build_function_submit_signature(leaves))
+    group.command(name=command_name, help=help_text, epilog=epilog, rich_help_panel=rich_help_panel)(_submit)
 
 
 def _build_function_submit_signature(leaves: list[SpecLeafField]) -> inspect.Signature:

packages/nemo_platform_plugin/src/nemo_platform_plugin/function.py

@@ -137,6 +137,14 @@ class NemoFunction(_NamedPlugin, Generic[SpecT]):
         relocate itself outside its plugin's URL namespace. Leave
         ``None`` to use the default.
 
+    .. attribute:: generate_legacy_verbs
+        :type: bool
+
+        Temporary CLI compatibility knob. ``True`` keeps the generated
+        ``<function> run|submit`` command group. ``False`` registers
+        ``<function>`` itself as the remote submit command and omits the
+        local ``run`` and legacy ``submit`` verbs.
+
     Stream response start:
 
     .. attribute:: send_headers_before_first_frame
@@ -188,6 +196,8 @@ class NemoFunction(_NamedPlugin, Generic[SpecT]):
 
     endpoint: ClassVar[str | None] = None
 
+    generate_legacy_verbs: ClassVar[bool] = True
+
     send_headers_before_first_frame: ClassVar[bool] = False
 
     # ------------------------------------------------------------------ #

packages/nemo_platform_plugin/src/nemo_platform_plugin/job.py

@@ -153,6 +153,14 @@ class NemoJob(_NamedPlugin):
         when ``name = "train"``; ``"/metric-jobs"`` for a legacy flat
         collection path.
 
+    .. attribute:: generate_legacy_verbs
+        :type: bool
+
+        Temporary CLI compatibility knob. ``True`` keeps the generated
+        ``<job> run|submit|explain`` command group. ``False`` registers
+        ``<job>`` itself as the remote submit command and omits the legacy
+        ``run``, ``submit``, and ``explain`` verbs.
+
     Plugin-owned options:
 
     .. attribute:: backend_options_schemas
@@ -188,6 +196,12 @@ class NemoJob(_NamedPlugin):
 
     job_collection_path: ClassVar[str | None] = None
 
+    # ------------------------------------------------------------------ #
+    # Temporary CLI compatibility                                        #
+    # ------------------------------------------------------------------ #
+
+    generate_legacy_verbs: ClassVar[bool] = True
+
     # ------------------------------------------------------------------ #
     # Plugin-owned options (inert; see class docstring)                  #
     # ------------------------------------------------------------------ #

packages/nemo_platform_plugin/tests/test_cli_hooks.py

@@ -49,6 +49,10 @@ def run(self, config: dict) -> dict:
         return {"message": f"Bye, {config.get('name', 'world')}!"}
 
 
+class _FlatGreetJob(_GreetJob):
+    generate_legacy_verbs: ClassVar[bool] = False
+
+
 class _GreetSpec(BaseModel):
     name: str
 
@@ -75,6 +79,10 @@ async def run(self, spec: _GreetSpec) -> _GreetResponse:
         return _GreetResponse(message=f"Bye, {spec.name}!")
 
 
+class _FlatGreetFunction(_GreetFunction):
+    generate_legacy_verbs: ClassVar[bool] = False
+
+
 class _NoOpCLI(NemoCLI):
     """Minimal NemoCLI subclass with no hook overrides."""
 
@@ -212,6 +220,28 @@ def custom() -> None:
         bye_help = runner.invoke(app, ["bye", "--help"])
         assert "custom" not in bye_help.output
 
+    def test_direct_mode_hook_can_replace_flat_job_command(self) -> None:
+        class _CLI(_NoOpCLI):
+            def update_job_cli(self, job_cls, group) -> None:
+                if job_cls is not _FlatGreetJob:
+                    return
+                original = next(c for c in group.registered_commands if c.name == "greet").callback
+                assert original is not None
+
+                @group.command("greet")
+                def greet() -> None:
+                    typer.echo("flat-job-replaced")
+
+        app = _app_with_jobs(_FlatGreetJob, cli=_CLI())
+
+        help_result = runner.invoke(app, ["--help"])
+        assert help_result.exit_code == 0
+        assert "greet" in help_result.output
+
+        result = runner.invoke(app, ["greet"])
+        assert result.exit_code == 0
+        assert result.output.strip() == "flat-job-replaced"
+
 
 # ---------------------------------------------------------------------------
 # update_function_cli
@@ -309,3 +339,25 @@ def custom() -> None:
 
         bye_help = runner.invoke(app, ["bye", "--help"])
         assert "custom" not in bye_help.output
+
+    def test_direct_mode_hook_can_replace_flat_function_command(self) -> None:
+        class _CLI(_NoOpCLI):
+            def update_function_cli(self, fn_cls, group) -> None:
+                if fn_cls is not _FlatGreetFunction:
+                    return
+                original = next(c for c in group.registered_commands if c.name == "greet").callback
+                assert original is not None
+
+                @group.command("greet")
+                def greet() -> None:
+                    typer.echo("flat-function-replaced")
+
+        app = _app_with_functions(_FlatGreetFunction, cli=_CLI())
+
+        help_result = runner.invoke(app, ["--help"])
+        assert help_result.exit_code == 0
+        assert "greet" in help_result.output
+
+        result = runner.invoke(app, ["greet"])
+        assert result.exit_code == 0
+        assert result.output.strip() == "flat-function-replaced"

packages/nemo_platform_plugin/tests/test_commands.py

@@ -82,6 +82,10 @@ def run(self, config: dict) -> dict:
         return config
 
 
+class _FlatGreetJob(_GreetJob):
+    generate_legacy_verbs: ClassVar[bool] = False
+
+
 runner = CliRunner()
 
 
@@ -163,6 +167,20 @@ def test_job_verb_help_text_does_not_repeat_job_name(self) -> None:
         assert "Run run locally" not in result.output
         assert "schemas for run" not in result.output
 
+    def test_non_legacy_job_registers_flat_submit_command(self) -> None:
+        app = _app_with_jobs(_FlatGreetJob)
+        result = runner.invoke(app, ["--help"])
+        assert result.exit_code == 0
+        assert "greet" in result.output
+
+        help_result = runner.invoke(app, ["greet", "--help"])
+        assert help_result.exit_code == 0
+        output = _plain(help_result.output)
+        assert "COMMAND" not in output
+        assert "--base-url" in output
+        assert "--profile" in output
+        assert "explain" not in output
+
 
 # ---------------------------------------------------------------------------
 # Bare form — usage + non-zero exit
@@ -367,6 +385,40 @@ def get_sdk_context(self) -> SimpleNamespace:
         assert result.exit_code == 0, result.output
         assert captured["headers"] == {"Authorization": "Bearer test-token"}
 
+    def test_non_legacy_job_name_submits_remotely(self, monkeypatch) -> None:
+        captured: dict[str, object] = {}
+
+        def _capture(_self, job_cls, spec, *, base_url=None, workspace=None, **_kwargs) -> dict:
+            captured.update(
+                {
+                    "job_cls": job_cls,
+                    "spec": spec,
+                    "base_url": base_url,
+                    "workspace": workspace,
+                }
+            )
+            return {"id": "job-123"}
+
+        monkeypatch.setattr("nemo_platform_plugin.scheduler.NemoJobScheduler.submit_remote", _capture)
+
+        app = _app_with_jobs(_FlatGreetJob)
+        result = runner.invoke(
+            app,
+            ["greet", "--spec", '{"name": "Ada"}', "--base-url", "http://platform", "--workspace", "team-alpha"],
+        )
+
+        assert result.exit_code == 0, result.output
+        assert json.loads(result.output) == {"id": "job-123"}
+        assert captured == {
+            "job_cls": _FlatGreetJob,
+            "spec": {"name": "Ada"},
+            "base_url": "http://platform",
+            "workspace": "team-alpha",
+        }
+
+        assert runner.invoke(app, ["greet", "run"]).exit_code != 0
+        assert runner.invoke(app, ["greet", "submit"]).exit_code != 0
+
 
 # ---------------------------------------------------------------------------
 # explain verb — phase 1 MR 1.2c stubs
@@ -556,6 +608,10 @@ async def run(self, spec: _WorkspaceSpec, *, is_local: bool) -> dict:
         return {"is_local": is_local}
 
 
+class _FlatGreetFunction(_GreetFunction):
+    generate_legacy_verbs: ClassVar[bool] = False
+
+
 def _app_with_functions(*function_classes: type[NemoFunction]) -> typer.Typer:
     app = typer.Typer()
 
@@ -599,6 +655,15 @@ def test_bare_function_name_exits_non_zero(self) -> None:
         result = runner.invoke(app, ["greet"])
         assert result.exit_code != 0
 
+    def test_non_legacy_function_registers_flat_submit_command(self) -> None:
+        app = _app_with_functions(_FlatGreetFunction)
+        help_result = runner.invoke(app, ["greet", "--help"])
+        assert help_result.exit_code == 0
+        output = _plain(help_result.output)
+        assert "COMMAND" not in output
+        assert "--base-url" in output
+        assert "--request-id" in output
+
 
 # ---------------------------------------------------------------------------
 # Function `run` verb
@@ -878,6 +943,39 @@ def _fake_post(url: str, body: dict, *, headers: dict, timeout: float = 30.0, **
         assert result.exit_code == 0, result.output
         assert captured_url[0].startswith("http://from-env:1234/")
 
+    def test_non_legacy_function_name_submits_remotely(self, monkeypatch) -> None:
+        captured: dict[str, object] = {}
+
+        def _fake_post(url: str, body: dict, *, headers: dict, timeout: float = 30.0, **_kwargs) -> None:
+            captured.update({"url": url, "body": body, "headers": headers})
+            del timeout
+            typer.echo(json.dumps({"message": "ok"}))
+
+        monkeypatch.setattr("nemo_platform_plugin.commands._post_function_submit", _fake_post)
+
+        app = _app_with_functions(_FlatGreetFunction)
+        result = runner.invoke(
+            app,
+            [
+                "greet",
+                "--spec",
+                '{"name": "Ada"}',
+                "--base-url",
+                "http://platform",
+                "--workspace",
+                "team-alpha",
+            ],
+        )
+
+        assert result.exit_code == 0, result.output
+        assert json.loads(result.output) == {"message": "ok"}
+        assert str(captured["url"]).startswith("http://platform/apis/")
+        assert str(captured["url"]).endswith("/v2/workspaces/team-alpha/greet")
+        assert captured["body"] == {"name": "Ada"}
+
+        assert runner.invoke(app, ["greet", "run"]).exit_code != 0
+        assert runner.invoke(app, ["greet", "submit"]).exit_code != 0
+
 
 # ---------------------------------------------------------------------------
 # Function URL derivation