fix(工作区): implicit next must not retarget via config.name fold

CHANGELOG.md

@@ -12,11 +12,14 @@
   collide. Bridge workspace list loads each row by stored root so a
   fold-twin does not mark the other stale. Implicit `next` and ready
   briefing never advertise a `--workspace` selector that would
-  fail-close; they use `--root` instead. `console` plan and apply share
-  that resolve (canonical on a unique fold; fail-closed on twins).
-  Console overview recommendations omit a colliding `--workspace` ad.
-  The operator copy path does not invent `--workspace … doctor` when
-  that field is blank.
+  fail-close; they use `--root` instead. Implicit and `--root` next /
+  briefing / repair also refuse a unique fold of the current profile
+  name when that selector would resolve to a different registered root;
+  those ads use `--root` for the current workspace. `console` plan and
+  apply share that resolve (canonical on a unique fold; fail-closed on
+  twins). Console overview recommendations omit a colliding or
+  cross-root `--workspace` ad. The operator copy path does not invent
+  `--workspace … doctor` when that field is blank.
 - JSON `doctor` / `status` / `next` no longer share a flat 5s observation
   deadline with Bridge. Those commands start at the documented 45s ceiling
   (not 5s) so a ~50+ worktree workspace that the text path finishes in ~7s

src/dyro/cli.py

@@ -50,7 +50,6 @@
     repair_commands,
 )
 from .continuation.ready_briefing import (
-    briefing_command,
     build_ready_briefing,
     scoped_briefing_command,
 )
@@ -665,6 +664,31 @@ def _timeout_status_rows(args: argparse.Namespace) -> list[object]:
     return rows
 
 
+def _timeout_verified_root(args: argparse.Namespace) -> Path | None:
+    """Return this invocation's root when it is already known, never a fold alias."""
+    resolved = getattr(args, "_control_plane_resolution", None)
+    profile = getattr(resolved, "profile", None)
+    config = getattr(profile, "config", None)
+    root = getattr(config, "root", None)
+    if root is not None:
+        return Path(root)
+    raw = getattr(args, "root", None)
+    if isinstance(raw, str) and raw.strip():
+        path = Path(raw).expanduser()
+        if not path.is_absolute():
+            path = Path.cwd() / path
+        return path
+    return None
+
+
+def _timeout_unscoped_repair_commands(args: argparse.Namespace) -> list[str]:
+    """Repair ads after config reload fails. Never emit a fold-matching --workspace."""
+    root = _timeout_verified_root(args)
+    if root is None:
+        return [shlex.join(("dyro", "doctor"))]
+    return [shlex.join(("dyro", "--root", str(root), "doctor"))]
+
+
 def _timeout_repair_commands(
     args: argparse.Namespace, findings: list[str]
 ) -> list[str]:
