Add project registration endpoint

.env.example

@@ -13,6 +13,7 @@ ATOCORE_SOURCE_DRIVE_ENABLED=true
 ATOCORE_LOG_DIR=./logs
 ATOCORE_BACKUP_DIR=./backups
 ATOCORE_RUN_DIR=./run
+ATOCORE_PROJECT_REGISTRY_DIR=./config
 ATOCORE_PROJECT_REGISTRY_PATH=./config/project-registry.json
 ATOCORE_HOST=127.0.0.1
 ATOCORE_PORT=8100

config/project-registry.example.json

@@ -0,0 +1,21 @@
+{
+  "projects": [
+    {
+      "id": "p07-example",
+      "aliases": ["p07", "example-project"],
+      "description": "Short description of the project and the staged source set.",
+      "ingest_roots": [
+        {
+          "source": "vault",
+          "subpath": "incoming/projects/p07-example",
+          "label": "Primary staged project docs"
+        },
+        {
+          "source": "drive",
+          "subpath": "projects/p07-example",
+          "label": "Trusted operational docs"
+        }
+      ]
+    }
+  ]
+}

deploy/dalidou/docker-compose.yml

@@ -17,6 +17,7 @@ services:
       - ${ATOCORE_LOG_DIR}:${ATOCORE_LOG_DIR}
       - ${ATOCORE_BACKUP_DIR}:${ATOCORE_BACKUP_DIR}
       - ${ATOCORE_RUN_DIR}:${ATOCORE_RUN_DIR}
+      - ${ATOCORE_PROJECT_REGISTRY_DIR}:${ATOCORE_PROJECT_REGISTRY_DIR}
       - ${ATOCORE_VAULT_SOURCE_DIR}:${ATOCORE_VAULT_SOURCE_DIR}:ro
       - ${ATOCORE_DRIVE_SOURCE_DIR}:${ATOCORE_DRIVE_SOURCE_DIR}:ro
     healthcheck:

docs/next-steps.md

@@ -31,6 +31,7 @@ AtoCore now has:
      explicit
    - move toward a project source registry and refresh workflow
    - foundation now exists via project registry + per-project refresh API
+   - registration policy + template are now the next normal path for new projects
 5. Define backup and export procedures for Dalidou
    - SQLite snapshot/backup strategy
    - Chroma backup or rebuild policy

docs/project-registration-policy.md

@@ -0,0 +1,109 @@
+# AtoCore Project Registration Policy
+
+## Purpose
+
+This document defines the normal path for adding a new project to AtoCore.
+
+The goal is to make `register + refresh` the standard workflow instead of
+relying on long custom ingestion prompts every time.
+
+## What Registration Means
+
+Registering a project does not ingest it by itself.
+
+Registration means:
+
+- the project gets a canonical AtoCore id
+- known aliases are recorded
+- the staged source roots for that project are defined
+- AtoCore and OpenClaw can later refresh that project consistently
+
+## Required Fields
+
+Each project registry entry must include:
+
+- `id`
+  - stable canonical project id
+  - prefer lowercase kebab-case
+  - examples:
+    - `p04-gigabit`
+    - `p05-interferometer`
+    - `p06-polisher`
+- `aliases`
+  - short common names or abbreviations
+  - examples:
+    - `p05`
+    - `interferometer`
+- `description`
+  - short explanation of what the registered source set represents
+- `ingest_roots`
+  - one or more staged roots under configured source layers
+
+## Allowed Source Roots
+
+Current allowed `source` values are:
+
+- `vault`
+- `drive`
+
+These map to the configured Dalidou source boundaries.
+
+## Recommended Registration Rules
+
+1. Prefer one canonical project id
+2. Keep aliases short and practical
+3. Start with the smallest useful staged roots
+4. Prefer curated high-signal docs before broad corpora
+5. Keep repo context selective at first
+6. Avoid registering noisy or generated trees
+7. Use `drive` for trusted operational material when available
+8. Use `vault` for curated staged PKM and repo-doc snapshots
+
+## Normal Workflow
+
+For a new project:
+
+1. stage the initial source docs on Dalidou
+2. inspect the expected shape with:
+   - `GET /projects/template`
+   - or `atocore.sh project-template`
+3. preview the entry without mutating state:
+   - `POST /projects/proposal`
+   - or `atocore.sh propose-project ...`
+4. register the approved entry:
+   - `POST /projects/register`
+   - or `atocore.sh register-project ...`
+5. verify the entry with:
+   - `GET /projects`
+   - or the T420 helper `atocore.sh projects`
+6. refresh it with:
+   - `POST /projects/{id}/refresh`
+   - or `atocore.sh refresh-project <id>`
+7. verify retrieval and context quality
+8. only later promote stable facts into Trusted Project State
+
+## What Not To Do
+
+Do not:
+
+- register giant noisy trees blindly
+- treat registration as equivalent to trusted state
+- dump the full PKM by default
+- rely on aliases that collide across projects
+- use the live machine DB as a source root
+
+## Template
+
+Use:
+
+- [project-registry.example.json](C:/Users/antoi/ATOCore/config/project-registry.example.json)
+
+And the API template endpoint:
+
+- `GET /projects/template`
+
+Other lifecycle endpoints:
+
+- `POST /projects/proposal`
+- `POST /projects/register`
+- `POST /projects/{id}/refresh`

