Connect selected capture requests to the held installer worker (P1086)

docs/retained-practice-inputs.md

@@ -1,10 +1,38 @@
-# Original-coordinate practice inputs — implementation checkpoint
-
-Core #3829 remains unfinished. Entry, selected saved-context and coordinator input
-evidence now have explicit PracticeCopy consumers in source. Complete coordinator
-relations, coherent aggregate capture, bounded peak materialization, native-journal
-reading and the prepared-root application run remain unfinished.
-This document records the source decisions, not an executable selection roster.
+# Original-coordinate practice inputs — source and verification history
+
+Core #3829 delivered the selected file/context/coordinator operation and its
+private-fixture worker controls. The sections below retain their original source
+and execution epochs; an older **UNRUN** statement describes that checkpoint,
+not a fresh result. Original-data capture, peak resources and the native/physical
+reader journey remain open under P1086.
+
+## Selected requests through the shared installer
+
+`InstallRequest.selected_inputs` accepts a JSON array of the existing encoded
+`SelectedInputsProbe` records through `--selected-inputs`. The launcher serializes
+the same typed request. `upgrade` consumes that request directly rather than
+threading each option through another signature.
+
+The operation runs on `staged.probe` after `mode.owners_stopped` enters the real
+retargeted maintenance hold and before `member.rehearsed` takes its first complete
+practice copy. Each result is retained in `RehearsedUpgrade.captured_inputs`.
+The default empty tuple performs no selected capture. A plain `--rehearse` does
+not gain a hold: an explicit selected probe still refuses at its existing
+maintenance boundary. Ordinary rehearsal and complete backup selection stay
+unchanged.
+
+A caller with an encoded probe file can form the array with
+`jq -s '.' selected-probe.json` and pass that JSON as the value of
+`--selected-inputs` to the existing `agent-comms install` operation. This is an
+explicit install request, not a standalone freezer or permission to launch native
+sessions from copied data.
+
+The CLI/launcher source check decoded the actual P1086 request, preserved its
+values through the launcher wire, kept an empty default, and round-tripped the
+returned measurements. It parsed five changed Python files. It did not run a
+probe or create the selected destination. The first check incorrectly compared
+the CLI subclass identity with the decoded base request; that failed epoch is
+retained beside the corrected declared-value check.
 
 ## Original recipient birth participates in owner matching
 

src/agent_comms/installer/launch.py

@@ -198,8 +198,7 @@ def launch(self, output: TextIO) -> int:
 
     def run(self, runtime: LaunchRuntime) -> RehearsedUpgrade:
         command = self.command
-        return upgrade(command.checkout, self.root, self.target, command.installer,
-                       command.mode, command.retire, command.spill_dir, self.scratch,
+        return upgrade(command, self.root, self.target, self.scratch,
                        runtime, self.harnesses, InstallReport(
                            inherited_phases=[launched.timing() for launched in self.phases]))
 

src/agent_comms/installer/request.py

@@ -1,3 +1,3 @@
 """The install operation's options, shared by CLI dispatch and teammate launches."""
 
 from __future__ import annotations
@@ -10,6 +10,7 @@
 from agent_comms.nominal.field_codec import AbsolutePathText, Argument, FieldCodec, Normalize, WireName
 from agent_comms.install_choices import InstallerChoice, LiveMode, RehearsalMode, RunMode, UnnamedInstaller
 from agent_comms.installer.probes.harnesses import HarnessInstallation
+from agent_comms.installer.probes.inputs import SelectedInputsProbe
 
 
 @dataclass(frozen=True, kw_only=True)
@@ -37,6 +38,16 @@
             "action": "store_const", "const": FieldCodec.encode(RehearsalMode()),
         }),
     ] = dataclasses.field(default_factory=LiveMode)
