"""Task cutover protections with real repositories, gitlinks and worktrees.""" import json import os from pathlib import Path import shutil import subprocess import tempfile import unittest SOURCE = Path(__file__).resolve().parents[1] class WorkspaceCutoverTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory(prefix="workspace-cutover-") self.addCleanup(self.tmp.cleanup) self.base = Path(self.tmp.name) self.repo = self.base / "repo" self.env = os.environ.copy() for key in list(self.env): if key.startswith("GIT_CONFIG_") or key in ("GIT_DIR", "GIT_WORK_TREE", "GIT_INDEX_FILE", "GOWORK", "PRECHECK_RECEIPT", "RECEIPT", "CMD", "IMPORT_RECEIPT", "COMPOSITION_SOURCE_MAP", "CI_DEPENDENCY_MIRRORS"): self.env.pop(key, None) self.env.update({"GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": "/dev/null", "GIT_ALLOW_PROTOCOL": "file", "GIT_TERMINAL_PROMPT": "0", "GIT_AUTHOR_NAME": "Cutover Test", "GIT_AUTHOR_EMAIL": "test@example.invalid", "GIT_COMMITTER_NAME": "Cutover Test", "GIT_COMMITTER_EMAIL": "test@example.invalid", "XDG_CACHE_HOME": str(self.base / "cache")}) self.call(self.base, "git", "init", "--initial-branch=main", str(self.repo)) shutil.copytree(SOURCE / "scripts", self.repo / "scripts") (self.repo / "Makefile").write_text("precheck:\n\t@bash scripts/task.sh precheck\n") (self.repo / ".gitignore").write_text(".workspace/\ngo.work\ngo.work.sum\nmodules/\n") (self.repo / "tracked").write_text("root\n") self.git(self.repo, "add", ".") self.git(self.repo, "commit", "-m", "Old project") self.remote = self.base / "project.git" self.call(self.base, "git", "init", "--bare", "--initial-branch=main", str(self.remote)) self.git(self.remote, "config", "gc.auto", "0") self.git(self.remote, "config", "receive.autogc", "false") self.git(self.repo, "remote", "add", "origin", str(self.remote)) self.git(self.repo, "push", "-u", "origin", "main") self.app, self.app_sha = self.component("App", {"content": "app v1\n", ".gitignore": "private/\n"}) self.rules, self.rules_sha = self.component("Rules", {"core/baseline.md": "# Baseline\n", "workflows/delivery.md": "# Delivery\n"}) def call(self, cwd, *args, check=True, env=None): result = subprocess.run(args, cwd=cwd, env=env or self.env, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=60) if check: self.assertEqual(result.returncode, 0, result.stdout) return result def git(self, cwd, *args, **kwargs): return self.call(cwd, "git", *args, **kwargs) def task(self, cwd, command, *args, **kwargs): return self.call(cwd, "bash", str(cwd / "scripts/task.sh"), command, *args, **kwargs) def component(self, name, files): path = self.base / name self.call(self.base, "git", "init", "--initial-branch=main", str(path)) for relative, content in files.items(): target = path / relative target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content) self.git(path, "add", ".") self.git(path, "commit", "-m", name + " v1") return path, self.git(path, "rev-parse", "HEAD").stdout.strip() def adopt(self): records = [{"id": name, "repo": str(path), "path": ".workspace/" + name, "legacy_path": "modules/" + name, "role": role, "modules": [], "target": "main"} for name, path, role in (("App", self.app, "application"), ("Rules", self.rules, "guidelines"))] (self.repo / "components.json").write_text(json.dumps({"schema": 1, "components": records})) (self.repo / "components.lock.json").write_text(json.dumps({"schema": 1, "components": {"App": {"commit": self.app_sha}, "Rules": {"commit": self.rules_sha}}})) self.git(self.repo, "add", "components.json", "components.lock.json") self.git(self.repo, "commit", "-m", "Adopt source manifests") self.git(self.repo, "push", "origin", "main") def worktree(self, name="feature/old"): path = self.base / "task" self.git(self.repo, "worktree", "add", "-b", name, str(path), "origin/main") return path def workspace(self, root, *args): return self.call(root, "python3", "scripts/workspace.py", "--root", str(root), *args) def test_precheck_prepares_fixed_rules_outside_project_before_ready(self): self.adopt() result = self.task(self.repo, "precheck") self.assertIn("[verdict] READY", result.stdout) self.assertFalse((self.repo / ".workspace").exists()) self.assertIn("[rules]", result.stdout) reading = next(line for line in result.stdout.splitlines() if line.startswith("[rules]")) self.assertIn(self.rules_sha, reading) self.assertTrue(list((self.base / "cache").rglob("core/baseline.md"))) self.assertTrue(list((self.base / "cache").rglob("workflows/delivery.md"))) def test_precheck_never_claims_ready_for_unavailable_fixed_rules(self): self.adopt() (self.rules / "core/baseline.md").unlink() self.git(self.rules, "commit", "-am", "Rules missing baseline") document = json.loads((self.repo / "components.lock.json").read_text()) document["components"]["Rules"]["commit"] = self.git(self.rules, "rev-parse", "HEAD").stdout.strip() (self.repo / "components.lock.json").write_text(json.dumps(document)) self.git(self.repo, "commit", "-am", "Pin invalid rules") self.git(self.repo, "push", "origin", "main") result = self.task(self.repo, "precheck", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertNotIn("[verdict] READY", result.stdout) self.assertFalse((self.repo / ".workspace").exists()) def test_start_prepares_fixed_sources_without_recursive_submodule_init(self): nested, _ = self.component("Nested", {"value": "nested\n"}) self.git(self.app, "submodule", "add", str(nested), "deps/Nested") self.git(self.app, "commit", "-am", "Pin nested source") self.app_sha = self.git(self.app, "rev-parse", "HEAD").stdout.strip() self.adopt() receipt = self.base / "precheck.json" self.call(self.repo, "bash", "scripts/evidence.sh", env=self.env | {"RECEIPT": str(receipt), "CMD": "make precheck"}) self.task(self.repo, "start", "feature/new", env=self.env | {"PRECHECK_RECEIPT": str(receipt)}) wt = self.base / "repo-wt-feature-new" self.assertTrue((wt / ".workspace/App/.git").is_dir(), "fixed App checkout is missing") self.assertEqual(self.git(wt / ".workspace/App", "rev-parse", "HEAD").stdout.strip(), self.app_sha) self.assertEqual(self.git(wt / ".workspace/Rules", "rev-parse", "HEAD").stdout.strip(), self.rules_sha) self.assertFalse((wt / ".workspace/App/deps/Nested/.git").exists()) self.assertTrue((wt / ".task/feature-new.md").exists()) def test_old_task_sync_adopts_new_sources_and_reports_explicit_transition(self): self.git(self.repo, "-c", "core.excludesfile=/dev/null", "submodule", "add", "--force", str(self.app), "modules/App") self.git(self.repo, "commit", "-m", "Pin old source") self.git(self.repo, "push", "origin", "main") wt = self.worktree() self.git(wt, "submodule", "update", "--init") legacy = wt / "modules/App" self.git(legacy, "switch", "-c", "feature/owned") before = self.git(legacy, "show-ref").stdout self.git(self.repo, "rm", "--cached", "modules/App") self.git(self.repo, "rm", ".gitmodules") self.git(self.repo, "commit", "-m", "Retire gitlink without removing checkout") self.adopt() result = self.task(wt, "sync") self.assertIn("[transition]", result.stdout) self.assertEqual(self.git(legacy, "show-ref").stdout, before) self.assertEqual(self.git(legacy, "branch", "--show-current").stdout.strip(), "feature/owned") self.assertTrue((wt / ".workspace/App/.git").is_dir()) self.assertEqual((legacy / "content").read_text(), "app v1\n") def test_sync_preserves_edit_branch_and_refuses_changed_declared_source(self): self.adopt() wt = self.worktree() self.workspace(wt, "edit", "--name", "App", "--branch", "feature/owned") edited = wt / ".workspace/App" (edited / "content").write_text("local unsaved\n") self.task(wt, "sync") (self.app / "content").write_text("app v2\n") self.git(self.app, "commit", "-am", "App v2") lock = json.loads((self.repo / "components.lock.json").read_text()) lock["components"]["App"]["commit"] = self.git(self.app, "rev-parse", "HEAD").stdout.strip() (self.repo / "components.lock.json").write_text(json.dumps(lock)) self.git(self.repo, "commit", "-am", "Update application lock") self.git(self.repo, "push", "origin", "main") result = self.task(wt, "sync", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertNotIn("[synced]", result.stdout) self.assertEqual((edited / "content").read_text(), "local unsaved\n") self.assertEqual(self.git(edited, "branch", "--show-current").stdout.strip(), "feature/owned") def test_finish_preserves_external_worktree_of_retired_legacy_checkout(self): self.adopt() wt = self.worktree() legacy = wt / "modules/App" self.git(wt, "clone", str(self.app), str(legacy)) external = self.base / "external" self.git(legacy, "worktree", "add", "-b", "feature/external", str(external)) (external / "content").write_text("external unique\n") self.git(external, "add", "content") before = self.git(external, "status", "--porcelain").stdout result = self.task(wt, "finish", check=False, env=self.env | {"TASK_WRITERS_STOPPED": "1"}) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("registered component worktree", result.stdout) self.assertEqual(self.git(external, "status", "--porcelain").stdout, before) self.assertEqual((external / "content").read_text(), "external unique\n") self.assertTrue(wt.exists()) def test_verify_detects_active_git_operation_in_nested_retired_root(self): self.adopt() wt = self.worktree() legacy = wt / "modules/App" self.git(wt, "clone", str(self.app), str(legacy)) nested = legacy / "retired/nested" self.git(wt, "clone", str(self.rules), str(nested)) lock = nested / ".git/index.lock" lock.write_text("another writer\n") result = self.call(wt, "python3", "scripts/check-worktree.py", "verify", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("active Git operation/lock", result.stdout) self.assertEqual(lock.read_text(), "another writer\n") def test_make_workspace_commands_prepare_edit_unlink_and_report(self): self.adopt() shutil.copyfile(SOURCE / "Makefile", self.repo / "Makefile") self.call(self.repo, "make", "workspace-prepare", "NAME=App") source = self.repo / ".workspace/App" self.assertEqual((source / "content").read_text(), "app v1\n") self.call(self.repo, "make", "component-edit", "NAME=App", "BRANCH=feature/from-make") self.assertEqual(self.git(source, "branch", "--show-current").stdout.strip(), "feature/from-make") self.call(self.repo, "make", "component-unlink", "NAME=App") self.assertTrue(source.is_dir()) status = self.call(self.repo, "make", "workspace-status", "NAME=App") self.assertIn('"feature/from-make"', status.stdout) transition = self.call(self.repo, "make", "workspace-transition", "NAME=App") self.assertIn('"legacy"', transition.stdout) self.assertFalse((self.repo / "modules/App").exists()) def test_finish_preserves_absorbed_metadata_after_legacy_checkout_disappears(self): self.git(self.repo, "submodule", "add", "--force", str(self.app), "modules/App") self.git(self.repo, "commit", "-m", "Old gitlink") self.git(self.repo, "push", "origin", "main") wt = self.worktree() self.git(wt, "submodule", "update", "--init") legacy = wt / "modules/App" external = self.base / "external" self.git(legacy, "worktree", "add", "-b", "feature/external", str(external)) (external / "content").write_text("independent owner\n") self.git(external, "add", "content") before = self.git(external, "status", "--porcelain").stdout self.git(self.repo, "rm", "--cached", "modules/App") self.git(self.repo, "rm", ".gitmodules") self.git(self.repo, "commit", "-m", "Remove legacy index entries") self.adopt() self.task(wt, "sync") shutil.rmtree(legacy) # Simulate an absent checkout while its Git metadata remains. result = self.task(wt, "finish", check=False, env=self.env | {"TASK_WRITERS_STOPPED": "1"}) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("retained metadata", result.stdout) self.assertEqual(self.git(external, "status", "--porcelain").stdout, before) self.assertEqual((external / "content").read_text(), "independent owner\n") def test_finish_preserves_nonbranch_legacy_refs_without_remote_backup(self): self.adopt() wt = self.worktree() legacy = wt / "modules/App" self.git(wt, "clone", str(self.app), str(legacy)) (legacy / "content").write_text("unpublished\n") self.git(legacy, "commit", "-am", "Unique migration object") unique = self.git(legacy, "rev-parse", "HEAD").stdout.strip() self.git(legacy, "update-ref", "refs/migration/unique", unique) self.git(legacy, "reset", "--hard", "origin/main") self.git(legacy, "reflog", "expire", "--expire=now", "--all") result = self.task(wt, "finish", check=False, env=self.env | {"TASK_WRITERS_STOPPED": "1"}) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("unpreserved local refs", result.stdout) self.assertEqual(self.git(legacy, "rev-parse", "refs/migration/unique").stdout.strip(), unique) @unittest.skipUnless(shutil.which("go"), "Go is required for architecture gate") def test_architecture_gate_preserves_edges_under_workspace_paths_and_new_module_identity(self): self.adopt() policy = self.repo / "policy" policy.mkdir(exist_ok=True) (policy / "architecture.yaml").write_text("backend:\n root: .workspace/Agent-System-Backend/internal\n import_prefix: git-ssh.u1680832.nyat.app/weichen/Agent-System-Backend.git/internal/\nfrontend:\n root: .workspace/Agent-System-Frontend/src\n groups: components\n") (policy / "architecture-baseline.json").write_text('["backend/app backend/store"]') backend = self.repo / ".workspace/Agent-System-Backend" (backend / "internal/app").mkdir(parents=True) (backend / "internal/store").mkdir() (backend / "go.mod").write_text("module git-ssh.u1680832.nyat.app/weichen/Agent-System-Backend.git\n\ngo 1.23\n") source = backend / "internal/app/app.go" source.write_text('package app\nimport _ "git-ssh.u1680832.nyat.app/weichen/Agent-System-Backend.git/internal/store"\n') (self.repo / ".workspace/Agent-System-Frontend/src").mkdir(parents=True) self.call(self.repo, "bash", "scripts/lint-architecture.sh") (backend / "internal/server").mkdir() source.write_text(source.read_text() + 'import _ "git-ssh.u1680832.nyat.app/weichen/Agent-System-Backend.git/internal/server"\n') result = self.call(self.repo, "bash", "scripts/lint-architecture.sh", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("backend/app backend/server", result.stdout) self.assertEqual(json.loads((policy / "architecture-baseline.json").read_text()), ["backend/app backend/store"]) @unittest.skipUnless(shutil.which("go"), "Go is required for boundary gates") def test_boundary_gate_uses_isolated_map_and_rejects_new_backend_internal_import(self): self.adopt() names = ("Agent-System-Backend", "Agent-System-Frontend", "Agent-Intent", "Agent-Native", "Agent-Squad") modules = {name: "git-ssh.u1680832.nyat.app/weichen/" + name + ".git" for name in names} entries = [{"id": name, "repo": str(self.app), "path": ".workspace/" + name, "role": "application" if name in names[:2] else "go-library", "modules": [], "target": "main"} for name in names] (self.repo / "components.json").write_text(json.dumps({"schema": 1, "components": entries})) (self.repo / "components.lock.json").write_text(json.dumps({"schema": 1, "components": {name: {"commit": self.app_sha} for name in names[:2]}})) policy = self.repo / "policy" policy.mkdir() (policy / "architecture.yaml").write_text("backend: .workspace/Agent-System-Backend/internal\nfrontend: .workspace/Agent-System-Frontend/src\ncomponents: {}\n") (policy / "architecture-baseline.json").write_text("[]") (policy / "lib-deps-allowlist.txt").write_text("# No sibling-library imports\n") mapping = {"schema": 1, "root_commit": self.git(self.repo, "rev-parse", "HEAD").stdout.strip(), "components": {}} for name in names: path = self.base / "isolated" / name path.mkdir(parents=True) if name != "Agent-System-Frontend": (path / "go.mod").write_text("module " + modules[name] + "\n\ngo 1.23\n") (path / "source.go").write_text("package fixture\n") mapping["components"][name] = {"path": str(path), "commit": self.app_sha, "repo": str(self.app)} backend = self.base / "isolated/Agent-System-Backend" (backend / "internal/store").mkdir(parents=True) (backend / "internal/store/store.go").write_text("package store\n") sdk = backend / "pkg/extension" sdk.mkdir(parents=True) source = sdk / "sdk.go" source.write_text("package extension\n") (self.base / "isolated/Agent-System-Frontend/src").mkdir() source_map = self.base / "source-map.json" source_map.write_text(json.dumps(mapping)) env = self.env | {"COMPOSITION_SOURCE_MAP": str(source_map)} self.call(self.repo, "bash", "scripts/lint-boundaries.sh", env=env) source.write_text('package extension\nimport _ "' + modules["Agent-System-Backend"] + '/internal/store"\n') result = self.call(self.repo, "bash", "scripts/lint-boundaries.sh", env=env, check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("boundary: pkg/extension imports", result.stdout) # A forged source identity cannot substitute for the locked application. source.write_text("package extension\n") mapping["components"]["Agent-System-Backend"]["commit"] = "0" * 40 source_map.write_text(json.dumps(mapping)) result = self.call(self.repo, "bash", "scripts/lint-boundaries.sh", env=env, check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("locked source", result.stdout) def test_verify_detects_nested_git_operation_inside_new_workspace(self): self.adopt() wt = self.worktree() self.workspace(wt, "prepare") nested = wt / ".workspace/App/private/Rules" self.git(wt, "clone", str(self.rules), str(nested)) lock = nested / ".git/index.lock" lock.write_text("active nested writer\n") result = self.call(wt, "python3", "scripts/check-worktree.py", "verify", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertIn("active Git operation/lock", result.stdout) self.assertEqual(lock.read_text(), "active nested writer\n") def test_old_task_sync_preserves_unique_legacy_commit_staging_dirty_and_ignored_files(self): self.git(self.repo, "submodule", "add", "--force", str(self.app), "modules/App") self.git(self.repo, "commit", "-m", "Old gitlink") self.git(self.repo, "push", "origin", "main") wt = self.worktree() self.git(wt, "submodule", "update", "--init") legacy = wt / "modules/App" self.git(legacy, "switch", "-c", "feature/unique") (legacy / "content").write_text("unpublished commit\n") self.git(legacy, "commit", "-am", "Unique component history") (legacy / "content").write_text("staged component change\n") self.git(legacy, "add", "content") (legacy / "content").write_text("unstaged component change\n") (legacy / "private").mkdir() (legacy / "private/unique").write_text("ignored unique data\n") before = (self.git(legacy, "show-ref").stdout, self.git(legacy, "ls-files", "--stage").stdout, self.git(wt, "rev-parse", "HEAD").stdout) self.git(self.repo, "rm", "--cached", "modules/App") self.git(self.repo, "rm", ".gitmodules") self.git(self.repo, "commit", "-m", "Retire old gitlink") self.adopt() result = self.task(wt, "sync", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertNotIn("[synced]", result.stdout) self.assertEqual((self.git(legacy, "show-ref").stdout, self.git(legacy, "ls-files", "--stage").stdout, self.git(wt, "rev-parse", "HEAD").stdout), before) self.assertEqual((legacy / "content").read_text(), "unstaged component change\n") self.assertEqual((legacy / "private/unique").read_text(), "ignored unique data\n") def test_old_task_text_conflict_keeps_legacy_source_and_reports_no_success(self): self.git(self.repo, "submodule", "add", "--force", str(self.app), "modules/App") self.git(self.repo, "commit", "-m", "Old gitlink") self.git(self.repo, "push", "origin", "main") wt = self.worktree() self.git(wt, "submodule", "update", "--init") (wt / "tracked").write_text("task branch choice\n") self.git(wt, "commit", "-am", "Task text change") legacy = wt / "modules/App" before = self.git(legacy, "show-ref").stdout (self.repo / "tracked").write_text("main branch choice\n") self.git(self.repo, "commit", "-am", "Main text change") self.git(self.repo, "rm", "--cached", "modules/App") self.git(self.repo, "rm", ".gitmodules") self.git(self.repo, "commit", "-m", "Retire legacy index") self.adopt() result = self.task(wt, "sync", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertNotIn("[synced]", result.stdout) self.assertTrue(self.git(wt, "ls-files", "-u").stdout.strip()) self.assertEqual(self.git(legacy, "show-ref").stdout, before) self.assertEqual((legacy / "content").read_text(), "app v1\n") def test_precheck_rejects_borrowed_rules_cache_symlink(self): self.adopt() self.task(self.repo, "precheck") source = next(path for path in (self.base / "cache/agent-system/rules").iterdir() if path.is_dir()) borrowed = self.base / "borrowed-rules" source.rename(borrowed) source.symlink_to(borrowed, target_is_directory=True) before = self.git(borrowed, "show-ref").stdout result = self.task(self.repo, "precheck", check=False) self.assertNotEqual(result.returncode, 0, result.stdout) self.assertNotIn("[verdict] READY", result.stdout) self.assertEqual(self.git(borrowed, "show-ref").stdout, before) self.assertEqual((borrowed / "core/baseline.md").read_text(), "# Baseline\n") if __name__ == "__main__": unittest.main()