docs/source-refresh-model.md

@@ -84,6 +84,9 @@ The first concrete foundation for this now exists in AtoCore:
 
 - a project registry file records known project ids, aliases, and ingest roots
 - the API can list those registered projects
+- the API can return a registration template for new projects
+- the API can preview a proposed registration before writing it
+- the API can persist an approved registration to the registry
 - the API can refresh a single registered project from its configured roots
 
 This is not full source automation yet, but it gives the refresh model a real

src/atocore/api/routes.py

@@ -34,7 +34,10 @@ from atocore.memory.service import (
 )
 from atocore.observability.logger import get_logger
 from atocore.projects.registry import (
+    build_project_registration_proposal,
+    get_project_registry_template,
     list_registered_projects,
+    register_project,
     refresh_registered_project,
 )
 from atocore.retrieval.retriever import retrieve
@@ -67,6 +70,13 @@ class ProjectRefreshResponse(BaseModel):
     roots: list[dict]
 
 
+class ProjectRegistrationProposalRequest(BaseModel):
+    project_id: str
+    aliases: list[str] = []
+    description: str = ""
+    ingest_roots: list[dict]
+
+
 class QueryRequest(BaseModel):
     prompt: str
     top_k: int = 10
@@ -169,6 +179,44 @@ def api_projects() -> dict:
     }
 
 
+@router.get("/projects/template")
+def api_projects_template() -> dict:
+    """Return a starter template for project registry entries."""
+    return {
+        "template": get_project_registry_template(),
+        "registry_path": str(_config.settings.resolved_project_registry_path),
+        "allowed_sources": ["vault", "drive"],
+    }
+
+
+@router.post("/projects/proposal")
+def api_project_registration_proposal(req: ProjectRegistrationProposalRequest) -> dict:
+    """Return a normalized project registration proposal without writing it."""
+    try:
+        return build_project_registration_proposal(
+            project_id=req.project_id,
+            aliases=req.aliases,
+            description=req.description,
+            ingest_roots=req.ingest_roots,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
+@router.post("/projects/register")
+def api_project_registration(req: ProjectRegistrationProposalRequest) -> dict:
+    """Persist a validated project registration to the registry file."""
+    try:
+        return register_project(
+            project_id=req.project_id,
+            aliases=req.aliases,
+            description=req.description,
+            ingest_roots=req.ingest_roots,
+        )
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+
+
 @router.post("/projects/{project_name}/refresh", response_model=ProjectRefreshResponse)
 def api_refresh_project(project_name: str, purge_deleted: bool = False) -> ProjectRefreshResponse:
     """Refresh one registered project from its configured ingest roots."""

src/atocore/config.py

@@ -106,6 +106,7 @@ class Settings(BaseSettings):
             self.resolved_log_dir,
             self.resolved_backup_dir,
             self.resolved_run_dir,
+            self.resolved_project_registry_path.parent,
         ]
 
     @property

src/atocore/projects/registry.py

