diff --git a/CHANGELOG.md b/CHANGELOG.md index d519d9d..befa929 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/dyro/cli.py b/src/dyro/cli.py index c065836..f265ced 100644 --- a/src/dyro/cli.py +++ b/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) diff --git a/src/dyro/console/_inspect_worker.py b/src/dyro/console/_inspect_worker.py index fc37326..f014077 100644 --- a/src/dyro/console/_inspect_worker.py +++ b/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,9 +112,9 @@ 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 ( @@ -110,13 +122,15 @@ def _parse_child_result( 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], }, diff --git a/src/dyro/console/overview.py b/src/dyro/console/overview.py index 1d24635..72ce3b3 100644 --- a/src/dyro/console/overview.py +++ b/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 { diff --git a/src/dyro/continuation/ready_briefing.py b/src/dyro/continuation/ready_briefing.py index f5928a2..6a26fa3 100644 --- a/src/dyro/continuation/ready_briefing.py +++ b/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) diff --git a/src/dyro/hub.py b/src/dyro/hub.py index 489ea36..c3dda3c 100644 --- a/src/dyro/hub.py +++ b/src/dyro/hub.py @@ -84,6 +84,40 @@ def unique_registered_alias(name: str, names: tuple[str, ...]) -> str | None: return None +def same_workspace_root(left: Path | None, right: Path | None) -> bool: + """True when both paths resolve to the same workspace root.""" + if left is None or right is None: + return False + try: + return Path(left).resolve() == Path(right).resolve() + except OSError: + return False + + +def workspace_alias_retargets_root( + alias: str, + root: Path, + workspaces: tuple[WorkspaceRecord, ...] | None = None, +) -> bool: + """True when ``--workspace alias`` uniquely fold-resolves to another root. + + A miss or fold collision is not a working selector for a different root. + Registry read failures cannot prove the selector stays on this root. + """ + if not isinstance(alias, str) or not alias: + return False + try: + records = ( + workspaces if workspaces is not None else load_registry().workspaces + ) + except (DyroError, ValidationError, OSError, TypeError, AttributeError): + return True + matches = workspace_alias_matches(records, alias) + if len(matches) != 1: + return False + return not same_workspace_root(matches[0].root, root) + + class WorkspaceAliasCollisionError(DyroError): """More than one registered alias folds to the same lookup key.""" diff --git a/tests/support.py b/tests/support.py index 2d2156e..07fb46b 100644 --- a/tests/support.py +++ b/tests/support.py @@ -1,9 +1,11 @@ from __future__ import annotations +import os from pathlib import Path import subprocess import tempfile import unittest +from unittest.mock import patch CONFIG = '''schema_version = 1 @@ -63,15 +65,23 @@ class WorkspaceCase(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory(prefix="dyro-test-") self.root = Path(self.tmp.name) + self.registry_tmp = tempfile.TemporaryDirectory(prefix="dyro-registry-") + self.registry_environment = patch.dict( + os.environ, {"DYRO_HOME": self.registry_tmp.name}, clear=False + ) + self.registry_environment.start() (self.root / "dyro.toml").write_text(CONFIG, encoding="utf-8") self.anchor = self.root / "repositories/api" self.anchor.mkdir(parents=True) shell("git", "init", "-b", "main", cwd=self.anchor) shell("git", "config", "user.name", "Test User", cwd=self.anchor) shell("git", "config", "user.email", "test@example.com", cwd=self.anchor) + shell("git", "config", "commit.gpgsign", "false", cwd=self.anchor) (self.anchor / "README.md").write_text("anchor\n", encoding="utf-8") shell("git", "add", "README.md", cwd=self.anchor) shell("git", "commit", "-m", "chore: initial", cwd=self.anchor) def tearDown(self) -> None: + self.registry_environment.stop() + self.registry_tmp.cleanup() self.tmp.cleanup() diff --git a/tests/test_console_inspection.py b/tests/test_console_inspection.py index 4ab09e4..2379b85 100644 --- a/tests/test_console_inspection.py +++ b/tests/test_console_inspection.py @@ -844,6 +844,50 @@ def test_fold_twin_cards_do_not_advertise_fail_closed_workspace_selector(self) - self.assertNotIn(str(self.root), command) self.assertNotIn(str(other), command) + def test_isolated_root_console_does_not_advertise_other_root_fold_match( + self, + ) -> None: + from dyro.hub import remove_workspace + + remove_workspace("demo") + (self.root / "dyro.toml").write_text( + (self.root / "dyro.toml") + .read_text(encoding="utf-8") + .replace('name = "test-workspace"', 'name = "Demo"') + .replace( + 'mount = "services/api"', + 'mount = "services/api"\nremote = "https://example.invalid/api.git"', + ), + encoding="utf-8", + ) + other = self.root.parent / f"{self.root.name}-fold-other" + other.mkdir() + other.joinpath("dyro.toml").write_text( + (self.root / "dyro.toml") + .read_text(encoding="utf-8") + .replace('name = "Demo"', 'name = "other"'), + encoding="utf-8", + ) + (other / "repositories/api").mkdir(parents=True) + add_workspace(self.root, name="current", make_default=True) + add_workspace(other, name="demo") + self.anchor.rename(self.root / "api-missing") + service = IsolatedOverviewService( + registry_state_home=self.home, + timeout_seconds=5, + cursor_secret=b"q" * 32, + target_root=self.root, + ) + + page = service.page() + cards = page["data"]["workspaces"] + self.assertEqual(len(cards), 1) + command = cards[0]["recommendation"]["command"] + self.assertNotIn("--workspace demo", command) + self.assertNotIn("--workspace Demo", command) + self.assertNotIn("bootstrap --yes", command) + self.assertNotIn(str(other), command) + def test_isolated_summary_worker_passes_next_commands_loader(self) -> None: from dyro.config import load from dyro.continuation.next_step import next_commands diff --git a/tests/test_hub.py b/tests/test_hub.py index 513b83f..b7292c1 100644 --- a/tests/test_hub.py +++ b/tests/test_hub.py @@ -1,5 +1,6 @@ from __future__ import annotations +import argparse from contextlib import redirect_stderr, redirect_stdout from io import StringIO import json @@ -10,7 +11,13 @@ import unittest from unittest.mock import patch -from dyro.cli import _route_experiment_surface, build_parser, main +from dyro.cli import ( + _print_json_observation_timeout, + _route_experiment_surface, + _timeout_repair_commands, + build_parser, + main, +) from dyro.config import load from dyro.home import ( HomeTarget, @@ -33,6 +40,7 @@ remove_workspace, set_default_workspace, unique_registered_alias, + workspace_alias_retargets_root, ) from dyro.tooling import ( ToolPreferences, @@ -174,6 +182,28 @@ def test_alias_fold_helpers_distinguish_unique_and_collision(self) -> None: self.assertIsNone(unique_registered_alias("demo", colliding)) self.assertIsNone(unique_registered_alias("missing", names)) + def test_unique_fold_of_profile_name_retargets_when_root_differs(self) -> None: + other = self._second_workspace("other") + add_workspace(self.workspace, name="current", make_default=True) + add_workspace(other, name="demo") + self.assertTrue( + workspace_alias_retargets_root("Demo", self.workspace.resolve()) + ) + self.assertFalse( + workspace_alias_retargets_root("demo", other.resolve()) + ) + self.assertFalse( + workspace_alias_retargets_root("current", self.workspace.resolve()) + ) + self.assertFalse(workspace_alias_retargets_root("missing", self.workspace)) + + def test_registry_read_failure_cannot_prove_workspace_stays_on_root(self) -> None: + add_workspace(self.workspace, name="current", make_default=True) + with patch("dyro.hub.load_registry", side_effect=OSError("registry unread")): + self.assertTrue( + workspace_alias_retargets_root("Demo", self.workspace.resolve()) + ) + def test_get_workspace_fails_closed_on_case_fold_collision(self) -> None: from dyro.errors import DyroError @@ -1696,6 +1726,203 @@ def test_implicit_json_next_does_not_advertise_colliding_alias(self) -> None: self.assertNotIn("--workspace demo", command) self.assertTrue(any("--root" in item for item in advertised)) + def _profile_named(self, root: Path, name: str, *, remote: bool = False) -> None: + text = root.joinpath("dyro.toml").read_text(encoding="utf-8") + text = text.replace('name = "test-workspace"', f'name = "{name}"') + if remote and 'remote = "' not in text: + text = text.replace( + 'mount = "services/api"', + 'mount = "services/api"\nremote = "https://example.invalid/api.git"', + ) + root.joinpath("dyro.toml").write_text(text, encoding="utf-8") + + def _fold_retarget_registry(self) -> Path: + """Default A named Demo; unique fold of Demo is registered at another root.""" + self._profile_named(self.root, "Demo", remote=True) + other = self.root.parent / f"{self.root.name}-fold-other" + other.mkdir() + other.joinpath("dyro.toml").write_text( + (self.root / "dyro.toml") + .read_text(encoding="utf-8") + .replace('name = "Demo"', 'name = "other"'), + encoding="utf-8", + ) + (other / "repositories/api").mkdir(parents=True) + add_workspace(self.root, name="current", make_default=True) + add_workspace(other, name="demo") + self.anchor.rename(self.root / "api-missing") + return other + + def _advertised_next(self, argv: list[str]) -> tuple[dict[str, object], list[str]]: + output = StringIO() + with redirect_stdout(output): + main(argv) + payload = json.loads(output.getvalue()) + briefing = payload.get("briefing") or {} + briefing_command = ( + briefing.get("command") if isinstance(briefing, dict) else None + ) + advertised = [ + item + for item in ( + *(payload.get("commands") or []), + *(payload.get("diagnostic_commands") or []), + briefing_command, + ) + if isinstance(item, str) + ] + return payload, advertised + + def test_implicit_and_root_next_do_not_advertise_other_root_fold_match( + self, + ) -> None: + other = self._fold_retarget_registry() + unrelated = self.root.parent / f"{self.root.name}-unrelated" + unrelated.mkdir() + previous = Path.cwd() + try: + os.chdir(unrelated) + implicit, implicit_ads = self._advertised_next(["next", "--format", "json"]) + finally: + os.chdir(previous) + rooted, root_ads = self._advertised_next( + ["--root", str(self.root), "next", "--format", "json"] + ) + current_root = str(self.root.resolve()) + other_root = str(other.resolve()) + for payload, advertised in ( + (implicit, implicit_ads), + (rooted, root_ads), + ): + self.assertEqual(payload["kind"], "next_step") + self.assertEqual(payload["state"], "needs_repair") + self.assertTrue(advertised, payload) + joined = "\n".join(advertised) + self.assertNotIn("--workspace demo", joined) + self.assertNotIn("--workspace Demo", joined) + self.assertNotIn(other_root, joined) + self.assertNotIn("dyro --workspace demo bootstrap --yes", advertised) + self.assertNotIn("dyro --workspace Demo bootstrap --yes", advertised) + self.assertNotIn("dyro --workspace demo doctor", advertised) + self.assertNotIn("dyro --workspace Demo doctor", advertised) + self.assertTrue( + any("--root" in item and current_root in item for item in advertised), + advertised, + ) + for command in advertised: + if "bootstrap" in command or "doctor" in command: + self.assertIn("--root", command) + self.assertIn(current_root, command) + + def _timeout_args(self, **overrides: object) -> argparse.Namespace: + values = { + "workspace_alias": "Demo", + "root": None, + "format": "json", + "command": "next", + "all": False, + "_control_plane_read_budget": None, + "_stashed_findings": None, + } + values.update(overrides) + return argparse.Namespace(**values) + + def _assert_no_other_root_fold_ad( + self, advertised: list[str], other: Path + ) -> None: + joined = "\n".join(advertised) + self.assertNotIn("--workspace demo", joined) + self.assertNotIn("--workspace Demo", joined) + self.assertNotIn(str(other.resolve()), joined) + self.assertNotIn("dyro --workspace demo bootstrap --yes", advertised) + self.assertNotIn("dyro --workspace Demo bootstrap --yes", advertised) + self.assertNotIn("dyro --workspace demo doctor", advertised) + self.assertNotIn("dyro --workspace Demo doctor", advertised) + + def test_timeout_fallback_after_config_fail_does_not_advertise_other_root_fold( + self, + ) -> None: + other = self._fold_retarget_registry() + args = self._timeout_args() + with patch("dyro.cli._config", side_effect=OSError("profile unread")): + commands = _timeout_repair_commands(args, ["FAIL observation deadline"]) + output = StringIO() + with redirect_stdout(output): + _print_json_observation_timeout(args) + payload = json.loads(output.getvalue()) + advertised = [ + item + for item in ( + *commands, + *(payload.get("commands") or []), + *(payload.get("diagnostic_commands") or []), + ) + if isinstance(item, str) + ] + self._assert_no_other_root_fold_ad(advertised, other) + self.assertTrue(advertised) + self.assertTrue(all("doctor" in item for item in advertised), advertised) + self.assertTrue( + all("--workspace" not in item for item in advertised), advertised + ) + rooted = self._timeout_args(root=str(self.root)) + with patch("dyro.cli._config", side_effect=OSError("profile unread")): + rooted_commands = _timeout_repair_commands( + rooted, ["FAIL observation deadline"] + ) + self._assert_no_other_root_fold_ad(rooted_commands, other) + if rooted_commands: + self.assertTrue( + any( + "--root" in item and str(self.root) in item + for item in rooted_commands + ), + rooted_commands, + ) + + def test_registry_read_failure_uses_root_not_other_workspace_fold(self) -> None: + other = self._fold_retarget_registry() + from dyro.continuation.ready_briefing import scoped_briefing_command + + config = load(self.root) + with ( + patch( + "dyro.continuation.ready_briefing.load_registry", + side_effect=OSError("registry unread"), + ), + patch("dyro.hub.load_registry", side_effect=OSError("registry unread")), + ): + command = scoped_briefing_command(config, "Demo", "bootstrap", "--yes") + rooted, advertised = self._advertised_next( + ["--root", str(self.root), "next", "--format", "json"] + ) + self.assertIn("--root", command) + self.assertIn(str(config.root), command) + self.assertNotIn("--workspace Demo", command) + self.assertNotIn("--workspace demo", command) + self.assertNotIn(str(other.resolve()), command) + self.assertEqual(rooted["kind"], "next_step") + self._assert_no_other_root_fold_ad(advertised, other) + self.assertTrue( + any("--root" in item and str(self.root) in item for item in advertised), + advertised, + ) + + def test_unavailable_summary_with_root_omits_other_root_fold_workspace( + self, + ) -> None: + other = self._fold_retarget_registry() + from dyro.console._inspect_worker import _unavailable_summary + from dyro.console.overview import WORKSPACE_TIMEOUT + + summary = _unavailable_summary( + "Demo", WORKSPACE_TIMEOUT, ("Demo",), root=self.root + ) + command = summary["recommendation"]["command"] + self.assertIsInstance(command, str) + self._assert_no_other_root_fold_ad([command], other) + self.assertNotIn(str(self.root), command) + def test_console_unique_fold_plan_and_apply_share_canonical_alias(self) -> None: add_workspace(self.root, name="Demo") output = StringIO()