@@ -675,7 +699,7 @@ def _timeout_repair_commands(
         commands = deadline_repair_commands(config, alias, failures)
         return commands or [_briefing_command(args, config, "doctor")]
     except (DyroError, OSError, ValidationError, TypeError, AttributeError):
-        return [briefing_command(alias, "doctor")]
+        return _timeout_unscoped_repair_commands(args)
 
 
 def _print_json_observation_timeout(args: argparse.Namespace) -> None:
@@ -729,7 +753,7 @@ def _print_json_observation_timeout(args: argparse.Namespace) -> None:
     try:
         diagnostic_commands = [_briefing_command(args, _config(args), "doctor")]
     except (DyroError, OSError, ValidationError, TypeError, AttributeError):
-        diagnostic_commands = [briefing_command(workspace, "doctor")]
+        diagnostic_commands = _timeout_unscoped_repair_commands(args)
     _print_control_plane_json(
         "next_step",
         state="needs_repair",
@@ -827,7 +851,7 @@ def _scoped_command(
 def _briefing_command(
     args: argparse.Namespace, config: Config, *command: str
 ) -> str:
-    """Scope a read-only briefing command without a fail-closed selector."""
+    """Scope a briefing command without a fail-closed or cross-root selector."""
     alias = getattr(args, "workspace_alias", None) or config.name
     return scoped_briefing_command(config, str(alias), *command)
 

src/dyro/console/_inspect_worker.py

@@ -41,9 +41,15 @@
 
 
 def _unavailable_summary(
-    alias: str, code: str, names: tuple[str, ...] = ()
+    alias: str,
+    code: str,
+    names: tuple[str, ...] = (),
+    *,
+    root: Path | None = None,
 ) -> dict[str, object]:
-    return unavailable_workspace_summary(alias, False, reason=code, names=names)
+    return unavailable_workspace_summary(
+        alias, False, reason=code, names=names, root=root
+    )
 
 
 def _capture_workspace_summary(
@@ -63,7 +69,10 @@ def _capture_workspace_summary(
             result_queue.put(
                 {
                     "summary": _unavailable_summary(
-                        record.name, WORKSPACE_MISSING_ROOT, names
+                        record.name,
+                        WORKSPACE_MISSING_ROOT,
+                        names,
+                        root=record.root,
                     ),
                     "warnings": [WORKSPACE_MISSING_ROOT],
                 }
@@ -85,7 +94,10 @@ def _capture_workspace_summary(
         result_queue.put(
             {
                 "summary": _unavailable_summary(
-                    record.name, WORKSPACE_UNAVAILABLE, names
+                    record.name,
+                    WORKSPACE_UNAVAILABLE,
+                    names,
+                    root=record.root,
                 ),
                 "warnings": [WORKSPACE_UNAVAILABLE],
             }
@@ -100,23 +112,25 @@ def _parse_child_result(
     names: tuple[str, ...] = (),
 ) -> tuple[dict[str, object], set[str]]:
     if not isinstance(value, dict):
-        return _unavailable_summary(record.name, WORKSPACE_UNAVAILABLE, names), {
-            WORKSPACE_UNAVAILABLE
-        }
+        return _unavailable_summary(
+            record.name, WORKSPACE_UNAVAILABLE, names, root=record.root
+        ), {WORKSPACE_UNAVAILABLE}
     summary = value.get("summary")
     warnings = value.get("warnings")
     if (
         not isinstance(summary, dict)
         or not isinstance(warnings, list)
         or not all(isinstance(item, str) for item in warnings)
     ):
-        return _unavailable_summary(record.name, WORKSPACE_UNAVAILABLE, names), {
-            WORKSPACE_UNAVAILABLE
-        }
+        return _unavailable_summary(
+            record.name, WORKSPACE_UNAVAILABLE, names, root=record.root
+        ), {WORKSPACE_UNAVAILABLE}
     copied = dict(summary)
     copied["alias"] = record.name
     copied["is_default"] = is_default
-    return omit_colliding_workspace_command(copied, names), set(warnings)
+    return omit_colliding_workspace_command(
+        copied, names, root=record.root
+    ), set(warnings)
 
 
 def _isolated_summaries(
@@ -152,7 +166,10 @@ def finish(record: WorkspaceRecord, value: object, *, default: bool) -> None:
                     record,
                     {
                         "summary": _unavailable_summary(
-                            record.name, WORKSPACE_TIMEOUT, names
+                            record.name,
+                            WORKSPACE_TIMEOUT,
+                            names,
+                            root=record.root,
                         ),
                         "warnings": [WORKSPACE_TIMEOUT],
                     },
@@ -164,7 +181,10 @@ def finish(record: WorkspaceRecord, value: object, *, default: bool) -> None:
                     record,
                     {
                         "summary": _unavailable_summary(
-                            record.name, WORKSPACE_TIMEOUT, names
+                            record.name,
+                            WORKSPACE_TIMEOUT,
+                            names,
+                            root=record.root,
                         ),
                         "warnings": [WORKSPACE_TIMEOUT],
                     },
@@ -216,7 +236,7 @@ def finish(record: WorkspaceRecord, value: object, *, default: bool) -> None:
                         record,
                         {
                             "summary": _unavailable_summary(
-                                record.name, code, names
+                                record.name, code, names, root=record.root
                             ),
                             "warnings": [code],
                         },

src/dyro/console/overview.py

@@ -27,6 +27,7 @@
     WorkspaceRegistry,
     alias_fold_collides,
     load_registry,
+    workspace_alias_retargets_root,
 )
 from ..continuation.briefing import follow_up_from_kind
 from ..updates import UpdateState, classify_update, load_update_state
@@ -146,22 +147,32 @@ def workspace_root_missing(root: Path) -> bool:
         return False
 
 
-def _workspace_ad(alias: str, *parts: str, names: tuple[str, ...]) -> str:
-    """Return a ``--workspace`` command only when that selector would resolve."""
+def _workspace_ad(
+    alias: str,
+    *parts: str,
+    names: tuple[str, ...],
+    root: Path | None = None,
+) -> str:
+    """Return a ``--workspace`` command only when that selector stays here."""
     if not isinstance(alias, str) or alias_fold_collides(alias, names):
         return ""
+    if root is not None and workspace_alias_retargets_root(alias, root):
+        return ""
     return " ".join(("dyro", "--workspace", alias, *parts))
 
 
 def omit_colliding_workspace_command(
-    summary: dict[str, object], names: tuple[str, ...]
+    summary: dict[str, object],
+    names: tuple[str, ...],
+    root: Path | None = None,
 ) -> dict[str, object]:
-    """Blank a fail-closed ``--workspace`` ad after list-by-root capture."""
+    """Blank a fail-closed or cross-root ``--workspace`` ad after capture."""
     alias = summary.get("alias")
     recommendation = summary.get("recommendation")
     if not isinstance(alias, str) or not isinstance(recommendation, dict):
         return summary
-    if not alias_fold_collides(alias, names):
+    retargets = root is not None and workspace_alias_retargets_root(alias, root)
+    if not alias_fold_collides(alias, names) and not retargets:
         return summary
     command = recommendation.get("command")
     if not isinstance(command, str) or "--workspace" not in command:
@@ -177,6 +188,7 @@ def unavailable_workspace_summary(
     *,
     reason: str,
     names: tuple[str, ...] = (),
+    root: Path | None = None,
 ) -> dict[str, object]:
     """Path-free unread card. Isolated still requires an allowlisted command."""
     safe_alias = _safe_code(alias)
@@ -202,7 +214,7 @@ def unavailable_workspace_summary(
         "attention_counts": _empty_attention_counts(),
         "recommendation": {
             "reason": code,
-            "command": _workspace_ad(safe_alias, "doctor", names=names),
+            "command": _workspace_ad(safe_alias, "doctor", names=names, root=root),
         },
         "findings": [],
         "snapshot_sha256": "",
@@ -786,6 +798,7 @@ def _capture(
                     is_default,
                     reason=reason,
                     names=self._registry_names(),
+                    root=root,
                 ),
                 {reason},
                 _empty_inventory(),
@@ -839,7 +852,11 @@ def _capture(
             "task_status_counts": dict(sorted(task_status_counts.items())),
             "attention_counts": attention["counts"],
             "recommendation": self._recommendation(
-                safe_alias, attention["items"], findings=findings, commands=commands
+                safe_alias,
+                attention["items"],
+                findings=findings,
+                commands=commands,
+                root=getattr(config, "root", None) or root,
             ),
             "findings": findings,
             "snapshot_sha256": str(envelope.get("snapshot_sha256", "")),
@@ -903,10 +920,11 @@ def _recommendation(
         attention: object,
         findings: object = None,
         commands: object = None,
+        root: Path | None = None,
     ) -> dict[str, str] | None:
         names = self._registry_names()
         collide = alias_fold_collides(alias, names)
-        doctor = _workspace_ad(alias, "doctor", names=names)
+        doctor = _workspace_ad(alias, "doctor", names=names, root=root)
         next_command = ""
         if isinstance(commands, list) and not collide:
             for raw in commands:
@@ -936,6 +954,7 @@ def _recommendation(
             alias,
             *follow_up_from_kind(_safe_code(item.get("kind")), objective_id),
             names=names,
+            root=root,
         )
         command = _console_command(follow_up, alias) or next_command or doctor
         return {

src/dyro/continuation/ready_briefing.py

@@ -6,7 +6,12 @@
 
 from ..config import Config
 from ..errors import DyroError, ValidationError
-from ..hub import alias_fold_collides, load_registry, unique_registered_alias
+from ..hub import (
+    alias_fold_collides,
+    load_registry,
+    unique_registered_alias,
+    workspace_alias_retargets_root,
+)
 from ..read_limits import ReadBudget, ReadLimitError
 from .briefing import (
     briefing_payload,
@@ -24,25 +29,38 @@ def briefing_command(alias: str, *command: str) -> str:
     return shlex.join(("dyro", "--workspace", alias, *command))
 
 
+def _root_scoped_command(config: Config, *command: str) -> str:
+    return shlex.join(("dyro", "--root", str(config.root), *command))
+
+
 def scoped_briefing_command(
     config: Config,
     alias: str,
     *command: str,
     names: tuple[str, ...] | None = None,
 ) -> str:
-    """Advertise ``--workspace`` only when that selector would resolve.
+    """Advertise ``--workspace`` only when that selector stays on this root.
 
-    A unique fold uses the canonical registered spelling. An unregistered
-    profile name keeps ``--workspace`` so path-free next ads stay path-free.
-    A fold collision fail-closes at resolve, so the ad switches to ``--root``.
+    A unique fold uses the canonical registered spelling when that record is
+    the current workspace. An unregistered profile name keeps ``--workspace``
+    so path-free next ads stay path-free. A fold collision, a unique fold
+    that would resolve to a different root, or a registry read that cannot
+    prove the selector stays here, switches the ad to ``--root``.
     """
-    registered = (
-        names
-        if names is not None
-        else tuple(item.name for item in load_registry().workspaces)
-    )
+    try:
+        records = tuple(load_registry().workspaces)
+    except (DyroError, ValidationError, OSError, TypeError, AttributeError):
+        if getattr(config, "root", None) is not None:
+            return _root_scoped_command(config, *command)
+        records = ()
+    registered = names if names is not None else tuple(item.name for item in records)
     if alias_fold_collides(alias, registered):
-        return shlex.join(("dyro", "--root", str(config.root), *command))
+        return _root_scoped_command(config, *command)
+    root = getattr(config, "root", None)
+    if root is not None and workspace_alias_retargets_root(
+        alias, root, workspaces=records or None
+    ):
+        return _root_scoped_command(config, *command)
     canonical = unique_registered_alias(alias, registered) or alias
     return briefing_command(canonical, *command)