diff --git a/cli/remote_provider.py b/cli/remote_provider.py
index cb4c003..685696b 100644
--- a/cli/remote_provider.py
+++ b/cli/remote_provider.py
@@ -139,6 +139,68 @@ def _reject_api_conflicts(args: argparse.Namespace) -> None:
         )
 
 
+class _TaskServing(NamedTuple):
+    """What this join decided about task serving, and what the operator must be told (issue 58)."""
+
+    #: The operator opted in. False means they did not, and nothing here applies.
+    requested: bool
+    #: Why this provider cannot run a task, in the words of whichever check found out.
+    problem: str | None
+
+    @property
+    def allowed(self) -> bool:
+        return self.requested and self.problem is None
+
+
+def _decide_task_serving() -> _TaskServing:
+    """Ask, in the parent, whether the child about to be spawned could actually run a task.
+
+    Every one of these answers exists already — and until this ran, `run_task` was the only place
+    that asked, which is *after a member is waiting*. So `grid join` printed "serving",
+    `grid project status` said online, and the provider looked healthy right up until somebody
+    else's task died on it. The fail-closed work of issues 22 and 23 made these failures loud; they
+    were loud at the wrong moment and to the wrong person.
+
+    ⚠️ **The refusal cannot be printed by the serve child.** Measured: `_spawn_remote_engine`
+    detaches it with BOTH stdout and stderr redirected into the engine log, and this process then
+    waits 3s only to tell "alive" from "died" — tailing that log only when it died. Task serving
+    must not kill the child, so anything it printed would land in a file nobody has a reason to
+    open. The parent is the only place the sentence reaches the person who typed the command.
+
+    Costs nothing when the opt-in is off, which is the default: no `claude --version`, no probe of a
+    root nobody configured.
+
+    `(Exception, SystemExit)` because these checks use both — `task_sandbox` raises `SystemExit` as
+    a clean-error idiom, and a *bug* in any of them must degrade to "task serving off, inference
+    fine" rather than taking down a provider that was only asked to serve inference.
+    """
+    from remote import task_opt_in
+
+    if not task_opt_in.serving_enabled():
+        return _TaskServing(requested=False, problem=None)
+    from remote import task_agent
+
+    try:
+        task_agent.preflight_before_serving()
+    except (Exception, SystemExit) as exc:  # noqa: BLE001 — inference must survive any of them
+        return _TaskServing(requested=True, problem=str(exc) or exc.__class__.__name__)
+    return _TaskServing(requested=True, problem=None)
+
+
+def _task_serving_override(decision: _TaskServing) -> dict[str, str] | None:
+    """What to change in the serve child's environment, or `None` to hand it over untouched.
+
+    Only ever an OFF. Turning the opt-in *on* for somebody is the one thing this feature may not do
+    — a task loop spends the operator's own agent subscription — so a join that decided task serving
+    is fine passes the operator's own environment through rather than asserting it.
+    """
+    from remote import task_opt_in
+
+    if decision.requested and not decision.allowed:
+        return {task_opt_in.SERVING_ENV: "0"}
+    return None
+
+
 def cmd_remote_join(args: argparse.Namespace) -> int:
     from remote import credentials
 
@@ -203,6 +265,21 @@ def cmd_remote_join(args: argparse.Namespace) -> int:
     engine_id = _REMOTE_IDENTITY
     meta_name = getattr(args, "name", None) or socket.gethostname()
 
+    # ⚠️ **Before the lock, and before anything is stopped.** A provider that was serving inference
+    # a second ago must still be serving it after a task check that failed — so this may not run
+    # from inside the block that has already terminated the prior child on its way to finding out.
+    # It is also the last point where nothing has been written: a pure read, then the mutation.
+    task_serving = _decide_task_serving()
+    if task_serving.problem:
+        print(
+            f"Task serving is off for this join: {task_serving.problem}\n"
+            f"Inference is unaffected. Fix that and re-run `grid join --respawn` to claim tasks.",
+            file=sys.stderr,
+        )
+    # Withheld from the CHILD, never from this process: the opt-in travels in the environment the
+    # parent hands over, and the child reads it once at startup.
+    engine_env = _task_serving_override(task_serving)
+
     # Remote has ONE identity per grid (the token pins the relay node_id), so `grid join` is additive:
     # merge this join's engines into whatever is already serving, then respawn the single detached engine.
     # The read-merge-write is serialized so two concurrent joins can't lost-update the union (ADR 0010).
@@ -302,7 +379,8 @@ def cmd_remote_join(args: argparse.Namespace) -> int:
         if reloaded:
             reloaded = _hot_reload_identity(network_id, record, live)  # False if it fell back to a respawn
         else:
-            _respawn_identity(network_id, record, live)  # stops prior process(es) then respawns; aborts on failure
+            # stops prior process(es) then respawns; aborts on failure
+            _respawn_identity(network_id, record, live, env_overrides=engine_env)
 
     appended = bool(live)
     verb = "Appended to" if appended else "Joining"
@@ -315,6 +393,8 @@ def cmd_remote_join(args: argparse.Namespace) -> int:
         print(f"models={','.join(record['models'])}")
     if media:  # the comfyui:* models are resolved from bundle gating at serve time, not here
         print("media=on (serving comfyui:* workflows via the relay)")
+    if task_serving.allowed:  # said only when it is true — the refusal has already gone to stderr
+        print("tasks=on (claiming tasks for this grid)")
     print(f"log={paths.engines_dir(network_id) / f'{engine_id}.log'}")
     if reloaded:  # the live process re-advertised in place — nothing restarted, nothing dropped
         print(f"(hot-reloaded — no in-flight requests dropped; stop with `grid leave {label}`)")