@@ -25,18 +25,113 @@ class RegisteredProject:
     ingest_roots: tuple[ProjectSourceRef, ...]
 
 
+def get_project_registry_template() -> dict:
+    """Return a minimal template for registering a new project."""
+    return {
+        "projects": [
+            {
+                "id": "p07-example",
+                "aliases": ["p07", "example-project"],
+                "description": "Short description of the project and staged corpus.",
+                "ingest_roots": [
+                    {
+                        "source": "vault",
+                        "subpath": "incoming/projects/p07-example",
+                        "label": "Primary staged project docs",
+                    }
+                ],
+            }
+        ]
+    }
+
+
+def build_project_registration_proposal(
+    project_id: str,
+    aliases: list[str] | tuple[str, ...] | None = None,
+    description: str = "",
+    ingest_roots: list[dict] | tuple[dict, ...] | None = None,
+) -> dict:
+    """Build a normalized project registration proposal without mutating state."""
+    normalized_id = project_id.strip()
+    if not normalized_id:
+        raise ValueError("Project id must be non-empty")
+
+    normalized_aliases = _normalize_aliases(aliases or [])
+    normalized_roots = _normalize_ingest_roots(ingest_roots or [])
+    if not normalized_roots:
+        raise ValueError("At least one ingest root is required")
+
+    collisions = _find_name_collisions(normalized_id, normalized_aliases)
+    resolved_roots = []
+    for root in normalized_roots:
+        source_ref = ProjectSourceRef(
+            source=root["source"],
+            subpath=root["subpath"],
+            label=root.get("label", ""),
+        )
+        resolved_path = _resolve_ingest_root(source_ref)
+        resolved_roots.append(
+            {
+                **root,
+                "path": str(resolved_path),
+                "exists": resolved_path.exists(),
+                "is_dir": resolved_path.is_dir(),
+            }
+        )
+
+    return {
+        "project": {
+            "id": normalized_id,
+            "aliases": normalized_aliases,
+            "description": description.strip(),
+            "ingest_roots": normalized_roots,
+        },
+        "resolved_ingest_roots": resolved_roots,
+        "collisions": collisions,
+        "registry_path": str(_config.settings.resolved_project_registry_path),
+        "valid": not collisions,
+    }
+
+
+def register_project(
+    project_id: str,
+    aliases: list[str] | tuple[str, ...] | None = None,
+    description: str = "",
+    ingest_roots: list[dict] | tuple[dict, ...] | None = None,
+) -> dict:
+    """Persist a validated project registration to the registry file."""
+    proposal = build_project_registration_proposal(
+        project_id=project_id,
+        aliases=aliases,
+        description=description,
+        ingest_roots=ingest_roots,
+    )
+    if not proposal["valid"]:
+        collision_names = ", ".join(collision["name"] for collision in proposal["collisions"])
+        raise ValueError(f"Project registration has collisions: {collision_names}")
+
+    registry_path = _config.settings.resolved_project_registry_path
+    payload = _load_registry_payload(registry_path)
+    payload.setdefault("projects", []).append(proposal["project"])
+    _write_registry_payload(registry_path, payload)
+
+    return {
+        **proposal,
+        "status": "registered",
+    }
+
+
 def load_project_registry() -> list[RegisteredProject]:
     """Load project registry entries from JSON config."""
     registry_path = _config.settings.resolved_project_registry_path
-    if not registry_path.exists():
+    payload = _load_registry_payload(registry_path)
-        return []
-
-    payload = json.loads(registry_path.read_text(encoding="utf-8"))
     entries = payload.get("projects", [])
     projects: list[RegisteredProject] = []
 
     for entry in entries:
         project_id = str(entry["id"]).strip()
+        if not project_id:
+            raise ValueError("Project registry entry is missing a non-empty id")
         aliases = tuple(
             alias.strip()
             for alias in entry.get("aliases", [])
@@ -53,6 +148,8 @@ def load_project_registry() -> list[RegisteredProject]:
             if str(root.get("source", "")).strip()
             and str(root.get("subpath", "")).strip()
         )
+        if not ingest_roots:
+            raise ValueError(f"Project registry entry '{project_id}' has no ingest_roots")
         projects.append(
             RegisteredProject(
                 project_id=project_id,
@@ -62,6 +159,7 @@ def load_project_registry() -> list[RegisteredProject]:
             )
         )
 
