feat(projects): projects repo + group owner reads; wire into ApplyService

Read side of §7 ownership, ahead of the claim logic.

- project_repo: `ProjectRepository` trait + `PostgresProjectRepository`
  (list / get_by_id / get_by_slug). Writes are NOT here — a claim registers
  its project via an in-tx upsert (next commit).
- group_repo: `list_with_owner()` LEFT-JOINs projects for each group's owner
  slug (backs `pic groups ls`); columns qualified since id/slug/name exist on
  both tables.
- ApplyService gains a `projects` handle; constructed in the picloud binary.
- tests/projects_repo: round-trip pinning list_with_owner (claimed→slug,
  unclaimed→None) and that ancestors() surfaces owner_project nearest-first —
  the fold input the claim path walks.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

crates/manager-core/src/apply_service.rs

@@ -653,6 +653,10 @@ pub struct ApplyService {
     pub apps: Arc<dyn AppRepository>,
     /// Group lookups (Phase 5): resolve a group slug→id for group/tree apply.
     pub groups: Arc<dyn crate::group_repo::GroupRepository>,
+    /// Project registry (§7 multi-repo ownership): the READ side. The claim
+    /// write path registers projects via `upsert_project_tx` inside the apply
+    /// transaction, not through this repo.
+    pub projects: Arc<dyn crate::project_repo::ProjectRepository>,
     /// Module resolver (Phase 4b): lexical, origin-aware module lookup used by
     /// the single-node dangling-import plan check (§5.5).
     pub modules: Arc<dyn picloud_shared::ModuleSource>,

crates/manager-core/src/group_repo.rs

@@ -58,6 +58,10 @@ pub trait GroupRepository: Send + Sync {
     /// Every group on the instance, ordered by name. The tree is small
     /// (org structure), so callers assemble the hierarchy in memory.
     async fn list(&self) -> Result<Vec<Group>, GroupRepositoryError>;
+    /// Every group paired with its owning project's slug (`None` when the node
+    /// is unclaimed — no manifest owns it). One LEFT JOIN instead of an N+1
+    /// per-group project lookup; backs `pic groups ls`' owner column (§7).
+    async fn list_with_owner(&self) -> Result<Vec<(Group, Option<String>)>, GroupRepositoryError>;
     async fn get_by_id(&self, id: GroupId) -> Result<Option<Group>, GroupRepositoryError>;
     async fn get_by_slug(&self, slug: &str) -> Result<Option<Group>, GroupRepositoryError>;
     /// Direct children groups of `parent`.
@@ -119,6 +123,24 @@ impl GroupRepository for PostgresGroupRepository {
         Ok(rows.into_iter().map(Into::into).collect())
     }
 
+    async fn list_with_owner(&self) -> Result<Vec<(Group, Option<String>)>, GroupRepositoryError> {
+        // Columns are qualified (`g.`) because `id`/`slug`/`name` exist on both
+        // groups and projects — a bare GROUP_COLS would be ambiguous here.
+        let rows = sqlx::query_as::<_, GroupWithOwnerRow>(
+            "SELECT g.id, g.parent_id, g.slug, g.name, g.description, \
+                    g.structure_version, g.created_at, g.updated_at, g.owner_project, \
+                    p.slug AS owner_slug \
+             FROM groups g LEFT JOIN projects p ON p.id = g.owner_project \
+             ORDER BY g.name",
+        )
+        .fetch_all(&self.pool)
+        .await?;
+        Ok(rows
+            .into_iter()
+            .map(|r| (r.group.into(), r.owner_slug))
+            .collect())
+    }
+
     async fn get_by_id(&self, id: GroupId) -> Result<Option<Group>, GroupRepositoryError> {
         let row = sqlx::query_as::<_, GroupRow>(&format!(
             "SELECT {GROUP_COLS} FROM groups WHERE id = $1"
@@ -353,6 +375,15 @@ struct GroupRow {
     owner_project: Option<Uuid>,
 }
 
+/// `list_with_owner` row: a `GroupRow` flattened with its owning project's
+/// slug (`NULL` when the group is unclaimed).
+#[derive(sqlx::FromRow)]
+struct GroupWithOwnerRow {
+    #[sqlx(flatten)]
+    group: GroupRow,
+    owner_slug: Option<String>,
+}
+
 impl From<GroupRow> for Group {
     fn from(r: GroupRow) -> Self {
         Self {

crates/manager-core/src/lib.rs

@@ -79,6 +79,7 @@ pub mod module_source;
 pub mod outbox_event_emitter;
 pub mod outbox_repo;
 pub mod principal_resolver;
+pub mod project_repo;
 pub mod pubsub_repo;
 pub mod pubsub_service;
 pub mod queue_repo;
@@ -223,6 +224,7 @@ pub use outbox_repo::{
     NewOutboxRow, OutboxRepo, OutboxRepoError, OutboxRow, OutboxSourceKind, PostgresOutboxRepo,
 };
 pub use principal_resolver::{AdminPrincipalResolver, PrincipalResolver, PrincipalResolverError};
+pub use project_repo::{PostgresProjectRepository, ProjectRepository, ProjectRepositoryError};
 pub use pubsub_repo::{PostgresPubsubRepo, PublishCtx, PubsubRepo, PubsubRepoError};
 pub use pubsub_service::{PubsubServiceImpl, SubscriberTokenConfig};
 pub use realtime_authority::RealtimeAuthorityImpl;

crates/manager-core/src/project_repo.rs

@@ -0,0 +1,92 @@
+//! Read access over the `projects` registry (§7 multi-repo ownership).
+//!
+//! A project is registered by the APPLY path — an upsert-by-slug inside the
+//! reconcile transaction (see `apply_service`), so a claim registers the
+//! project atomically with the write. This repo is the READ side only, used
+//! for ownership visibility (`pic projects ls`, the `pic groups ls` owner
+//! column) and tests. Writes deliberately do not live here.
+
+use async_trait::async_trait;
+use picloud_shared::{Project, ProjectId};
+use sqlx::PgPool;
+use uuid::Uuid;
+
+#[derive(Debug, thiserror::Error)]
+pub enum ProjectRepositoryError {
+    #[error("database error: {0}")]
+    Db(#[from] sqlx::Error),
+}
+
+#[async_trait]
+pub trait ProjectRepository: Send + Sync {
+    /// Every project on the instance, ordered by slug.
+    async fn list(&self) -> Result<Vec<Project>, ProjectRepositoryError>;
+    async fn get_by_id(&self, id: ProjectId) -> Result<Option<Project>, ProjectRepositoryError>;
+    async fn get_by_slug(&self, slug: &str) -> Result<Option<Project>, ProjectRepositoryError>;
+}
+
+pub struct PostgresProjectRepository {
+    pool: PgPool,
+}
+
+impl PostgresProjectRepository {
+    #[must_use]
+    pub fn new(pool: PgPool) -> Self {
+        Self { pool }
+    }
+}
+
+const PROJECT_COLS: &str = "id, slug, name, created_by, created_at";
+
+#[async_trait]
+impl ProjectRepository for PostgresProjectRepository {
+    async fn list(&self) -> Result<Vec<Project>, ProjectRepositoryError> {
+        let rows = sqlx::query_as::<_, ProjectRow>(&format!(
+            "SELECT {PROJECT_COLS} FROM projects ORDER BY slug"
+        ))
+        .fetch_all(&self.pool)
+        .await?;
+        Ok(rows.into_iter().map(Into::into).collect())
+    }
+
+    async fn get_by_id(&self, id: ProjectId) -> Result<Option<Project>, ProjectRepositoryError> {
+        let row = sqlx::query_as::<_, ProjectRow>(&format!(
+            "SELECT {PROJECT_COLS} FROM projects WHERE id = $1"
+        ))
+        .bind(id.into_inner())
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(row.map(Into::into))
+    }
+
+    async fn get_by_slug(&self, slug: &str) -> Result<Option<Project>, ProjectRepositoryError> {
+        let row = sqlx::query_as::<_, ProjectRow>(&format!(
+            "SELECT {PROJECT_COLS} FROM projects WHERE slug = $1"
+        ))
+        .bind(slug)
+        .fetch_optional(&self.pool)
+        .await?;
+        Ok(row.map(Into::into))
+    }
+}
+
+#[derive(sqlx::FromRow)]
+struct ProjectRow {
+    id: Uuid,
+    slug: String,
+    name: String,
+    created_by: Option<Uuid>,
+    created_at: chrono::DateTime<chrono::Utc>,
+}
+
+impl From<ProjectRow> for Project {
+    fn from(r: ProjectRow) -> Self {
+        Self {
+            id: r.id.into(),
+            slug: r.slug,
+            name: r.name,
+            created_by: r.created_by.map(Into::into),
+            created_at: r.created_at,
+        }
+    }
+}

crates/manager-core/tests/projects_repo.rs

@@ -0,0 +1,106 @@
+//! §7 repo round-trip: `list_with_owner` pairs each group with its owning
+//! project's slug (`None` when unclaimed), and `ancestors()` surfaces
+//! `owner_project` for every node — the nearest-claimed fold the ownership
+//! claim path (M1.3) relies on.
+//!
+//! Deterministic; skips cleanly when `DATABASE_URL` is unset.
+
+use picloud_manager_core::group_repo::{GroupRepository, PostgresGroupRepository};
+use picloud_manager_core::project_repo::{PostgresProjectRepository, ProjectRepository};
+use sqlx::postgres::PgPoolOptions;
+use sqlx::PgPool;
+use uuid::Uuid;
+
+async fn pool_or_skip() -> Option<PgPool> {
+    let Ok(url) = std::env::var("DATABASE_URL") else {
+        eprintln!("projects_repo: DATABASE_URL unset — skipping");
+        return None;
+    };
+    let pool = PgPoolOptions::new()
+        .max_connections(2)
+        .connect(&url)
+        .await
+        .expect("connect to DATABASE_URL");
+    sqlx::migrate!("./migrations")
+        .run(&pool)
+        .await
+        .expect("apply migrations");
+    Some(pool)
+}
+
+fn uniq(prefix: &str) -> String {
+    format!("{prefix}-{}", &Uuid::new_v4().to_string()[..8])
+}
+
+#[tokio::test]
+async fn list_with_owner_and_ancestors_surface_ownership() {
+    let Some(pool) = pool_or_skip().await else {
+        return;
+    };
+    let groups = PostgresGroupRepository::new(pool.clone());
+    let projects = PostgresProjectRepository::new(pool.clone());
+
+    // A project + a parent/child group pair; claim the CHILD only (a parent is
+    // deliberately left unclaimed to pin the `None` path).
+    let proj_slug = uniq("proj");
+    let (proj_id,): (Uuid,) =
+        sqlx::query_as("INSERT INTO projects (slug, name) VALUES ($1, $1) RETURNING id")
+            .bind(&proj_slug)
+            .fetch_one(&pool)
+            .await
+            .expect("insert project");
+
+    let parent = groups
+        .create(&uniq("parent"), "Parent", None, None)
+        .await
+        .expect("create parent");
+    let child_slug = uniq("child");
+    let child = groups
+        .create(&child_slug, "Child", None, Some(parent.id))
+        .await
+        .expect("create child");
+    sqlx::query("UPDATE groups SET owner_project = $1 WHERE id = $2")
+        .bind(proj_id)
+        .bind(child.id.into_inner())
+        .execute(&pool)
+        .await
+        .expect("claim child");
+
+    // list_with_owner: the child pairs with the project slug; the parent is
+    // unclaimed (None). (The list spans the whole instance — filter by slug.)
+    let listed = groups.list_with_owner().await.expect("list_with_owner");
+    let child_row = listed
+        .iter()
+        .find(|(g, _)| g.slug == child_slug)
+        .expect("child present in list_with_owner");
+    assert_eq!(child_row.1.as_deref(), Some(proj_slug.as_str()));
+    let parent_row = listed
+        .iter()
+        .find(|(g, _)| g.id == parent.id)
+        .expect("parent present");
+    assert_eq!(parent_row.1, None, "unclaimed group has no owner slug");
+
+    // ancestors(child) is nearest-first and carries owner_project on each node —
+    // the child holds its claim, the parent is None. This is the exact fold
+    // input the nearest-claimed resolution walks.
+    let chain = groups.ancestors(child.id).await.expect("ancestors");
+    assert_eq!(chain[0].id, child.id, "ancestors is nearest-first");
+    assert!(
+        chain[0].owner_project.is_some(),
+        "child ancestor row carries its claim"
+    );
+    let parent_in_chain = chain
+        .iter()
+        .find(|g| g.id == parent.id)
+        .expect("parent in chain");
+    assert!(parent_in_chain.owner_project.is_none());
+
+    // Project reads round-trip by slug and by id.
+    let by_slug = projects.get_by_slug(&proj_slug).await.expect("get_by_slug");
+    assert_eq!(by_slug.as_ref().map(|p| p.id.into_inner()), Some(proj_id));
+    let by_id = projects
+        .get_by_id(by_slug.unwrap().id)
+        .await
+        .expect("get_by_id");
+    assert_eq!(by_id.map(|p| p.slug), Some(proj_slug));
+}

crates/picloud/src/lib.rs

@@ -31,12 +31,13 @@ use picloud_manager_core::{
     PostgresDocsRepo, PostgresExecutionLogRepository, PostgresExecutionLogSink,
     PostgresGroupCollectionResolver, PostgresGroupDocsRepo, PostgresGroupKvRepo,
     PostgresGroupMembersRepository, PostgresGroupQueueRepo, PostgresGroupRepository,
-    PostgresKvRepo, PostgresOutboxRepo, PostgresPubsubRepo, PostgresRouteRepository,
-    PostgresScriptRepository, PostgresSecretsRepo, PostgresTopicRepo, PostgresTriggerRepo,
-    PostgresVarsRepo, PrincipalResolver, PubsubServiceImpl, RealtimeAuthorityImpl, RepoResolver,
-    RouteAdminState, SandboxCeiling, ScriptRepository, SecretsConfig, SecretsServiceImpl,
-    SecretsState, SubscriberTokenConfig, TopicRepo, TopicsState, TriggerConfig, TriggerRepo,
-    TriggersState, UsersServiceConfig, UsersServiceImpl, VarsApiState, VarsServiceImpl,
+    PostgresKvRepo, PostgresOutboxRepo, PostgresProjectRepository, PostgresPubsubRepo,
+    PostgresRouteRepository, PostgresScriptRepository, PostgresSecretsRepo, PostgresTopicRepo,
+    PostgresTriggerRepo, PostgresVarsRepo, PrincipalResolver, PubsubServiceImpl,
+    RealtimeAuthorityImpl, RepoResolver, RouteAdminState, SandboxCeiling, ScriptRepository,
+    SecretsConfig, SecretsServiceImpl, SecretsState, SubscriberTokenConfig, TopicRepo, TopicsState,
+    TriggerConfig, TriggerRepo, TriggersState, UsersServiceConfig, UsersServiceImpl, VarsApiState,
+    VarsServiceImpl,
 };
 use picloud_orchestrator_core::realtime::DEFAULT_GC_INTERVAL_SECS;
 use picloud_orchestrator_core::routing::{AppDomainTable, RouteTable};
@@ -537,6 +538,7 @@ pub async fn build_app(
         vars: Arc::new(PostgresVarsRepo::new(pool.clone())),
         apps: apps_repo.clone(),
         groups: groups_repo.clone(),
+        projects: Arc::new(PostgresProjectRepository::new(pool.clone())),
         modules: modules.clone(),
         domains: domains_repo.clone(),
         authz: authz.clone(),