@@ -997,7 +1077,8 @@ def _hot_reload_identity(
 
 
 def _respawn_identity(
-    network_id: str, record: dict[str, object], priors: list[dict[str, object]]
+    network_id: str, record: dict[str, object], priors: list[dict[str, object]],
+    *, env_overrides: dict[str, str] | None = None,
 ) -> None:
     """Stop the prior process(es), then write ``record`` and (re)spawn the one detached engine, setting
     ``record["pid"]``. Shared by join-append and leave-shrink (respawn is Slice 1's update mechanism).
@@ -1041,7 +1122,7 @@ def _respawn_identity(
     service_truth.clear_service_truth(record)
     record["started_at"] = runtime.utc_now()
     run_records.write_record(network_id, engine_id, record)
-    proc = _spawn_remote_engine(network_id, engine_id)
+    proc = _spawn_remote_engine(network_id, engine_id, env_overrides=env_overrides)
     # Stamp the identity, not just the pid: this is the ONLY stamp a `grid leave` racing this join
     # can see, because the child's own self-stamp has not run yet (POSIX `flock` gives the lock to
     # whoever asks, in no order). With it, that leave can verify the pid it is about to signal and —
@@ -1223,7 +1304,16 @@ def _build_record(
     }
 
 
-def _spawn_remote_engine(network_id: str, engine_id: str) -> subprocess.Popen:
+def _spawn_remote_engine(
+    network_id: str, engine_id: str, *, env_overrides: dict[str, str] | None = None
+) -> subprocess.Popen:
+    """Start the detached serve child. ``env_overrides`` wins over this process's own environment.
+
+    The child reads its opt-ins from the environment it inherits, so an override is how the parent
+    withholds one it has just decided this provider cannot honour (issue 58). A new dict rather than
+    a mutation of `os.environ`: this process may still have work to do, and a knob turned off for a
+    child must not be turned off for its parent.
+    """
     from local import runtime
 
     log_path = paths.engines_dir(network_id) / f"{engine_id}.log"
@@ -1237,7 +1327,7 @@ def _spawn_remote_engine(network_id: str, engine_id: str) -> subprocess.Popen:
         stdout=log,
         stderr=subprocess.STDOUT,
         start_new_session=True,
-        env={**os.environ, "PYTHONUNBUFFERED": "1"},
+        env={**os.environ, "PYTHONUNBUFFERED": "1", **(env_overrides or {})},
     )
 
 
diff --git a/cli/remote_task.py b/cli/remote_task.py
index bb2ce46..5fb69c7 100644
--- a/cli/remote_task.py
+++ b/cli/remote_task.py
@@ -15,6 +15,7 @@
 import base64
 import datetime
 import json
+import shlex
 import sys
 from pathlib import Path
 from typing import Any
@@ -1844,8 +1845,64 @@ def _task_get(args: argparse.Namespace) -> int:
         print(f"\nThe grid could not combine this work with a colleague's change. Your work is "
               f"safe in this conversation — ask for it again with:\n"
               f"  grid task send {conversation or '<conversation-id>'} --prompt '<what to do>'")
+    for line in _changed_since_note(task):
+        print(line)
     result = task.get("result_text")
     if result:
         print("\n--- result ---")
         print(result.rstrip("\n"))
     return code
+
+
+# How many of the relay's sample this prints. The relay already caps what it sends
+# (`turn_superseded.MAX_PATHS`); this is the second, smaller bound on what a person reads in a
+# terminal, and the count beside it is what stays true either way.
+_CHANGED_SINCE_NAMED = 5
+
+
+def _changed_since_note(task: dict) -> list[str]:
+    """Say when the project no longer holds what this turn changed (ND-16/F-3). Lines, or none.
+
+    ⚠️ **The failure this closes is a turn that reads like a success everywhere.** Under ADR 0034
+    D-d the relay applies every finished turn itself, and D-g hands a collision to a turn in the
+    OTHER conversation — whose decision to drop somebody's work is recorded where that somebody
+    cannot read it, because `grid task list --project` returns only the asker's own turns. Measured
+    live 2026-08-20: `state=completed`, `grid task diff` showing their own change, and the project
+    holding somebody else's line.
+
+    ⚠️ **BOTH keys, and both type-checked, or this says nothing.** They arrive together from a relay
+    that can answer at all, and absence is what every relay deployed before this feature sends — so a
+    missing or unreadable pair must read as *nothing to show*, never as a warning made up here. A
+    count of `0` is the relay's positive answer that the work still stands, and it is the ordinary
+    case: a note that appeared on every task would be furniture within a day.
+
+    ⚠️ `bool` is refused explicitly. It is a subclass of `int`, so `True` would otherwise print
+    *1 of the files …* for a relay that sent a flag where a count belongs.
+    """
+    count = task.get("changed_since_count")
+    paths = task.get("changed_since_paths")
+    if isinstance(count, bool) or not isinstance(count, int) or count <= 0:
+        return []
+    if not isinstance(paths, list):
+        return []
+    named = [path for path in paths[:_CHANGED_SINCE_NAMED] if isinstance(path, str) and path]
+    if not named:
+        return []
+    files = "file" if count == 1 else "files"
+    lines = ["",
+             f"{count} {files} this task changed {'has' if count == 1 else 'have'} been changed "
+             f"again since it ran:"]
+    lines += [f"  {path}" for path in named]
+    if count > len(named):
+        lines.append(f"  … and {count - len(named)} more")
+    # The project id comes off the task rather than from the caller's arguments: `grid task get`
+    # takes only a task id, and a person who has to go and find the project id first is a person who
+    # reads the warning and does nothing about it.
+    project = task.get("project_id")
+    lines.append("What the project holds now may not be what this result describes. Read what is "
+                 "there with:")
+    # `shlex.quote` leaves an ordinary path exactly as it is and wraps one with a space in it. A
+    # command somebody is told to run has to be a command they can paste, and a repository that came
+    # in through `POST …/import` can hold any name its author gave it.
+    lines.append(f"  grid project file {project or '<project-id>'} {shlex.quote(named[0])}")
+    return lines
diff --git a/docs/adr/0034-a-task-is-a-conversation-and-nobody-merges-by-hand.md b/docs/adr/0034-a-task-is-a-conversation-and-nobody-merges-by-hand.md
index 3093f50..1f8fb6c 100644
--- a/docs/adr/0034-a-task-is-a-conversation-and-nobody-merges-by-hand.md
+++ b/docs/adr/0034-a-task-is-a-conversation-and-nobody-merges-by-hand.md
@@ -427,6 +427,36 @@ allowlist with no domain constraint, so this rule is **wider** than the change r
 reaches every non-private project, including imported company repositories. That is not the supported
 topology, and `visibility` is the only instrument for it. Named here rather than discovered.
 
+⚠️ **CORRECTED — measured 2026-08-23 (ND-21). "Grid *is* domain, structurally" is true of the DOMAIN
+and false of the ROSTER, and it is false on the supported topology.** grid-apis
+`store.member_for_access` checks an allowlist row **before** anything else, on every network type,
+and grid-src `grid_auth._requires_allowlist` names three ways onto a `private-domain` grid — the
+right domain, an allowlist row, or owning the grid — carrying the sentence *"an invited account from
+another domain keeps working after the switch"*. An invited outsider is therefore **authenticated**,
+so D-k gives them every non-private project on the grid, imported company repositories included,
+plus a `project_members` row minted on their first read that needs a key.
+
+**The relay does not compare domains and cannot.** `grid_auth.GridAuthContext` carries `email`;
+`relay.AuthContext`, the object a route is handed, does not. The premise is enforced entirely at the
+control plane, and the rule there **admits** rather than restricts.
+
+**Decided 2026-08-23: correct the claim, do not narrow the gate.** D-k's premise reads *authenticated
+on this grid ⇒ a colleague **or somebody a colleague invited***, and the blast radius above is
+accepted with it. Three reasons, in order of weight: the alternative asks the control plane for a new
+value saying WHY a caller was admitted (domain vs allowlist vs owner), which is a lockstep value and
+a rollout order for a boundary that is already live; narrowing it would revoke access from invited
+accounts that work today, which grid-apis explicitly promises will keep working; and `visibility`
+already exists as the instrument for a project that must not be grid-wide. What is **not** accepted
+is the justification standing while being false — that is what this correction removes.
+
+⚖️ In fairness to the shape being accepted: the minted row is **visible** afterwards in
+`grid project member list`, so an owner can see who reached the project. Nobody is notified, and the
+access is granted before anyone looks.
+
+Pinned by grid-src `test_project_visibility.TestAnInvitedOutsiderIsAColleagueToThisRule` — a
+characterization test, so a future narrowing announces itself here rather than in a support ticket.
+Narrowing is tracked as follow-up work, not as a defect against this decision.
+
 ⚠️ **`TASK_SERVED_DOMAINS` is untouched.** It is 0033 issue 24's *a provider serves only its own
 company's domain* gate — a different axis with a different failure mode. This is the third thing in
 the codebase called "domain".
diff --git a/docs/cli.md b/docs/cli.md
index df4e360..066a839 100644
--- a/docs/cli.md
+++ b/docs/cli.md
@@ -1626,6 +1626,25 @@ can `get`, `list` and `follow` any task in it. That is deliberate — a member c
 project and read any task branch — and it is what makes reviewing a colleague's run possible without
 one.
 
+`get` also says when the project **no longer holds** what a task changed:
+
+```
+2 files this task changed have been changed again since it ran:
+  shared.txt
+  notes.md
+What the project holds now may not be what this result describes. Read what is there with:
+  grid project file 4f0e… shared.txt
+```
+
+The grid applies every finished task to the project by itself, so a colleague's later work can land
+over yours — and when two tasks collide, the step that combines them runs in **their** conversation,
+where you never see it. Without this line your task reads `completed` on every surface you have,
+including `grid task diff`, which shows what your task changed rather than what the project holds
+now. The line appears only when something really has changed, and it counts a later task of your own
+the same as anybody else's: what it means is *look before you rely on this*. `--json` carries the
+same fact as `changed_since_count` and `changed_since_paths`; a grid whose relay predates this sends
+neither key and the line simply does not appear.
+
 ### Acting on the outcome from a script
 
 Both `get` and `follow` exit with the task's own outcome, so a shell can branch on it without
@@ -2167,6 +2186,15 @@ a 24.04 host that a task can then read its workspace, install a dependency from
 read a file outside the workspace. **Do not** task that sysctl off to "fix" a task that fails to run
 commands; check `bubblewrap` and `socat` first.
 
+**`grid join` checks all of this before it serves.** With `GRID_TASKS` on, the join asks the same
+questions a task would — is Claude Code installed and new enough, is the permission mode and
+passthrough list valid, can the sandbox start, and can this account create a workspace under
+`GRID_TASK_ROOT` — and it asks them *in the terminal you typed the command in*. A provider that
+fails any of them still **serves inference**: the join lands, the answer says what to change, and
+only task serving is withheld until you fix it and re-run `grid join --respawn`. Before this, every
+one of those was checked only after a task had been claimed, so the first person to find out was a
+member of your grid whose task died.
+
 The environment variables that tune a provider, all optional:
 
 | variable | default | what it does |
diff --git a/remote/serve.py b/remote/serve.py
index 4f083a2..7e1a7da 100644
--- a/remote/serve.py
+++ b/remote/serve.py
@@ -26,7 +26,7 @@
 
 from remote import (
     api_keys, bringup, control_plane, credentials, engine_health, probe, relay, service_truth,
-    codex_auth, codex_oauth, task_capacity, throughput,
+    codex_auth, codex_oauth, task_capacity, task_opt_in, throughput,
 )
 from shared.handlers import HANDLERS
 from shared import run_records
@@ -2825,53 +2825,6 @@ def _supervise(loop: Callable[[_ServeState], None], state: _ServeState) -> None:
         state.stop.set()
 
 
-def _task_serving_enabled() -> bool:
-    """Whether this provider claims distributed tasks (ADR 0032). Opt-in, and off by default.
-
-    Read from the environment at serve time rather than baked into the run record: the detached
-    serve child inherits the parent's environment (`_spawn_remote_engine` passes ``{**os.environ}``),
-    so ``GRID_TASKS=1 grid join …`` reaches here. Opt-in is not a convenience — a task loop spends
-    the operator's own agent subscription, so it may never turn itself on.
-    """
-    return os.getenv("GRID_TASKS", "").strip().lower() in ("1", "true", "yes", "on")
-
-
-TASK_WORKERS_ENV = "GRID_MAX_TASKS"
-# The count that changes nothing. Turning task serving on may not also change how much of the
-# operator's subscription it spends, so the pool starts where it has always been and only the
-# operator moves it. Deliberately NOT a benchmarked ceiling — the ceiling that matters is the
-# subscription's own, read at runtime by `remote/task_capacity.py` (ADR 0032 issue 09).
-_DEFAULT_TASK_WORKERS = 1
-
-
-def _task_worker_count() -> int:
-    """How many tasks this provider runs at once (ADR 0032 issue 09).
-
-    Misconfiguration falls back rather than failing, the same rule `tasks.task_timeout()` states: a
-    provider that refused to serve tasks because an operator typed `three` would take task serving
-    down for the life of the process, which is a far worse answer than running with the default and
-    saying so.
-
-    There is **no upper clamp**, and that is deliberate. A number picked here would be exactly the
-    guessed constant this issue exists to remove, and it would be guessed about the wrong thing —
-    what binds is the operator's own subscription, not this process's opinion of their machine. The
-    machine's real limit is discovered instead: a thread that cannot start is reported, and the
-    workers that did start keep serving.
-    """
-    raw = (os.getenv(TASK_WORKERS_ENV) or "").strip()
-    if not raw:
-        return _DEFAULT_TASK_WORKERS
-    try:
-        count = int(raw)
-    except ValueError:
-        count = 0
-    if count < 1:
-        print(f"\n[tasks] {TASK_WORKERS_ENV}={raw!r} is not a positive whole number of tasks; "
-              f"using {_DEFAULT_TASK_WORKERS}", file=sys.stderr)
-        return _DEFAULT_TASK_WORKERS
-    return count
-
-
 def _start_task_worker(state: _ServeState) -> list[threading.Thread]:
     """Start the distributed-tasks claim loops, if enabled. Returns the threads that started.
 
@@ -2897,11 +2850,11 @@ def _start_task_worker(state: _ServeState) -> list[threading.Thread]:
     makes every worker throttle on ONE reading of the one subscription they all spend.
     """
     try:
-        if not _task_serving_enabled():
+        if not task_opt_in.serving_enabled():
             return []
         from . import tasks
 
-        count = _task_worker_count()
+        count = task_opt_in.worker_count()
     except (Exception, SystemExit) as exc:
         print(f"\nCould not start task serving ({exc!r}); inference is unaffected.", file=sys.stderr)
         state.tasks_stop.set()
diff --git a/remote/task_agent.py b/remote/task_agent.py
index d39df11..3369e04 100644
--- a/remote/task_agent.py
+++ b/remote/task_agent.py
@@ -745,6 +745,61 @@ def preflight() -> None:
         task_sandbox.preflight(task_root=workspace_root())
 
 
+def preflight_before_serving() -> None:
+    """Everything a task will need, asked before there is a task — for `grid join` (issue 58).
+
+    `run_task` asks `preflight()` and `resolve_binary()` at claim time, which is after a member is
+    already waiting on this provider. Until this existed, that was the ONLY moment either was asked:
+    `grid join` printed its banner, `grid project status` said online, and the provider looked
+    healthy right up until somebody else's task died on it.
+
+    The **same two functions**, never a copy of them. A second opinion here about the sandbox, the
+    permission mode or the version floor would be one more thing to keep in step, and every failure
+    this plane has recorded is two readings of one rule that got edited apart.
+
+    One question is asked here and nowhere else — whether the workspace root is reachable at all —
+    and it is asked **unconditionally**, unlike the checks inside `preflight()` that belong to the
+    sandbox. An unwritable root fails every task with the sandbox on or off.
+    """
+    preflight()
+    resolve_binary()
+    _require_a_reachable_workspace_root()
+
+
+def _require_a_reachable_workspace_root() -> None:
+    """Refuse a workspace root this provider could not put a task under.
+
+    The macOS default is the shape this exists for: `DEFAULT_WORKSPACE_ROOT` is under `/var`, which
+    is root-owned there, so a provider without `sudo` fails EVERY task with "could not create
+    /var/grid/…" — one member's task at a time, forever, with every health signal green.
+
+    **A probe, never a creation.** Which directory the root should be, and with what mode and what
+    refusals, is issue 62's decision; making one here would pre-empt it and would also mean a
+    `grid join` that failed left a directory behind.
+
+    `os.access` is advisory — it knows nothing about ACLs, a read-only mount, or a full disk — so
+    this is an early warning that can MISS, never a guarantee. A root it passes and `mkdir` then
+    refuses fails exactly where it does today, at `ensure_workspace`. The direction to be wrong in
+    is letting a working provider through, because the alternative is refusing one.
+    """
+    root = workspace_root()
+    if root.exists() and not root.is_dir():
+        raise OSError(
+            f"the task workspace root {root} is not a directory, so no task can be checked out "
+            f"under it; point {WORKSPACE_ROOT_ENV} somewhere else")
+
+    reachable = root
+    while not reachable.exists():
+        if reachable.parent == reachable:  # reached the filesystem root without finding anything
+            break
+        reachable = reachable.parent
+    if reachable.is_dir() and os.access(reachable, os.W_OK | os.X_OK):
+        return
+    raise OSError(
+        f"this provider cannot create a task workspace under {root}: {reachable} is not writable "
+        f"by it. Point {WORKSPACE_ROOT_ENV} at a short path this account can write")
+
+
 def _require_version_for_the_sandbox(binary: str) -> None:
     """Refuse a binary that would accept the confinement policy and ignore it.
 
diff --git a/remote/task_evict.py b/remote/task_evict.py
index c4ed573..f607c16 100644
--- a/remote/task_evict.py
+++ b/remote/task_evict.py
@@ -13,7 +13,7 @@
 a bound that cannot be met is simply not met.
 
 Two knobs, both TUNABLES — a bad value warns on stderr and falls back, the convention
-`serve._task_worker_count` and `tasks.task_timeout` follow, rather than the refuse-outright one
+`task_opt_in.worker_count` and `tasks.task_timeout` follow, rather than the refuse-outright one
 `GRID_TASK_PERMISSION_MODE` follows. Refusing to start over a misconfigured *cap* would take task
 serving down for the life of the process, which is a far larger fault than the one it reports.
 """
diff --git a/remote/task_opt_in.py b/remote/task_opt_in.py
new file mode 100644
index 0000000..b24c010
--- /dev/null
+++ b/remote/task_opt_in.py
@@ -0,0 +1,66 @@
+"""Whether this provider claims distributed tasks, and how many at once (ADR 0032, issue 09).
+
+Its own module because **two processes ask**. The serve child asks at startup, which is where these
+answers began (`remote/serve.py`); the CLI parent asks before it spawns that child, so `grid join`
+can tell an operator their task configuration is broken while they are still looking at the terminal
+instead of after a member's task has died on it.
+
+The parent cannot get them from `serve`: that module is the provider runtime and pulls httpx and the
+whole relay client with it. But the import weight is the smaller reason. The real one is that the
+alternative — the parent reading `GRID_TASKS` for itself — is a **second reading of one rule**, and
+two readings of one rule get edited apart while every test stays green. `tests/test_task_opt_in.py`
+enforces that there is exactly one of each, by scanning the source rather than by asking nicely.
+"""
+from __future__ import annotations
+
+import os
+import sys
+
+SERVING_ENV = "GRID_TASKS"
+WORKERS_ENV = "GRID_MAX_TASKS"
+
+# The count that changes nothing. Turning task serving on may not also change how much of the
+# operator's subscription it spends, so the pool starts where it has always been and only the
+# operator moves it. Deliberately NOT a benchmarked ceiling — the ceiling that matters is the
+# subscription's own, read at runtime by `remote/task_capacity.py` (ADR 0032 issue 09).
+DEFAULT_WORKERS = 1
+
+
+def serving_enabled() -> bool:
+    """Whether this provider claims distributed tasks (ADR 0032). Opt-in, and off by default.
+
+    Read from the environment at serve time rather than baked into the run record: the detached
+    serve child inherits the parent's environment (`cli/remote_provider._spawn_remote_engine` passes
+    ``{**os.environ}``), so ``GRID_TASKS=1 grid join …`` reaches the child that way. Opt-in is not a
+    convenience — a task loop spends the operator's own agent subscription, so it may never turn
+    itself on.
+    """
+    return os.getenv(SERVING_ENV, "").strip().lower() in ("1", "true", "yes", "on")
+
+
+def worker_count() -> int:
+    """How many tasks this provider runs at once (ADR 0032 issue 09).
+
+    Misconfiguration falls back rather than failing, the same rule `tasks.task_timeout()` states: a
+    provider that refused to serve tasks because an operator typed `three` would take task serving
+    down for the life of the process, which is a far worse answer than running with the default and
+    saying so.
+
+    There is **no upper clamp**, and that is deliberate. A number picked here would be exactly the
+    guessed constant this issue exists to remove, and it would be guessed about the wrong thing —
+    what binds is the operator's own subscription, not this process's opinion of their machine. The
+    machine's real limit is discovered instead: a thread that cannot start is reported, and the
+    workers that did start keep serving.
+    """
+    raw = (os.getenv(WORKERS_ENV) or "").strip()
+    if not raw:
+        return DEFAULT_WORKERS
+    try:
+        count = int(raw)
+    except ValueError:
+        count = 0
+    if count < 1:
+        print(f"\n[tasks] {WORKERS_ENV}={raw!r} is not a positive whole number of tasks; "
+              f"using {DEFAULT_WORKERS}", file=sys.stderr)
+        return DEFAULT_WORKERS
+    return count
diff --git a/tests/e2e_cross_repo/provider_process.py b/tests/e2e_cross_repo/provider_process.py
index 1e47911..ffa7553 100644
--- a/tests/e2e_cross_repo/provider_process.py
+++ b/tests/e2e_cross_repo/provider_process.py
@@ -70,7 +70,7 @@ def __init__(self, state, task_id, *, interval=None, on_beat=None):
     # process has exactly one worker. Every test that does not ask for concurrency must keep it.
     #
     # One `task_loop` per worker, on its own thread, sharing one `_State` — which is what
-    # `remote/serve.py` does for `GRID_MAX_TASKS`. Not a rebinding of that variable: this file
+    # `remote/serve.py` does for `task_opt_in.worker_count()`. Not a rebinding of that variable: this file
     # deliberately does not go through `serve`, so reading it here would name a knob nothing in this
     # process consults.
     workers = int(os.environ.get("GRID_E2E_TASK_WORKERS", "1"))
diff --git a/tests/test_local_cli.py b/tests/test_local_cli.py
index 4ec9bb4..1facd76 100644
--- a/tests/test_local_cli.py
+++ b/tests/test_local_cli.py
@@ -6865,6 +6865,9 @@ def fake_popen(cmd, **kw):
         # `spawned["cmd"]` absent, and the assertions raise KeyError rather than passing.
         if run_records.REMOTE_ENGINE_MARKER in cmd:
             spawned["cmd"] = cmd
+            # The child's environment, kept beside its argv: task serving is opted into through the
+            # environment the parent hands over, so this is where issue 58's decision is observable.
+            spawned["env"] = kw.get("env") or {}
         return type("P", (), {"pid": pid})()
 
     monkeypatch.setattr(cli.remote_provider.subprocess, "Popen", fake_popen)
@@ -13036,10 +13039,12 @@ def test_task_worker_that_fails_to_START_does_not_take_the_engine_down(monkeypat
     would propagate out of `_serve_loop` to the top-level handler, unregister the node and kill the
     process. A task-plane STARTUP fault must be as survivable as a task-plane runtime fault.
     """