+    _validate_unique_project_names(projects)
     return projects
 
 
@@ -122,6 +220,35 @@ def refresh_registered_project(project_name: str, purge_deleted: bool = False) -
     }
 
 
+def _normalize_aliases(aliases: list[str] | tuple[str, ...]) -> list[str]:
+    deduped: list[str] = []
+    seen: set[str] = set()
+    for alias in aliases:
+        candidate = alias.strip()
+        if not candidate:
+            continue
+        key = candidate.lower()
+        if key in seen:
+            continue
+        seen.add(key)
+        deduped.append(candidate)
+    return deduped
+
+
+def _normalize_ingest_roots(ingest_roots: list[dict] | tuple[dict, ...]) -> list[dict]:
+    normalized: list[dict] = []
+    for root in ingest_roots:
+        source = str(root.get("source", "")).strip()
+        subpath = str(root.get("subpath", "")).strip()
+        label = str(root.get("label", "")).strip()
+        if not source or not subpath:
+            continue
+        if source not in {"vault", "drive"}:
+            raise ValueError(f"Unsupported source root: {source}")
+        normalized.append({"source": source, "subpath": subpath, "label": label})
+    return normalized
+
+
 def _project_to_dict(project: RegisteredProject) -> dict:
     return {
         "id": project.project_id,
@@ -150,3 +277,50 @@ def _resolve_ingest_root(source_ref: ProjectSourceRef) -> Path:
         raise ValueError(f"Unsupported source root: {source_ref.source}") from exc
 
     return (base_dir / source_ref.subpath).resolve(strict=False)
+
+
+def _validate_unique_project_names(projects: list[RegisteredProject]) -> None:
+    seen: dict[str, str] = {}
+    for project in projects:
+        names = [project.project_id, *project.aliases]
+        for name in names:
+            key = name.lower()
+            if key in seen and seen[key] != project.project_id:
+                raise ValueError(
+                    f"Project registry name collision: '{name}' is used by both "
+                    f"'{seen[key]}' and '{project.project_id}'"
+                )
+            seen[key] = project.project_id
+
+
+def _find_name_collisions(project_id: str, aliases: list[str]) -> list[dict]:
+    collisions: list[dict] = []
+    existing = load_project_registry()
+    requested_names = [project_id, *aliases]
+    for requested in requested_names:
+        requested_key = requested.lower()
+        for project in existing:
+            project_names = [project.project_id, *project.aliases]
+            if requested_key in {name.lower() for name in project_names}:
+                collisions.append(
+                    {
+                        "name": requested,
+                        "existing_project": project.project_id,
+                    }
+                )
+                break
+    return collisions
+
+
+def _load_registry_payload(registry_path: Path) -> dict:
+    if not registry_path.exists():
+        return {"projects": []}
+    return json.loads(registry_path.read_text(encoding="utf-8"))
+
+
+def _write_registry_payload(registry_path: Path, payload: dict) -> None:
+    registry_path.parent.mkdir(parents=True, exist_ok=True)
+    registry_path.write_text(
+        json.dumps(payload, indent=2, ensure_ascii=True) + "\n",
+        encoding="utf-8",
+    )

tests/test_api_storage.py

@@ -150,3 +150,147 @@ def test_project_refresh_endpoint_uses_registered_roots(tmp_data_dir, monkeypatc
     assert response.status_code == 200
     assert calls == [("p05", False)]
     assert response.json()["project"] == "p05-interferometer"
+
+
+def test_projects_template_endpoint_returns_template(tmp_data_dir, monkeypatch):
+    config.settings = config.Settings()
+
+    client = TestClient(app)
+    response = client.get("/projects/template")
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["allowed_sources"] == ["vault", "drive"]
+    assert body["template"]["projects"][0]["id"] == "p07-example"
+
+
+def test_project_proposal_endpoint_returns_normalized_preview(tmp_data_dir, monkeypatch):
+    vault_dir = tmp_data_dir / "vault-source"
+    drive_dir = tmp_data_dir / "drive-source"
+    config_dir = tmp_data_dir / "config"
+    staged = vault_dir / "incoming" / "projects" / "p07-example"
+    staged.mkdir(parents=True)
+    drive_dir.mkdir()
+    config_dir.mkdir()
+
+    registry_path = config_dir / "project-registry.json"
+    registry_path.write_text('{"projects": []}', encoding="utf-8")
+
+    monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir))
+    monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir))
+    monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path))
+    config.settings = config.Settings()
+
+    client = TestClient(app)
+    response = client.post(
+        "/projects/proposal",
+        json={
+            "project_id": "p07-example",
+            "aliases": ["p07", "example-project", "p07"],
+            "description": "Example project",
+            "ingest_roots": [
+                {
+                    "source": "vault",
+                    "subpath": "incoming/projects/p07-example",
+                    "label": "Primary docs",
+                }
+            ],
+        },
+    )
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["project"]["aliases"] == ["p07", "example-project"]
+    assert body["resolved_ingest_roots"][0]["exists"] is True
+    assert body["valid"] is True
+
+
+def test_project_register_endpoint_persists_entry(tmp_data_dir, monkeypatch):
+    vault_dir = tmp_data_dir / "vault-source"
+    drive_dir = tmp_data_dir / "drive-source"
+    config_dir = tmp_data_dir / "config"
+    staged = vault_dir / "incoming" / "projects" / "p07-example"
+    staged.mkdir(parents=True)
+    drive_dir.mkdir()
+    config_dir.mkdir()
+
+    registry_path = config_dir / "project-registry.json"
+    registry_path.write_text('{"projects": []}', encoding="utf-8")
+
+    monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir))
+    monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir))
+    monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path))
+    config.settings = config.Settings()
+
+    client = TestClient(app)
+    response = client.post(
+        "/projects/register",
+        json={
+            "project_id": "p07-example",
+            "aliases": ["p07", "example-project"],
+            "description": "Example project",
+            "ingest_roots": [
+                {
+                    "source": "vault",
+                    "subpath": "incoming/projects/p07-example",
+                    "label": "Primary docs",
+                }
+            ],
+        },
+    )
+
+    assert response.status_code == 200
+    body = response.json()
+    assert body["status"] == "registered"
+    assert body["project"]["id"] == "p07-example"
+    assert '"p07-example"' in registry_path.read_text(encoding="utf-8")
+
+
+def test_project_register_endpoint_rejects_collisions(tmp_data_dir, monkeypatch):
+    vault_dir = tmp_data_dir / "vault-source"
+    drive_dir = tmp_data_dir / "drive-source"
+    config_dir = tmp_data_dir / "config"
+    vault_dir.mkdir()
+    drive_dir.mkdir()
+    config_dir.mkdir()
+
+    registry_path = config_dir / "project-registry.json"
+    registry_path.write_text(
+        """
+{
+  "projects": [
+    {
+      "id": "p05-interferometer",
+      "aliases": ["p05", "interferometer"],
+      "ingest_roots": [
+        {"source": "vault", "subpath": "incoming/projects/p05-interferometer"}
+      ]
+    }
+  ]
+}
+""".strip(),
+        encoding="utf-8",
+    )
+
+    monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir))
+    monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir))
+    monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path))
+    config.settings = config.Settings()
+
+    client = TestClient(app)
+    response = client.post(
+        "/projects/register",
+        json={
+            "project_id": "p07-example",
+            "aliases": ["interferometer"],
+            "ingest_roots": [
+                {
+                    "source": "vault",
+                    "subpath": "incoming/projects/p07-example",
+                }
+            ],
+        },
+    )
+
+    assert response.status_code == 400
+    assert "collisions" in response.json()["detail"]

