teams: feat: team webhooks and worker demand

backend/gradient-ci/src/webhooks/routing.rs

@@ -20,6 +20,7 @@ pub fn routes_to(hook: &MWebhook, owner: &EventOwner, name: &str, personal: bool
         _ if personal => false,
         WebhookScope::Project => hook.project.is_some() && hook.project == owner.project,
         WebhookScope::Cache => hook.cache.is_some() && hook.cache == owner.cache,
+        WebhookScope::Team => hook.team.is_some() && hook.team == owner.team,
     }
 }
 
@@ -34,6 +35,9 @@ pub async fn candidates<C: ConnectionTrait>(
     if let Some(cache) = owner.cache {
         scope = scope.add(CWebhook::Cache.eq(cache));
     }
+    if let Some(team) = owner.team {
+        scope = scope.add(CWebhook::Team.eq(team));
+    }
     EWebhook::find()
         .filter(CWebhook::Active.eq(true))
         .filter(scope)
@@ -116,6 +120,33 @@ mod tests {
         ));
     }
 
+    fn team_hook(team: TeamId) -> MWebhook {
+        MWebhook {
+            team: Some(team),
+            ..hook(WebhookScope::Team, None, None, &[])
+        }
+    }
+
+    #[test]
+    fn team_hook_receives_its_team_event() {
+        let a = TeamId::new(Uuid::from_u128(4));
+        assert!(routes_to(
+            &team_hook(a),
+            &EventOwner::team(a),
+            "worker.demand",
+            false
+        ));
+    }
+
+    #[test]
+    fn team_hook_ignores_other_team_and_project_events() {
+        let h = team_hook(TeamId::new(Uuid::from_u128(4)));
+        let other_team = EventOwner::team(TeamId::new(Uuid::from_u128(5)));
+        let project = EventOwner::project(ProjectId::new(Uuid::from_u128(1)));
+        assert!(!routes_to(&h, &other_team, "worker.demand", false));
+        assert!(!routes_to(&h, &project, "build.failed", false));
+    }
+
     #[test]
     fn an_empty_event_list_subscribes_to_everything() {
         let h = hook(WebhookScope::Instance, None, None, &[]);

backend/gradient-db/src/graph/runtime_can_start.rs

@@ -7,6 +7,7 @@
 //! The seed cannot read the transition off the row it is writing.
 //! The commit making an output present wrote `cached_path` one statement earlier.
 //! `freshly_present` is the caller's endpoint for that transition.
+//! A freshly present dependency still counts as missing for a referrer in the same seed, since its own ripple subtracts it.
 //! A re-push of an already backed path must not count as fresh.
 
 use std::collections::BTreeMap;
@@ -21,14 +22,15 @@ use crate::sql::StatusWrite;
 
 fn seed_sql() -> String {
     format!(
-        "UPDATE derivation_build db SET missing_runtime_deps = x.n \
+        "WITH s(derivation, fresh) AS (SELECT * FROM unnest($1::uuid[], $2::bool[])) \
+         UPDATE derivation_build db SET missing_runtime_deps = x.n \
          FROM (SELECT b.id, ({was_complete} AND NOT s.fresh) AS was_complete, \
                       (SELECT count(*) FROM derivation_dependency e \
                          JOIN derivation_build dep ON dep.derivation = e.dependency \
+                         LEFT JOIN s landed ON landed.derivation = e.dependency AND landed.fresh \
                         WHERE e.derivation = b.derivation AND e.kind IN (1, 2) \
-                          AND NOT {dep_complete})::int AS n \
-               FROM unnest($1::uuid[], $2::bool[]) AS s(derivation, fresh) \
-               JOIN derivation_build b ON b.derivation = s.derivation) x \
+                          AND NOT ({dep_complete} AND landed.derivation IS NULL))::int AS n \
+               FROM s JOIN derivation_build b ON b.derivation = s.derivation) x \
          WHERE db.id = x.id \
          RETURNING db.derivation, x.was_complete, {complete} AS complete",
         was_complete = shared_build_complete_predicate("b"),

backend/gradient-db/src/teams/demand.rs

@@ -0,0 +1,213 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Wavelens GmbH <info@wavelens.io>
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+use super::workers::projects_granted_with_workers;
+use gradient_entity::evaluation::EvaluationStatus;
+use gradient_types::*;
+use sea_orm::sea_query::Query;
+use sea_orm::{ColumnTrait, ConnectionTrait, DbErr, EntityTrait, QueryFilter};
+use serde::Serialize;
+use std::collections::BTreeMap;
+
+#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, utoipa::ToSchema)]
+pub struct TeamDemand {
+    pub builds: Vec<UnmetRequirement>,
+    pub fetch: u32,
+    pub eval: u32,
+}
+
+pub async fn team_demand(
+    db: &impl ConnectionTrait,
+    team: TeamId,
+    projects: Option<&[ProjectId]>,
+) -> Result<TeamDemand, DbErr> {
+    let granting: Vec<ProjectId> = projects_granted_with_workers(db, team)
+        .await?
+        .into_iter()
+        .filter(|p| projects.is_none_or(|only| only.contains(p)))
+        .collect();
+    if granting.is_empty() {
+        return Ok(TeamDemand::default());
+    }
+
+    let tasks = Query::select()
+        .column(CTask::Id)
+        .from(ETask::default())
+        .and_where(CTask::Project.is_in(granting))
+        .to_owned();
+    let reasons = EEvaluation::find()
+        .filter(CEvaluation::Status.eq(EvaluationStatus::Waiting))
+        .filter(CEvaluation::Task.in_subquery(tasks))
+        .all(db)
+        .await?
+        .into_iter()
+        .filter_map(|e| e.waiting_reason.as_ref().and_then(WaitingReason::from_json));
+    Ok(aggregate(reasons))
+}
+
+fn aggregate(reasons: impl IntoIterator<Item = WaitingReason>) -> TeamDemand {
+    let mut demand = TeamDemand::default();
+    let mut builds: BTreeMap<(String, Vec<String>), u32> = BTreeMap::new();
+    for reason in reasons {
+        match reason {
+            WaitingReason::Workers { unmet, .. } => {
+                for requirement in unmet {
+                    let mut features = requirement.required_features;
+                    features.sort();
+                    *builds
+                        .entry((requirement.architecture, features))
+                        .or_default() += requirement.build_count;
+                }
+            }
+            WaitingReason::EvalWorkers { capability, .. } => match capability {
+                EvalCapability::Fetch => demand.fetch += 1,
+                EvalCapability::Eval => demand.eval += 1,
+            },
+            _ => {}
+        }
+    }
+    demand.builds = builds
+        .into_iter()
+        .map(
+            |((architecture, required_features), build_count)| UnmetRequirement {
+                architecture,
+                required_features,
+                build_count,
+            },
+        )
+        .collect();
+    demand
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use gradient_fixtures::mock::{all_statements, mock_db};
+
+    fn grant(project: ProjectId) -> MTeamProject {
+        MTeamProject {
+            project,
+            includes_workers: true,
+            ..Default::default()
+        }
+    }
+
+    fn waiting(reason: WaitingReason) -> MEvaluation {
+        MEvaluation {
+            status: EvaluationStatus::Waiting,
+            waiting_reason: Some(reason.to_json()),
+            ..Default::default()
+        }
+    }
+
+    fn unmet(architecture: &str, features: &[&str], build_count: u32) -> UnmetRequirement {
+        UnmetRequirement {
+            architecture: architecture.into(),
+            required_features: features.iter().map(|f| f.to_string()).collect(),
+            build_count,
+        }
+    }
+
+    #[tokio::test]
+    async fn merges_unmet_builds_by_system_and_features() {
+        let project = ProjectId::now_v7();
+        let db = mock_db()
+            .append_query_results([vec![grant(project)]])
+            .append_query_results([vec![
+                waiting(WaitingReason::workers(
+                    vec![unmet("x86_64-linux", &["kvm", "big"], 2)],
+                    1,
+                    vec![],
+                )),
+                waiting(WaitingReason::workers(
+                    vec![
+                        unmet("x86_64-linux", &["big", "kvm"], 3),
+                        unmet("aarch64-darwin", &[], 1),
+                    ],
+                    1,
+                    vec![],
+                )),
+            ]])
+            .into_connection();
+
+        let demand = team_demand(&db, TeamId::now_v7(), None)
+            .await
+            .expect("demand");
+
+        assert_eq!(
+            demand.builds,
+            vec![
+                unmet("aarch64-darwin", &[], 1),
+                unmet("x86_64-linux", &["big", "kvm"], 5),
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn counts_eval_and_fetch_gaps() {
+        let db = mock_db()
+            .append_query_results([vec![grant(ProjectId::now_v7())]])
+            .append_query_results([vec![
+                waiting(WaitingReason::eval_workers(EvalCapability::Eval, 0)),
+                waiting(WaitingReason::eval_workers(EvalCapability::Eval, 2)),
+                waiting(WaitingReason::eval_workers(EvalCapability::Fetch, 0)),
+                waiting(WaitingReason::Draining),
+            ]])
+            .into_connection();
+
+        let demand = team_demand(&db, TeamId::now_v7(), None)
+            .await
+            .expect("demand");
+
+        assert_eq!((demand.eval, demand.fetch), (2, 1));
+        assert!(demand.builds.is_empty());
+    }
+
+    #[tokio::test]
+    async fn ignores_projects_without_worker_grant() {
+        let db = mock_db()
+            .append_query_results([Vec::<MTeamProject>::new()])
+            .into_connection();
+
+        let demand = team_demand(&db, TeamId::now_v7(), None)
+            .await
+            .expect("demand");
+
+        assert_eq!(demand, TeamDemand::default());
+        let statements = all_statements(db.into_transaction_log());
+        assert!(
+            statements
+                .iter()
+                .all(|s| !s.sql.contains("FROM \"evaluation\""))
+        );
+        assert!(
+            statements
+                .iter()
+                .any(|s| s.sql.contains("\"includes_workers\" = $"))
+        );
+    }
+
+    #[tokio::test]
+    async fn project_filter_limits_aggregation() {
+        let (kept, dropped) = (ProjectId::now_v7(), ProjectId::now_v7());
+        let db = mock_db()
+            .append_query_results([vec![grant(kept), grant(dropped)]])
+            .append_query_results([Vec::<MEvaluation>::new()])
+            .into_connection();
+
+        team_demand(&db, TeamId::now_v7(), Some(&[kept]))
+            .await
+            .expect("demand");
+
+        let evaluations = all_statements(db.into_transaction_log())
+            .into_iter()
+            .find(|s| s.sql.contains("FROM \"evaluation\""))
+            .expect("an evaluation query");
+        let values = format!("{:?}", evaluations.values);
+        assert!(values.contains(&kept.to_string()));
+        assert!(!values.contains(&dropped.to_string()));
+    }
+}

backend/gradient-db/src/teams/mod.rs

@@ -4,6 +4,7 @@
  * SPDX-License-Identifier: AGPL-3.0-only
  */
 
+pub mod demand;
 pub mod evaluations;
 pub mod grants;
 pub mod invitations;

backend/gradient-db/src/teams/workers.rs

@@ -44,18 +44,25 @@ pub async fn projects_granted_with_workers<C: ConnectionTrait>(
         .collect())
 }
 
-pub async fn team_workers_for_project<C: ConnectionTrait>(
+pub async fn teams_granting_workers<C: ConnectionTrait>(
     db: &C,
     project: ProjectId,
-) -> Result<Vec<(String, MTeamWorker)>, DbErr> {
-    let teams: Vec<TeamId> = ETeamProject::find()
+) -> Result<Vec<TeamId>, DbErr> {
+    Ok(ETeamProject::find()
         .filter(CTeamProject::Project.eq(project))
         .filter(CTeamProject::IncludesWorkers.eq(true))
         .all(db)
         .await?
         .into_iter()
         .map(|grant| grant.team)
-        .collect();
+        .collect())
+}
+
+pub async fn team_workers_for_project<C: ConnectionTrait>(
+    db: &C,
+    project: ProjectId,
+) -> Result<Vec<(String, MTeamWorker)>, DbErr> {
+    let teams = teams_granting_workers(db, project).await?;
     if teams.is_empty() {
         return Ok(Vec::new());
     }

backend/gradient-db/src/webhooks.rs

@@ -16,6 +16,7 @@ pub struct Owner {
     pub scope: WebhookScope,
     pub project: Option<ProjectId>,
     pub cache: Option<CacheId>,
+    pub team: Option<TeamId>,
 }
 
 impl Owner {
@@ -30,6 +31,10 @@ impl Owner {
                 Some(c) => CWebhook::Cache.eq(c),
                 None => CWebhook::Cache.is_null(),
             })
+            .add(match self.team {
+                Some(t) => CWebhook::Team.eq(t),
+                None => CWebhook::Team.is_null(),
+            })
     }
 }
 

backend/gradient-entity/src/webhook.rs

@@ -9,7 +9,7 @@ use num_enum::{IntoPrimitive, TryFromPrimitive};
 use sea_orm::entity::prelude::*;
 use serde::{Deserialize, Serialize};
 
-use crate::ids::{CacheId, ProjectId, UserId, WebhookId};
+use crate::ids::{CacheId, ProjectId, TeamId, UserId, WebhookId};
 
 #[repr(i16)]
 #[derive(
@@ -37,6 +37,8 @@ pub enum WebhookScope {
     Cache = 1,
     #[sea_orm(num_value = 2)]
     Instance = 2,
+    #[sea_orm(num_value = 3)]
+    Team = 3,
 }
 
 #[derive(Clone, Debug, Default, PartialEq, DeriveEntityModel, Deserialize, Serialize)]
@@ -47,6 +49,7 @@ pub struct Model {
     pub scope: WebhookScope,
     pub project: Option<ProjectId>,
     pub cache: Option<CacheId>,
+    pub team: Option<TeamId>,
     pub name: String,
     pub url: String,
     pub secret: String,