diff --git a/CHANGELOG.md b/CHANGELOG.md index aba608c..d519d9d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,21 @@ ## Unreleased +- Resolve `--workspace` aliases case-insensitively when exactly one + registered name folds to the same key, and keep that canonical spelling + in outputs. Two registered aliases that differ only by case fail closed + with the colliding names listed; a total miss still suggests nearby + names. Implicit JSON resolve does not stamp a `--workspace` selector. + `workspace default` / `workspace remove` plan and apply share that + policy; remove can still target each exact registered name when twins + 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. - 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/bridge/observations.py b/src/dyro/bridge/observations.py index 30ca3a2..d9f5cb5 100644 --- a/src/dyro/bridge/observations.py +++ b/src/dyro/bridge/observations.py @@ -9,6 +9,7 @@ from ..config import Config from ..continuation.resolution import ( WorkspaceResolutionError, + load_registered_profile, resolve_workspace_readonly, ) from ..continuation.store import get_objective, list_objectives @@ -78,12 +79,7 @@ def list_workspaces_observation(*, budget: ReadBudget | None = None) -> dict[str failures: list[dict[str, str]] = [] for record in registry.workspaces: try: - profile = resolve_workspace_readonly( - start=None, - workspace=record.name, - cwd=Path("/"), - budget=limits, - ).profile + profile = load_registered_profile(record, limits) items.append( { "alias": record.name, diff --git a/src/dyro/cli.py b/src/dyro/cli.py index 3206d00..c065836 100644 --- a/src/dyro/cli.py +++ b/src/dyro/cli.py @@ -49,7 +49,11 @@ deadline_repair_commands, repair_commands, ) -from .continuation.ready_briefing import briefing_command, build_ready_briefing +from .continuation.ready_briefing import ( + briefing_command, + build_ready_briefing, + scoped_briefing_command, +) from .continuation.engine import ( build_scheduler_tick, render_scheduler_tick_text, @@ -343,12 +347,20 @@ def _config(args: argparse.Namespace) -> Config: cwd=Path.cwd().absolute(), budget=budget, ) + if ( + workspace_arg + and resolved.source is WorkspaceResolutionSource.EXPLICIT + and resolved.registry_alias is not None + ): + setattr(args, "workspace_alias", resolved.registry_alias) setattr(args, "_control_plane_resolution", resolved) return resolved.profile.config if root_arg: root = Path(root_arg).expanduser() elif workspace_arg: - root = get_workspace(workspace_arg).root + record = get_workspace(workspace_arg) + setattr(args, "workspace_alias", record.name) + root = record.root else: interactive = sys.stdin.isatty() and sys.stdout.isatty() return resolve_workspace( @@ -661,9 +673,9 @@ def _timeout_repair_commands( try: config = _config(args) commands = deadline_repair_commands(config, alias, failures) + return commands or [_briefing_command(args, config, "doctor")] except (DyroError, OSError, ValidationError, TypeError, AttributeError): - commands = [briefing_command(alias, "doctor")] - return commands or [briefing_command(alias, "doctor")] + return [briefing_command(alias, "doctor")] def _print_json_observation_timeout(args: argparse.Namespace) -> None: @@ -714,12 +726,16 @@ def _print_json_observation_timeout(args: argparse.Namespace) -> None: return commands = _timeout_repair_commands(args, findings) failures = [item for item in findings if item.startswith("FAIL")] + try: + diagnostic_commands = [_briefing_command(args, _config(args), "doctor")] + except (DyroError, OSError, ValidationError, TypeError, AttributeError): + diagnostic_commands = [briefing_command(workspace, "doctor")] _print_control_plane_json( "next_step", state="needs_repair", summary="工作区还不能开始任务。", commands=commands, - diagnostic_commands=[briefing_command(workspace, "doctor")], + diagnostic_commands=diagnostic_commands, mutation_available=False, partial=True, findings=[ @@ -811,9 +827,9 @@ def _scoped_command( def _briefing_command( args: argparse.Namespace, config: Config, *command: str ) -> str: - """Scope a read-only briefing command without embedding --root paths.""" + """Scope a read-only briefing command without a fail-closed selector.""" alias = getattr(args, "workspace_alias", None) or config.name - return briefing_command(str(alias), *command) + return scoped_briefing_command(config, str(alias), *command) def _workspace_ready_briefing( @@ -1926,8 +1942,16 @@ def cmd_console(args: argparse.Namespace) -> None: initial_workspace = getattr(args, "workspace_alias", None) root_arg = getattr(args, "root", None) target_root: Path | None = None + if root_arg: + if args.dry_run: + target_root = Path(root_arg).expanduser().absolute() + else: + config = load(Path(root_arg).expanduser()) + target_root = config.root + initial_workspace = config.name + elif initial_workspace: + initial_workspace = get_workspace(initial_workspace).name if args.dry_run: - target_root = Path(root_arg).expanduser().absolute() if root_arg else None render_console_plan( port=args.port, no_open=args.no_open, @@ -1935,12 +1959,6 @@ def cmd_console(args: argparse.Namespace) -> None: target_root=target_root, ) return - if root_arg: - config = load(Path(root_arg).expanduser()) - target_root = config.root - initial_workspace = config.name - elif initial_workspace: - get_workspace(initial_workspace) launch_console( port=args.port, no_open=args.no_open, @@ -2022,23 +2040,23 @@ def cmd_workspace_list(args: argparse.Namespace) -> None: def cmd_workspace_default(args: argparse.Namespace) -> None: + record = get_workspace(args.name) if args.dry_run: - get_workspace(args.name) - print(f"DRY RUN: 将默认工作区设为 {args.name}") + print(f"DRY RUN: 将默认工作区设为 {record.name}") return - set_default_workspace(args.name) - print(f"默认工作区：{args.name}") + set_default_workspace(record.name) + print(f"默认工作区：{record.name}") def cmd_workspace_remove(args: argparse.Namespace) -> None: - get_workspace(args.name) + record = get_workspace(args.name, exact_on_collision=True) if not args.yes and not args.dry_run: raise DyroError("移除只会删除全局首页入口，不会删除项目文件；确认后请加 --yes") if args.dry_run: - print(f"DRY RUN: 将移除工作区入口 {args.name}；不会删除项目文件") + print(f"DRY RUN: 将移除工作区入口 {record.name}；不会删除项目文件") return - remove_workspace(args.name) - print(f"已移除工作区入口：{args.name}；项目文件未改动") + remove_workspace(record.name) + print(f"已移除工作区入口：{record.name}；项目文件未改动") def _blueprint_document(args: argparse.Namespace): diff --git a/src/dyro/console/_inspect_worker.py b/src/dyro/console/_inspect_worker.py index 3f48854..fc37326 100644 --- a/src/dyro/console/_inspect_worker.py +++ b/src/dyro/console/_inspect_worker.py @@ -27,6 +27,7 @@ WORKSPACE_MISSING_ROOT, WORKSPACE_TIMEOUT, WORKSPACE_UNAVAILABLE, + omit_colliding_workspace_command, unavailable_workspace_summary, workspace_root_missing, ) @@ -39,31 +40,41 @@ _WORKER_RESPONSE_LIMIT = 2 * 1024 * 1024 -def _unavailable_summary(alias: str, code: str) -> dict[str, object]: - return unavailable_workspace_summary(alias, False, reason=code) +def _unavailable_summary( + alias: str, code: str, names: tuple[str, ...] = () +) -> dict[str, object]: + return unavailable_workspace_summary(alias, False, reason=code, names=names) def _capture_workspace_summary( result_queue: Any, record: WorkspaceRecord, is_default: bool, + registry: WorkspaceRegistry | None = None, ) -> None: """Capture one workspace only, returning a JSON-safe value through IPC.""" + names = ( + tuple(item.name for item in registry.workspaces) + if registry is not None + else (record.name,) + ) try: if workspace_root_missing(record.root): result_queue.put( { - "summary": _unavailable_summary(record.name, WORKSPACE_MISSING_ROOT), + "summary": _unavailable_summary( + record.name, WORKSPACE_MISSING_ROOT, names + ), "warnings": [WORKSPACE_MISSING_ROOT], } ) return - registry = WorkspaceRegistry( + capture_registry = registry or WorkspaceRegistry( default=record.name if is_default else "", workspaces=(record,), ) service = ConsoleOverviewService( - registry_loader=lambda: registry, + registry_loader=lambda: capture_registry, commands_loader=next_commands, ) payload = service.workspace(record.name) @@ -73,17 +84,23 @@ def _capture_workspace_summary( except Exception: result_queue.put( { - "summary": _unavailable_summary(record.name, WORKSPACE_UNAVAILABLE), + "summary": _unavailable_summary( + record.name, WORKSPACE_UNAVAILABLE, names + ), "warnings": [WORKSPACE_UNAVAILABLE], } ) def _parse_child_result( - value: object, record: WorkspaceRecord, *, is_default: bool + value: object, + record: WorkspaceRecord, + *, + is_default: bool, + names: tuple[str, ...] = (), ) -> tuple[dict[str, object], set[str]]: if not isinstance(value, dict): - return _unavailable_summary(record.name, WORKSPACE_UNAVAILABLE), { + return _unavailable_summary(record.name, WORKSPACE_UNAVAILABLE, names), { WORKSPACE_UNAVAILABLE } summary = value.get("summary") @@ -93,13 +110,13 @@ 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), { + return _unavailable_summary(record.name, WORKSPACE_UNAVAILABLE, names), { WORKSPACE_UNAVAILABLE } copied = dict(summary) copied["alias"] = record.name copied["is_default"] = is_default - return copied, set(warnings) + return omit_colliding_workspace_command(copied, names), set(warnings) def _isolated_summaries( @@ -115,9 +132,12 @@ def _isolated_summaries( summaries: list[dict[str, object]] = [] warnings: set[str] = set() deadline = time.monotonic() + total_timeout + names = tuple(item.name for item in registry.workspaces) def finish(record: WorkspaceRecord, value: object, *, default: bool) -> None: - summary, codes = _parse_child_result(value, record, is_default=default) + summary, codes = _parse_child_result( + value, record, is_default=default, names=names + ) summaries.append(summary) warnings.update(codes) @@ -131,7 +151,9 @@ def finish(record: WorkspaceRecord, value: object, *, default: bool) -> None: finish( record, { - "summary": _unavailable_summary(record.name, WORKSPACE_TIMEOUT), + "summary": _unavailable_summary( + record.name, WORKSPACE_TIMEOUT, names + ), "warnings": [WORKSPACE_TIMEOUT], }, default=record.name == registry.default, @@ -141,7 +163,9 @@ def finish(record: WorkspaceRecord, value: object, *, default: bool) -> None: finish( record, { - "summary": _unavailable_summary(record.name, WORKSPACE_TIMEOUT), + "summary": _unavailable_summary( + record.name, WORKSPACE_TIMEOUT, names + ), "warnings": [WORKSPACE_TIMEOUT], }, default=record.name == registry.default, @@ -153,7 +177,12 @@ def finish(record: WorkspaceRecord, value: object, *, default: bool) -> None: result_queue = context.Queue(maxsize=1) process = context.Process( target=_capture_workspace_summary, - args=(result_queue, record, record.name == registry.default), + args=( + result_queue, + record, + record.name == registry.default, + registry, + ), daemon=True, ) process.start() @@ -186,7 +215,9 @@ def finish(record: WorkspaceRecord, value: object, *, default: bool) -> None: finish( record, { - "summary": _unavailable_summary(record.name, code), + "summary": _unavailable_summary( + record.name, code, names + ), "warnings": [code], }, default=record.name == registry.default, diff --git a/src/dyro/console/assets.py b/src/dyro/console/assets.py index dbf8c4b..d5e46e7 100644 --- a/src/dyro/console/assets.py +++ b/src/dyro/console/assets.py @@ -27,8 +27,8 @@ class ConsoleAsset: ), "app.js": ( "text/javascript; charset=utf-8", - "d3a2787202bd1b42dc924c7b7417dedaf2d3415f1efd5bcfab68b63f010bbe07", - 97664, + "859472c4513e7ce0f11ed3cd6bf8e8b8fe5ee1295916999bc5de2417189f60dd", + 97461, ), "styles.css": ( "text/css; charset=utf-8", diff --git a/src/dyro/console/assets/app.js b/src/dyro/console/assets/app.js index 06b5feb..d84f8c6 100644 --- a/src/dyro/console/assets/app.js +++ b/src/dyro/console/assets/app.js @@ -534,21 +534,16 @@ function recommendedCommand(summary) { const unread = unavailableReason(summary); if (unread === "missing_root" || unread === "read_timeout") return ""; const command = text(summary && summary.recommendation && summary.recommendation.command); - const doctor = `dyro --workspace ${alias} doctor`; const yes = "--" + "yes"; const push = "--" + "push"; - if (workspaceHasFail(summary)) { - if ( - command - && !isBareWorkspaceCommand(command, alias) - && !command.includes(yes) - && !command.includes(push) - ) { - return command; - } - return doctor; + if ( + !command + || isBareWorkspaceCommand(command, alias) + || command.includes(yes) + || command.includes(push) + ) { + return ""; } - if (!command || isBareWorkspaceCommand(command, alias)) return ""; return command; } diff --git a/src/dyro/console/inspection.py b/src/dyro/console/inspection.py index 5edcffd..2c9944f 100644 --- a/src/dyro/console/inspection.py +++ b/src/dyro/console/inspection.py @@ -1125,6 +1125,8 @@ def _safe_code(value: object) -> bool: @staticmethod def _safe_command(command: object, alias: object) -> bool: + if command == "": + return True if not isinstance(command, str) or not isinstance(alias, str): return False escaped_alias = re.escape(alias) diff --git a/src/dyro/console/overview.py b/src/dyro/console/overview.py index b4c1ad3..1d24635 100644 --- a/src/dyro/console/overview.py +++ b/src/dyro/console/overview.py @@ -23,7 +23,11 @@ from ..canonical import canonical_json_bytes from ..config import Config, load, validate_id from ..errors import DyroError, ValidationError -from ..hub import WorkspaceRegistry, load_registry +from ..hub import ( + WorkspaceRegistry, + alias_fold_collides, + load_registry, +) from ..continuation.briefing import follow_up_from_kind from ..updates import UpdateState, classify_update, load_update_state from ..observations import ( @@ -142,11 +146,37 @@ 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.""" + if not isinstance(alias, str) or alias_fold_collides(alias, names): + return "" + return " ".join(("dyro", "--workspace", alias, *parts)) + + +def omit_colliding_workspace_command( + summary: dict[str, object], names: tuple[str, ...] +) -> dict[str, object]: + """Blank a fail-closed ``--workspace`` ad after list-by-root 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): + return summary + command = recommendation.get("command") + if not isinstance(command, str) or "--workspace" not in command: + return summary + copied = dict(summary) + copied["recommendation"] = {**recommendation, "command": ""} + return copied + + def unavailable_workspace_summary( alias: str, is_default: bool, *, reason: str, + names: tuple[str, ...] = (), ) -> dict[str, object]: """Path-free unread card. Isolated still requires an allowlisted command.""" safe_alias = _safe_code(alias) @@ -172,7 +202,7 @@ def unavailable_workspace_summary( "attention_counts": _empty_attention_counts(), "recommendation": { "reason": code, - "command": f"dyro --workspace {safe_alias} doctor", + "command": _workspace_ad(safe_alias, "doctor", names=names), }, "findings": [], "snapshot_sha256": "", @@ -751,7 +781,12 @@ def _capture( else WORKSPACE_UNAVAILABLE ) return ( - unavailable_workspace_summary(safe_alias, is_default, reason=reason), + unavailable_workspace_summary( + safe_alias, + is_default, + reason=reason, + names=self._registry_names(), + ), {reason}, _empty_inventory(), ) @@ -856,6 +891,12 @@ def _invoke_commands_loader(self, config: Config, alias: str) -> object: except TypeError: return loader(config) + def _registry_names(self) -> tuple[str, ...]: + try: + return tuple(item.name for item in self._load_registry().workspaces) + except ConsoleOverviewError: + return () + def _recommendation( self, alias: str, @@ -863,9 +904,11 @@ def _recommendation( findings: object = None, commands: object = None, ) -> dict[str, str] | None: - doctor = f"dyro --workspace {alias} doctor" + names = self._registry_names() + collide = alias_fold_collides(alias, names) + doctor = _workspace_ad(alias, "doctor", names=names) next_command = "" - if isinstance(commands, list): + if isinstance(commands, list) and not collide: for raw in commands: next_command = _console_command(raw, alias) if next_command: @@ -889,13 +932,10 @@ def _recommendation( if not isinstance(item, dict): return {"reason": "HOME_GUIDANCE", "command": next_command or doctor} objective_id = _safe_code(item.get("objective_id")) - follow_up = " ".join( - ( - "dyro", - "--workspace", - alias, - *follow_up_from_kind(_safe_code(item.get("kind")), objective_id), - ) + follow_up = _workspace_ad( + alias, + *follow_up_from_kind(_safe_code(item.get("kind")), objective_id), + names=names, ) command = _console_command(follow_up, alias) or next_command or doctor return { diff --git a/src/dyro/continuation/next_step.py b/src/dyro/continuation/next_step.py index 4483a48..ca1baa9 100644 --- a/src/dyro/continuation/next_step.py +++ b/src/dyro/continuation/next_step.py @@ -7,7 +7,7 @@ from ..onboarding import validate_bootstrap_destination from ..read_limits import ReadBudget, ReadLimitCode, ReadLimitError from ..workspace import OBSERVATION_DEADLINE_FINDING, doctor, list_lines -from .ready_briefing import briefing_command +from .ready_briefing import scoped_briefing_command def next_commands( @@ -48,7 +48,7 @@ def next_commands( except (DyroError, ValidationError, OSError, TypeError, AttributeError): return [] if not lines: - return [briefing_command(token, "line", "create", "dev", "--yes")] + return [scoped_briefing_command(config, token, "line", "create", "dev", "--yes")] return [] @@ -65,7 +65,7 @@ def deadline_repair_commands( if OBSERVATION_DEADLINE_FINDING not in failures: failures.append(OBSERVATION_DEADLINE_FINDING) commands = repair_commands(config, alias, failures) - return commands or [briefing_command(alias, "doctor")] + return commands or [scoped_briefing_command(config, alias, "doctor")] def repair_commands(config: Config, alias: str, failures: list[str]) -> list[str]: @@ -73,8 +73,8 @@ def repair_commands(config: Config, alias: str, failures: list[str]) -> list[str if not failures: return [] if bootstrap_repair_applicable(config, failures): - return [briefing_command(alias, "bootstrap", "--yes")] - return [briefing_command(alias, "doctor")] + return [scoped_briefing_command(config, alias, "bootstrap", "--yes")] + return [scoped_briefing_command(config, alias, "doctor")] def bootstrap_repair_applicable(config: Config, failures: list[str]) -> bool: diff --git a/src/dyro/continuation/ready_briefing.py b/src/dyro/continuation/ready_briefing.py index 2808dc7..f5928a2 100644 --- a/src/dyro/continuation/ready_briefing.py +++ b/src/dyro/continuation/ready_briefing.py @@ -6,6 +6,7 @@ from ..config import Config from ..errors import DyroError, ValidationError +from ..hub import alias_fold_collides, load_registry, unique_registered_alias from ..read_limits import ReadBudget, ReadLimitError from .briefing import ( briefing_payload, @@ -23,6 +24,29 @@ def briefing_command(alias: str, *command: str) -> str: return shlex.join(("dyro", "--workspace", alias, *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. + + 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``. + """ + registered = ( + names + if names is not None + else tuple(item.name for item in load_registry().workspaces) + ) + if alias_fold_collides(alias, registered): + return shlex.join(("dyro", "--root", str(config.root), *command)) + canonical = unique_registered_alias(alias, registered) or alias + return briefing_command(canonical, *command) + + def _read_plan( config: Config, objective_id: str, @@ -61,20 +85,22 @@ def build_ready_briefing( if record.operator_state != "stopped" ] except (DyroError, ValidationError, OSError, ReadLimitError): - command = briefing_command(alias, "objective", "list") + command = scoped_briefing_command(config, alias, "objective", "list") return unread_briefing(command), [command] if not records: return None, [] if len(records) > 1: - command = briefing_command(alias, "objective", "list") + command = scoped_briefing_command(config, alias, "objective", "list") return inventory_briefing(command, len(records)), [command] record = records[0] - explain = briefing_command(alias, "objective", "explain", record.objective.id) + explain = scoped_briefing_command( + config, alias, "objective", "explain", record.objective.id + ) try: stored, plan = _read_plan(config, record.objective.id, read_budget) except (DyroError, ValidationError, OSError, ReadLimitError): return unread_briefing(explain), [explain] - command = briefing_command(alias, *follow_up_argv(plan)) + command = scoped_briefing_command(config, alias, *follow_up_argv(plan)) return ( briefing_payload(plan, command=command, title=stored.objective.title), [command], diff --git a/src/dyro/continuation/resolution.py b/src/dyro/continuation/resolution.py index ca01326..07b17c3 100644 --- a/src/dyro/continuation/resolution.py +++ b/src/dyro/continuation/resolution.py @@ -12,12 +12,13 @@ from ..config import CONFIG_NAME, Config, LoadedProfile, load, load_profile_exact, validate_id from ..errors import DyroError, ValidationError from ..hub import ( + WorkspaceAliasCollisionError, WorkspaceRecord, get_workspace, load_registry, load_registry_bounded, looks_like_workspace_path, - unregistered_workspace_error, + select_workspace_record, workspace_path_as_alias_error, ) from ..read_limits import ReadBudget, ReadLimitCode, ReadLimitError @@ -199,6 +200,11 @@ def _bounded_registry(budget: ReadBudget): ) from exc +def load_registered_profile(record: WorkspaceRecord, budget: ReadBudget) -> LoadedProfile: + """Load one registry row by its stored root; do not re-resolve the alias.""" + return _registered_profile(record, budget) + + def _registered_profile(record: WorkspaceRecord, budget: ReadBudget) -> LoadedProfile: try: return load_profile_exact(record.root, budget) @@ -231,18 +237,18 @@ def resolve_workspace_readonly( raise workspace_path_as_alias_error(workspace) validate_id(workspace, "工作区别名") registry = _bounded_registry(budget) - matches = tuple(item for item in registry.workspaces if item.name == workspace) - if len(matches) != 1: + try: + record = select_workspace_record(registry.workspaces, workspace) + except WorkspaceAliasCollisionError as exc: + raise WorkspaceResolutionError( + WorkspaceResolutionFailure.AMBIGUOUS_WORKSPACE, + message=str(exc), + ) from exc + except DyroError as exc: raise WorkspaceResolutionError( WorkspaceResolutionFailure.WORKSPACE_NOT_REGISTERED, - message=str( - unregistered_workspace_error( - workspace, - tuple(item.name for item in registry.workspaces), - ) - ), - ) - record = matches[0] + message=str(exc), + ) from exc return ResolvedWorkspace( _registered_profile(record, budget), WorkspaceResolutionSource.EXPLICIT, diff --git a/src/dyro/hub.py b/src/dyro/hub.py index afd44a9..489ea36 100644 --- a/src/dyro/hub.py +++ b/src/dyro/hub.py @@ -57,8 +57,77 @@ def workspace_path_as_alias_error(value: str) -> ValidationError: ) +def _alias_fold(name: str) -> str: + return name.casefold() + + +def workspace_alias_matches( + workspaces: tuple[WorkspaceRecord, ...], name: str +) -> tuple[WorkspaceRecord, ...]: + """Return registered records whose aliases fold equal to ``name``.""" + key = _alias_fold(name) + return tuple(record for record in workspaces if _alias_fold(record.name) == key) + + +def alias_fold_collides(name: str, names: tuple[str, ...]) -> bool: + """True when two or more registered aliases fold equal to ``name``.""" + key = _alias_fold(name) + return sum(1 for item in names if _alias_fold(item) == key) > 1 + + +def unique_registered_alias(name: str, names: tuple[str, ...]) -> str | None: + """Return the sole registered spelling that folds equal to ``name``.""" + key = _alias_fold(name) + matches = tuple(item for item in names if _alias_fold(item) == key) + if len(matches) == 1: + return matches[0] + return None + + +class WorkspaceAliasCollisionError(DyroError): + """More than one registered alias folds to the same lookup key.""" + + +def colliding_workspace_aliases_error( + name: str, colliding: tuple[str, ...] +) -> WorkspaceAliasCollisionError: + listed = "、".join(colliding) + return WorkspaceAliasCollisionError( + f"工作区别名大小写冲突：{name} 同时匹配 {listed}" + ) + + +def select_workspace_record( + workspaces: tuple[WorkspaceRecord, ...], + name: str, + *, + exact_on_collision: bool = False, +) -> WorkspaceRecord: + """Resolve one registered alias. + + A unique case-insensitive match returns the canonical registered record. + Two or more aliases that fold equal fail closed unless + ``exact_on_collision`` is set and ``name`` equals one registered spelling. + A total miss keeps the existing unregistered suggestion. + """ + matches = workspace_alias_matches(workspaces, name) + if len(matches) == 1: + return matches[0] + if matches: + if exact_on_collision: + exact = next((record for record in matches if record.name == name), None) + if exact is not None: + return exact + raise colliding_workspace_aliases_error( + name, tuple(record.name for record in matches) + ) + raise unregistered_workspace_error(name, tuple(record.name for record in workspaces)) + + def _close_workspace_names(name: str, names: tuple[str, ...]) -> tuple[str, ...]: - exact_ci = tuple(item for item in names if item.lower() == name.lower() and item != name) + exact_ci = tuple( + item for item in names if _alias_fold(item) == _alias_fold(name) and item != name + ) if exact_ci: return exact_ci return tuple(difflib.get_close_matches(name, names, n=5, cutoff=0.4)) @@ -335,26 +404,22 @@ def ensure_workspace(path: str | Path) -> WorkspaceRecord: return add_workspace(root, name=alias, make_default=not registry.default) -def get_workspace(name: str) -> WorkspaceRecord: +def get_workspace(name: str, *, exact_on_collision: bool = False) -> WorkspaceRecord: if looks_like_workspace_path(name): raise workspace_path_as_alias_error(name) validate_id(name, "工作区别名") registry = load_registry() - try: - return next(record for record in registry.workspaces if record.name == name) - except StopIteration as exc: - raise unregistered_workspace_error( - name, tuple(record.name for record in registry.workspaces) - ) from exc + return select_workspace_record( + registry.workspaces, name, exact_on_collision=exact_on_collision + ) def set_default_workspace(name: str) -> None: validate_id(name, "工作区别名") def update(current: WorkspaceRegistry) -> WorkspaceRegistry: - if name not in {record.name for record in current.workspaces}: - raise DyroError(f"未登记工作区：{name}") - return replace(current, default=name) + selected = select_workspace_record(current.workspaces, name) + return replace(current, default=selected.name) _update_registry(update) @@ -363,14 +428,15 @@ def remove_workspace(name: str) -> None: validate_id(name, "工作区别名") def update(current: WorkspaceRegistry) -> WorkspaceRegistry: - if name not in {record.name for record in current.workspaces}: - raise DyroError(f"未登记工作区：{name}") + selected = select_workspace_record( + current.workspaces, name, exact_on_collision=True + ) remaining = tuple( - record for record in current.workspaces if record.name != name + record for record in current.workspaces if record.name != selected.name ) default = ( current.default - if current.default != name + if current.default != selected.name else (remaining[0].name if remaining else "") ) return WorkspaceRegistry(default, remaining) diff --git a/tests/support/console_operator.mjs b/tests/support/console_operator.mjs index cdf6a20..7c99be4 100644 --- a/tests/support/console_operator.mjs +++ b/tests/support/console_operator.mjs @@ -305,17 +305,83 @@ if (action === "fail_overview") { { status: "FAIL", reason: "MISSING_ORIGIN", line: "core_pay" }, { status: "FAIL", reason: "MISSING_ORIGIN", line: "release_a" }, ], - recommendation: { reason: "HOME_GUIDANCE", command: "dyro --workspace core" }, + recommendation: { reason: "HOME_GUIDANCE", command: "" }, attention_counts: emptyAttention(), }, ], }, }); + const emptyFail = { + heading: nodesById.get("overview-heading").textContent, + primary: nodesById.get("primary-command").textContent, + command: nodesById.get("primary-copy").dataset.command, + needsYou: collectText(nodesById.get("needs-you")), + }; + const uniqueCard = { + alias: "core", + display_name: "core", + availability: "available", + health: "degraded", + findings: [ + { status: "FAIL", reason: "MISSING_ORIGIN", line: "core" }, + ], + recommendation: { + reason: "MISSING_ORIGIN", + command: "dyro --workspace core doctor", + }, + attention_counts: emptyAttention(), + unavailable_reason: "", + }; + api.renderOverview({ + captured_at: "2026-08-21T07:00:00Z", + freshness: { partial: false, warnings: [] }, + data: { + total_workspaces: 1, + attention_counts: emptyAttention(), + task_status_counts: {}, + workspaces: [uniqueCard], + }, + }); + result = { + ...emptyFail, + uniqueCommand: nodesById.get("primary-copy").dataset.command, + uniquePrimary: nodesById.get("primary-command").textContent, + uniqueRecommended: api.recommendedCommand(uniqueCard), + }; +} else if (action === "fold_twin_fail_overview") { + const findings = [{ status: "FAIL", reason: "MISSING_ORIGIN", line: "core" }]; + const demo = { + alias: "Demo", + display_name: "Demo", + availability: "available", + health: "degraded", + findings, + recommendation: { reason: "MISSING_ORIGIN", command: "" }, + attention_counts: emptyAttention(), + unavailable_reason: "", + }; + const twin = { + ...demo, + alias: "demo", + display_name: "demo", + }; + api.renderOverview({ + captured_at: "2026-08-21T07:00:00Z", + freshness: { partial: false, warnings: [] }, + data: { + total_workspaces: 2, + attention_counts: emptyAttention(), + task_status_counts: {}, + workspaces: [demo, twin], + }, + }); result = { heading: nodesById.get("overview-heading").textContent, primary: nodesById.get("primary-command").textContent, command: nodesById.get("primary-copy").dataset.command, needsYou: collectText(nodesById.get("needs-you")), + recommendedDemo: api.recommendedCommand(demo), + recommendedTwin: api.recommendedCommand(twin), }; } else if (action === "tabs") { const live = api.renderLivePanes("core", { diff --git a/tests/test_bridge_resolution.py b/tests/test_bridge_resolution.py index aa55db8..d585d67 100644 --- a/tests/test_bridge_resolution.py +++ b/tests/test_bridge_resolution.py @@ -72,3 +72,21 @@ def test_list_marks_stale_registered_root_without_paths(self) -> None: self.assertTrue(payload["partial"]) self.assertNotIn(str(self.root.resolve()), str(payload)) self.assertNotIn(str(stale.resolve()), str(payload)) + + def test_list_does_not_mark_fold_twins_ambiguous(self) -> None: + add_workspace(self.root, name="Acme", make_default=True) + twin = self.root / "acme-twin" + twin.mkdir() + (twin / "dyro.toml").write_text( + CONFIG.replace('name = "test-workspace"', 'name = "acme-twin"'), + encoding="utf-8", + ) + add_workspace(twin, name="acme") + payload = list_workspaces_observation() + by_alias = {item["alias"]: item for item in payload["workspaces"]} + self.assertEqual(by_alias["Acme"]["status"], "ok") + self.assertEqual(by_alias["acme"]["status"], "ok") + self.assertFalse( + any(item["code"] == "AMBIGUOUS_WORKSPACE" for item in payload["failures"]) + ) + self.assertFalse(payload["partial"]) diff --git a/tests/test_cli.py b/tests/test_cli.py index 324d967..43e20a7 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1222,6 +1222,35 @@ def test_control_plane_next_preserves_an_explicit_workspace_selector(self) -> No payload, ) + def test_control_plane_next_uses_canonical_alias_spelling(self) -> None: + with tempfile.TemporaryDirectory(prefix="dyro-registry-") as registry_home: + with patch.dict(os.environ, {"DYRO_HOME": registry_home}, clear=False): + main( + [ + "workspace", + "add", + str(self.root), + "--name", + "Acme", + "--default", + ] + ) + output = StringIO() + with redirect_stdout(output): + main( + [ + "--workspace", + "acme", + "next", + "--format", + "json", + ] + ) + self.assertEqual(load_registry().workspaces[0].name, "Acme") + + payload = json.loads(output.getvalue()) + self.assertEqual(payload["commands"], ["dyro --workspace Acme doctor"]) + def test_control_plane_json_runtime_errors_use_one_stable_envelope(self) -> None: with tempfile.TemporaryDirectory(prefix="dyro-registry-") as registry_home: stdout = StringIO() diff --git a/tests/test_console_inspection.py b/tests/test_console_inspection.py index c5c5129..4ab09e4 100644 --- a/tests/test_console_inspection.py +++ b/tests/test_console_inspection.py @@ -761,6 +761,7 @@ def test_isolated_command_allowlist_rejects_task_next(self) -> None: self.assertFalse( IsolatedOverviewService._safe_command("dyro --workspace demo", "demo") ) + self.assertTrue(IsolatedOverviewService._safe_command("", "demo")) def test_missing_origin_fail_is_not_ready_or_a_bare_workspace_command(self) -> None: from dyro.config import load @@ -804,6 +805,45 @@ def test_missing_origin_fail_is_not_ready_or_a_bare_workspace_command(self) -> N "dyro --workspace demo", ) + def test_fold_twin_cards_do_not_advertise_fail_closed_workspace_selector(self) -> None: + from dyro.config import load + from dyro.hub import remove_workspace + from dyro.workspace import create_line + + remove_workspace("demo") + (self.root / "dyro.toml").write_text( + (self.root / "dyro.toml") + .read_text(encoding="utf-8") + .replace('name = "test-workspace"', 'name = "Demo"'), + encoding="utf-8", + ) + other = self.root.parent / f"{self.root.name}-twin" + other.mkdir() + other.joinpath("dyro.toml").write_text( + (self.root / "dyro.toml") + .read_text(encoding="utf-8") + .replace('name = "Demo"', 'name = "demo"'), + encoding="utf-8", + ) + add_workspace(self.root, name="Demo", make_default=True) + add_workspace(other, name="demo") + create_line(load(self.root), line_id="core", branch="feat/core", base="main") + service = IsolatedOverviewService( + registry_state_home=self.home, + timeout_seconds=5, + cursor_secret=b"q" * 32, + ) + + page = service.page() + cards = page["data"]["workspaces"] + self.assertEqual({card["alias"] for card in cards}, {"Demo", "demo"}) + for card in cards: + command = card["recommendation"]["command"] + self.assertNotIn("--workspace Demo", command) + self.assertNotIn("--workspace demo", command) + self.assertNotIn(str(self.root), 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_console_operator.py b/tests/test_console_operator.py index 24e7373..6cba6c1 100644 --- a/tests/test_console_operator.py +++ b/tests/test_console_operator.py @@ -33,10 +33,29 @@ def test_fail_findings_and_empty_commands_are_not_unknown_or_bare(self) -> None: self.assertNotEqual(result["heading"], "关注项未知") self.assertEqual(result["heading"], "需要修复") self.assertNotEqual(result["command"], "dyro --workspace core") - self.assertEqual(result["command"], "dyro --workspace core doctor") + self.assertNotEqual(result["command"], "dyro --workspace core doctor") + self.assertEqual(result["command"], "") + self.assertNotIn("dyro --workspace core doctor", result["primary"]) self.assertNotIn("摘要未列出关注项", result["needsYou"]) self.assertIn("core", result["needsYou"]) self.assertIn("release_a", result["needsYou"]) + self.assertEqual(result["uniqueCommand"], "dyro --workspace core doctor") + self.assertEqual(result["uniqueRecommended"], "dyro --workspace core doctor") + self.assertIn("dyro --workspace core doctor", result["uniquePrimary"]) + + def test_fold_twin_fail_empty_command_does_not_invent_workspace_doctor(self) -> None: + result = _run("fold_twin_fail_overview") + + self.assertEqual(result["heading"], "需要修复") + self.assertEqual(result["command"], "") + self.assertEqual(result["recommendedDemo"], "") + self.assertEqual(result["recommendedTwin"], "") + self.assertNotIn("dyro --workspace Demo", result["primary"]) + self.assertNotIn("dyro --workspace demo", result["primary"]) + self.assertNotIn("dyro --workspace Demo doctor", result["needsYou"]) + self.assertNotIn("dyro --workspace demo doctor", result["needsYou"]) + self.assertIn("Demo", result["needsYou"]) + self.assertIn("demo", result["needsYou"]) def test_tablist_switch_changes_visible_section_ids(self) -> None: result = _run("tabs") diff --git a/tests/test_console_overview.py b/tests/test_console_overview.py index 8ec95f6..a9c73fe 100644 --- a/tests/test_console_overview.py +++ b/tests/test_console_overview.py @@ -203,6 +203,49 @@ def test_rejects_tampered_or_stale_cursor_without_falling_back_to_an_offset(self with self.assertRaisesRegex(ConsoleOverviewError, "OVERVIEW_CURSOR_INVALID"): self.service.page(cursor=cursor, limit=1) + def test_fold_twin_cards_do_not_recommend_fail_closed_workspace_selector(self) -> None: + demo_root = Path("/private/demo-workspace") + twin_root = Path("/private/demo-twin-workspace") + registry = WorkspaceRegistry( + default="Demo", + workspaces=( + WorkspaceRecord("Demo", demo_root), + WorkspaceRecord("demo", twin_root), + ), + ) + configurations = { + demo_root: SimpleNamespace(name="Demo", repositories={"api": object()}), + twin_root: SimpleNamespace(name="demo", repositories={"web": object()}), + } + snapshots = { + "Demo": _snapshot(name="Demo", attention=()), + "demo": _snapshot(name="demo", attention=()), + } + service = ConsoleOverviewService( + registry_loader=lambda: registry, + config_loader=lambda root: configurations[root], + snapshot_loader=lambda config: snapshots[config.name], + clock=lambda: datetime(2026, 8, 4, 12, 5, tzinfo=timezone.utc), + cursor_secret=b"k" * 32, + doctor_loader=lambda config: [ + "FAIL line:core/api: missing origin/feat/core", + ], + commands_loader=lambda config, alias=None: [ + f"dyro --workspace {alias} doctor" + ], + ) + + page = service.page() + cards = page["data"]["workspaces"] + self.assertEqual({card["alias"] for card in cards}, {"Demo", "demo"}) + for card in cards: + command = card["recommendation"]["command"] + self.assertNotIn("--workspace Demo", command) + self.assertNotIn("--workspace demo", command) + self.assertNotIn("/private", command) + unique = self.service._recommendation("core", []) + self.assertEqual(unique["command"], "dyro --workspace core doctor") + def test_empty_attention_recommends_doctor_not_a_bare_workspace_invocation(self) -> None: recommendation = self.service._recommendation("core", []) diff --git a/tests/test_continuation_resolution.py b/tests/test_continuation_resolution.py index 9021ee3..c2dbc38 100644 --- a/tests/test_continuation_resolution.py +++ b/tests/test_continuation_resolution.py @@ -6,10 +6,18 @@ from unittest.mock import patch from dyro.config import load -from dyro.continuation.resolution import resolve_line, resolve_objective, resolve_workspace +from dyro.continuation.resolution import ( + WorkspaceResolutionFailure, + WorkspaceResolutionError, + resolve_line, + resolve_objective, + resolve_workspace, + resolve_workspace_readonly, +) from dyro.continuation.store import create_objective from dyro.errors import DyroError, ValidationError from dyro.hub import add_workspace +from dyro.read_limits import ObservationLimits, ReadBudget from dyro.tasks import task_template from dyro.workspace import create_line, line_root @@ -121,3 +129,57 @@ def test_multiple_active_objectives_require_selector_in_non_interactive_mode(sel with self.assertRaisesRegex(DyroError, "非交互模式必须显式指定"): resolve_objective(self.config, interactive=False) self.assertEqual(resolve_objective(self.config, objective_id="observe-a", interactive=False).objective.id, "observe-a") + + def test_readonly_explicit_alias_matches_case_insensitively(self) -> None: + home = self.root / "dyro-home" + with patch.dict(os.environ, {"DYRO_HOME": str(home)}, clear=False): + add_workspace(self.root, name="Acme", make_default=True) + resolved = resolve_workspace_readonly( + start=None, + workspace="acme", + cwd=self.root, + budget=ReadBudget(ObservationLimits()), + ) + self.assertEqual(resolved.registry_alias, "Acme") + self.assertEqual(resolved.profile.config.root, self.root.resolve()) + + def test_readonly_explicit_alias_fails_closed_on_case_fold_collision(self) -> None: + with tempfile.TemporaryDirectory(prefix="dyro-alias-collision-") as tmp: + other = Path(tmp) / "other" + other.mkdir() + (other / "dyro.toml").write_text( + CONFIG.replace('name = "test-workspace"', 'name = "other"'), + encoding="utf-8", + ) + home = self.root / "dyro-home" + with patch.dict(os.environ, {"DYRO_HOME": str(home)}, clear=False): + add_workspace(self.root, name="Acme") + add_workspace(other, name="acme") + with self.assertRaises(WorkspaceResolutionError) as ctx: + resolve_workspace_readonly( + start=None, + workspace="Acme", + cwd=self.root, + budget=ReadBudget(ObservationLimits()), + ) + self.assertEqual( + ctx.exception.code, WorkspaceResolutionFailure.AMBIGUOUS_WORKSPACE + ) + self.assertIn("Acme", str(ctx.exception)) + self.assertIn("acme", str(ctx.exception)) + + def test_readonly_explicit_alias_unrelated_miss_stays_unregistered(self) -> None: + home = self.root / "dyro-home" + with patch.dict(os.environ, {"DYRO_HOME": str(home)}, clear=False): + add_workspace(self.root, name="Acme", make_default=True) + with self.assertRaises(WorkspaceResolutionError) as ctx: + resolve_workspace_readonly( + start=None, + workspace="missing", + cwd=self.root, + budget=ReadBudget(ObservationLimits()), + ) + self.assertEqual( + ctx.exception.code, WorkspaceResolutionFailure.WORKSPACE_NOT_REGISTERED + ) + self.assertIn("未登记工作区：missing", str(ctx.exception)) diff --git a/tests/test_hub.py b/tests/test_hub.py index e8456f7..513b83f 100644 --- a/tests/test_hub.py +++ b/tests/test_hub.py @@ -2,6 +2,7 @@ from contextlib import redirect_stderr, redirect_stdout from io import StringIO +import json import os from pathlib import Path import subprocess @@ -24,12 +25,14 @@ ) from dyro.hub import ( add_workspace, + alias_fold_collides, get_workspace, load_registry, mark_workspace_used, registry_home, remove_workspace, set_default_workspace, + unique_registered_alias, ) from dyro.tooling import ( ToolPreferences, @@ -137,6 +140,94 @@ def test_malformed_registry_fails_closed(self) -> None: with self.assertRaisesRegex(Exception, "工作区记录"): load_registry() + def _second_workspace(self, name: str) -> Path: + other = self.base / name + other.mkdir() + other.joinpath("dyro.toml").write_text( + self.workspace.joinpath("dyro.toml") + .read_text(encoding="utf-8") + .replace('name = "demo"', f'name = "{name}"'), + encoding="utf-8", + ) + return other + + def test_get_workspace_exact_alias_match(self) -> None: + add_workspace(self.workspace, name="Acme") + record = get_workspace("Acme") + self.assertEqual(record.name, "Acme") + self.assertEqual(record.root, self.workspace.resolve()) + self.assertEqual(load_registry().workspaces[0].name, "Acme") + + def test_get_workspace_matches_alias_case_insensitively(self) -> None: + add_workspace(self.workspace, name="Acme") + record = get_workspace("acme") + self.assertEqual(record.name, "Acme") + self.assertEqual(record.root, self.workspace.resolve()) + self.assertEqual(load_registry().workspaces[0].name, "Acme") + + def test_alias_fold_helpers_distinguish_unique_and_collision(self) -> None: + names = ("Demo", "other") + self.assertFalse(alias_fold_collides("demo", names)) + self.assertEqual(unique_registered_alias("demo", names), "Demo") + colliding = ("Demo", "demo") + self.assertTrue(alias_fold_collides("Demo", colliding)) + self.assertIsNone(unique_registered_alias("demo", colliding)) + self.assertIsNone(unique_registered_alias("missing", names)) + + def test_get_workspace_fails_closed_on_case_fold_collision(self) -> None: + from dyro.errors import DyroError + + other = self._second_workspace("other") + add_workspace(self.workspace, name="Acme") + add_workspace(other, name="acme") + with self.assertRaises(DyroError) as ctx: + get_workspace("Acme") + message = str(ctx.exception) + self.assertIn("Acme", message) + self.assertIn("acme", message) + self.assertNotIn("你是不是指", message) + + def test_get_workspace_unrelated_miss_stays_unregistered(self) -> None: + from dyro.errors import DyroError + + add_workspace(self.workspace, name="Acme") + with self.assertRaisesRegex(DyroError, "未登记工作区：missing"): + get_workspace("missing") + + def test_set_default_unique_fold_writes_canonical_name(self) -> None: + other = self._second_workspace("other") + add_workspace(other, name="other", make_default=True) + add_workspace(self.workspace, name="Acme") + set_default_workspace("acme") + self.assertEqual(load_registry().default, "Acme") + + def test_set_default_collision_fails_closed(self) -> None: + from dyro.errors import DyroError + + other = self._second_workspace("other") + add_workspace(self.workspace, name="Acme", make_default=True) + add_workspace(other, name="acme") + with self.assertRaises(DyroError) as ctx: + set_default_workspace("acme") + self.assertIn("Acme", str(ctx.exception)) + self.assertIn("acme", str(ctx.exception)) + self.assertEqual(load_registry().default, "Acme") + + def test_remove_unique_fold_deletes_canonical_row(self) -> None: + add_workspace(self.workspace, name="Acme") + remove_workspace("acme") + self.assertEqual(load_registry().workspaces, ()) + + def test_remove_exact_registered_name_works_under_collision(self) -> None: + other = self._second_workspace("other") + add_workspace(self.workspace, name="Acme") + add_workspace(other, name="acme") + remove_workspace("Acme") + remaining = tuple(item.name for item in load_registry().workspaces) + self.assertEqual(remaining, ("acme",)) + remove_workspace("acme") + self.assertEqual(load_registry().workspaces, ()) + def test_malformed_registry_rejects_non_string_alias(self) -> None: self.state.mkdir(parents=True) registry_home().joinpath("workspaces.json").write_text( @@ -1484,14 +1575,152 @@ def test_unknown_alias_suggests_close_registered_name(self) -> None: from dyro.errors import DyroError from dyro.hub import get_workspace - add_workspace(self.root, name="DyroEngineeringFlow", make_default=True) - with self.assertRaisesRegex(DyroError, "你是不是指 DyroEngineeringFlow"): - get_workspace("dyroengineeringflow") + add_workspace(self.root, name="AcmeLab", make_default=True) + with self.assertRaisesRegex(DyroError, "你是不是指 AcmeLab"): + get_workspace("acme-labs") stderr = StringIO() with redirect_stderr(stderr), self.assertRaises(SystemExit) as raised: - main(["--workspace", "dyroengineeringflow", "next"]) + main(["--workspace", "acme-labs", "next"]) self.assertEqual(raised.exception.code, 2) - self.assertIn("你是不是指 DyroEngineeringFlow", stderr.getvalue()) + self.assertIn("你是不是指 AcmeLab", stderr.getvalue()) + + def _second_registered_workspace(self, alias: str) -> Path: + other = self.root.parent / f"{self.root.name}-{alias}" + other.mkdir() + other.joinpath("dyro.toml").write_text( + (self.root / "dyro.toml") + .read_text(encoding="utf-8") + .replace('name = "test-workspace"', f'name = "{alias}"'), + encoding="utf-8", + ) + return other + + def test_workspace_default_unique_fold_plan_and_apply(self) -> None: + other = self._second_registered_workspace("other") + add_workspace(other, name="other", make_default=True) + add_workspace(self.root, name="Acme") + output = StringIO() + with redirect_stdout(output): + main(["--dry-run", "workspace", "default", "acme"]) + self.assertIn("Acme", output.getvalue()) + self.assertEqual(load_registry().default, "other") + output = StringIO() + with redirect_stdout(output): + main(["workspace", "default", "acme"]) + self.assertEqual(load_registry().default, "Acme") + self.assertIn("Acme", output.getvalue()) + + def test_workspace_remove_unique_fold_plan_and_apply(self) -> None: + add_workspace(self.root, name="Acme") + output = StringIO() + with redirect_stdout(output): + main(["--dry-run", "workspace", "remove", "acme"]) + self.assertEqual(load_registry().workspaces[0].name, "Acme") + self.assertIn("Acme", output.getvalue()) + main(["workspace", "remove", "acme", "--yes"]) + self.assertEqual(load_registry().workspaces, ()) + + def test_workspace_default_collision_refuses_plan_and_apply(self) -> None: + other = self._second_registered_workspace("other") + add_workspace(self.root, name="Acme", make_default=True) + add_workspace(other, name="acme") + stderr = StringIO() + with redirect_stderr(stderr), self.assertRaises(SystemExit) as planned: + main(["--dry-run", "workspace", "default", "acme"]) + self.assertEqual(planned.exception.code, 2) + self.assertIn("acme", stderr.getvalue()) + self.assertEqual(load_registry().default, "Acme") + stderr = StringIO() + with redirect_stderr(stderr), self.assertRaises(SystemExit) as applied: + main(["workspace", "default", "acme"]) + self.assertEqual(applied.exception.code, 2) + self.assertEqual(load_registry().default, "Acme") + + def test_workspace_remove_exact_names_work_under_collision(self) -> None: + other = self._second_registered_workspace("other") + add_workspace(self.root, name="Acme") + add_workspace(other, name="acme") + main(["workspace", "remove", "Acme", "--yes"]) + self.assertEqual( + tuple(item.name for item in load_registry().workspaces), ("acme",) + ) + main(["workspace", "remove", "acme", "--yes"]) + self.assertEqual(load_registry().workspaces, ()) + + def test_implicit_json_next_does_not_advertise_colliding_alias(self) -> None: + create_line(load(self.root), line_id="alpha", branch="feat/alpha", base="main") + (self.root / "dyro.toml").write_text( + (self.root / "dyro.toml") + .read_text(encoding="utf-8") + .replace('name = "test-workspace"', 'name = "Demo"'), + encoding="utf-8", + ) + other = self.root.parent / f"{self.root.name}-twin" + other.mkdir() + other.joinpath("dyro.toml").write_text( + (self.root / "dyro.toml") + .read_text(encoding="utf-8") + .replace('name = "Demo"', 'name = "demo"'), + encoding="utf-8", + ) + add_workspace(self.root, name="Demo", make_default=True) + add_workspace(other, name="demo") + unrelated = self.root.parent / f"{self.root.name}-unrelated" + unrelated.mkdir() + output = StringIO() + previous = Path.cwd() + try: + os.chdir(unrelated) + with redirect_stdout(output): + main(["next", "--format", "json"]) + finally: + os.chdir(previous) + rendered = output.getvalue() + payload = json.loads(rendered) + self.assertEqual(payload["kind"], "next_step") + self.assertEqual(payload["state"], "needs_repair") + commands = payload.get("commands") or [] + diagnostic = payload.get("diagnostic_commands") or [] + briefing = payload.get("briefing") or {} + briefing_command = ( + briefing.get("command") if isinstance(briefing, dict) else None + ) + advertised = [ + item + for item in (*commands, *diagnostic, briefing_command) + if isinstance(item, str) + ] + self.assertTrue(advertised) + for command in advertised: + self.assertNotIn("--workspace Demo", command) + self.assertNotIn("--workspace demo", command) + self.assertTrue(any("--root" in item for item in advertised)) + + def test_console_unique_fold_plan_and_apply_share_canonical_alias(self) -> None: + add_workspace(self.root, name="Demo") + output = StringIO() + with redirect_stdout(output): + main(["--dry-run", "--workspace", "demo", "console"]) + self.assertIn("初始焦点：Demo", output.getvalue()) + self.assertNotIn("初始焦点：demo", output.getvalue()) + with patch("dyro.cli.launch_console") as launch: + main(["--workspace", "demo", "console"]) + launch.assert_called_once() + self.assertEqual(launch.call_args.kwargs["initial_workspace"], "Demo") + + def test_console_collision_refuses_plan_and_apply(self) -> None: + other = self._second_registered_workspace("other") + add_workspace(self.root, name="Demo") + add_workspace(other, name="demo") + stderr = StringIO() + with redirect_stderr(stderr), self.assertRaises(SystemExit) as planned: + main(["--dry-run", "--workspace", "Demo", "console"]) + self.assertEqual(planned.exception.code, 2) + self.assertIn("Demo", stderr.getvalue()) + stderr = StringIO() + with redirect_stderr(stderr), self.assertRaises(SystemExit) as applied: + main(["--workspace", "Demo", "console"]) + self.assertEqual(applied.exception.code, 2) def test_status_and_next_disclose_disabled_push(self) -> None: self._create_line()