tests/test_config.py

@@ -50,6 +50,9 @@ def test_ensure_runtime_dirs_creates_machine_dirs_only(tmp_path, monkeypatch):
     monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(tmp_path / "drive-source"))
     monkeypatch.setenv("ATOCORE_LOG_DIR", str(tmp_path / "logs"))
     monkeypatch.setenv("ATOCORE_BACKUP_DIR", str(tmp_path / "backups"))
+    monkeypatch.setenv(
+        "ATOCORE_PROJECT_REGISTRY_PATH", str(tmp_path / "config" / "project-registry.json")
+    )
 
     original_settings = config.settings
     try:
@@ -63,6 +66,7 @@ def test_ensure_runtime_dirs_creates_machine_dirs_only(tmp_path, monkeypatch):
         assert config.settings.resolved_log_dir.exists()
         assert config.settings.resolved_backup_dir.exists()
         assert config.settings.resolved_run_dir.exists()
+        assert config.settings.resolved_project_registry_path.parent.exists()
         assert not config.settings.resolved_vault_source_dir.exists()
         assert not config.settings.resolved_drive_source_dir.exists()
     finally:

tests/test_project_registry.py

@@ -4,8 +4,11 @@ import json
 
 import atocore.config as config
 from atocore.projects.registry import (
+    build_project_registration_proposal,
     get_registered_project,
+    get_project_registry_template,
     list_registered_projects,
+    register_project,
     refresh_registered_project,
 )
 