-    from remote import serve
+    from remote import serve, task_opt_in
 
     monkeypatch.setenv("GRID_TASKS", "1")
-    monkeypatch.setattr(serve, "_task_serving_enabled", _raise(RuntimeError("cannot start thread")))
+    # Patched where the opt-in now LIVES (issue 57). `_start_task_worker` looks the name up on the
+    # module at call time, so this reaches the same guarded block the pre-move patch did.
+    monkeypatch.setattr(task_opt_in, "serving_enabled", _raise(RuntimeError("cannot start thread")))
     monkeypatch.setattr(serve, "_poll_loop", lambda state: state.stop.set())
     monkeypatch.setattr(serve, "_heartbeat_loop", lambda state: None)
     state = _serve_state(monkeypatch, tmp_path)
@@ -32744,6 +32749,106 @@ def test_task_get_on_an_ordinary_failed_turn_says_no_such_thing(monkeypatch, tmp
     assert "combine" not in (captured.out + captured.err).lower()
 
 
+def test_task_get_says_when_the_project_no_longer_holds_what_the_turn_changed(
+        monkeypatch, tmp_path, capsys):
+    """ND-16/F-3. A turn whose work a later one overwrote still reads `completed` everywhere.
+
+    Measured live 2026-08-20 (scenario A2): a merge turn in SOMEBODY ELSE's conversation dropped B's
+    line on purpose and said so in its own result — which B cannot read, because
+    `grid task list --project` returns only the asker's own turns. `task get` said `completed`,
+    `task diff` showed B's own change, and only `grid project file` revealed the project did not
+    hold it. The PRD's person fires one task and walks away, so "your next turn will be told" is not
+    a way for them to find out.
+    """
+    _seed_running_remote_grid(monkeypatch, tmp_path)
+    state.set_mode("remote")
+    _mock_relay(monkeypatch, lambda r: httpx.Response(200, json={
+        "id": "t-1", "project_id": "p-1", "state": "completed",
+        "result_text": "shared.txt now contains exactly one line: STATUS: bravo",
+        "changed_since_count": 1, "changed_since_paths": ["shared.txt"],
+    }))
+
+    rc = cli.main(["task", "get", "t-1"])
+
+    assert rc == 0, "a turn that completed did not stop completing because somebody edited it after"
+    out = capsys.readouterr().out
+    assert "shared.txt" in out
+    assert "changed again" in out, out
+    assert "grid project file p-1 shared.txt" in out, (
+        "the person is told their work may be gone with no way to look at what is there now")
+
+
+def test_task_get_quotes_a_path_it_tells_somebody_to_paste(monkeypatch, tmp_path, capsys):
+    """A repository that arrived through `import` holds whatever names its author gave it, and the
+    line below is a command a person copies. Unquoted, `my notes.txt` reaches `grid project file` as
+    two arguments and is refused — a warning about lost work whose one next step does not run."""
+    _seed_running_remote_grid(monkeypatch, tmp_path)
+    state.set_mode("remote")
+    _mock_relay(monkeypatch, lambda r: httpx.Response(200, json={
+        "id": "t-1", "project_id": "p-1", "state": "completed",
+        "changed_since_count": 1, "changed_since_paths": ["my notes.txt"]}))
+
+    cli.main(["task", "get", "t-1"])
+
+    out = capsys.readouterr().out
+    assert "grid project file p-1 'my notes.txt'" in out, out
+
+
+def test_task_get_says_nothing_when_the_turn_still_stands(monkeypatch, tmp_path, capsys):
+    """The control, and the one that decides whether the sentence means anything: `0` is the
+    ordinary answer for the ordinary turn, and a warning that appears on every task is furniture."""
+    _seed_running_remote_grid(monkeypatch, tmp_path)
+    state.set_mode("remote")
+    _mock_relay(monkeypatch, lambda r: httpx.Response(200, json={
+        "id": "t-1", "project_id": "p-1", "state": "completed", "result_text": "done",
+        "changed_since_count": 0, "changed_since_paths": [],
+    }))
+
+    cli.main(["task", "get", "t-1"])
+
+    assert "changed again" not in capsys.readouterr().out
+
+
+def test_task_get_on_a_relay_that_cannot_answer_that_question_says_nothing(
+        monkeypatch, tmp_path, capsys):
+    """The degrade direction, and it is the one this pair exists to pin. An older relay sends
+    NEITHER key, and absence must read as "nothing to show" — never as a warning invented client
+    side. The relay is deployed before this CLI (see the register in CLAUDE.md), so this is the
+    ordinary state of every grid for the length of a rollout."""
+    _seed_running_remote_grid(monkeypatch, tmp_path)
+    state.set_mode("remote")
+    _mock_relay(monkeypatch, lambda r: httpx.Response(200, json={
+        "id": "t-1", "project_id": "p-1", "state": "completed", "result_text": "done"}))
+
+    rc = cli.main(["task", "get", "t-1"])
+
+    assert rc == 0
+    assert "changed again" not in capsys.readouterr().out
+
+
+def test_task_get_ignores_a_changed_since_pair_it_cannot_read(monkeypatch, tmp_path, capsys):
+    """A count without paths, a string where a number belongs, a `null` list. None of these can be
+    rendered into a true sentence, and the reading for all of them is the one absence already has.
+
+    Not defensive tidiness: this is the shape a proxy, a partial write or a future relay produces,
+    and the alternative is an `AttributeError` on a read whose exit code means *the task ended
+    badly* — a client-side parse fault delivered to a script as a verdict about somebody's work."""
+    _seed_running_remote_grid(monkeypatch, tmp_path)
+    state.set_mode("remote")
+    for broken in ({"changed_since_count": 2},
+                   {"changed_since_count": "2", "changed_since_paths": ["a.txt"]},
+                   {"changed_since_count": 2, "changed_since_paths": None},
+                   {"changed_since_count": True, "changed_since_paths": ["a.txt"]},
+                   {"changed_since_paths": ["a.txt"]}):
+        _mock_relay(monkeypatch, lambda r, extra=broken: httpx.Response(200, json={
+            "id": "t-1", "project_id": "p-1", "state": "completed", **extra}))
+
+        rc = cli.main(["task", "get", "t-1"])
+
+        assert rc == 0, broken
+        assert "changed again" not in capsys.readouterr().out, broken
+
+
 def test_task_list_shows_a_merge_turn_as_a_step_rather_than_as_something_somebody_typed(
         monkeypatch, tmp_path, capsys):
     """ADR 0034 D-g (issue 42). A merge turn's `prompt` is the RELAY's — `merge_tiers._MERGE_PROMPT`,
@@ -33289,6 +33394,56 @@ def test_project_status_says_a_provider_has_withdrawn_and_when_it_returns(
     assert "paused" in out.lower() or "withdraw" in out.lower()
 
 
+def test_project_status_does_not_tell_a_team_to_add_a_provider_for_a_paused_one(
+        monkeypatch, tmp_path, capsys):
+    """B7.2. The two states look identical from the outside — a queue that is not moving — and they
+    want OPPOSITE actions. Nobody is serving means find another machine; everybody is serving and
+    out of headroom means wait, and buying a second subscription for the same account would not have
+    helped. Sending a team to `grid join` for the second is the expensive kind of wrong.
+
+    The distinction lives in the code as two branches of `_print_providers`, and until this case
+    nothing held them apart: the `grid join` line could migrate into the withdrawn sentence and
+    every existing test would stay green, because they assert what IS said and not what is not.
+    """
+    _seed_running_remote_grid(monkeypatch, tmp_path)
+    state.set_mode("remote")
+    _mock_relay(monkeypatch, lambda r: httpx.Response(200, json={
+        "project_id": "P1", "member_key": "def456", "trunk": "main",
+        "main_commit": "a" * 40, "active_turns": [], "members": [],
+        "queue": {"queued": 3, "running": 0, "oldest_queued_at": "2026-08-09T09:00:00+00:00"},
+        "providers": {"online": 2, "paused": 2, "resumes_at": "2026-08-09T11:30:00+00:00"}}))
+
+    cli.main(["project", "status", "P1"])
+
+    out = capsys.readouterr().out
+    assert "withdrawn" in out, "the reason the queue is not moving went missing entirely"
+    assert "headroom" in out, "a team is told work is stuck without being told why"
+    assert "grid join" not in out, (
+        f"a team whose providers are out of subscription headroom is being told to add another "
+        f"provider, which does not help and costs money:\n{out}")
+
+
+def test_project_status_DOES_say_to_add_one_when_nobody_is_serving(
+        monkeypatch, tmp_path, capsys):
+    """The positive control, and B7.2 means nothing without it: the advice has to appear in the one
+    state it is right for, or the case above would pass on a build that never gives advice at all."""
+    _seed_running_remote_grid(monkeypatch, tmp_path)
+    state.set_mode("remote")
+    _mock_relay(monkeypatch, lambda r: httpx.Response(200, json={
+        "project_id": "P1", "member_key": "def456", "trunk": "main",
+        "main_commit": "a" * 40, "active_turns": [], "members": [],
+        "queue": {"queued": 3, "running": 0, "oldest_queued_at": "2026-08-09T09:00:00+00:00"},
+        "providers": {"online": 0, "paused": 0, "resumes_at": None}}))
+
+    cli.main(["project", "status", "P1"])
+
+    out = capsys.readouterr().out
+    assert "grid join" in out, out
+    assert "withdrawn" not in out, (
+        "a fleet with nobody online is being described as withdrawn, which tells a team to wait for "
+        "a return that is not coming")
+
+
 def test_project_status_stays_quiet_about_a_fleet_that_is_all_serving(
         monkeypatch, tmp_path, capsys):
     """Nothing is wrong, so nothing is said. A line reading "0 paused" on every healthy poll is the
@@ -37829,3 +37984,116 @@ def test_leaving_under_json_prints_the_relays_document_and_nothing_else(monkeypa
     assert cli.main(["project", "leave", "P1", "--yes", "--json"]) == 0
 
     assert json.loads(capsys.readouterr().out) == _leave_reply()
+
+
+# --- issue 58: `grid join` says why it will not serve tasks ---------------------------------------
+
+
+def _child_env(spawned: dict) -> dict[str, str]:
+    """What the detached serve child was handed. `_mock_remote_spawn` keeps it beside the argv."""
+    assert "env" in spawned, "the join did not spawn a serve child"
+    return spawned["env"]
+
+
+def _the_child_would_claim_tasks(spawned: dict) -> bool:
+    """Ask the child's OWN predicate against the child's OWN environment.
+
+    Never `env["GRID_TASKS"] == "0"`: that is a second reading of the opt-in living in a test, and
+    it would keep passing after somebody widened or narrowed the spellings `serving_enabled` accepts.
+    """
+    from unittest import mock
+
+    from remote import task_opt_in
+
+    with mock.patch.dict(os.environ, _child_env(spawned), clear=True):
+        return task_opt_in.serving_enabled()
+
+
+def test_remote_join_without_the_task_opt_in_asks_the_provider_nothing(monkeypatch, tmp_path):
+    """Opt-in is off by default and costs nothing when it is off — no `claude --version`, no probe
+    of a workspace root nobody configured."""
+    from remote import task_agent
+
+    _seed_running_remote_grid(monkeypatch, tmp_path)
+    _mock_remote_spawn(monkeypatch)
+    monkeypatch.delenv("GRID_TASKS", raising=False)
+    monkeypatch.setattr(task_agent, "preflight_before_serving",
+                        lambda: pytest.fail("the join checked task serving nobody asked for"))
+
+    assert cli.main(["join", "--serve", "m"]) == 0
+
+
+@pytest.mark.parametrize("fault", [
+    # What each check actually raises, and one that nothing raises on purpose.
+    RuntimeError,   # `resolve_binary` — not installed, or below the floor with the sandbox on
+    SystemExit,     # ⚠️ `task_sandbox`'s clean-error idiom. An `except Exception` alone MISSES this
+                    # one, and missing it aborts the whole join — inference included
+    OSError,        # the workspace root, and `link_transcript`'s containment refusals
+    ValueError,     # a malformed `GRID_TASK_ENV_PASSTHROUGH` or config directory
+    KeyError,       # a BUG in one of the checks, which must still cost inference nothing
+])
+def test_remote_join_with_a_broken_task_config_still_serves_inference(
+        monkeypatch, tmp_path, capsys, fault):
+    """The whole point of the issue. Refusing the join would take a working inference provider down
+    over a task misconfiguration — including for the operator who keeps `GRID_TASKS=1` in a shell
+    profile — so the join lands, the child serves, and only task serving is withheld."""
+    from remote import task_agent
+
+    _seed_running_remote_grid(monkeypatch, tmp_path)
+    spawned = _mock_remote_spawn(monkeypatch)
+    monkeypatch.setenv("GRID_TASKS", "1")
+    monkeypatch.setattr(task_agent, "preflight_before_serving", _raise(
+        fault("Claude Code isn't installed on this provider; install it with: curl -fsSL …")))
+
+    assert cli.main(["join", "--serve", "m"]) == 0
+
+    record = cli.provider._read_records("n1")["remote"]
+    assert record["models"] == ["m"], "the inference join did not land"
+    assert spawned["cmd"][-3:-2] == ["__remote-engine"], "no serve child was spawned"
+    err = capsys.readouterr().err
+    assert "install it with" in err, f"the refusal does not say what to fix: {err!r}"
+    assert "respawn" in err, f"the refusal does not say how to retry it: {err!r}"
+    assert not _the_child_would_claim_tasks(spawned), (
+        "the child was spawned still claiming tasks it cannot finish")
+
+
+def test_remote_join_with_a_working_task_config_says_task_serving_is_on(monkeypatch, tmp_path, capsys):
+    """Today the banner says nothing about tasks in any case, so an operator who typed the variable
+    correctly gets exactly the same output as one who typed it wrong."""
+    from remote import task_agent
+
+    _seed_running_remote_grid(monkeypatch, tmp_path)
+    spawned = _mock_remote_spawn(monkeypatch)
+    monkeypatch.setenv("GRID_TASKS", "1")
+    monkeypatch.setattr(task_agent, "preflight_before_serving", lambda: None)
+
+    assert cli.main(["join", "--serve", "m"]) == 0
+
+    assert "tasks=on" in capsys.readouterr().out
+    assert _the_child_would_claim_tasks(spawned), "the opt-in did not reach the child"
+
+
+def test_the_task_check_runs_before_the_join_stops_or_spawns_anything(monkeypatch, tmp_path):
+    """⚠️ The ordering IS the property. A provider that was serving inference a second ago must
+    still be serving it after a task check that failed — so the check may not run from inside the
+    lock, after the prior child has been terminated on the way to finding out."""
+    from remote import task_agent
+    from shared import run_records
+
+    _seed_running_remote_grid(monkeypatch, tmp_path)
+    spawned = _mock_remote_spawn(monkeypatch)
+    monkeypatch.setenv("GRID_TASKS", "1")
+    order: list[str] = []
+    monkeypatch.setattr(task_agent, "preflight_before_serving", lambda: order.append("checked"))
+    real_terminate = run_records.terminate_recorded
+    monkeypatch.setattr(run_records, "terminate_recorded",
+                        lambda prior: (order.append("stopped"), real_terminate(prior))[1])
+    monkeypatch.setattr(cli.remote_provider, "_spawn_remote_engine", lambda *a, **k: (
+        order.append("spawned"), type("P", (), {"pid": 4242})())[1])
+
+    assert cli.main(["join", "--serve", "m"]) == 0
+
+    assert order and order[0] == "checked", (
+        f"the join acted before it asked whether this provider can run a task: {order}")
+    assert "spawned" in order, "the join never spawned, so this proves nothing about the order"
+    assert spawned is not None
diff --git a/tests/test_task_agent.py b/tests/test_task_agent.py
index 987a568..7ad7743 100644
--- a/tests/test_task_agent.py
+++ b/tests/test_task_agent.py
@@ -1587,45 +1587,73 @@ def test_a_misconfigured_deadline_falls_back_instead_of_retiring_task_serving(mo
     assert "1h" in capsys.readouterr().err
 
 
-def test_tool_activity_is_published_while_the_agent_is_still_running(agent):
+def test_tool_activity_is_published_while_the_agent_is_still_running(agent, tmp_path):
     """Issue 03's first acceptance criterion — "while it is still running, not after".
 
-    Proven by ordering against the child's own clock: the tool call is emitted, then the child sleeps
-    before writing anything else. A publisher that buffered to the end would see the tool event only
-    after that sleep, so asserting the event arrived BEFORE the child exited is the whole test.
+    Proven by ORDERING against a gate this test holds shut, not by a wall clock. The child emits the
+    tool call and then blocks until the test creates `gate`, so it cannot reach its result line — let
+    alone exit — while the gate is shut. An event that arrives in that window therefore arrived while
+    the agent was still running, on any machine at any load.
 
-    The sleep and the bound are deliberately a factor of THREE apart. A threshold sitting exactly on
-    the child's own sleep has no margin in either direction: it fails on any scheduling hiccup, and
-    the temptation is then to widen it until it no longer distinguishes a live publisher from a
-    buffered one. Live is ~0.05s and buffered is ~3s, so 1.5s separates them with room on both sides.
+    ⚠️ The bound this replaces was `at < 1.5s` against a child that slept 3s, and it read as a
+    regression under the very load this feature creates: 2.90s with the machine at load 9.03 and 29
+    `claude` processes of its own test round (ND-19). The property it named still held — 2.90s was
+    still inside the child's sleep — so the threshold was wrong, not the publisher. Any absolute
+    bound here measures the machine as much as the code; a gate measures only the code.
+
+    Both controls are machine-checked rather than eyeballed:
+    · a publisher that BUFFERED to the end delivers nothing until the child exits, and the child
+      cannot exit while the gate is shut, so the wait below runs out and the test fails;
+    · `task.result` must be ABSENT from the same snapshot — that is what proves the child was still
+      running rather than already finished, which an arrival time alone can never establish.
     """
+    import threading
     import time
 
     from remote import tasks
 
+    gate = tmp_path / "release-the-child"
     seen = []
     agent(
         "printf '{\"type\":\"assistant\",\"message\":{\"content\":[{\"type\":\"tool_use\","
         "\"id\":\"t1\",\"name\":\"Edit\",\"input\":{\"file_path\":\"/w/app.py\"}}]}}\\n'\n"
-        "sleep 3\n"
+        f"while [ ! -f '{gate}' ]; do sleep 0.05; done\n"
         "printf '{\"type\":\"result\",\"subtype\":\"success\",\"is_error\":false,"
         "\"result\":\"done\"}\\n'\n"
     )
 
-    started = time.monotonic()
-    outcome = tasks.run_task(
-        _job(), publish=lambda kind, **f: seen.append((kind, f, time.monotonic() - started)))
-    whole_run = time.monotonic() - started
+    finished = {}
+    worker = threading.Thread(
+        target=lambda: finished.update(outcome=tasks.run_task(
+            _job(), publish=lambda kind, **f: seen.append((kind, f)))),
+        daemon=True)
+    worker.start()
+    try:
+        # Generous, and bounded only so a broken publisher fails instead of hanging: the event
+        # arrives in ~0.2s idle and took 2.90s under the worst load measured. It is not a property
+        # of the code under test — every assertion below reads the snapshot, not the clock.
+        deadline = time.monotonic() + 15
+        while (time.monotonic() < deadline and worker.is_alive()
+               and not any(kind == "task.tool_use" for kind, _ in seen)):
+            time.sleep(0.02)
+        while_the_child_was_blocked = list(seen)
+    finally:
+        # In a finally so a failed assertion costs a second rather than the run's whole deadline.
+        gate.write_text("go", encoding="utf-8")
 
-    assert outcome.state == "completed"
-    tool_events = [e for e in seen if e[0] == "task.tool_use"]
-    assert tool_events, f"no tool activity was published at all: {seen}"
-    kind, fields, at = tool_events[0]
-    assert (fields["tool"], fields["path"]) == ("Edit", "/w/app.py")
-    assert at < 1.5, f"the tool call surfaced only after the child's 3s sleep ({at:.2f}s)"
-    # The child really did outlive the event — without this the bound above would also be satisfied
-    # by a child that exited immediately, which proves nothing about publishing DURING a run.
-    assert whole_run > 2.5, f"the child did not actually sleep ({whole_run:.2f}s)"
+    worker.join(timeout=30)
+    assert not worker.is_alive(), "the run never finished after the gate was opened"
+
+    tool_events = [e for e in while_the_child_was_blocked if e[0] == "task.tool_use"]
+    assert tool_events, (
+        "no tool activity was published while the child was blocked — a publisher that buffers to "
+        f"the end looks exactly like this: {while_the_child_was_blocked}")
+    assert tool_events[0][1]["tool"] == "Edit"
+    assert tool_events[0][1]["path"] == "/w/app.py"
+    assert not any(kind == "task.result" for kind, _ in while_the_child_was_blocked), (
+        "the child had already finished, so this proves nothing about publishing DURING a run: "
+        f"{while_the_child_was_blocked}")
+    assert finished["outcome"].state == "completed"
 
 
 def test_a_large_burst_neither_stalls_nor_drops_events(agent):
@@ -5426,3 +5454,96 @@ def test_checkout_result_still_refuses_when_there_is_no_branch(tmp_path):
     with pytest.raises(task_repo.CheckoutError):
         task_repo.checkout_result(tmp_path / "d", url=remote.url, token="tok",
                                   branch="", commit=commit)
+
+
+# --- issue 58: the provider is asked before a member is waiting -----------------------------------
+
+
+def test_preflight_before_serving_asks_what_a_claim_would_ask(monkeypatch, tmp_path):
+    """The delegation, pinned. `grid join` must not grow a second opinion about any of this."""
+    from remote import task_agent
+
+    monkeypatch.setenv("GRID_TASK_ROOT", str(tmp_path / "root"))
+    asked: list[str] = []
+    monkeypatch.setattr(task_agent, "preflight", lambda: asked.append("preflight"))
+    monkeypatch.setattr(task_agent, "resolve_binary", lambda: asked.append("binary") or "claude")
+
+    task_agent.preflight_before_serving()
+
+    assert asked == ["preflight", "binary"]
+
+
+def test_preflight_before_serving_refuses_a_workspace_root_it_cannot_write(monkeypatch, tmp_path):
+    """The macOS shape: `/var` is root-owned, so a provider without sudo fails EVERY task with
+    "could not create /var/grid/…" — one member's task at a time, forever, every signal green."""
+    from remote import task_agent
+
+    denied = tmp_path / "denied"
+    denied.mkdir(mode=0o500)
+    monkeypatch.setenv("GRID_TASK_ROOT", str(denied / "grid" / "tasks"))
+    monkeypatch.setattr(task_agent, "preflight", lambda: None)
+    monkeypatch.setattr(task_agent, "resolve_binary", lambda: "claude")
+
+    with pytest.raises(OSError) as excinfo:
+        task_agent.preflight_before_serving()
+
+    assert "GRID_TASK_ROOT" in str(excinfo.value), (
+        f"the refusal does not say what to change: {excinfo.value}")
+    assert str(denied) in str(excinfo.value), (
+        f"the refusal does not say which directory refused it: {excinfo.value}")
+
+
+def test_preflight_before_serving_accepts_a_root_that_does_not_exist_yet(monkeypatch, tmp_path):
+    """The positive control, and the ordinary case: a root under a writable parent is fine, and
+    nothing is created here — which directory the root should be, and with what mode, is issue 62's
+    decision and this must not pre-empt it."""
+    from remote import task_agent
+
+    root = tmp_path / "grid" / "tasks"
+    monkeypatch.setenv("GRID_TASK_ROOT", str(root))
+    monkeypatch.setattr(task_agent, "preflight", lambda: None)
+    monkeypatch.setattr(task_agent, "resolve_binary", lambda: "claude")
+
+    task_agent.preflight_before_serving()
+
+    assert not root.exists(), "the probe created the root; that is issue 62's decision to make"
+
+
+def test_preflight_before_serving_refuses_a_root_that_is_a_file(monkeypatch, tmp_path):
+    """A root that exists and is not a directory is a different fault from an unwritable one, and
+    saying "not writable" about it would send an operator to `chmod`."""
+    from remote import task_agent
+
+    root = tmp_path / "root"
+    root.write_text("not a directory", encoding="utf-8")
+    monkeypatch.setenv("GRID_TASK_ROOT", str(root))
+    monkeypatch.setattr(task_agent, "preflight", lambda: None)
+    monkeypatch.setattr(task_agent, "resolve_binary", lambda: "claude")
+
+    with pytest.raises(OSError) as excinfo:
+        task_agent.preflight_before_serving()
+
+    assert "not a directory" in str(excinfo.value).lower(), str(excinfo.value)
+
+
+def test_preflight_before_serving_does_not_move_the_version_floor(monkeypatch, tmp_path):
+    """⚠️ The floor is enforced only while the sandbox is ON, and that is the SHAPE of the rule
+    rather than a convenience — it protects a control that fails open, so an operator who turned the
+    control off deliberately gets the provider that existed before issue 23, older agent included.
+
+    Asking earlier may not quietly tighten it. This is the check that would be "fixed" back.
+
+    ⚠️ A REAL binary reporting a real old version, resolved the real way — not a patched
+    `_binary_version`. With the sandbox off that reader is never reached, so patching it would have
+    left this test green against a `preflight_before_serving` that grew a version check of its own
+    through some other door, which is exactly the regression it exists to catch.
+    """
+    from remote import task_agent, task_sandbox
+
+    binary, _ = _claude_reporting(tmp_path, "2.0.1 (Claude Code)")
+    _resolving_to(monkeypatch, binary)
+    monkeypatch.setenv(task_sandbox.SANDBOX_ENV, "0")
+    monkeypatch.setenv("GRID_TASK_ROOT", str(tmp_path / "root"))
+    monkeypatch.setattr(task_agent, "preflight", lambda: None)
+
+    task_agent.preflight_before_serving()  # an ancient binary, and the sandbox is off: allowed
diff --git a/tests/test_task_evict.py b/tests/test_task_evict.py
index e20f106..f1c7e9c 100644
--- a/tests/test_task_evict.py
+++ b/tests/test_task_evict.py
@@ -221,6 +221,53 @@ def test_a_held_workspace_counts_against_the_cap_rather_than_costing_a_colder_on
         f"expected the held one and the newest to survive a cap of two, got {remaining}")
 
 
+def test_three_held_workspaces_against_a_cap_of_one_leaves_three(
+        tmp_path, short_task_root, monkeypatch):
+    """E4.3 — the all-held shape, which the case above states in prose and never runs.
+
+    ⚠️ **What is unique here is the RELEASE assertion, not the count.** Measured by mutation, and the
+    result was the opposite of what this case was written for:
+
+      * *three remain* is NOT a discriminator. Under a sweep that wrongly counted a skip as progress
+        the answer is still three — there is nothing colder to evict, so the fault is invisible in
+        this shape. The two cases above catch that one, because they have a cold workspace to lose.
+      * *stopping at the first held workspace* is caught here and by both of them.
+      * **releasing a reservation the sweep never took is caught by NOTHING ELSE.** That is the
+        expensive one: eviction takes the same reservation a worker takes
+        (`tasks._reserve_workspace`), so a spurious release hands away a workspace somebody is
+        running a turn inside — one registry, one owner, and this is what keeps it that way.
+
+    The `asked` count is the harness's own control: without it the answer *three remain* is also
+    what a sweep that never looked at anything produces, which is exactly how a bound that has
+    quietly stopped working reads.
+    """
+    from remote import task_evict
+
+    monkeypatch.setenv(task_agent_root_env(), str(short_task_root))
+    monkeypatch.setenv(task_evict.MAX_WORKSPACES_ENV, "1")
+    for conversation in _CONVERSATIONS[:3]:
+        _real_workspace(tmp_path, short_task_root, conversation)
+
+    asked, released = [], []
+
+    def _all_busy(triple):
+        asked.append(triple)
+        return False
+
+    task_evict.sweep(short_task_root, keep=None,
+                     reserve=_all_busy, release=lambda triple: released.append(triple))
+
+    assert _conversation_dirs(short_task_root) == sorted(_CONVERSATIONS[:3]), (
+        "a workspace somebody is holding was collected, or a colder one was evicted to pay for one "
+        "the sweep could not touch")
+    assert len(asked) >= 2, (
+        f"the sweep stopped after {len(asked)} candidate(s), so it treated a skip as progress "
+        f"towards the cap rather than moving on to the next one")
+    assert released == [], (
+        "a reservation that was never taken was released, which on a shared registry hands away a "
+        "workspace a worker is actually running in")
+
+
 def test_the_conversation_a_turn_is_about_to_run_in_is_never_evicted(
         tmp_path, short_task_root, monkeypatch):
     """`keep`, which covers what the reservation cannot.
diff --git a/tests/test_task_lease.py b/tests/test_task_lease.py
index cf443b1..9fa1bc8 100644
--- a/tests/test_task_lease.py
+++ b/tests/test_task_lease.py
@@ -985,6 +985,45 @@ def test_the_task_view_names_the_conversation_a_follow_up_is_addressed_to():
         "cancel or fetch")
 
 
+def test_the_single_turn_read_says_when_a_turns_work_no_longer_stands():
+    """ND-16/F-3: `changed_since_count` / `changed_since_paths` on `GET /tasks/{id}`.
+
+    ⚠️ **A rename on the relay side is SILENT here, in the direction that costs somebody their
+    work.** `cli/remote_task._changed_since_note` requires both keys, type-checked, and prints
+    NOTHING otherwise — the correct degrade for a relay predating this slice, and byte-identical to
+    a relay that renamed them. So a drift raises nothing, warns nobody and fails no request: the
+    warning simply stops appearing, and the person whose line a colleague's turn dropped goes back
+    to reading `completed` on every surface they have.
+
+    ⚠️ **Pinned on `get_task` and NOT on `task_view`, deliberately.** The pair is added by the route
+    rather than by the shared view because the answer costs two git reads, and `task_view` is also
+    what `GET /tasks` builds every row of — a listing of forty turns would pay eighty. A future
+    tidy-up that moved it into the view would make the listing quietly expensive, so this check
+    names the place the cost is bounded.
+
+    Rollout is **relay before CLI**, with the project routes: an old relay sends neither key and the
+    CLI says nothing, which is the pre-slice behaviour exactly.
+    """
+    import ast
+
+    source = _relay_module("tasks.py")
+    if not source.exists():
+        pytest.skip("grid-src worktree is not beside this one; the lockstep cannot be checked here")
+    found = [node for node in ast.walk(ast.parse(source.read_text()))
+             if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == "get_task"]
+    assert found, (
+        "get_task is no longer defined in grid-src's tasks.py — it was renamed, so teach this check "
+        "the new name rather than deleting it")
+    literals = {node.value for node in ast.walk(found[0])
+                if isinstance(node, ast.Constant) and isinstance(node.value, str)}
+
+    for key in ("changed_since_count", "changed_since_paths"):
+        assert key in literals, (
+            f"grid-src's `GET /tasks/{{id}}` no longer reports `{key}`, so `grid task get` silently "
+            f"stops saying that the project may no longer hold what a turn changed — the CLI reads "
+            f"a missing key as 'an old relay' and prints nothing at all")
+
+
 def test_the_status_view_reports_where_a_member_stands_against_their_running_cap():
     """ADR 0034 D-i (issue 49): `member_running_turns` / `member_running_cap` on the status view.
 
diff --git a/tests/test_task_opt_in.py b/tests/test_task_opt_in.py
new file mode 100644
index 0000000..cd3f710
--- /dev/null
+++ b/tests/test_task_opt_in.py
@@ -0,0 +1,156 @@
+"""The task opt-in is read in ONE place (issue 57).
+
+`GRID_TASKS` and `GRID_MAX_TASKS` began life inside `remote/serve.py`, which is the serve child's
+own module. Two processes need the answers now — the child at startup, and the CLI parent before it
+spawns that child (issues 58, 60, 61) — and the parent cannot ask `serve` for them.
+
+The hazard this file exists for is not the import: it is the second READING. A parent that
+re-implements "is `GRID_TASKS` on" compiles, passes every test, and then gets edited apart from the
+copy in the serve child, which is the silent-divergence failure this repository keeps paying for.
+So the rule is enforced by a scan, not by a convention.
+
+⚠️ **The scan measures a reading, not a mention.** `remote/task_evict.py` names `GRID_MAX_TASKS` in
+its module docstring to say what bounds what, and `docs/` names both everywhere — those are
+cross-references, and a rule that forced them out would make the tree less legible to enforce a
+property they do not violate. So the scan parses each file and looks at STRING LITERALS IN CODE,
+with docstrings excluded.
+"""
+from __future__ import annotations
+
+import ast
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+_REPO = Path(__file__).resolve().parent.parent
+# The packages shipped in the wheel. ⚠️ `build/lib/` and `.venv/` hold *copies* of this tree; walking
+# either would count every reading twice and make the scan unfixable.
+_SOURCE_PACKAGES = ("cli", "local", "remote", "shared", "doggi", "train")
+# A walk that visits nothing reports "read in one place" for a tree that reads it in ten. Well below
+# the real count, so it pins the walk without breaking on ordinary growth.
+_MINIMUM_FILES_WALKED = 100
+
+_OPT_IN_MODULE = "remote/task_opt_in.py"
+
+
+def _source_files() -> list[Path]:
+    files: list[Path] = []
+    for package in _SOURCE_PACKAGES:
+        files.extend(sorted((_REPO / package).rglob("*.py")))
+    return files
+
+
+def _code_string_literals(path: Path) -> set[str]:
+    """Every string literal in `path` that is CODE. Docstrings are prose and are excluded.
+
+    A name in prose is a cross-reference and must stay readable; a name in a code literal is a
+    reading of the variable, and that is what may exist only once.
+    """
+    tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+    prose: set[int] = set()
+    for node in ast.walk(tree):
+        if not isinstance(node, (ast.Module, ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
+            continue
+        body = node.body
+        if (body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant)
+                and isinstance(body[0].value.value, str)):
+            prose.add(id(body[0].value))
+    return {
+        node.value for node in ast.walk(tree)
+        if isinstance(node, ast.Constant) and isinstance(node.value, str) and id(node) not in prose
+    }
+
+
+def _files_reading(literal: str, files: list[Path] | None = None) -> list[str]:
+    """Every source file that spells `literal` in code, as repo-relative paths."""
+    found = []
+    for path in (files if files is not None else _source_files()):
+        if literal in _code_string_literals(path):
+            relative = path.relative_to(_REPO) if path.is_relative_to(_REPO) else path
+            found.append(str(relative).replace("\\", "/"))
+    return sorted(found)
+
+
+def test_the_walk_really_visited_the_source():
+    """The floor. Every assertion below is "found in exactly one file", and a walk that found
+    nothing at all reports the same thing — with the rule completely unenforced."""
+    walked = _source_files()
+    assert len(walked) >= _MINIMUM_FILES_WALKED, (
+        f"the walk visited only {len(walked)} source files; it is not walking the tree")
+    for package in _SOURCE_PACKAGES:
+        assert any(path.is_relative_to(_REPO / package) for path in walked), (
+            f"the walk contributed no file from {package}/")
+    assert not any("build/lib" in str(path) or ".venv" in str(path) for path in walked), (
+        "the walk reached a COPY of the tree; every reading would be counted twice")
+
+
+def test_every_source_file_could_actually_be_PARSED():
+    """A file the scan cannot parse is a file it cannot police, and skipping one in silence is how
+    an enforcement scan comes to certify a tree it never read."""
+    unparseable = []
+    for path in _source_files():
+        try:
+            _code_string_literals(path)
+        except (SyntaxError, UnicodeDecodeError) as exc:
+            unparseable.append(f"{path.relative_to(_REPO)}: {exc}")
+    assert not unparseable, "the scan could not read:\n  " + "\n  ".join(unparseable)
+
+
+@pytest.mark.parametrize("literal", ["GRID_TASKS", "GRID_MAX_TASKS"])
+def test_the_task_opt_in_is_read_in_exactly_one_module(literal):
+    """One reading, two callers. A second spelling anywhere in the source is the divergence."""
+    reading = _files_reading(literal)
+    assert reading == [_OPT_IN_MODULE], (
+        f"{literal} is read outside {_OPT_IN_MODULE}: {reading}. Import it from there instead — two "
+        f"readings of one environment variable get edited apart, silently.")
+
+
+def test_the_scan_catches_a_planted_second_reading(tmp_path):
+    """The positive control. Every assertion above is "nothing else was found", which is exactly
+    what a broken matcher also reports."""
+    planted = tmp_path / "planted.py"
+    planted.write_text('import os\nopt_in = os.getenv("GRID_TASKS")\n', encoding="utf-8")
+
+    assert _files_reading("GRID_TASKS", [planted]) == [str(planted).replace("\\", "/")]
+
+
+def test_the_scan_does_not_fire_on_a_MENTION(tmp_path):
+    """The control for the exclusion above — and it is a real file, not a hypothetical one:
+    `remote/task_evict.py` names `GRID_MAX_TASKS` in its docstring to say what bounds what."""
+    prose = tmp_path / "prose.py"
+    prose.write_text(
+        '"""`GRID_TASKS` turns this on."""\n# and GRID_MAX_TASKS sizes the pool\n', encoding="utf-8")
+
+    assert _files_reading("GRID_TASKS", [prose]) == []
+    assert _files_reading("GRID_MAX_TASKS", [prose]) == []
+
+
+def test_the_scan_matches_the_whole_NAME(tmp_path):
+    """`GRID_TASK_ROOT` is a different variable that shares a prefix with `GRID_TASKS`. A substring
+    match would report the provider's workspace root as a second reading of the opt-in."""
+    other = tmp_path / "other.py"
+    other.write_text('import os\nroot = os.getenv("GRID_TASK_ROOT")\n', encoding="utf-8")
+
+    assert _files_reading("GRID_TASKS", [other]) == []
+
+
+def test_the_opt_in_module_does_not_pull_the_provider_runtime():
+    """Asserted, not assumed. The whole point of the move is that a CLI command can ask this
+    question without importing the serve child's world.
+
+    A subprocess, because in-process `sys.modules` is whatever the rest of the suite already
+    imported — measured: `remote.serve` costs 83 ms and 273 modules and pulls httpx, against 16 ms
+    and 119 modules for `remote.task_agent`.
+    """
+    probe = (
+        "import sys, remote.task_opt_in; "
+        "print('httpx' in sys.modules, 'remote.serve' in sys.modules, 'remote.relay' in sys.modules)"
+    )
+    result = subprocess.run(
+        [sys.executable, "-c", probe], cwd=_REPO, capture_output=True, text=True, timeout=60)
+
+    assert result.returncode == 0, result.stderr
+    assert result.stdout.strip() == "False False False", (
+        f"the opt-in module dragged the provider runtime in with it: {result.stdout.strip()}")