From af17ae9d0e67599a354fccdac976209408d53e52 Mon Sep 17 00:00:00 2001 From: Layne Penney Date: Fri, 21 Aug 2026 05:05:30 -0500 Subject: [PATCH 1/3] feat(workspace): initialize spec from topology --- gr2/python_cli/app.py | 133 +++++++++++++- .../test_workspace_init_from_topology.py | 165 ++++++++++++++++++ 2 files changed, 290 insertions(+), 8 deletions(-) create mode 100644 gr2/tests/test_workspace_init_from_topology.py diff --git a/gr2/python_cli/app.py b/gr2/python_cli/app.py index 095fa30..d691734 100644 --- a/gr2/python_cli/app.py +++ b/gr2/python_cli/app.py @@ -4,6 +4,7 @@ import io import json import os +import tomllib from pathlib import Path from types import SimpleNamespace from typing import Optional @@ -291,32 +292,67 @@ def _configured_merge_method(workspace_root: Path) -> str | None: return value +def _toml_basic_string(value: str) -> str: + """Render one string through the TOML basic-string grammar. + + WorkspaceSpec values are operator-controlled at several call sites. Keeping + the escaping here makes every value written by ``_write_workspace_spec`` + parseable, rather than relying on each caller to reject a partial set of + characters. + """ + if not isinstance(value, str): + raise TypeError(f"TOML basic string requires str, got {type(value).__name__}") + + escapes = { + '"': '\\"', + "\\": "\\\\", + "\b": "\\b", + "\t": "\\t", + "\n": "\\n", + "\f": "\\f", + "\r": "\\r", + } + rendered: list[str] = ['"'] + for character in value: + code_point = ord(character) + if character in escapes: + rendered.append(escapes[character]) + elif code_point < 0x20 or code_point == 0x7F: + rendered.append(f"\\u{code_point:04X}") + elif 0xD800 <= code_point <= 0xDFFF: + raise ValueError("TOML basic strings cannot contain surrogate code points") + else: + rendered.append(character) + rendered.append('"') + return "".join(rendered) + + def _write_workspace_spec(workspace_root: Path, repos: list[dict[str, str]], default_unit: str) -> Path: spec_path = _workspace_spec_path(workspace_root) - spec_path.parent.mkdir(parents=True, exist_ok=True) lines = [ - f'workspace_name = "{workspace_root.name}"', + f"workspace_name = {_toml_basic_string(workspace_root.name)}", "", ] for repo in repos: lines.extend( [ "[[repos]]", - f'name = "{repo["name"]}"', - f'path = "{repo["path"]}"', - f'url = "{repo["url"]}"', + f"name = {_toml_basic_string(repo['name'])}", + f"path = {_toml_basic_string(repo['path'])}", + f"url = {_toml_basic_string(repo['url'])}", "", ] ) lines.extend( [ "[[units]]", - f'name = "{default_unit}"', - f'path = "agents/{default_unit}/home"', - "repos = [" + ", ".join(f'"{repo["name"]}"' for repo in repos) + "]", + f"name = {_toml_basic_string(default_unit)}", + f"path = {_toml_basic_string(f'agents/{default_unit}/home')}", + "repos = [" + ", ".join(_toml_basic_string(repo["name"]) for repo in repos) + "]", "", ] ) + spec_path.parent.mkdir(parents=True, exist_ok=True) spec_path.write_text("\n".join(lines)) return spec_path @@ -343,6 +379,53 @@ def _scan_existing_repos(workspace_root: Path) -> list[dict[str, str]]: return repos +def _declared_repos_from_workspace_topology(workspace_root: Path) -> list[dict[str, str]]: + """Lower neutral ``workspace.toml`` repository declarations for gr2. + + This deliberately reads only the fields the existing WorkspaceSpec writer + accepts. Declarations can carry fields such as ``default_ref`` as well, + but those do not belong in the WorkspaceSpec emission path. + """ + topology_path = workspace_root / "workspace.toml" + if not topology_path.is_file(): + raise SystemExit(f"workspace topology not found: {topology_path}") + try: + with topology_path.open("rb") as topology_file: + document = tomllib.load(topology_file) + except tomllib.TOMLDecodeError as exc: + raise SystemExit(f"workspace.toml at {topology_path} is not valid TOML: {exc}") from exc + + raw_repos = document.get("repos", []) + if not isinstance(raw_repos, list): + raise SystemExit("workspace.toml repos must be an array of tables") + if not raw_repos: + raise SystemExit("workspace.toml declares no [[repos]] entries") + + repos: list[dict[str, str]] = [] + for index, raw_repo in enumerate(raw_repos): + if not isinstance(raw_repo, dict): + raise SystemExit(f"workspace.toml repos[{index}] must be a table") + key = str(raw_repo.get("key", "")) + for field in ("key", "path", "url"): + value = raw_repo.get(field) + if not value: + raise SystemExit( + f"workspace.toml repos[{index}] ({key!r}) is missing {field!r}" + ) + if not isinstance(value, str): + raise SystemExit( + f"workspace.toml repos[{index}] ({key!r}) must declare {field!r} as a string" + ) + repos.append( + { + "name": str(raw_repo["key"]), + "path": str(raw_repo["path"]), + "url": str(raw_repo["url"]), + } + ) + return repos + + def _exit(code: int) -> None: if code != 0: raise typer.Exit(code=code) @@ -414,6 +497,40 @@ def workspace_init( typer.echo("\n".join(lines)) +@workspace_app.command("init-from-topology") +def workspace_init_from_topology( + workspace_root: Path, + default_unit: str = typer.Option("default", help="Default owner unit for declared repos"), + json_output: bool = typer.Option(False, "--json", help="Emit machine-readable JSON"), +) -> None: + """Create WorkspaceSpec from neutral ``workspace.toml`` repo declarations.""" + workspace_root = workspace_root.resolve() + repos = _declared_repos_from_workspace_topology(workspace_root) + spec_path = _write_workspace_spec(workspace_root, repos, default_unit) + payload = { + "workspace_root": str(workspace_root), + "spec_path": str(spec_path), + "repo_count": len(repos), + "repos": repos, + "default_unit": default_unit, + "source": "workspace.toml", + } + if json_output: + typer.echo(json.dumps(payload, indent=2)) + else: + lines = [ + "WorkspaceInitFromTopology", + f"workspace_root = {workspace_root}", + f"spec_path = {spec_path}", + f"default_unit = {default_unit}", + f"repo_count = {len(repos)}", + "source = workspace.toml", + "REPOS", + ] + lines.extend(f"- {repo['name']}\t{repo['path']}\t{repo['url']}" for repo in repos) + typer.echo("\n".join(lines)) + + @workspace_app.command("materialize") def workspace_materialize( workspace_root: Path, diff --git a/gr2/tests/test_workspace_init_from_topology.py b/gr2/tests/test_workspace_init_from_topology.py new file mode 100644 index 0000000..2149c54 --- /dev/null +++ b/gr2/tests/test_workspace_init_from_topology.py @@ -0,0 +1,165 @@ +from __future__ import annotations + +import json +import tomllib +from pathlib import Path + +from gr2.python_cli.app import app +from typer.testing import CliRunner + +runner = CliRunner() + + +def _write_topology(workspace_root: Path) -> None: + (workspace_root / "workspace.toml").write_text( + """\ +schema_version = 2 +workspace_name = "declarative-team" + +[[repos]] +key = "product" +url = "https://example.invalid/product.git" +path = "repos/product" +default_ref = "main" + +[[repos]] +key = "config" +url = "https://example.invalid/config.git" +path = "config" +default_ref = "dev" +""" + ) + + +def test_workspace_init_from_topology_writes_declared_repos_in_an_empty_directory( + tmp_path: Path, +) -> None: + """The zero-to-team wire must not fall back to scanning local Git repos. + + Mutation: replace the declared-topology reader with `_scan_existing_repos`. + This root has no Git repositories, so the old adoption path refuses and this + witness goes red before a WorkspaceSpec can be written. + """ + workspace_root = tmp_path / "empty-team" + workspace_root.mkdir() + _write_topology(workspace_root) + + result = runner.invoke( + app, + [ + "workspace", + "init-from-topology", + str(workspace_root), + "--default-unit", + "team", + "--json", + ], + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert payload["workspace_root"] == str(workspace_root) + assert payload["spec_path"] == str(workspace_root / ".grip" / "workspace_spec.toml") + assert payload["repo_count"] == 2 + assert [repo["name"] for repo in payload["repos"]] == ["product", "config"] + assert [repo["path"] for repo in payload["repos"]] == ["repos/product", "config"] + assert [repo["url"] for repo in payload["repos"]] == [ + "https://example.invalid/product.git", + "https://example.invalid/config.git", + ] + assert payload["default_unit"] == "team" + assert payload["source"] == "workspace.toml" + assert not list(workspace_root.glob("*/.git")) + assert not (workspace_root / "repos" / "product").exists() + assert not (workspace_root / "config").exists() + + with (workspace_root / ".grip" / "workspace_spec.toml").open("rb") as spec_file: + spec = tomllib.load(spec_file) + assert spec["repos"] == payload["repos"] + assert spec["units"] == [ + { + "name": "team", + "path": "agents/team/home", + "repos": ["product", "config"], + } + ] + + text_result = runner.invoke( + app, + ["workspace", "init-from-topology", str(workspace_root), "--default-unit", "team"], + ) + assert text_result.exit_code == 0, text_result.output + assert "source = workspace.toml" in text_result.output + + +def test_workspace_init_from_topology_refuses_an_incomplete_declared_repo_before_writing( + tmp_path: Path, +) -> None: + workspace_root = tmp_path / "incomplete-team" + workspace_root.mkdir() + (workspace_root / "workspace.toml").write_text( + """\ +[[repos]] +key = "product" +path = "repos/product" +default_ref = "main" +""" + ) + + result = runner.invoke(app, ["workspace", "init-from-topology", str(workspace_root)]) + + assert result.exit_code != 0 + assert "workspace.toml repos[0] ('product') is missing 'url'" in result.output + assert not (workspace_root / ".grip" / "workspace_spec.toml").exists() + + +def test_workspace_init_from_topology_serializes_every_writer_value( + tmp_path: Path, +) -> None: + """Hostile strings must round-trip through every WorkspaceSpec value slot. + + Mutation: replace the writer serializer with raw interpolation. The command + still exits 0, but this direct parse of its bytes fails. The witness thus + catches the WRONG-BUT-GREEN failure mode rather than only a refusal path. + """ + workspace_root = tmp_path / 'unsafe"team' + workspace_root.mkdir() + (workspace_root / "workspace.toml").write_text( + r''' +[[repos]] +key = 'product"\\name' +url = 'https://example.invalid/a"\\b.git' +path = 'repos/product"\\path' +''' + ) + + default_unit = 'team"\\unit' + result = runner.invoke( + app, + [ + "workspace", + "init-from-topology", + str(workspace_root), + "--default-unit", + default_unit, + ], + ) + + assert result.exit_code == 0, result.output + with (workspace_root / ".grip" / "workspace_spec.toml").open("rb") as spec_file: + spec = tomllib.load(spec_file) + assert spec["workspace_name"] == workspace_root.name + assert spec["repos"] == [ + { + "name": r'product"\\name', + "path": r'repos/product"\\path', + "url": r'https://example.invalid/a"\\b.git', + } + ] + assert spec["units"] == [ + { + "name": default_unit, + "path": f"agents/{default_unit}/home", + "repos": [r'product"\\name'], + } + ] From 01ef91181bbc6038b6c59ed3b6b9cc7675bd1d82 Mon Sep 17 00:00:00 2001 From: Layne Penney Date: Fri, 21 Aug 2026 07:54:14 -0500 Subject: [PATCH 2/3] test(workspace): cover encoding refusal before directory creation --- .../test_workspace_init_from_topology.py | 35 +++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/gr2/tests/test_workspace_init_from_topology.py b/gr2/tests/test_workspace_init_from_topology.py index 2149c54..be1b8a0 100644 --- a/gr2/tests/test_workspace_init_from_topology.py +++ b/gr2/tests/test_workspace_init_from_topology.py @@ -113,6 +113,41 @@ def test_workspace_init_from_topology_refuses_an_incomplete_declared_repo_before assert not (workspace_root / ".grip" / "workspace_spec.toml").exists() +def test_workspace_init_from_topology_refuses_an_encoding_error_before_creating_grip_directory( + tmp_path: Path, +) -> None: + """A writer encoding refusal must leave neither the spec nor its directory. + + TOML itself rejects a surrogate escape in ``workspace.toml`` before the + command reaches the writer, so this uses the CLI-owned default-unit value, + one of the writer's seven string positions, to exercise the serializer. + + Mutation: move ``spec_path.parent.mkdir`` above ``lines`` construction. + The command still refuses, but this witness reaches its directory assertion + and fails while the three pre-existing witnesses remain green. + """ + workspace_root = tmp_path / "encoding-refusal-team" + workspace_root.mkdir() + _write_topology(workspace_root) + + result = runner.invoke( + app, + [ + "workspace", + "init-from-topology", + str(workspace_root), + "--default-unit", + "team-\ud800", + ], + ) + + assert result.exit_code != 0 + assert isinstance(result.exception, ValueError) + assert "surrogate" in str(result.exception) + assert not (workspace_root / ".grip" / "workspace_spec.toml").exists() + assert not (workspace_root / ".grip").exists() + + def test_workspace_init_from_topology_serializes_every_writer_value( tmp_path: Path, ) -> None: From 3bce8c25433608733a11dac95b71048802173186 Mon Sep 17 00:00:00 2001 From: Layne Penney Date: Fri, 21 Aug 2026 08:49:48 -0500 Subject: [PATCH 3/3] fix(workspace): honor declared topology name --- gr2/python_cli/app.py | 30 ++++++++++++---- .../test_workspace_init_from_topology.py | 34 ++++++++++++++++--- 2 files changed, 54 insertions(+), 10 deletions(-) diff --git a/gr2/python_cli/app.py b/gr2/python_cli/app.py index d691734..2a7aad0 100644 --- a/gr2/python_cli/app.py +++ b/gr2/python_cli/app.py @@ -327,10 +327,17 @@ def _toml_basic_string(value: str) -> str: return "".join(rendered) -def _write_workspace_spec(workspace_root: Path, repos: list[dict[str, str]], default_unit: str) -> Path: +def _write_workspace_spec( + workspace_root: Path, + repos: list[dict[str, str]], + default_unit: str, + *, + workspace_name: str | None = None, +) -> Path: spec_path = _workspace_spec_path(workspace_root) + emitted_workspace_name = workspace_root.name if workspace_name is None else workspace_name lines = [ - f"workspace_name = {_toml_basic_string(workspace_root.name)}", + f"workspace_name = {_toml_basic_string(emitted_workspace_name)}", "", ] for repo in repos: @@ -379,7 +386,9 @@ def _scan_existing_repos(workspace_root: Path) -> list[dict[str, str]]: return repos -def _declared_repos_from_workspace_topology(workspace_root: Path) -> list[dict[str, str]]: +def _declared_workspace_topology( + workspace_root: Path, +) -> tuple[str | None, list[dict[str, str]]]: """Lower neutral ``workspace.toml`` repository declarations for gr2. This deliberately reads only the fields the existing WorkspaceSpec writer @@ -395,6 +404,10 @@ def _declared_repos_from_workspace_topology(workspace_root: Path) -> list[dict[s except tomllib.TOMLDecodeError as exc: raise SystemExit(f"workspace.toml at {topology_path} is not valid TOML: {exc}") from exc + workspace_name = document.get("workspace_name") + if workspace_name is not None and not isinstance(workspace_name, str): + raise SystemExit("workspace.toml workspace_name must be a string when declared") + raw_repos = document.get("repos", []) if not isinstance(raw_repos, list): raise SystemExit("workspace.toml repos must be an array of tables") @@ -423,7 +436,7 @@ def _declared_repos_from_workspace_topology(workspace_root: Path) -> list[dict[s "url": str(raw_repo["url"]), } ) - return repos + return workspace_name, repos def _exit(code: int) -> None: @@ -505,8 +518,13 @@ def workspace_init_from_topology( ) -> None: """Create WorkspaceSpec from neutral ``workspace.toml`` repo declarations.""" workspace_root = workspace_root.resolve() - repos = _declared_repos_from_workspace_topology(workspace_root) - spec_path = _write_workspace_spec(workspace_root, repos, default_unit) + workspace_name, repos = _declared_workspace_topology(workspace_root) + spec_path = _write_workspace_spec( + workspace_root, + repos, + default_unit, + workspace_name=workspace_name, + ) payload = { "workspace_root": str(workspace_root), "spec_path": str(spec_path), diff --git a/gr2/tests/test_workspace_init_from_topology.py b/gr2/tests/test_workspace_init_from_topology.py index be1b8a0..1c77ca8 100644 --- a/gr2/tests/test_workspace_init_from_topology.py +++ b/gr2/tests/test_workspace_init_from_topology.py @@ -10,11 +10,18 @@ runner = CliRunner() -def _write_topology(workspace_root: Path) -> None: +def _write_topology( + workspace_root: Path, + *, + workspace_name: str | None = "declarative-team", +) -> None: + declared_workspace_name = ( + f'workspace_name = "{workspace_name}"\n' if workspace_name is not None else "" + ) (workspace_root / "workspace.toml").write_text( - """\ + f"""\ schema_version = 2 -workspace_name = "declarative-team" +{declared_workspace_name} [[repos]] key = "product" @@ -75,6 +82,7 @@ def test_workspace_init_from_topology_writes_declared_repos_in_an_empty_director with (workspace_root / ".grip" / "workspace_spec.toml").open("rb") as spec_file: spec = tomllib.load(spec_file) + assert spec["workspace_name"] == "declarative-team" assert spec["repos"] == payload["repos"] assert spec["units"] == [ { @@ -92,6 +100,22 @@ def test_workspace_init_from_topology_writes_declared_repos_in_an_empty_director assert "source = workspace.toml" in text_result.output +def test_workspace_init_from_topology_falls_back_to_root_name_when_name_is_absent( + tmp_path: Path, +) -> None: + """The pre-existing scan-style fallback remains explicit when topology omits a name.""" + workspace_root = tmp_path / "fallback-team" + workspace_root.mkdir() + _write_topology(workspace_root, workspace_name=None) + + result = runner.invoke(app, ["workspace", "init-from-topology", str(workspace_root)]) + + assert result.exit_code == 0, result.output + with (workspace_root / ".grip" / "workspace_spec.toml").open("rb") as spec_file: + spec = tomllib.load(spec_file) + assert spec["workspace_name"] == "fallback-team" + + def test_workspace_init_from_topology_refuses_an_incomplete_declared_repo_before_writing( tmp_path: Path, ) -> None: @@ -161,6 +185,8 @@ def test_workspace_init_from_topology_serializes_every_writer_value( workspace_root.mkdir() (workspace_root / "workspace.toml").write_text( r''' +workspace_name = 'declared"\\workspace' + [[repos]] key = 'product"\\name' url = 'https://example.invalid/a"\\b.git' @@ -183,7 +209,7 @@ def test_workspace_init_from_topology_serializes_every_writer_value( assert result.exit_code == 0, result.output with (workspace_root / ".grip" / "workspace_spec.toml").open("rb") as spec_file: spec = tomllib.load(spec_file) - assert spec["workspace_name"] == workspace_root.name + assert spec["workspace_name"] == r'declared"\\workspace' assert spec["repos"] == [ { "name": r'product"\\name',