@@ -150,3 +153,231 @@ def test_refresh_registered_project_ingests_registered_roots(tmp_path, monkeypat
     assert calls[0][0].endswith("p06-polisher")
     assert calls[0][1] is False
     assert result["roots"][0]["status"] == "ingested"
+
+
+def test_project_registry_template_has_expected_shape():
+    template = get_project_registry_template()
+    assert "projects" in template
+    assert template["projects"][0]["id"] == "p07-example"
+    assert template["projects"][0]["ingest_roots"][0]["source"] == "vault"
+
+
+def test_project_registry_rejects_alias_collision(tmp_path, monkeypatch):
+    vault_dir = tmp_path / "vault"
+    drive_dir = tmp_path / "drive"
+    config_dir = tmp_path / "config"
+    vault_dir.mkdir()
+    drive_dir.mkdir()
+    config_dir.mkdir()
+
+    registry_path = config_dir / "project-registry.json"
+    registry_path.write_text(
+        json.dumps(
+            {
+                "projects": [
+                    {
+                        "id": "p04-gigabit",
+                        "aliases": ["shared"],
+                        "ingest_roots": [
+                            {"source": "vault", "subpath": "incoming/projects/p04-gigabit"}
+                        ],
+                    },
+                    {
+                        "id": "p05-interferometer",
+                        "aliases": ["shared"],
+                        "ingest_roots": [
+                            {"source": "vault", "subpath": "incoming/projects/p05-interferometer"}
+                        ],
+                    },
+                ]
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir))
+    monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir))
+    monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path))
+
+    original_settings = config.settings
+    try:
+        config.settings = config.Settings()
+        try:
+            list_registered_projects()
+        except ValueError as exc:
+            assert "collision" in str(exc)
+        else:
+            raise AssertionError("Expected project registry collision to raise")
+    finally:
+        config.settings = original_settings
+
+
+def test_project_registration_proposal_normalizes_and_resolves_paths(tmp_path, monkeypatch):
+    vault_dir = tmp_path / "vault"
+    drive_dir = tmp_path / "drive"
+    config_dir = tmp_path / "config"
+    staged = vault_dir / "incoming" / "projects" / "p07-example"
+    staged.mkdir(parents=True)
+    drive_dir.mkdir()
+    config_dir.mkdir()
+    registry_path = config_dir / "project-registry.json"
+    registry_path.write_text(json.dumps({"projects": []}), encoding="utf-8")
+
+    monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir))
+    monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir))
+    monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path))
+
+    original_settings = config.settings
+    try:
+        config.settings = config.Settings()
+        proposal = build_project_registration_proposal(
+            project_id="p07-example",
+            aliases=["p07", "example-project", "p07"],
+            description="Example project",
+            ingest_roots=[
+                {
+                    "source": "vault",
+                    "subpath": "incoming/projects/p07-example",
+                    "label": "Primary docs",
+                }
+            ],
+        )
+    finally:
+        config.settings = original_settings
+
+    assert proposal["project"]["aliases"] == ["p07", "example-project"]
+    assert proposal["resolved_ingest_roots"][0]["exists"] is True
+    assert proposal["valid"] is True
+
+
+def test_project_registration_proposal_reports_collisions(tmp_path, monkeypatch):
+    vault_dir = tmp_path / "vault"
+    drive_dir = tmp_path / "drive"
+    config_dir = tmp_path / "config"
+    vault_dir.mkdir()
+    drive_dir.mkdir()
+    config_dir.mkdir()
+    registry_path = config_dir / "project-registry.json"
+    registry_path.write_text(
+        json.dumps(
+            {
+                "projects": [
+                    {
+                        "id": "p05-interferometer",
+                        "aliases": ["p05", "interferometer"],
+                        "ingest_roots": [
+                            {"source": "vault", "subpath": "incoming/projects/p05-interferometer"}
+                        ],
+                    }
+                ]
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir))
+    monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir))
+    monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path))
+
+    original_settings = config.settings
+    try:
+        config.settings = config.Settings()
+        proposal = build_project_registration_proposal(
+            project_id="p08-example",
+            aliases=["interferometer"],
+            ingest_roots=[
+                {"source": "vault", "subpath": "incoming/projects/p08-example"}
+            ],
+        )
+    finally:
+        config.settings = original_settings
+
+    assert proposal["valid"] is False
+    assert proposal["collisions"][0]["existing_project"] == "p05-interferometer"
+
+
+def test_register_project_persists_new_entry(tmp_path, monkeypatch):
+    vault_dir = tmp_path / "vault"
+    drive_dir = tmp_path / "drive"
+    config_dir = tmp_path / "config"
+    staged = vault_dir / "incoming" / "projects" / "p07-example"
+    staged.mkdir(parents=True)
+    drive_dir.mkdir()
+    config_dir.mkdir()
+    registry_path = config_dir / "project-registry.json"
+    registry_path.write_text(json.dumps({"projects": []}), encoding="utf-8")
+
+    monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir))
+    monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir))
+    monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path))
+
+    original_settings = config.settings
+    try:
+        config.settings = config.Settings()
+        result = register_project(
+            project_id="p07-example",
+            aliases=["p07", "example-project"],
+            description="Example project",
+            ingest_roots=[
+                {
+                    "source": "vault",
+                    "subpath": "incoming/projects/p07-example",
+                    "label": "Primary docs",
+                }
+            ],
+        )
+    finally:
+        config.settings = original_settings
+
+    assert result["status"] == "registered"
+    payload = json.loads(registry_path.read_text(encoding="utf-8"))
+    assert payload["projects"][0]["id"] == "p07-example"
+    assert payload["projects"][0]["aliases"] == ["p07", "example-project"]
+
+
+def test_register_project_rejects_collisions(tmp_path, monkeypatch):
+    vault_dir = tmp_path / "vault"
+    drive_dir = tmp_path / "drive"
+    config_dir = tmp_path / "config"
+    vault_dir.mkdir()
+    drive_dir.mkdir()
+    config_dir.mkdir()
+    registry_path = config_dir / "project-registry.json"
+    registry_path.write_text(
+        json.dumps(
+            {
+                "projects": [
+                    {
+                        "id": "p05-interferometer",
+                        "aliases": ["p05", "interferometer"],
+                        "ingest_roots": [
+                            {"source": "vault", "subpath": "incoming/projects/p05-interferometer"}
+                        ],
+                    }
+                ]
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    monkeypatch.setenv("ATOCORE_VAULT_SOURCE_DIR", str(vault_dir))
+    monkeypatch.setenv("ATOCORE_DRIVE_SOURCE_DIR", str(drive_dir))
+    monkeypatch.setenv("ATOCORE_PROJECT_REGISTRY_PATH", str(registry_path))
+
+    original_settings = config.settings
+    try:
+        config.settings = config.Settings()
+        try:
+            register_project(
+                project_id="p07-example",
+                aliases=["interferometer"],
+                ingest_roots=[
+                    {"source": "vault", "subpath": "incoming/projects/p07-example"}
+                ],
+            )
+        except ValueError as exc:
+            assert "collisions" in str(exc)
+        else:
+            raise AssertionError("Expected collision to prevent project registration")
+    finally:
+        config.settings = original_settings