feat(portal): answer halts on a phone without any customer ingress

docs/DECISION_PORTAL.md

@@ -1,8 +1,17 @@
 # The mobile decision portal
 
 When a governed run cannot confirm something, OpenAdapt halts instead of
-guessing. This portal is how that question reaches a staff member's phone
-without moving protected evidence off the runner.
+guessing. This portal is how that question reaches a staff member's phone with
+**full evidence** -- the screen crops, the gated control label, the whole halt
+detail -- without moving any of it off the runner.
+
+It is not the only way to reach a phone, and it is not the right one for a
+customer with no IT department. Serving full evidence to a phone requires an
+HTTPS origin the customer terminates themselves, which a dental practice will
+not stand up. For them, Flow's hosted lane dials **out** to the control plane
+and needs nothing on their network; it carries the signed PHI-free task and the
+closed halt context, never pixels. See
+[Answering a halt with no ingress](#answering-a-halt-with-no-ingress).
 
 Desktop owns the **lifecycle, the network boundary, device pairing, and the
 generic notification**. It owns no decision semantics: the question, the
@@ -48,6 +57,37 @@ The rules, all enforced in `engine/portal/ingress.py` and all fail-closed:
 There is no self-signed-certificate bypass and no test-only wide bind. The test
 suite exercises the shipped loopback configuration on a real socket.
 
+## Answering a halt with no ingress
+
+Everything above is about publishing **this** surface, which serves protected
+evidence. A customer who does not operate an ingress uses Flow's hosted lane
+instead: the engine makes outbound HTTPS requests only, so there is no inbound
+port, no port forward, no certificate, no reverse proxy and no static address,
+and it works behind NAT.
+
+Desktop turns it on when the operator's `deployment.json` sets
+`human_decisions.remote.enabled: true`. It then passes `--remote-decisions` to
+the attended console and hands it the runner credential from the keychain in the
+child process's environment -- never in `argv`, where it would sit in the
+process table for every user on the machine.
+
+Two things are checked **before** the console is spawned, because afterwards the
+failure is an opaque "the local decision service did not start":
+
+- this computer must be registered with the control plane, or Desktop refuses
+  and names the host; and
+- the resolved Flow must be at least `MIN_FLOW_FOR_REMOTE_DECISIONS`
+  (`engine/portal/service.py`), because an older one exits on the unknown flag
+  before printing its capability banner.
+
+`enabled` must be literally `true`. A truthy string or a `1` means no. Neither
+check degrades: a lane that looks on and is not is worse than one that is
+plainly off, because nothing on either surface would say the phone will never
+ring.
+
+The two paths are independent. A deployment may run the local portal, the hosted
+lane, both, or neither.
+
 ## Pairing a phone
 
 Desktop shows a QR code; the phone shows a short code back. Following RFC

engine/portal/service.py

@@ -71,6 +71,7 @@
 
 from loguru import logger
 
+from engine.auth.store import load_runner_credential
 from engine.flow_bridge import FLOW_BIN, _flow_command, _subprocess_env
 from engine.portal.flow_client import FlowConsoleClient, FlowConsoleUnavailable
 from engine.portal.ingress import IngressError, PortalIngress, resolve_ingress
@@ -96,6 +97,17 @@
 #: How long to wait for the console to announce itself before failing loud.
 CONSOLE_START_TIMEOUT_S = 60.0
 
+#: The first Flow release whose attended console accepts ``--remote-decisions``.
+#: Passing the flag to an older Flow makes argparse exit before the banner, so
+#: the operator would see "the console did not start" instead of the real cause.
+#: Checking the version first turns that into a sentence that names the fix.
+MIN_FLOW_FOR_REMOTE_DECISIONS = (1, 26, 0)
+
+#: Environment variable ``openadapt_flow.console.decision_relay`` reads the
+#: runner credential from. It is passed to the child process only, never
+#: written to a file and never logged.
+RUNNER_TOKEN_ENV = "OPENADAPT_RUNNER_TOKEN"
+
 #: How long uvicorn may take to bind *after* the banner is printed.  Flow
 #: prints the capability immediately before ``uvicorn.run()``, so the first
 #: request can legitimately arrive at a closed port.
@@ -415,9 +427,67 @@ def _start_console(self) -> ConsoleProcess:
         # The staged secret-bearing config lives only until the banner proves
         # Flow has read it; leaving this ``with`` block removes the file.
         with stage_private_yaml(staging_dir, prepared=prepared) as config_path:
-            return self._spawn_console(prefix, config_path)
+            return self._spawn_console(
+                prefix, config_path, remote_decisions=prepared.remote_decisions
+            )
+
+    def _remote_decision_env(self) -> dict[str, str]:
+        """The child's environment with the runner credential, or a refusal.
+
+        A deployment that enabled remote decisions and has no runner credential
+        must stop here. Starting the console without the credential would give
+        the operator a working local portal and a phone lane that is silently
+        absent -- the failure that is worse than the gap, because nothing on
+        either surface says the phone will never ring.
+        """
+
+        host = str(getattr(self.config, "hosted_host", "") or "").strip()
+        credential = load_runner_credential(host) if host else None
+        token = str((credential or {}).get("runner_token") or "").strip()
+        if not token:
+            where = host or "the control plane"
+            raise PortalError(
+                "This deployment answers halts on a phone through OpenAdapt "
+                f"Cloud, but this computer is not registered with {where} yet. "
+                "Connect it once, then start the portal again."
+            )
+        env = _subprocess_env()
+        env[RUNNER_TOKEN_ENV] = token
+        return env
 
-    def _spawn_console(self, prefix: list[str], config_path: Path) -> ConsoleProcess:
+    def _assert_flow_supports_remote_decisions(self) -> None:
+        """Refuse before spawning when the resolved Flow has no such flag."""
+
+        from importlib.metadata import PackageNotFoundError, version
+
+        try:
+            raw = version("openadapt-flow")
+        except PackageNotFoundError:  # pragma: no cover - defensive
+            raise PortalError(
+                "The OpenAdapt Flow runtime version could not be read, so "
+                "phone decisions cannot be enabled safely."
+            ) from None
+        parts: list[int] = []
+        for piece in raw.split(".")[:3]:
+            digits = "".join(ch for ch in piece if ch.isdigit())
+            parts.append(int(digits) if digits else 0)
+        while len(parts) < 3:
+            parts.append(0)
+        if tuple(parts) < MIN_FLOW_FOR_REMOTE_DECISIONS:
+            wanted = ".".join(str(part) for part in MIN_FLOW_FOR_REMOTE_DECISIONS)
+            raise PortalError(
+                "This deployment answers halts on a phone, which needs "
+                f"openadapt-flow {wanted} or newer. This computer has {raw}. "
+                "Update OpenAdapt, or turn off human_decisions.remote."
+            )
+
+    def _spawn_console(
+        self,
+        prefix: list[str],
+        config_path: Path,
+        *,
+        remote_decisions: bool = False,
+    ) -> ConsoleProcess:
         command = [
             *prefix,
             "console",
@@ -432,12 +502,20 @@ def _spawn_console(self, prefix: list[str], config_path: Path) -> ConsoleProcess
             "--port",
             str(int(getattr(self.config, "portal_console_port", 7863))),
         ]
+        env = _subprocess_env()
+        if remote_decisions:
+            # Both checks happen BEFORE the spawn. A console that starts and
+            # then dies on an unknown flag reports "the console did not start",
+            # which names neither the missing credential nor the old runtime.
+            self._assert_flow_supports_remote_decisions()
+            env = self._remote_decision_env()
+            command.append("--remote-decisions")
         process = self._popen(
             command,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             text=True,
-            env=_subprocess_env(),
+            env=env,
         )
         # Both pipes are drained for the process's lifetime. A blocking
         # readline() here would ignore the timeout entirely (it is only checked

engine/private_flow_config.py

@@ -80,10 +80,18 @@ class PrivateFlowConfigError(ValueError):
 
 @dataclass(frozen=True)
 class PreparedPrivateYaml:
-    """One immutable serialization and its same-snapshot log redactions."""
+    """One immutable serialization and its same-snapshot log redactions.
+
+    ``remote_decisions`` is read from the SAME snapshot as ``payload``, not by
+    re-opening the operator's file. Deciding "does this deployment answer halts
+    on a phone?" from a second read would reintroduce exactly the TOCTOU gap
+    this class exists to close: Flow could be launched with the outbound lane on
+    while executing a config that never enabled it.
+    """
 
     payload: str
     redactions: tuple[str, ...]
+    remote_decisions: bool = False
 
 
 def _load_mapping(source: Path | None) -> dict[str, Any]:
@@ -209,9 +217,27 @@ def prepare_flow_config(
     return PreparedPrivateYaml(
         payload=yaml.safe_dump(deployment, sort_keys=False),
         redactions=_redactions_for_mapping(deployment),
+        remote_decisions=_remote_decisions_enabled(deployment),
     )
 
 
+def _remote_decisions_enabled(deployment: Mapping[str, Any]) -> bool:
+    """Whether this deployment answers halts through the hosted lane.
+
+    Strictly ``True``: a truthy string, a 1, or a missing section all mean "no".
+    Turning on an outbound lane that carries decision context is not a default
+    and is not inferred.
+    """
+
+    human_decisions = deployment.get("human_decisions")
+    if not isinstance(human_decisions, Mapping):
+        return False
+    remote = human_decisions.get("remote")
+    if not isinstance(remote, Mapping):
+        return False
+    return remote.get("enabled") is True
+
+
 def prepare_flow_record_request(
     *,
     target: ExecutionTarget,

tests/test_engine/test_private_flow_config.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 import os
 import stat
 from pathlib import Path
@@ -235,3 +236,40 @@ def test_prepare_and_stage_use_one_immutable_source_snapshot(tmp_path: Path) ->
     assert "first-secret" in prepared.redactions
     assert "second.example" not in prepared.payload
     assert "second-secret" not in prepared.redactions
+
+
+# --------------------------------------------------- the hosted decision lane
+
+
+def test_remote_decisions_is_read_from_the_same_snapshot_as_the_payload(tmp_path):
+    """One read decides both what Flow executes and whether the lane is on."""
+    source = tmp_path / "deployment.json"
+    source.write_text(
+        json.dumps({"human_decisions": {"remote": {"enabled": True}}}),
+        encoding="utf-8",
+    )
+    prepared = prepare_flow_config(source, None)
+    assert prepared is not None
+    assert prepared.remote_decisions is True
+
+
+@pytest.mark.parametrize(
+    "deployment",
+    [
+        {},
+        {"human_decisions": {}},
+        {"human_decisions": {"remote": {}}},
+        {"human_decisions": {"remote": {"enabled": False}}},
+        # Truthy is not True. An outbound lane carrying decision context is
+        # never inferred from a stray string or a 1.
+        {"human_decisions": {"remote": {"enabled": "yes"}}},
+        {"human_decisions": {"remote": {"enabled": 1}}},
+        {"human_decisions": "remote"},
+    ],
+)
+def test_remote_decisions_defaults_to_off_and_is_never_inferred(tmp_path, deployment):
+    source = tmp_path / "deployment.json"
+    source.write_text(json.dumps(deployment), encoding="utf-8")
+    prepared = prepare_flow_config(source, None)
+    assert prepared is not None
+    assert prepared.remote_decisions is False

tests/test_portal/test_service.py

@@ -512,3 +512,111 @@ def test_stopping_the_console_terminates_directly_off_windows(monkeypatch) -> No
     process = FakeProcess("")
     service_mod._kill_tree(process)
     assert process.terminated is True
+
+
+# ------------------------------------------------- phone decisions, no ingress
+#
+# The whole point of the hosted lane is that a practice with no reverse proxy
+# gets a phone. Everything below is about it failing LOUDLY rather than starting
+# a console whose phone lane is silently absent.
+
+REMOTE_DEPLOYMENT = {
+    **DEPLOYMENT,
+    "human_decisions": {
+        "remote": {
+            "enabled": True,
+            "tenant_id": "tenant_exact_01",
+            "runner_id": "runner_exact_01",
+        }
+    },
+}
+
+RUNNER_TOKEN = "oar_" + "Z" * 40
+
+
+def remote_configured(data_dir: Path, **overrides) -> EngineConfig:
+    data_dir.mkdir(parents=True, exist_ok=True)
+    (data_dir / "deployment.json").write_text(
+        json.dumps(REMOTE_DEPLOYMENT), encoding="utf-8"
+    )
+    return EngineConfig(data_dir=data_dir, **overrides)
+
+
+def _capture_spawn(monkeypatch, banner_token: str = "b" * 43):
+    """Record the command and env the console would have been spawned with."""
+    seen: dict = {}
+
+    def popen(command, **kwargs):
+        seen["command"] = list(command)
+        seen["env"] = dict(kwargs.get("env") or {})
+        return FakeProcess(f"http://127.0.0.1:7863/#token={banner_token}")
+
+    monkeypatch.setattr(
+        "engine.portal.service._flow_command", lambda _bin: ["openadapt-flow"]
+    )
+    return seen, popen
+
+
+def test_a_deployment_without_remote_decisions_never_asks_for_the_flag(
+    monkeypatch, tmp_path
+) -> None:
+    seen, popen = _capture_spawn(monkeypatch)
+    service = PortalService(configured(tmp_path / "data"), popen=popen)
+    try:
+        service.start()
+    except PortalError:
+        pass  # the socket/flow-client half is not what this asserts
+    assert "--remote-decisions" not in seen["command"]
+    assert "OPENADAPT_RUNNER_TOKEN" not in seen["env"]
+
+
+def test_remote_decisions_refuses_when_this_computer_is_not_registered(
+    monkeypatch, tmp_path
+) -> None:
+    seen, popen = _capture_spawn(monkeypatch)
+    monkeypatch.setattr("engine.portal.service.load_runner_credential", lambda _h: None)
+    monkeypatch.setattr(
+        "engine.portal.service.PortalService._assert_flow_supports_remote_decisions",
+        lambda _self: None,
+    )
+    service = PortalService(remote_configured(tmp_path / "data"), popen=popen)
+    with pytest.raises(PortalError, match="not registered"):
+        service.start()
+    assert "command" not in seen  # nothing was spawned
+
+
+def test_remote_decisions_refuses_a_flow_runtime_without_the_flag(
+    monkeypatch, tmp_path
+) -> None:
+    seen, popen = _capture_spawn(monkeypatch)
+    monkeypatch.setattr(
+        "importlib.metadata.version", lambda _name: "1.25.0", raising=False
+    )
+    service = PortalService(remote_configured(tmp_path / "data"), popen=popen)
+    with pytest.raises(PortalError, match="1.26.0 or newer"):
+        service.start()
+    assert "command" not in seen
+
+
+def test_remote_decisions_passes_the_flag_and_the_credential(
+    monkeypatch, tmp_path
+) -> None:
+    seen, popen = _capture_spawn(monkeypatch)
+    monkeypatch.setattr(
+        "engine.portal.service.load_runner_credential",
+        lambda _h: {"runner_id": "runner_exact_01", "runner_token": RUNNER_TOKEN},
+    )
+    monkeypatch.setattr(
+        "engine.portal.service.PortalService._assert_flow_supports_remote_decisions",
+        lambda _self: None,
+    )
+    service = PortalService(remote_configured(tmp_path / "data"), popen=popen)
+    try:
+        service.start()
+    except PortalError:
+        pass
+    assert "--remote-decisions" in seen["command"]
+    assert seen["env"]["OPENADAPT_RUNNER_TOKEN"] == RUNNER_TOKEN
+    # The credential goes to the child process only. It is never an argument,
+    # where it would appear in the process table for every user on the machine.
+    assert RUNNER_TOKEN not in " ".join(seen["command"])