from __future__ import annotations import subprocess from datetime import date from pathlib import Path import httpx from repo_manager.commands.task import add_adhoc_task, mutate_task from repo_manager.commands.workplan import create_workplan from repo_manager.identifiers import ( derive_work_record_uuid, ensure_missing_work_record_identifiers, ) from repo_manager.projection_sync import sync_repository_projection def _git(repo: Path, *args: str) -> None: subprocess.run(["git", *args], cwd=repo, check=True, capture_output=True) def _fixture(tmp_path: Path, *, remote: bool = False) -> Path: repo = tmp_path / "pilot" repo.mkdir() _git(repo, "init") _git(repo, "config", "user.email", "test@example.com") _git(repo, "config", "user.name", "Test") (repo / "workplans").mkdir() (repo / ".repo-classification.yaml").write_text( "repo_classification:\n category: tooling\n domain: infotech\n", encoding="utf-8", ) (repo / "workplans" / "P-WP-0001-first.md").write_text( """--- id: P-WP-0001 type: workplan title: First domain: infotech repo: pilot status: active owner: codex --- # First ## Existing task ```task id: P-WP-0001-T01 status: todo priority: high ``` Existing task details. """, encoding="utf-8", ) _git(repo, "add", ".") _git(repo, "commit", "-m", "seed") if remote: bare = tmp_path / "remote.git" subprocess.run(["git", "init", "--bare", str(bare)], check=True, capture_output=True) _git(repo, "remote", "add", "origin", str(bare)) _git(repo, "push", "-u", "origin", "HEAD") return repo def test_adhoc_ids_are_canonical_and_stable() -> None: identifier = "P-WP-ADHOC-2026-08-30" task_id = f"{identifier}-T01" assert derive_work_record_uuid("helixforge", identifier) == derive_work_record_uuid( "helixforge", identifier ) assert derive_work_record_uuid("helixforge", task_id).version == 5 def test_missing_assignment_preserves_existing_uuid(tmp_path: Path) -> None: repo = _fixture(tmp_path) existing = "11111111-1111-4111-8111-111111111111" path = repo / "workplans" / "P-WP-0001-first.md" text = path.read_text(encoding="utf-8").replace( "owner: codex\n", f'owner: codex\nstate_hub_workstream_id: "{existing}"\n' ) path.write_text(text, encoding="utf-8") result = ensure_missing_work_record_identifiers(repo) assert result["ok"] is True assert {item["record_id"] for item in result["assignments"]} == {"P-WP-0001-T01"} assert existing in path.read_text(encoding="utf-8") def test_create_adhoc_uses_conventional_filename_and_uuid(tmp_path: Path) -> None: repo = _fixture(tmp_path) result = create_workplan( repo, "P-WP-ADHOC-2026-08-30", "Ad Hoc — 2026-08-30", "Small fixes.", ) assert result.status == "applied" path = repo / "workplans" / "ADHOC-2026-08-30.md" assert path.is_file() assert str(derive_work_record_uuid("helixforge", "P-WP-ADHOC-2026-08-30")) in path.read_text( encoding="utf-8" ) def test_task_add_update_and_adhoc_replay(tmp_path: Path, monkeypatch) -> None: repo = _fixture(tmp_path) monkeypatch.setenv("RM_IDEMPOTENCY_PATH", str(tmp_path / "idempotency.json")) added = mutate_task( repo, None, operation="add", workplan_id="P-WP-0001", title="Second task", description="Do the second thing.", priority="high", ) assert added.status == "applied" assert added.evidence["task_id"] == "P-WP-0001-T02" updated = mutate_task( repo, "P-WP-0001-T02", operation="update", status="progress", needs_human=True, intervention_note="Review the external effect.", ) assert updated.status == "applied" text = (repo / "workplans" / "P-WP-0001-first.md").read_text(encoding="utf-8") assert "status: progress" in text assert "needs_human: true" in text first = add_adhoc_task( repo, "Tiny fix", "Complete the tiny fix.", on_date=date(2026, 8, 30), idempotency_key="adhoc-1", ) replay = add_adhoc_task( repo, "Tiny fix", "Complete the tiny fix.", on_date=date(2026, 8, 30), idempotency_key="adhoc-1", ) assert first.status == replay.status == "applied" assert first.evidence["task_id"] == replay.evidence["task_id"] def test_sync_uses_two_requests_and_exact_pushed_commit(tmp_path: Path) -> None: repo = _fixture(tmp_path, remote=True) seen: list[str] = [] def handler(request: httpx.Request) -> httpx.Response: seen.append(request.url.path) if request.url.path == "/state/health": return httpx.Response( 200, json={"status": "ok", "instance_role": "primary", "instance_label": "railliance01"}, ) payload = __import__("json").loads(request.content) assert ( payload["expected_commit"] == subprocess.run( ["git", "rev-parse", "HEAD"], cwd=repo, check=True, capture_output=True, text=True ).stdout.strip() ) return httpx.Response( 200, json={ "outcome": {"status": "applied", "counts": {"created": 1}}, "derived_commit": payload["expected_commit"], }, ) result = sync_repository_projection( repo, api_base="http://hub.test", push=True, transport=httpx.MockTransport(handler), ) assert result["ok"] is True assert result["requests"] == 2 assert seen == ["/state/health", "/repos/pilot/work-record-projection/reconcile"] def test_sync_never_assigns_or_commits_over_dirty_workplans(tmp_path: Path) -> None: repo = _fixture(tmp_path, remote=True) path = repo / "workplans" / "P-WP-0001-first.md" original = path.read_text(encoding="utf-8") path.write_text(original.replace("First\n", "Locally edited\n", 1), encoding="utf-8") result = sync_repository_projection(repo, api_base="http://hub.test", push=True) assert result["ok"] is False assert result["status"] == "pending_commit" assert "state_hub_workstream_id" not in path.read_text(encoding="utf-8") assert ( subprocess.run( ["git", "log", "-1", "--pretty=%s"], cwd=repo, check=True, capture_output=True, text=True, ).stdout.strip() == "seed" ) def test_sync_queues_instead_of_writing_to_wrong_instance(tmp_path: Path) -> None: repo = _fixture(tmp_path, remote=True) def handler(_request: httpx.Request) -> httpx.Response: return httpx.Response( 200, json={"status": "ok", "instance_role": "cache", "instance_label": "workstation"}, ) result = sync_repository_projection( repo, api_base="http://cache.test", push=True, transport=httpx.MockTransport(handler), ) assert result["ok"] is True assert result["status"] == "queued" assert result["reason"] == "wrong_instance" assert Path(result["pending_path"]).is_file()