+    selected_inputs: Annotated[
+        tuple[SelectedInputsProbe, ...],
+        Argument(
+            flags=("--selected-inputs",),
+            options={
+                "type": FieldCodec.parse,
+                "help": "JSON array of selected-input probes run under the live install hold before rehearsal",
+            },
+        ),
+    ] = ()
     retire: Annotated[
         bool,
         Argument(

src/agent_comms/installer/upgrade.py

@@ -1,3 +1,3 @@
 """Upgrade a retained bus and the checkout that runs it: what ``agent-comms install`` runs.
 
 ``agent-comms install`` checks out the target revision and runs this package from that
@@ -53,7 +53,7 @@
 from agent_comms.nominal.field_codec import PathText
 from agent_comms.typed_table import ExternalStore
 from agent_comms.host_services import HostService, RestartedService
-from agent_comms.install_choices import InstallerChoice, RunMode
+from agent_comms.install_choices import InstallerChoice
 from agent_comms.installer.backup import (
     BackupReport,
     RestoreSpace,
@@ -101,6 +101,7 @@
     TargetProgram,
 )
 from agent_comms.installer.owner_starts import OwnerStart
+from agent_comms.installer.request import InstallRequest
 from agent_comms.installer.practice import (
     PracticeCopy,
     PracticeMeasurement,
@@ -421,6 +422,8 @@
     rehearsal: Rehearsal
     phases: list[PhaseTiming] = field(kw_only=True)
     """Every phase shares the one clock until installation and cleanup end."""
+    captured_inputs: tuple[PracticeMeasurement, ...] = field(default=(), kw_only=True)
+    """Selected copies completed under this install's hold, before its rehearsal."""
 
     finished: ClassVar[str] = "Install rehearsed"
 
@@ -776,14 +779,13 @@
         return self.copy.copied_names
 
 
-def upgrade(checkout: Path, root: Path, target: str, installer: InstallerChoice, mode: RunMode,
-            retire: bool, spill_dir: Path, scratch: Path,
+def upgrade(command: InstallRequest, root: Path, target: str, scratch: Path,
             runtime: LaunchRuntime,
             harnesses: Callable[[Stack], InstallResult], report: InstallReport) -> RehearsedUpgrade:
     """``scratch`` holds the target's worktree the launcher runs this source from; this
     run owns it and removes it. A live run ends by registering the ``harnesses``."""
     commands = InstallCommands()
-    tree = TargetTree(scratch, target, repo=checkout)
+    tree = TargetTree(scratch, target, repo=command.checkout)
     bridges = report.bridges
     with (
         InstallRunPhase().running(commands.clock),
@@ -793,22 +795,22 @@
     ):
         with PreparePhase().running(commands.clock):
             StoreBackup.complete_interrupted(
-                root, spill_dir, Stack(checkout, commands).restore_stores)
-            current = commands.git(checkout, "rev-parse", "HEAD").strip()
+                root, command.spill_dir, Stack(command.checkout, commands).restore_stores)
+            current = commands.git(command.checkout, "rev-parse", "HEAD").strip()
             with commands.timing(GitPhase(
-                checkout, ("merge-base", "--is-ancestor", current, target),
+                command.checkout, ("merge-base", "--is-ancestor", current, target),
             )):
-                follows = Repo.descends(checkout, current, target)
+                follows = Repo.descends(command.checkout, current, target)
             if not follows:
                 raise UpgradeRefused(f"{target[:12]} does not follow the checkout's {current[:12]}")
-            if commands.git(checkout, "status", "--porcelain",
+            if commands.git(command.checkout, "status", "--porcelain",
                             "--untracked-files=no").strip():
                 raise UpgradeRefused("The checkout has local changes")
             try:
                 chain = CarryChain.at(tree.path)
             except ValueError as error:
                 raise UpgradeRefused(str(error)) from None
-            member = Install.member(root, retire)
+            member = Install.member(root, command.retire)
             installed = Stack(tree.path, commands)
             if not installed.python.exists():
                 raise UpgradeRefused(f"No installed stack at {installed.python}")
@@ -820,13 +822,16 @@
             raise UpgradeRefused("The prepared install must run in its built target runtime")
         staged.complete()
         with StartingReportPublicationPhase().running(commands.clock):
-            report.starting(mode.destination(member.initial_writer(root, installed)), target)
-        with mode.owners_stopped(member, root, installed, staged, bridges) as report.owners:
+            report.starting(command.mode.destination(member.initial_writer(root, installed)), target)
+        with command.mode.owners_stopped(member, root, installed, staged, bridges) as report.owners:
+            captured = tuple(staged.probe(probe) for probe in command.selected_inputs)
             with PracticePhase().running(commands.clock):
-                install, rehearsal = member.rehearsed(checkout, root, current, target, chain,
-                                                      installed, staged, tree, installer,
-                                                      spill_dir, lifetime=practice_lifetime)
-            result = report.complete(mode.finish(install, rehearsal, bridges, harnesses, report))
+                install, rehearsal = member.rehearsed(
+                    command.checkout, root, current, target, chain, installed, staged, tree,
+                    command.installer, command.spill_dir, lifetime=practice_lifetime)
+            result = report.complete(replace(
+                command.mode.finish(install, rehearsal, bridges, harnesses, report),
+                captured_inputs=captured))
     bridges.require_resumed()
     return result
 

tests/test_install_announcements.py

@@ -19,6 +19,7 @@
 from agent_comms.installer.refusals import InterpreterWorkerDied, UpgradeRefused
 from agent_comms.installer.probes.carry import Carried
 from agent_comms.installer.launch import InstallReport
+from agent_comms.installer.request import InstallRequest
 from agent_comms.installer.commands import InstallCommands
 from agent_comms.installer.interpreter_operation import CommandPhase, PhaseTiming
 from agent_comms.installer.interpreter_operation import InstalledProgram
@@ -29,7 +30,7 @@
 from agent_comms.installer.refusals import InstallRolledBack
 from agent_comms.installer.target_tree import TargetTree
 from agent_comms.installer.owner_starts import RefusedOwner, StartedOwner
-from agent_comms.install_choices import LiveMode, RehearsalMode, UnnamedInstaller
+from agent_comms.install_choices import LiveMode, RehearsalMode
 from agent_comms.launch_runtime import LaunchRuntime, RestoredRuntime
 from agent_comms.root_runtime import RootRuntimeStore
 from agent_comms.thread_execution import ExternalThreadExecution, ServiceThreadExecution
@@ -65,8 +66,9 @@ def test_failed_rehearsal_does_not_publish_to_the_source_bus(private_comms, tmp_
 
     with pytest.raises(UpgradeRefused, match="local changes"):
         upgrade(
-            source.checkout, private_comms.root, target, UnnamedInstaller(), RehearsalMode(), False,
-            tmp_path / "spill", tmp_path / "scratch",
+            InstallRequest(checkout=source.checkout, mode=RehearsalMode(),
+                           spill_dir=tmp_path / "spill"),
+            private_comms.root, target, tmp_path / "scratch",
             LaunchRuntime.running(),
             lambda stack: pytest.fail("A refused rehearsal cannot install harnesses"),
             InstallReport(),

tests/test_interrupted_rollback.py

@@ -13,10 +13,11 @@
 
 from agent_comms.comms import Comms, wire
 from agent_comms.active_route import active_route_path
-from agent_comms.install_choices import RehearsalMode, UnnamedInstaller
+from agent_comms.install_choices import RehearsalMode
 from agent_comms.installer.backup import StoreBackup
 from agent_comms.installer.probes.carry import Carried, Carry
 from agent_comms.installer.launch import InstallReport
+from agent_comms.installer.request import InstallRequest
 from agent_comms.installer.upgrade import upgrade
 from agent_comms.launch_runtime import LaunchRuntime
 from agent_comms.locked_store import LiveFilesRequest, RegistryFile
@@ -99,8 +100,9 @@ def refused_install(self) -> None:
         rollback recovery must finish before a checkout can be inspected or a stack built."""
         with pytest.raises(UpgradeRefused, match="missing-checkout"):
             upgrade(
-                self.tmp_path / "missing-checkout", self.comms.root, "HEAD", UnnamedInstaller(),
-                RehearsalMode(), False, self.spill, self.tmp_path / "install-scratch",
+                InstallRequest(checkout=self.tmp_path / "missing-checkout",
+                               mode=RehearsalMode(), spill_dir=self.spill),
+                self.comms.root, "HEAD", self.tmp_path / "install-scratch",
                 LaunchRuntime.running(), lambda: None, InstallReport(),
             )
 

tests/test_retained_capture.py

@@ -1,4 +1,5 @@
 """Selected capture borrows a real hold and drains real store transactions."""
+import json
 from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import AbstractContextManager, nullcontext
@@ -9,19 +10,22 @@
 
 from agent_comms.bus import Bus
 from agent_comms.comms import Comms, wire
+from agent_comms.cli import build_parser
 from agent_comms.context_store import ContextStore
 from agent_comms.coordination_database import CoordinationStore
 from agent_comms.errors import RelationViolationError
 from agent_comms.input_disposition import InputDispositions
 from agent_comms.input_key import TurnInputKey
 from agent_comms.installer.practice import PracticeCopy
+from agent_comms.installer.launch import InstallRun
+from agent_comms.installer.request import InstallRequest
 from agent_comms.installer.refusals import UpgradeRefused
 from agent_comms.installer.probes.inputs import SelectedInputsProbe
 from agent_comms.launch_runtime import LaunchRuntime
 from agent_comms.maintenance_barrier import (
     MaintenanceBarrier, MaintenanceNotHeldError, MaintenanceOperator, NoMaintenanceHolder,
 )
-from agent_comms.nominal.field_codec import TurnId
+from agent_comms.nominal.field_codec import FieldCodec, TurnId
 from agent_comms.repo import Checkout
 from agent_comms.threads import Thread
 from agent_comms.typed_table import WalStore
@@ -42,6 +46,21 @@ def capture_source(tmp_path: Path) -> Comms:
     return comms
 
 
+def test_selected_capture_request_survives_cli_and_launcher_wire(tmp_path: Path) -> None:
+    probe = SelectedInputsProbe(tmp_path / 'source', tmp_path / 'selected', 3 * 1024 ** 3)
+    arguments = build_parser().parse_args((
+        'install', '--checkout', str(tmp_path),
+        '--selected-inputs', json.dumps(FieldCodec.encode((probe,))),
+    ))
+    command = arguments.command_declaration.from_arguments(arguments)
+    assert command.selected_inputs == (probe,)
+    run = InstallRun(command, probe.root, tmp_path / 'installer', 'a' * 40)
+    decoded = FieldCodec.decode(InstallRun, FieldCodec.encode(run))
+    assert decoded.command == FieldCodec.decode(InstallRequest, FieldCodec.encode(command))
+    assert decoded.command.selected_inputs == (probe,)
+    assert not probe.destination.exists()
+
+
 def test_capture_refuses_a_destination_inside_its_source(
     capture_source: Comms,
 ) -> None: