fix(toolkit): repair G5 and G6 contracts

Author: Ldsystem

scripts/work-bundle/control_plane.py

@@ -1083,6 +1083,26 @@ def _publish_evidence(control: Path, payload: dict[str, object]) -> Path:
     return path
 
 
+def _control_plane_gitlink_paths(control: Path) -> list[str]:
+    paths = {
+        str(marker.parent.relative_to(control)).replace("\\", "/")
+        for marker in control.rglob(".git")
+        if marker.parent != control and (marker.is_dir() or marker.is_file())
+    }
+    if (control / ".git").exists():
+        result = subprocess.run(
+            ["git", "-C", str(control), "ls-files", "--stage", "-z"],
+            check=False,
+            capture_output=True,
+        )
+        if result.returncode == 0:
+            for entry in result.stdout.split(b"\0"):
+                if not entry.startswith(b"160000 ") or b"\t" not in entry:
+                    continue
+                paths.add(entry.split(b"\t", 1)[1].decode("utf-8", errors="surrogateescape"))
+    return sorted(paths)
+
+
 def cmd_publish_control_plane(args: list[str]) -> int:
     parser = argparse.ArgumentParser(prog="wb.py publish-control-plane")
     parser.add_argument("workspace_root")
@@ -1109,6 +1129,16 @@ def cmd_publish_control_plane(args: list[str]) -> int:
     if not remote:
         out({"command": "publish-control-plane", "status": "issues-found", "failure_code": "WB_CONTROL_PLANE_REMOTE_REQUIRED", "changed_files": []})
         return 1
+    gitlink_paths = _control_plane_gitlink_paths(control)
+    if gitlink_paths:
+        out({
+            "command": "publish-control-plane",
+            "status": "issues-found",
+            "failure_code": "WB_CONTROL_PLANE_GITLINK_FORBIDDEN",
+            "gitlink_paths": gitlink_paths,
+            "changed_files": [],
+        })
+        return 1
     if parsed.dry_run:
         out({"command": "publish-control-plane", "status": "passed", "dry_run": True, "remote": remote, "changed_files": [], "git_actions": ["init", "configure-origin", "commit", "push"]})
         return 0

scripts/work-bundle/project.py

@@ -1895,7 +1895,7 @@ def assess_legacy_topology(
     )
     registry_sources = registry_sources if isinstance(registry_sources, list) else []
 
-    def identities(sources: object) -> list[dict[str, str]]:
+    def identities(sources: object, identity_kind: str) -> list[dict[str, str]]:
         result: list[dict[str, str]] = []
         if not isinstance(sources, list):
             return result
@@ -1906,11 +1906,14 @@ def identities(sources: object) -> list[dict[str, str]]:
             result.append({
                 'id': str(source.get('id') or ''),
                 'path': str(Path(str(raw_path)).expanduser().resolve()) if raw_path else '',
+                'kind': identity_kind,
             })
         return result
 
-    metadata_identities = identities(metadata_sources)
-    registry_identities = identities([*registry_sources, *(registry_origin_data or [])])
+    metadata_identities = identities(metadata_sources, 'member')
+    registry_member_identities = identities(registry_sources, 'member')
+    registry_origin_identities = identities(registry_origin_data or [], 'origin')
+    registry_identities = [*registry_member_identities, *registry_origin_identities]
     registry_identities = [
         dict(identity)
         for identity in {
@@ -1920,7 +1923,7 @@ def identities(sources: object) -> list[dict[str, str]]:
     ]
     conflicts: list[str] = []
     by_id: dict[str, set[str]] = {}
-    for identity in [*metadata_identities, *registry_identities]:
+    for identity in [*metadata_identities, *registry_member_identities]:
         if identity['id'] and identity['path']:
             by_id.setdefault(identity['id'], set()).add(identity['path'])
     for source_id, paths in sorted(by_id.items()):

tests/test_control_plane_v4.py

@@ -1550,6 +1550,49 @@ def test_publish_existing_dirty_control_plane_fails_closed_without_mutation(tmp_
     assert evidence["recovery_required"] is True
 
 
+def test_publish_rejects_nested_gitlink_before_staging(tmp_path: Path) -> None:
+    config = config_root(tmp_path / "config-root")
+    source_remote, _, _ = make_remote(tmp_path / "source-fixture", "source")
+    workspace = tmp_path / "workspace"
+    assert run_wb(
+        config,
+        "init-workspace",
+        str(workspace),
+        "--slug",
+        "demo",
+        "--repository",
+        f"source-main={source_remote}",
+        "--apply",
+    ).returncode == 0
+    control = workspace / ".work-bundle"
+    nested = control / "knowledge/notes/nested-repository"
+    nested.mkdir(parents=True)
+    git(nested, "init", "-q", "-b", "main")
+    git(nested, "config", "user.email", "test@example.com")
+    git(nested, "config", "user.name", "Test")
+    (nested / "README.md").write_text("nested\n", encoding="utf-8")
+    git(nested, "add", "README.md")
+    git(nested, "commit", "-q", "-m", "nested")
+    remote = tmp_path / "control.git"
+    subprocess.run(["git", "init", "--bare", "-q", str(remote)], check=True)
+
+    failed = run_wb(
+        config,
+        "publish-control-plane",
+        str(workspace),
+        "--remote",
+        str(remote),
+        "--apply",
+    )
+
+    assert failed.returncode == 1
+    data = json.loads(failed.stdout)
+    assert data["failure_code"] == "WB_CONTROL_PLANE_GITLINK_FORBIDDEN"
+    assert data["gitlink_paths"] == ["knowledge/notes/nested-repository"]
+    assert not (control / ".git").exists()
+    assert (nested / ".git").is_dir()
+
+
 def test_publish_failed_git_snapshot_fails_closed_without_mutation(tmp_path: Path) -> None:
     config = config_root(tmp_path / "config-root")
     source_remote, _, _ = make_remote(tmp_path / "source-fixture", "source")

tests/test_project_initialization.py

@@ -1440,6 +1440,60 @@ def test_migrate_project_routes_registry_multi_source_to_workspace_migration(tmp
     assert metadata_path.read_bytes() == before
 
 
+def test_migrate_project_accepts_same_id_origin_and_member_paths(tmp_path: Path) -> None:
+    config_root, project = _init_fixture_project(tmp_path)
+    origin = tmp_path / "origin-locator"
+    origin.mkdir()
+    registry_path = config_root / "registry" / "projects.yaml"
+    registry_path.write_text(
+        "\n".join(
+            [
+                "projects:",
+                "  - slug: demo",
+                "    name: demo",
+                f"    work_bundle_root: {project.resolve() / '.work-bundle'}",
+                f"    knowledge_root: {project.resolve() / '.work-bundle' / 'knowledge'}",
+                "    aliases: []",
+                "    repository_origins:",
+                "      - id: demo-main",
+                f"        origin_path: {origin.resolve()}",
+                "        git_repository: true",
+                "    source_repositories:",
+                "      - id: demo-main",
+                f"        path: {project.resolve()}",
+                "        work_dir: true",
+                '        remote: ""',
+                "        git_repository: true",
+                "    status: active",
+                "    updated_at: 2026-01-01",
+                "",
+            ]
+        ),
+        encoding="utf-8",
+    )
+    metadata_path = project / ".work-bundle/project.yaml"
+    metadata_path.write_text(
+        "\n".join(
+            [
+                "metadata_version: 1",
+                "authority: canonical",
+                f"project_root: {project.resolve()}",
+                "industry: legacy",
+                "",
+            ]
+        ),
+        encoding="utf-8",
+    )
+
+    migrated = run_wb(config_root, "migrate-project", str(project), "--dry-run", "--name", "demo")
+
+    assert migrated.returncode == 1
+    data = json.loads(migrated.stdout)
+    assert data["mode"] == "multi-repository-migration-required"
+    assert data["topology_assessment"]["conflicts"] == []
+    assert data["topology_assessment"]["required_command"] == "migrate-to-multi-repository"
+
+
 def _seed_committed_repository(path: Path) -> None:
     path.mkdir()
     git(path, "init", "-q", "-b", "main")