Apply InteractiveProcess restartability correctly (pantsbuild#21303)

The switch in pantsbuild#20874 to exposing intrinsics as python functions
meant that running an InteractiveProcess now occurs under a
separate engine Task than the goal that invokes it.

This introduced a bug where the `restartable` property was
applied to the InteractiveProcess intrinsic's Task, and not, as 
previously, to the goal's Task.

This change ensures that the restartability applies correctly,
via new rule helpers.

Fixes pantsbuild#21243.

src/python/pants/core/goals/deploy.py

@@ -14,8 +14,9 @@
 from pants.engine.console import Console
 from pants.engine.environment import EnvironmentName
 from pants.engine.goal import Goal, GoalSubsystem
-from pants.engine.process import InteractiveProcess, InteractiveProcessResult
-from pants.engine.rules import Effect, Get, MultiGet, collect_rules, goal_rule, rule
+from pants.engine.intrinsics import run_interactive_process
+from pants.engine.process import InteractiveProcess
+from pants.engine.rules import Get, MultiGet, collect_rules, goal_rule, rule
 from pants.engine.target import (
     FieldSet,
     FieldSetsPerTarget,
@@ -163,7 +164,7 @@ async def _invoke_process(
         return 0, tuple(results)
 
     logger.debug(f"Execute {process}")
-    res = await Effect(InteractiveProcessResult, InteractiveProcess, process)
+    res = await run_interactive_process(process)
     if res.exit_code == 0:
         sigil = console.sigil_succeeded()
         status = success_status

src/python/pants/core/goals/export.py

@@ -22,8 +22,9 @@
 from pants.engine.environment import EnvironmentName
 from pants.engine.fs import EMPTY_DIGEST, AddPrefix, Digest, MergeDigests, Workspace
 from pants.engine.goal import Goal, GoalSubsystem
-from pants.engine.internals.selectors import Effect, Get, MultiGet
-from pants.engine.process import InteractiveProcess, InteractiveProcessResult
+from pants.engine.internals.selectors import Get, MultiGet
+from pants.engine.intrinsics import run_interactive_process
+from pants.engine.process import InteractiveProcess
 from pants.engine.rules import collect_rules, goal_rule
 from pants.engine.target import FilteredTargets, Target
 from pants.engine.unions import UnionMembership, union
@@ -177,7 +178,7 @@ async def export(
                 env={"PATH": environment.get("PATH", ""), **cmd.extra_env},
                 run_in_workspace=True,
             )
-            ipr = await Effect(InteractiveProcessResult, InteractiveProcess, ip)
+            ipr = await run_interactive_process(ip)
             if ipr.exit_code:
                 raise ExportError(f"Failed to write {result.description} to {result_dir}")
         if result.resolve:

src/python/pants/core/goals/export_test.py

@@ -8,6 +8,8 @@
 from pathlib import Path
 from typing import List, Tuple
 
+from _pytest.monkeypatch import MonkeyPatch
+
 from pants.base.build_root import BuildRoot
 from pants.core.goals.export import (
     Export,
@@ -79,10 +81,17 @@ def _mock_run(rule_runner: RuleRunner, ip: InteractiveProcess) -> InteractivePro
     return InteractiveProcessResult(0)
 
 
-def run_export_rule(rule_runner: RuleRunner, resolves: List[str]) -> Tuple[int, str]:
+def run_export_rule(
+    rule_runner: RuleRunner, monkeypatch: MonkeyPatch, resolves: List[str]
+) -> Tuple[int, str]:
     union_membership = UnionMembership({ExportRequest: [MockExportRequest]})
     with open(os.path.join(rule_runner.build_root, "somefile"), "wb") as fp:
         fp.write(b"SOMEFILE")
+
+    def noop():
+        pass
+
+    monkeypatch.setattr("pants.engine.intrinsics.task_side_effected", noop)
     with mock_console(rule_runner.options_bootstrapper) as (console, stdio_reader):
         digest = rule_runner.request(Digest, [CreateDigest([FileContent("foo/bar", b"BAR")])])
         result: Export = run_rule_with_mocks(
@@ -132,7 +141,7 @@ def run_export_rule(rule_runner: RuleRunner, resolves: List[str]) -> Tuple[int,
         return result.exit_code, stdio_reader.get_stdout()
 
 
-def test_run_export_rule() -> None:
+def test_run_export_rule(monkeypatch) -> None:
     rule_runner = RuleRunner(
         rules=[
             UnionRule(ExportRequest, MockExportRequest),
@@ -142,7 +151,8 @@ def test_run_export_rule() -> None:
         ],
         target_types=[MockTarget],
     )
-    exit_code, stdout = run_export_rule(rule_runner, ["resolve"])
+
+    exit_code, stdout = run_export_rule(rule_runner, monkeypatch, ["resolve"])
     assert exit_code == 0
     assert "Wrote mock export for resolve to dist/export/mock" in stdout
     for filename in ["bar", "bar1", "bar2"]:

src/python/pants/core/goals/publish.py

@@ -35,8 +35,9 @@ async def publish_example(request: PublishToMyRepoRequest, ...) -> PublishProces
 from pants.engine.console import Console
 from pants.engine.environment import ChosenLocalEnvironmentName, EnvironmentName
 from pants.engine.goal import Goal, GoalSubsystem
-from pants.engine.process import InteractiveProcess, InteractiveProcessResult
-from pants.engine.rules import Effect, Get, MultiGet, collect_rules, goal_rule, rule
+from pants.engine.intrinsics import run_interactive_process_in_environment
+from pants.engine.process import InteractiveProcess
+from pants.engine.rules import Get, MultiGet, collect_rules, goal_rule, rule
 from pants.engine.target import (
     FieldSet,
     ImmutableValue,
@@ -244,10 +245,7 @@ async def run_publish(
             continue
 
         logger.debug(f"Execute {pub.process}")
-        res = await Effect(
-            InteractiveProcessResult,
-            {pub.process: InteractiveProcess, local_environment.val: EnvironmentName},
-        )
+        res = await run_interactive_process_in_environment(pub.process, local_environment.val)
         if res.exit_code == 0:
             sigil = console.sigil_succeeded()
             status = "published"

src/python/pants/core/goals/repl.py

@@ -14,8 +14,9 @@
 from pants.engine.environment import EnvironmentName
 from pants.engine.fs import Digest
 from pants.engine.goal import Goal, GoalSubsystem
-from pants.engine.process import InteractiveProcess, InteractiveProcessResult
-from pants.engine.rules import Effect, Get, collect_rules, goal_rule
+from pants.engine.intrinsics import run_interactive_process
+from pants.engine.process import InteractiveProcess
+from pants.engine.rules import Get, collect_rules, goal_rule
 from pants.engine.target import FilteredTargets, Target
 from pants.engine.unions import UnionMembership, union
 from pants.option.option_types import ArgsListOption, BoolOption, StrOption
@@ -147,8 +148,7 @@ async def run_repl(
 
     env = {**complete_env, **request.extra_env}
 
-    result = await Effect(
-        InteractiveProcessResult,
+    result = await run_interactive_process(
         InteractiveProcess(
             argv=(*request.args, *repl_subsystem.args),
             env=env,

src/python/pants/core/goals/run.py

@@ -21,8 +21,9 @@
     AmbiguousImplementationsException,
     TooManyTargetsException,
 )
-from pants.engine.process import InteractiveProcess, InteractiveProcessResult
-from pants.engine.rules import Effect, Get, Rule, _uncacheable_rule, collect_rules, goal_rule, rule
+from pants.engine.intrinsics import run_interactive_process
+from pants.engine.process import InteractiveProcess
+from pants.engine.rules import Get, Rule, _uncacheable_rule, collect_rules, goal_rule, rule
 from pants.engine.target import (
     BoolField,
     FieldSet,
@@ -254,8 +255,7 @@ async def run(
             )
         )
 
-    result = await Effect(
-        InteractiveProcessResult,
+    result = await run_interactive_process(
         InteractiveProcess(
             argv=(*request.args, *run_subsystem.args),
             env={**complete_env, **request.extra_env},
@@ -265,7 +265,7 @@ async def run(
             keep_sandboxes=global_options.keep_sandboxes,
             immutable_input_digests=request.immutable_input_digests,
             append_only_caches=request.append_only_caches,
-        ),
+        )
     )
 
     return Run(result.exit_code)

src/python/pants/core/goals/run_integration_test.py

@@ -1,6 +1,5 @@
 # Copyright 2021 Pants project contributors (see CONTRIBUTORS.md).
 # Licensed under the Apache License, Version 2.0 (see LICENSE).
-
 import time
 from pathlib import Path
 from textwrap import dedent
@@ -13,29 +12,34 @@
 
 
 @ensure_daemon
-def test_run_then_edit(use_pantsd: bool) -> None:
+def test_restartable(use_pantsd: bool) -> None:
     # These files must exist outside of a Pants `source_root` so that `coverage-py` doesn't try
     # to collect coverage metrics for them (as they are local to the chroot and coverage will
-    # error unable to find their source)
-    dirname = "not-a-source-root"
+    # error unable to find their source). We also need them to be in different locations for
+    # each parametrization of this test.
+    dirname = Path(f"test_restartable+{use_pantsd}/not-a-source-root").absolute()
+    dirname.mkdir(parents=True)
+
     files = {
-        f"{dirname}/slow.py": dedent(
+        dirname
+        / "slow.py": dedent(
             """\
             import time
             time.sleep(30)
             raise Exception("Should have been restarted by now!")
             """
         ),
-        f"{dirname}/BUILD": dedent(
+        dirname
+        / "BUILD": dedent(
             """\
             python_sources(name='lib')
             pex_binary(name='bin', entry_point='slow.py', restartable=True)
             """
         ),
     }
-    Path(dirname).mkdir(exist_ok=True)
-    for name, content in files.items():
-        Path(name).write_text(content)
+
+    for path, content in files.items():
+        path.write_text(content)
 
     with temporary_workdir() as workdir:
         client_handle = run_pants_with_workdir_without_waiting(
@@ -53,7 +57,70 @@ def test_run_then_edit(use_pantsd: bool) -> None:
         assert client_handle.process.poll() is None
 
         # Edit the file to restart the run, and check that it re-ran
-        Path(f"{dirname}/slow.py").write_text('print("No longer slow!")')
+        (dirname / "slow.py").write_text('print("No longer slow!")')
         result = client_handle.join()
         result.assert_success()
         assert result.stdout == "No longer slow!\n"
+
+
+@ensure_daemon
+def test_non_restartable(use_pantsd: bool) -> None:
+    # These files must exist outside of a Pants `source_root` so that `coverage-py` doesn't try
+    # to collect coverage metrics for them (as they are local to the chroot and coverage will
+    # error unable to find their source). We also need them to be in different locations for
+    # each parametrization of this test.
+    dirname = Path(f"test_non_restartable+{use_pantsd}/not-a-source-root").absolute()
+    dirname.mkdir(parents=True)
+
+    files = {
+        dirname
+        / "script.py": dedent(
+            """\
+            import os
+            import time
+
+            # Signal to the outside world that we've started.
+            touch_path = os.path.join("{dirname}", "touch")
+            with open(touch_path, "w") as fp:
+                fp.write("")
+            time.sleep(5)
+            print("Not restarted")
+            """.format(
+                dirname=dirname
+            )
+        ),
+        dirname
+        / "BUILD": dedent(
+            """\
+            python_sources(name="src", restartable=False)
+            """
+        ),
+    }
+
+    for path, content in files.items():
+        path.write_text(content)
+
+    with temporary_workdir() as workdir:
+        client_handle = run_pants_with_workdir_without_waiting(
+            [
+                "--backend-packages=['pants.backend.python']",
+                "run",
+                str(dirname / "script.py"),
+            ],
+            workdir=workdir,
+            use_pantsd=use_pantsd,
+        )
+
+        # Check that the pants run has actually started.
+        num_checks = 0
+        touch_path = dirname / "touch"
+        while not touch_path.exists():
+            time.sleep(1)
+            num_checks += 1
+            if num_checks > 30:
+                raise Exception("Failed to detect `pants run` process startup")
+
+        (dirname / "script.py").write_text('print("Restarted")')
+        result = client_handle.join()
+        result.assert_success()
+        assert result.stdout == "Not restarted\n"

src/python/pants/core/goals/run_test.py

@@ -8,6 +8,7 @@
 from typing import Iterable, Mapping, cast
 
 import pytest
+from _pytest.monkeypatch import MonkeyPatch
 
 from pants.core.goals.run import (
     Run,
@@ -81,6 +82,7 @@ class TestBinaryTarget(Target):
 
 def single_target_run(
     rule_runner: RuleRunner,
+    monkeypatch: MonkeyPatch,
     *,
     program_text: bytes,
     targets_to_field_sets: Mapping[Target, Iterable[FieldSet]] = FrozenDict(
@@ -89,6 +91,10 @@ def single_target_run(
 ) -> Run:
     workspace = Workspace(rule_runner.scheduler, _enforce_effects=False)
 
+    def noop():
+        pass
+
+    monkeypatch.setattr("pants.engine.intrinsics.task_side_effected", noop)
     with mock_console(rule_runner.options_bootstrapper) as (console, _):
         return run_rule_with_mocks(
             run,
@@ -139,16 +145,17 @@ def single_target_run(
         )
 
 
-def test_normal_run(rule_runner: RuleRunner) -> None:
+def test_normal_run(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:
     program_text = f'#!{sys.executable}\nprint("hello")'.encode()
     res = single_target_run(
         rule_runner,
+        monkeypatch,
         program_text=program_text,
     )
     assert res.exit_code == 0
 
 
-def test_materialize_input_files(rule_runner: RuleRunner) -> None:
+def test_materialize_input_files(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:
     program_text = f'#!{sys.executable}\nprint("hello")'.encode()
     binary = create_mock_run_request(rule_runner, program_text)
     with mock_console(rule_runner.options_bootstrapper):
@@ -162,30 +169,36 @@ def test_materialize_input_files(rule_runner: RuleRunner) -> None:
     assert result.exit_code == 0
 
 
-def test_failed_run(rule_runner: RuleRunner) -> None:
+def test_failed_run(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:
     program_text = f'#!{sys.executable}\nraise RuntimeError("foo")'.encode()
-    res = single_target_run(rule_runner, program_text=program_text)
+    res = single_target_run(rule_runner, monkeypatch, program_text=program_text)
     assert res.exit_code == 1
 
 
-def test_multi_target_error(rule_runner: RuleRunner) -> None:
+def test_multi_target_error(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:
     program_text = f'#!{sys.executable}\nprint("hello")'.encode()
     t1 = TestBinaryTarget({}, Address("some/addr"))
     t1_fs = TestRunFieldSet.create(t1)
     t2 = TestBinaryTarget({}, Address("some/other_addr"))
     t2_fs = TestRunFieldSet.create(t2)
     with pytest.raises(TooManyTargetsException):
         single_target_run(
-            rule_runner, program_text=program_text, targets_to_field_sets={t1: [t1_fs], t2: [t2_fs]}
+            rule_runner,
+            monkeypatch,
+            program_text=program_text,
+            targets_to_field_sets={t1: [t1_fs], t2: [t2_fs]},
         )
 
 
-def test_multi_field_set_error(rule_runner: RuleRunner) -> None:
+def test_multi_field_set_error(rule_runner: RuleRunner, monkeypatch: MonkeyPatch) -> None:
     program_text = f'#!{sys.executable}\nprint("hello")'.encode()
     target = TestBinaryTarget({}, Address("some/addr"))
     fs1 = TestRunFieldSet.create(target)
     fs2 = TestRunFieldSet.create(target)
     with pytest.raises(AmbiguousImplementationsException):
         single_target_run(
-            rule_runner, program_text=program_text, targets_to_field_sets={target: [fs1, fs2]}
+            rule_runner,
+            monkeypatch,
+            program_text=program_text,
+            targets_to_field_sets={target: [fs1, fs2]},
         )