feat(delivery): observe bounded node readiness and owner state

backend/src/kf_oj/modules/problems/operations.py

@@ -0,0 +1,64 @@
+"""Read-only import projection for the ordinary operator CLI; no private archive data."""
+
+from sqlalchemy import text
+
+from kf_oj.modules.problems.import_models import ImportPool
+from kf_oj.modules.problems.imports import ImportUnavailable, policy_digest
+
+
+class ImportOperationQueries:
+    def __init__(self, sessions, policy):
+        self.sessions, self.policy = sessions, policy
+
+    async def status(self) -> dict:
+        async with self.sessions() as session, session.begin():
+            await session.execute(text("SET TRANSACTION READ ONLY"))
+            await session.execute(text("SET LOCAL statement_timeout='1500'"))
+            await session.execute(text("SET LOCAL lock_timeout='500'"))
+            authorized = await session.scalar(
+                text("""
+                SELECT rolcanlogin AND NOT(rolsuper OR rolcreatedb OR rolcreaterole
+                  OR rolreplication OR rolbypassrls)
+                  AND pg_has_role(current_user,'kf_oj_problem_import','USAGE')
+                  AND NOT pg_has_role(current_user,'kf_oj_platform','MEMBER')
+                  AND NOT pg_has_role(current_user,'kf_oj_coordinator','MEMBER')
+                  AND NOT pg_has_role(current_user,'kf_oj_identity_operator','MEMBER')
+                  AND NOT pg_has_role(current_user,'kf_oj_identity_authority_owner','MEMBER')
+                FROM pg_roles WHERE rolname=session_user
+            """)
+            )
+            if authorized is not True:
+                raise ImportUnavailable("dedicated import observation role required")
+            pool = await session.get(ImportPool, 1)
+            row = (
+                (
+                    await session.execute(
+                        text("""
+                SELECT clock_timestamp() now,
+                  (SELECT count(*) FROM (SELECT 1 FROM problem_import_operations
+                    WHERE state='uploaded' LIMIT 129) q) uploaded,
+                  (SELECT count(*) FROM (SELECT 1 FROM problem_import_operations
+                    WHERE state='validating' LIMIT 129) q) validating,
+                  (SELECT coalesce(sum(reserved_bytes),0) FROM (
+                    SELECT reserved_bytes FROM problem_import_operations
+                    WHERE reserved_bytes>0 LIMIT 129) q) actual_reserved
+            """)
+                    )
+                )
+                .mappings()
+                .one()
+            )
+            if pool is None or pool.reserved_bytes != row["actual_reserved"]:
+                raise ImportUnavailable("import reservation consistency unknown")
+            return {
+                "schema_version": 1,
+                "worker_ready": bool(
+                    pool.worker_ready_until
+                    and pool.worker_ready_until > row["now"]
+                    and pool.worker_policy_digest == policy_digest(self.policy)
+                ),
+                "reserved_bytes": pool.reserved_bytes,
+                "uploaded_capped": min(row["uploaded"], 128),
+                "validating_capped": min(row["validating"], 128),
+                "truncated": row["uploaded"] > 128 or row["validating"] > 128,
+            }

backend/src/kf_oj/workers/problem_status.py

@@ -0,0 +1,33 @@
+"""Ordinary import-only read status; no parser, staging mutation or judging writes."""
+
+import asyncio
+import json
+
+from kf_oj.core.settings import Settings
+from kf_oj.infra.db.session import build_engine, build_session_factory
+from kf_oj.modules.problems.operations import ImportOperationQueries
+
+
+async def run() -> dict:
+    settings = Settings()
+    if settings.judge_admission is None:
+        raise ValueError("import admission required")
+    engine = build_engine(settings)
+    try:
+        return await ImportOperationQueries(
+            build_session_factory(engine), settings.judge_admission
+        ).status()
+    finally:
+        await engine.dispose()
+
+
+def main() -> None:
+    try:
+        result = asyncio.run(asyncio.wait_for(run(), timeout=6))
+        print(json.dumps(result, sort_keys=True))
+    except Exception:
+        raise SystemExit("import status unavailable") from None
+
+
+if __name__ == "__main__":
+    main()

backend/tests/test_delivery.py

@@ -23,6 +23,8 @@ factory_service = importlib.import_module("scripts.release.factory_service")
 dependencies = importlib.import_module("scripts.release.dependencies")
 dependency_install = importlib.import_module("scripts.release.dependency_install")
 dependency_initialize = importlib.import_module("scripts.release.dependency_initialize")
+node_health = importlib.import_module("scripts.release.node_health")
+broker_health = importlib.import_module("scripts.release.broker_health")
 
 
 def test_delivery_authorities_and_layout():
@@ -725,3 +727,119 @@ def test_database_initialization_receipt_does_not_prove_migration_or_authenticat
     receipt["cluster_identity"]["uid"] = 0
     with pytest.raises(policy.DeliveryError, match="cluster identity"):
         dependency_initialize.validate_receipt(receipt)
+
+
+@pytest.mark.parametrize(
+    "field,value",
+    [
+        ("superuser", True),
+        ("set_role", True),
+        ("connection_limit", -1),
+        ("login", "postgres"),
+        ("inherited", False),
+    ],
+)
+def test_readiness_rejects_wrong_or_excess_database_authority(field, value):
+    observed = {
+        "login": "kf_coordinator",
+        "database": "kf_onlinejudge",
+        "canlogin": True,
+        "superuser": False,
+        "createdb": False,
+        "createrole": False,
+        "replication": False,
+        "bypassrls": False,
+        "member": True,
+        "inherited": True,
+        "set_role": False,
+        "connection_limit": 8,
+        "privileges": node_health.PRIVILEGES["coordinator"].copy(),
+    }
+    node_health.validate_role("coordinator", observed)
+    observed[field] = value
+    with pytest.raises(policy.DeliveryError, match="role authority"):
+        node_health.validate_role("coordinator", observed)
+
+
+def test_readiness_refuses_context_swap_or_claimed_isolation_certificate():
+    value = {
+        "schema_version": 1,
+        "configuration_sha256": "b" * 64,
+        "supervisor_authorized": True,
+        "execution_gate": "required_per_execution",
+        "isolation_verified": False,
+        "max_concurrent_runs": 1,
+        "active_helpers": 0,
+        "cleanup_complete": True,
+        "remaining_requests": 4096,
+        "remaining_lifetime_seconds": 3600,
+    }
+    node_health.validate_factory(value, "b" * 64)
+    with pytest.raises(policy.DeliveryError, match="factory"):
+        node_health.validate_factory(value, "c" * 64)
+    value["isolation_verified"] = True
+    with pytest.raises(policy.DeliveryError, match="factory"):
+        node_health.validate_factory(value, "b" * 64)
+
+
+def test_busy_unknown_cleanup_or_expiring_factory_never_admits_new_work():
+    root = {
+        "cleanup_complete": True,
+        "active_helpers": 0,
+        "remaining_requests": 4096,
+        "remaining_lifetime_seconds": 3600,
+    }
+    counters = {
+        "in_flight": {"count_capped": 0, "truncated": False},
+        "cleanup_unconfirmed": {"count_capped": 0, "truncated": False},
+    }
+    admission = {
+        "legacy": {"max_job_wall_time_us": 60000000},
+        "multi_runtime": {"max_job_wall_time_us": 60000000},
+    }
+    assert node_health.admission_ready(root, counters, {"consumers": 1}, admission)
+    root["remaining_requests"] = 256
+    assert not node_health.admission_ready(root, counters, {"consumers": 1}, admission)
+    root["remaining_requests"] = 4096
+    root["remaining_lifetime_seconds"] = 119
+    assert not node_health.admission_ready(root, counters, {"consumers": 1}, admission)
+    root["remaining_lifetime_seconds"] = 3600
+    root["cleanup_complete"] = False
+    assert not node_health.admission_ready(root, counters, {"consumers": 1}, admission)
+    root["cleanup_complete"] = True
+    counters["cleanup_unconfirmed"]["count_capped"] = 1
+    assert not node_health.admission_ready(root, counters, {"consumers": 1}, admission)
+
+
+def test_actual_broker_argument_parser_keeps_typed_budgets_and_rejects_duplicates():
+    value = [
+        [key, "longstr" if key == "x-overflow" else "long", val]
+        for key, val in dependencies.QUEUE_ARGUMENTS.items()
+    ]
+    assert broker_health.queue_arguments(value) == dependencies.QUEUE_ARGUMENTS
+    value[-1] = value[0].copy()
+    with pytest.raises(policy.DeliveryError, match="duplicate"):
+        broker_health.queue_arguments(value)
+    with pytest.raises(policy.DeliveryError, match="budgets"):
+        broker_health.queue_arguments({**dependencies.QUEUE_ARGUMENTS, "x-max-length": 1024})
+
+
+def test_counter_projection_rejects_truncated_empty_and_source_admission_mismatch():
+    value = {
+        "schema_version": 1,
+        "admission": {"fixture": True},
+        "configured_workers": 1,
+        "lease_seconds": 120,
+        "counters": {
+            "queued": {"count_capped": 0, "truncated": False},
+            "in_flight": {"count_capped": 0, "truncated": False},
+            "cleanup_unconfirmed": {"count_capped": 0, "truncated": False},
+            "database_clock_unix_ms": 1791830000000,
+        },
+    }
+    node_health.validate_execution(value, value["admission"])
+    with pytest.raises(policy.DeliveryError, match="admission"):
+        node_health.validate_execution(value, {"fixture": False})
+    value["counters"]["cleanup_unconfirmed"]["truncated"] = True
+    with pytest.raises(policy.DeliveryError, match="bound"):
+        node_health.validate_execution(value, value["admission"])

judge/crates/coordinator/src/api.rs

@@ -132,6 +132,29 @@ async fn ready(State(app): State<App>) -> ApiResult<Json<Value>> {
     app.repository.database.ready().await?;
     Ok(Json(json!({"status":"ready"})))
 }
+async fn operations(State(app): State<App>, headers: HeaderMap) -> ApiResult<Json<Value>> {
+    // The general internal guard also recognizes Workers; this projection is
+    // separately restricted to platform/operator authority, never worker tokens.
+    platform(&app, &headers)?;
+    let counters = crate::operations::snapshot(&app.repository.database).await?;
+    Ok(Json(json!({
+        "schema_version": 1,
+        "admission": app.config.admission,
+        "configured_workers": app.config.workers.len(),
+        "lease_seconds": app.config.lease_seconds,
+        "counters": counters,
+    })))
+}
+async fn notification_probe_id(
+    State(app): State<App>,
+    headers: HeaderMap,
+) -> ApiResult<Json<Value>> {
+    platform(&app, &headers)?;
+    let id = crate::operations::absent_notification_id(&app.repository.database).await?;
+    Ok(Json(
+        json!({"schema_version":1,"evaluation_id":id,"absent":true}),
+    ))
+}
 async fn claim(State(app): State<App>, h: HeaderMap, b: Bytes) -> ApiResult<Response> {
     let w = worker(&app, &h)?;
     let n: JudgeTaskNotification = decode(&b)?;
@@ -198,6 +221,8 @@ pub fn application(app: App) -> Router {
     Router::new()
         .route("/health/live", get(live))
         .route("/health/ready", get(ready))
+        .route("/internal/operations/status", get(operations))
+        .route("/internal/operations/probe-id", get(notification_probe_id))
         .route("/internal/claim", post(claim))
         .route("/internal/renew", post(renew))
         .route("/internal/stage", post(stage))

judge/crates/coordinator/src/lib.rs

@@ -2,5 +2,6 @@
 pub mod api;
 pub mod config;
 pub mod database;
+pub mod operations;
 pub mod performance;
 pub mod repository;

judge/crates/coordinator/src/operations.rs

@@ -0,0 +1,91 @@
+//! Bounded read-only execution counters; no task identifiers or authority writes.
+use crate::database::{Database, Error, Result};
+use serde::Serialize;
+
+#[derive(Debug, Serialize)]
+pub struct BoundedCount {
+    pub count_capped: u16,
+    pub truncated: bool,
+}
+impl BoundedCount {
+    fn new(count: i64) -> Result<Self> {
+        if !(0..=129).contains(&count) {
+            return Err(Error::Invalid);
+        }
+        Ok(Self {
+            count_capped: count.min(128) as u16,
+            truncated: count == 129,
+        })
+    }
+}
+#[derive(Debug, Serialize)]
+pub struct ExecutionCounters {
+    pub queued: BoundedCount,
+    pub in_flight: BoundedCount,
+    pub cleanup_unconfirmed: BoundedCount,
+    pub database_clock_unix_ms: i64,
+}
+pub async fn snapshot(database: &Database) -> Result<ExecutionCounters> {
+    let mut session = database.session().await?;
+    let transaction = session
+        .client
+        .build_transaction()
+        .read_only(true)
+        .isolation_level(tokio_postgres::IsolationLevel::RepeatableRead)
+        .start()
+        .await?;
+    // LIMIT applies inside each count: a timeout or truncated projection can
+    // never be interpreted as empty capacity. The fixed DB deadline still
+    // bounds a planner that cannot produce the first 129 rows cheaply.
+    let row = transaction.query_one(
+        "SELECT
+          (SELECT count(*) FROM (SELECT 1 FROM evaluations WHERE state='queued' LIMIT 129) q) queued,
+          (SELECT count(*) FROM (SELECT 1 FROM evaluations WHERE state IN
+            ('claimed','compiling','running','cancelling') LIMIT 129) a) in_flight,
+          (SELECT count(*) FROM (SELECT 1 FROM evaluations e JOIN judge_attempts a
+            ON e.current_attempt_id=a.id AND a.evaluation_id=e.id
+            WHERE a.cleanup_complete IS DISTINCT FROM TRUE LIMIT 129) c) cleanup_unconfirmed,
+          floor(extract(epoch FROM clock_timestamp())*1000)::bigint database_clock_unix_ms",
+        &[],
+    ).await?;
+    let value = ExecutionCounters {
+        queued: BoundedCount::new(row.get("queued"))?,
+        in_flight: BoundedCount::new(row.get("in_flight"))?,
+        cleanup_unconfirmed: BoundedCount::new(row.get("cleanup_unconfirmed"))?,
+        database_clock_unix_ms: row.get("database_clock_unix_ms"),
+    };
+    transaction.commit().await?;
+    Ok(value)
+}
+
+pub async fn absent_notification_id(database: &Database) -> Result<uuid::Uuid> {
+    let session = database.session().await?;
+    for _ in 0..4 {
+        let identifier = uuid::Uuid::new_v4();
+        if session
+            .client
+            .query_opt("SELECT 1 FROM evaluations WHERE id=$1", &[&identifier])
+            .await?
+            .is_none()
+        {
+            return Ok(identifier);
+        }
+    }
+    Err(Error::Unavailable)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn truncated_counts_never_become_an_empty_drain_proof() {
+        let empty = BoundedCount::new(0).unwrap();
+        assert_eq!(empty.count_capped, 0);
+        assert!(!empty.truncated);
+        let crowded = BoundedCount::new(129).unwrap();
+        assert_eq!(crowded.count_capped, 128);
+        assert!(crowded.truncated);
+        assert!(BoundedCount::new(-1).is_err());
+        assert!(BoundedCount::new(130).is_err());
+    }
+}

judge/crates/sandbox/src/cgroup.rs

@@ -59,6 +59,45 @@ pub struct DelegatedCgroup {
 }
 
 impl DelegatedCgroup {
+    /// Read-only status: never repairs a missing controller or rewrites limits.
+    pub fn inspect(config: &ValidatedRuntimeConfig) -> Result<(), CgroupError> {
+        let policy = config.config();
+        let fd = open_directory(Path::new(&policy.delegated_cgroup_root))?;
+        let stat = rustix::fs::fstat(&fd).map_err(io::Error::from)?;
+        if stat.st_uid != rustix::process::geteuid().as_raw()
+            || stat.st_gid != 0
+            || stat.st_mode & 0o022 != 0
+            || rustix::fs::fstatfs(&fd).map_err(io::Error::from)?.f_type as u64 != CGROUP2_MAGIC
+            || read(&fd, "cgroup.type")?.trim() != "domain"
+            || !read(&fd, "cgroup.procs")?.trim().is_empty()
+            || read(&fd, "cgroup.max.depth")?.trim() != "1"
+            || read(&fd, "cgroup.max.descendants")?.trim() != policy.max_concurrent_runs.to_string()
+        {
+            return Err(invalid("delegation status differs").into());
+        }
+        for name in ["cgroup.controllers", "cgroup.subtree_control"] {
+            let observed = read(&fd, name)?;
+            if !["cpu", "memory", "pids"]
+                .iter()
+                .all(|required| observed.split_whitespace().any(|value| value == *required))
+            {
+                return Err(invalid("delegation controllers unavailable").into());
+            }
+        }
+        let budget = &policy.aggregate;
+        verify_limits(
+            &fd,
+            &AppliedLimits {
+                memory_bytes: budget.memory_bytes,
+                swap_bytes: 0,
+                processes: budget.processes,
+                cpu_quota_us: budget.cpu_quota_us,
+                cpu_period_us: budget.cpu_period_us,
+                oom_group: true,
+            },
+        )?;
+        Ok(())
+    }
     /// Config comes from the operator. Structural validation is not proof of
     /// config-file ownership, service delegation or an exec permit.
     /// The root must be empty, owned by this helper and not group/other writable.

judge/crates/sandbox/src/factory.rs

@@ -10,6 +10,7 @@ use rustix::{
     fs::{Mode, OFlags},
 };
 use serde::{Deserialize, Serialize};
+use sha2::{Digest as _, Sha256};
 use std::{
     fs::File,
     io::{self, Write},
@@ -30,6 +31,17 @@ struct Acquire {
     run_id: String,
     #[serde(default, skip_serializing_if = "is_false")]
     check_stopped: bool,
+    #[serde(
+        default,
+        skip_serializing_if = "Option::is_none",
+        deserialize_with = "health_version"
+    )]
+    health_version: Option<u8>,
+}
+fn health_version<'de, D: serde::Deserializer<'de>>(
+    deserializer: D,
+) -> Result<Option<u8>, D::Error> {
+    u8::deserialize(deserializer).map(Some)
 }
 fn is_false(v: &bool) -> bool {
     !v
@@ -37,16 +49,29 @@ fn is_false(v: &bool) -> bool {
 #[derive(Serialize, Deserialize)]
 #[serde(tag = "event", rename_all = "snake_case", deny_unknown_fields)]
 enum Allocation {
-    Ready { run_id: String, helper_pid: u32 },
+    Ready {
+        run_id: String,
+        helper_pid: u32,
+    },
     Refused {},
     Busy {},
-    Stopped { cleanup_complete: bool },
+    Stopped {
+        cleanup_complete: bool,
+    },
+    Health {
+        status: crate::factory_health::FactoryHealth,
+    },
 }
 impl crate::launcher::protocol::sealed::Sealed for Acquire {}
 impl crate::launcher::protocol::sealed::Sealed for Allocation {}
 impl WireMessage for Acquire {
     fn valid(&self) -> bool {
         crate::spec::identifier(&self.run_id, "run_id").is_ok()
+            && match self.health_version {
+                None => true,
+                Some(1) => self.run_id == "health-check" && !self.check_stopped,
+                Some(_) => false,
+            }
     }
 }
 impl WireMessage for Allocation {
@@ -56,6 +81,7 @@ impl WireMessage for Allocation {
                 *helper_pid > 1 && crate::spec::identifier(run_id, "run_id").is_ok()
             }
             Self::Refused {} | Self::Busy {} | Self::Stopped { .. } => true,
+            Self::Health { status } => status.valid(),
         }
     }
 }
@@ -86,12 +112,30 @@ impl FactoryClient {
         channel.send(&Acquire {
             run_id: "stop-check".into(),
             check_stopped: true,
+            health_version: None,
         })?;
         match channel.receive::<Allocation>()? {
             Allocation::Stopped { cleanup_complete } => Ok(cleanup_complete),
             _ => Err(denied()),
         }
     }
+    /// Root-authenticated status; does not recover, allocate or execute a task.
+    pub fn health(&self) -> io::Result<crate::factory_health::FactoryHealth> {
+        if rustix::process::geteuid().as_raw() < 1000 || rustix::process::getegid().as_raw() < 1000
+        {
+            return Err(denied());
+        }
+        let mut channel = rpc::channel(UnixStream::connect(&self.socket)?, 0, 0)?;
+        channel.send(&Acquire {
+            run_id: "health-check".into(),
+            check_stopped: false,
+            health_version: Some(1),
+        })?;
+        match channel.receive::<Allocation>()? {
+            Allocation::Health { status } => Ok(status),
+            _ => Err(denied()),
+        }
+    }
     /// The caller holds only ordinary artifact and Unix socket capabilities.
     pub fn execute(
         &self,
@@ -177,6 +221,7 @@ impl FactoryClient {
                 channel.send(&Acquire {
                     run_id: spec.run_id.clone(),
                     check_stopped: false,
+                    health_version: None,
                 })?;
                 let pid = match channel.receive::<Allocation>()? {
                     Allocation::Ready { run_id, helper_pid } if run_id == spec.run_id => helper_pid,
@@ -334,6 +379,13 @@ pub fn serve_with_limit(
     }
     clean_environment()?;
     let config = execution::load_config(config_path)?;
+    let configuration_sha256 = format!(
+        "{:x}",
+        Sha256::digest(execution::read_bounded_file(
+            config_path,
+            crate::runtime::MAX_CONFIG_BYTES
+        )?)
+    );
     execution::authorize_registry(&config)?;
     let policy = config.config();
     if policy.aggregate.supervisor_processes < 4 + 8 * policy.max_concurrent_runs {
@@ -417,6 +469,8 @@ pub fn serve_with_limit(
     let _delegated = crate::cgroup::DelegatedCgroup::open(&config).map_err(|_| denied())?;
     let mut active: Vec<Active> = Vec::new();
     let deadline = Instant::now() + Duration::from_secs(lifetime_seconds);
+    let mut health_checks = 0_usize;
+    let mut last_health = None;
     let result = (|| {
         while Instant::now() < deadline {
             let mut i = 0;
@@ -451,6 +505,47 @@ pub fn serve_with_limit(
             let Ok(request) = channel.receive::<Acquire>() else {
                 continue;
             };
+            if request.health_version == Some(1) {
+                if health_checks >= MAX_IDS
+                    || last_health
+                        .is_some_and(|last: Instant| last.elapsed() < Duration::from_secs(1))
+                {
+                    let _ = channel.send(&Allocation::Busy {});
+                    continue;
+                }
+                health_checks += 1;
+                last_health = Some(Instant::now());
+                // The fixed root supervisor context and delegation must still
+                // be authorized. Namespace/root/seccomp/exec acknowledgements
+                // remain mandatory for every actual execution, not inferred.
+                let current =
+                    execution::read_bounded_file(config_path, crate::runtime::MAX_CONFIG_BYTES)?;
+                if format!("{:x}", Sha256::digest(current)) != configuration_sha256
+                    || execution::authorize(&config).is_err()
+                    || crate::cgroup::DelegatedCgroup::inspect(&config).is_err()
+                {
+                    let _ = channel.send(&Allocation::Refused {});
+                    continue;
+                }
+                let cleanup_complete = active.is_empty() && drained(&config)?;
+                channel.send(&Allocation::Health {
+                    status: crate::factory_health::FactoryHealth {
+                        schema_version: 1,
+                        configuration_sha256: configuration_sha256.clone(),
+                        supervisor_authorized: true,
+                        execution_gate: "required_per_execution".into(),
+                        isolation_verified: false,
+                        max_concurrent_runs: policy.max_concurrent_runs,
+                        active_helpers: active.len() as u32,
+                        cleanup_complete,
+                        remaining_requests: (max_requests - used) as u32,
+                        remaining_lifetime_seconds: deadline
+                            .saturating_duration_since(Instant::now())
+                            .as_secs(),
+                    },
+                })?;
+                continue;
+            }
             if request.check_stopped {
                 if !active.is_empty() {
                     let _ = channel.send(&Allocation::Stopped {
@@ -578,4 +673,29 @@ mod tests {
                 .valid()
         );
     }
+    #[test]
+    fn health_is_explicitly_versioned_and_old_allocation_bytes_are_unchanged() {
+        let allocation = Acquire {
+            run_id: "a-1".into(),
+            check_stopped: false,
+            health_version: None,
+        };
+        assert_eq!(
+            serde_json::to_string(&allocation).unwrap(),
+            r#"{"run_id":"a-1"}"#
+        );
+        for data in [
+            r#"{"run_id":"health-check","health_version":2}"#,
+            r#"{"run_id":"work-id","health_version":1}"#,
+            r#"{"run_id":"health-check","health_version":1,"check_stopped":true}"#,
+        ] {
+            let request = serde_json::from_str::<Acquire>(data).unwrap();
+            assert!(!request.valid());
+        }
+        assert!(
+            serde_json::from_str::<Acquire>(r#"{"run_id":"health-check","health_version":1}"#)
+                .unwrap()
+                .valid()
+        );
+    }
 }

judge/crates/sandbox/src/factory_health.rs

@@ -0,0 +1,63 @@
+//! Versioned read-only factory observation, never an isolation/execution permit.
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct FactoryHealth {
+    pub schema_version: u8,
+    pub configuration_sha256: String,
+    pub supervisor_authorized: bool,
+    pub execution_gate: String,
+    pub isolation_verified: bool,
+    pub max_concurrent_runs: u32,
+    pub active_helpers: u32,
+    pub cleanup_complete: bool,
+    pub remaining_requests: u32,
+    pub remaining_lifetime_seconds: u64,
+}
+impl FactoryHealth {
+    pub(crate) fn valid(&self) -> bool {
+        self.schema_version == 1
+            && self.configuration_sha256.len() == 64
+            && self
+                .configuration_sha256
+                .bytes()
+                .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
+            && self.supervisor_authorized
+            && self.execution_gate == "required_per_execution"
+            && !self.isolation_verified
+            && (1..=32).contains(&self.max_concurrent_runs)
+            && self.active_helpers <= self.max_concurrent_runs
+            && (!self.cleanup_complete || self.active_helpers == 0)
+            && self.remaining_requests <= super::factory_cycles::MAX_IDS as u32
+            && self.remaining_lifetime_seconds <= 86400
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    #[test]
+    fn health_never_certifies_namespace_execution_or_busy_cleanup() {
+        let mut health = FactoryHealth {
+            schema_version: 1,
+            configuration_sha256: "b".repeat(64),
+            supervisor_authorized: true,
+            execution_gate: "required_per_execution".into(),
+            isolation_verified: false,
+            max_concurrent_runs: 1,
+            active_helpers: 0,
+            cleanup_complete: true,
+            remaining_requests: 4096,
+            remaining_lifetime_seconds: 3600,
+        };
+        assert!(health.valid());
+        health.isolation_verified = true;
+        assert!(!health.valid());
+        health.isolation_verified = false;
+        health.active_helpers = 1;
+        assert!(!health.valid());
+        health.cleanup_complete = false;
+        assert!(health.valid());
+    }
+}

judge/crates/sandbox/src/lib.rs

@@ -42,4 +42,6 @@ pub mod factory;
 #[cfg(target_os = "linux")]
 pub mod factory_cycles;
 #[cfg(target_os = "linux")]
+pub mod factory_health;
+#[cfg(target_os = "linux")]
 pub mod factory_preparation;

judge/crates/sandbox/src/main.rs

@@ -16,6 +16,11 @@ struct Cli {
 enum Command {
     /// Observe the current Linux environment without changing it.
     Probe,
+    /// Ordinary authenticated, read-only factory status; never certifies isolation.
+    FactoryHealth {
+        #[arg(long)]
+        socket: PathBuf,
+    },
     #[cfg(all(target_os = "linux", feature = "vm-acceptance"))]
     #[command(hide = true)]
     VmCase { case: String },
@@ -107,6 +112,24 @@ enum Command {
 
 fn main() -> ExitCode {
     match Cli::parse().command {
+        Command::FactoryHealth { socket } => {
+            #[cfg(target_os = "linux")]
+            match kf_sandbox::factory::FactoryClient::new(socket).health() {
+                Ok(status) => {
+                    println!(
+                        "{}",
+                        serde_json::to_string(&status).expect("bounded health")
+                    );
+                    ExitCode::SUCCESS
+                }
+                Err(_) => spec_error("factory_health_unavailable", None),
+            }
+            #[cfg(not(target_os = "linux"))]
+            {
+                let _ = socket;
+                spec_error("unsupported_platform", None)
+            }
+        }
         Command::Probe => match serde_json::to_string_pretty(&kf_sandbox::probe::observe()) {
             Ok(report) => {
                 println!("{report}");

scripts/release/README.md

@@ -243,3 +243,37 @@ reads the transient unit result and confirms stop; foreign/residual units and
 unrecorded/partial targets are refused. These are implementation interfaces, not
 actual VM acceptance. See [PostgreSQL initdb](https://www.postgresql.org/docs/18/app-initdb.html)
 and [systemd credentials](https://systemd.io/CREDENTIALS/) for the underlying interfaces.
+
+`python3.13 -m scripts.release.node_health` performs the installed service
+admission-precondition checks in a fixed root operator unit. It uses bounded
+ordinary broker/factory/import observer units and accepts no executable, account,
+port or command override. Each helper checks its effective node/service/process
+budgets, identity and no-new-privileges/capability boundary. Unknown or concurrent
+units are refused; transient helper stop is read back. This is an operator CLI,
+not an additional public HTTP control endpoint, and has no schema/state writer.
+
+The projection checks selected unit pointers and effective identities/limits,
+actual DB-role/column privileges, the import owner's fresh policy-bound heartbeat
+through its Unix socket, and broker queue semantics/permissions/vhost limits via
+ordinary-cookie read-only CLI queries. Coordinator owns the authenticated bounded
+execution/cleanup counters; truncated or unconfirmed data never proves a drain.
+Its `/internal/operations/` routes require platform authority and are not proxied
+publicly. Counters contain no submission IDs, source, archives or credential bytes.
+
+Factory health is a separate version-1 authenticated observation, rate/call-count
+bounded for the finite factory lifetime. It reads current supervisor/delegation
+authorization, the exact runtime-config digest, live helpers, remaining requests
+and lifetime, without recovery, task allocation, ledger writes or limit repair.
+The old allocation request bytes are unchanged. Health explicitly reports
+`isolation_verified=false` and `execution_gate=required_per_execution`: every real
+execution still needs its namespace/root/identity/seccomp/exec acknowledgements.
+Admission readiness requires confirmed empty execution/cleanup state and enough
+factory requests/lifetime for the configured maximum job. It does not certify
+TLS/browser behavior or replace the independent VM's execution acceptance.
+
+The optional explicit `--probe-notification` sends one small confirmed notification
+for a fresh identifier checked absent by Coordinator. It reads no message body,
+creates no evaluation/grade and starts no program; ordinary periodic observation
+does not publish messages. Probe failure is reported separately. Underlying
+interfaces: [RabbitMQ CLI](https://www.rabbitmq.com/docs/man/rabbitmqctl.8),
+[PostgreSQL privilege functions](https://www.postgresql.org/docs/18/functions-info.html).

scripts/release/broker_health.py

@@ -0,0 +1,218 @@
+"""Finite ordinary-cookie metadata queries; no queue/user mutation or message reads."""
+
+import argparse
+import json
+import stat
+import subprocess
+import tempfile
+from pathlib import Path
+
+from . import dependencies, operator_units, preflight
+from .policy import DeliveryError, canonical, fields, integer
+
+CTL = Path("/usr/lib/rabbitmq/sbin/rabbitmqctl")
+
+
+def queue_arguments(value) -> dict:
+    if isinstance(value, dict):
+        result = value
+    elif isinstance(value, list) and len(value) == 4:
+        result = {}
+        for row in value:
+            if not isinstance(row, list) or len(row) != 3 or row[0] in result:
+                raise DeliveryError("duplicate/unknown broker argument")
+            name, kind, argument = row
+            if name not in dependencies.QUEUE_ARGUMENTS or kind != (
+                "longstr" if name == "x-overflow" else "long"
+            ):
+                raise DeliveryError("broker argument type differs")
+            result[name] = argument
+    else:
+        raise DeliveryError("broker argument shape differs")
+    if canonical(result) != canonical(dependencies.QUEUE_ARGUMENTS):
+        raise DeliveryError("actual broker queue budgets differ")
+    return result
+
+
+def validate_snapshot(value: dict) -> dict:
+    fields(value, {"queues", "vhost_limits", "users", "permissions"}, "broker snapshot")
+    rows = value["queues"]
+    if not isinstance(rows, list) or len(rows) != 1:
+        raise DeliveryError("actual broker queue inventory differs")
+    queue = fields(
+        rows[0],
+        {
+            "name",
+            "durable",
+            "auto_delete",
+            "exclusive",
+            "type",
+            "state",
+            "arguments",
+            "messages_ready",
+            "messages_unacknowledged",
+            "message_bytes",
+            "consumers",
+            "policy",
+            "operator_policy",
+            "effective_policy_definition",
+        },
+        "broker queue",
+    )
+    if (
+        queue["name"] != dependencies.QUEUE
+        or queue["durable"] is not True
+        or queue["auto_delete"] is not False
+        or queue["exclusive"] is not False
+        or queue["type"] != "classic"
+        or queue["state"] != "running"
+        or queue["policy"] not in (None, "")
+        or queue["operator_policy"] not in (None, "")
+        or queue["effective_policy_definition"] not in ({}, [])
+    ):
+        raise DeliveryError("actual broker queue semantics differ")
+    queue_arguments(queue["arguments"])
+    limits = value["vhost_limits"]
+    if not isinstance(limits, list) or len(limits) != 1:
+        raise DeliveryError("actual broker vhost limits unavailable")
+    row = fields(limits[0], {"vhost", "limits"}, "broker vhost limits")
+    if row["vhost"] != "kf" or not isinstance(row["limits"], str):
+        raise DeliveryError("actual broker vhost binding differs")
+    if canonical(operator_units.decode(row["limits"].encode(), 1024)) != canonical(
+        {"max-connections": 4, "max-queues": 1}
+    ):
+        raise DeliveryError("actual broker vhost budgets differ")
+    users = value["users"]
+    if (
+        not isinstance(users, list)
+        or len(users) != 2
+        or canonical(sorted(users, key=lambda u: u.get("user", "")))
+        != canonical(
+            [
+                {"user": "kf_outbox", "tags": []},
+                {"user": "kf_worker", "tags": []},
+            ]
+        )
+    ):
+        raise DeliveryError("actual broker users/tags differ")
+    permissions = value["permissions"]
+    expected = [
+        {
+            "user": "kf_" + role,
+            "configure": r"^kf\.judge\.evaluations\.v1$",
+            "write": r"^(amq\.default|kf\.judge\.evaluations\.v1)$" if role == "outbox" else "^$",
+            "read": r"^kf\.judge\.evaluations\.v1$" if role == "worker" else "^$",
+        }
+        for role in ("outbox", "worker")
+    ]
+    if (
+        not isinstance(permissions, list)
+        or len(permissions) != 2
+        or canonical(sorted(permissions, key=lambda u: u.get("user", ""))) != canonical(expected)
+    ):
+        raise DeliveryError("actual broker authority differs")
+    return {
+        "schema_version": 1,
+        "metadata_verified": True,
+        "messages_ready": integer(queue["messages_ready"], 0, 64, "broker ready count"),
+        "messages_unacknowledged": integer(
+            queue["messages_unacknowledged"], 0, 1, "broker unacknowledged count"
+        ),
+        "message_bytes": integer(queue["message_bytes"], 0, 32768 + 4096, "broker bytes"),
+        "consumers": integer(queue["consumers"], 0, 1, "broker consumer count"),
+    }
+
+
+def command(arguments: list[str]):
+    # A file-backed capped stream prevents malicious metadata/log output from
+    # allocating an unbounded capture buffer. The unit adds a hard file limit.
+    env = {
+        "PATH": "/usr/bin:/bin",
+        "LC_ALL": "C.UTF-8",
+        "HOME": dependencies.STORAGE["rabbitmq"]["path"],
+        "ERL_EPMD_ADDRESS": "127.0.0.1",
+        "RABBITMQ_NODENAME": "kf@localhost",
+        "RABBITMQ_CTL_ERL_ARGS": (
+            "+S 1:1 +A 2 +P 1024 +Q 1024 -kernel inet_dist_use_interface {127,0,0,1}"
+        ),
+    }
+    with tempfile.TemporaryFile() as output:
+        result = subprocess.run(
+            [
+                str(CTL),
+                "-q",
+                "-n",
+                "kf@localhost",
+                "--timeout",
+                "5",
+                *arguments,
+                "--formatter=json",
+            ],
+            stdout=output,
+            stderr=subprocess.DEVNULL,
+            timeout=8,
+            env=env,
+        )
+        if result.returncode or output.tell() > 32768:
+            raise DeliveryError("bounded broker metadata query unavailable")
+        output.seek(0)
+        return operator_units.decode(output.read(32769))
+
+
+def observe() -> dict:
+    operator_units.guarded("broker-health")
+    preflight.protected(CTL)
+    cookie = Path(dependencies.STORAGE["rabbitmq"]["path"]) / ".erlang.cookie"
+    info = cookie.lstat()
+    if (
+        not stat.S_ISREG(info.st_mode)
+        or info.st_nlink != 1
+        or (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) != (1204, 1204, 0o400)
+    ):
+        raise DeliveryError("ordinary private broker cookie required")
+    data = {
+        "queues": command(
+            [
+                "list_queues",
+                "-p",
+                "kf",
+                "name",
+                "durable",
+                "auto_delete",
+                "exclusive",
+                "type",
+                "state",
+                "arguments",
+                "messages_ready",
+                "messages_unacknowledged",
+                "message_bytes",
+                "consumers",
+                "policy",
+                "operator_policy",
+                "effective_policy_definition",
+            ]
+        ),
+        "vhost_limits": command(["list_vhost_limits", "--global"]),
+        "users": command(["list_users"]),
+        "permissions": command(["list_permissions", "-p", "kf"]),
+    }
+    return validate_snapshot(data)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Bounded ordinary broker metadata")
+    parser.add_argument("--inside", action="store_true", required=True)
+    parser.parse_args()
+    try:
+        print(json.dumps(observe(), sort_keys=True))
+    except (OSError, ValueError, subprocess.SubprocessError) as error:
+        parser.exit(
+            1,
+            "Broker health refused: "
+            + (str(error) if isinstance(error, DeliveryError) else type(error).__name__)
+            + "\n",
+        )
+
+
+if __name__ == "__main__":
+    main()

scripts/release/factory_health.py

@@ -0,0 +1,46 @@
+"""Ordinary bounded status client; no execution request, credentials or privileges."""
+
+import argparse
+import os
+import re
+from pathlib import Path
+
+from . import install, operator_units, preflight
+from .policy import DeliveryError
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Fixed ordinary factory observation")
+    parser.add_argument("--inside", action="store_true", required=True)
+    parser.parse_args()
+    try:
+        operator_units.guarded("factory-health")
+        payload = Path(__file__).resolve().parents[3]
+        if (
+            payload.parent.parent != install.RELEASES
+            or payload.name != "payload"
+            or not re.fullmatch(r"sd-[0-9a-f]{12}", payload.parent.name)
+        ):
+            raise DeliveryError("factory observer is not an installed delivery")
+        binary = payload / "bin/kf-sandbox"
+        preflight.protected(binary)
+        os.execve(
+            binary,
+            [str(binary), "factory-health", "--socket", "/var/lib/kf-judge/factory/control.sock"],
+            {
+                "PATH": "/usr/bin:/bin",
+                "LC_ALL": "C.UTF-8",
+                "LD_LIBRARY_PATH": str(payload / "lib"),
+            },
+        )
+    except (OSError, ValueError) as error:
+        parser.exit(
+            1,
+            "Factory health refused: "
+            + (str(error) if isinstance(error, DeliveryError) else type(error).__name__)
+            + "\n",
+        )
+
+
+if __name__ == "__main__":
+    main()

scripts/release/import_health.py

@@ -0,0 +1,44 @@
+"""Fixed ordinary import-only read projection in its private PID/network view."""
+
+import argparse
+import os
+from pathlib import Path
+
+from . import install, operator_units, service_bootstrap
+from .policy import DeliveryError
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Private ordinary import observation")
+    parser.add_argument("--inside", action="store_true", required=True)
+    parser.parse_args()
+    try:
+        operator_units.guarded("import-health")
+        status = dict(
+            line.split(":", 1)
+            for line in Path("/proc/self/status").read_text().splitlines()
+            if ":" in line
+        )
+        interfaces = {
+            line.split(":", 1)[0].strip()
+            for line in Path("/proc/net/dev").read_text().splitlines()[2:]
+        }
+        if status.get("NSpid", "").split()[-1:] != ["1"] or interfaces != {"lo"}:
+            raise DeliveryError("private import observation view required")
+        payload = Path(__file__).resolve().parents[3]
+        if payload.parent.parent != install.RELEASES or payload.name != "payload":
+            raise DeliveryError("installed import observer required")
+        command, env = service_bootstrap.python_command("problem-import", payload.parent.name)
+        command[-1] = "kf_oj.workers.problem_status"
+        os.execve(command[0], command, env)
+    except (OSError, ValueError, KeyError) as error:
+        parser.exit(
+            1,
+            "Import health refused: "
+            + (str(error) if isinstance(error, DeliveryError) else type(error).__name__)
+            + "\n",
+        )
+
+
+if __name__ == "__main__":
+    main()

scripts/release/node_health.py

@@ -0,0 +1,584 @@
+"""Bounded operator readiness: fixed services, ACLs, broker and gated factory capacity."""
+
+import argparse
+import asyncio
+import json
+import logging
+import os
+import subprocess
+import urllib.request
+from pathlib import Path
+from uuid import UUID
+
+from . import (
+    configuration,
+    database,
+    dependencies,
+    dependency_install,
+    factory_service,
+    install,
+    operator_units,
+    preflight,
+    service_install,
+    services,
+)
+from .policy import DeliveryError, canonical, digest, fields, read_json
+
+# Table/column privilege introspection only. Business state counters come from
+# the execution owner, not a second privileged database state reader/writer.
+PRIVILEGES = {
+    "api": {
+        "accounts:SELECT": True,
+        "evaluations:SELECT": True,
+        "evaluations:state:UPDATE": False,
+        "accounts:is_administrator:UPDATE": False,
+    },
+    "coordinator": {
+        "evaluations:SELECT": True,
+        "evaluations:state:UPDATE": True,
+        "accounts:SELECT": False,
+        "problems:active_version_id:UPDATE": False,
+    },
+    "import": {
+        "problem_import_operations:SELECT": True,
+        "problem_import_operations:state:UPDATE": True,
+        "evaluations:state:UPDATE": False,
+        "accounts:SELECT": False,
+    },
+    "identity": {
+        "accounts:SELECT": True,
+        "accounts:is_administrator:UPDATE": True,
+        "evaluations:SELECT": False,
+        "evaluations:state:UPDATE": False,
+    },
+}
+
+
+def binding() -> tuple[dict, dict, Path]:
+    selected = install.selection(check_configuration=False)
+    payload = install.RELEASES / selected["release_id"] / "payload"
+    for name in (
+        "node_health.py",
+        "broker_health.py",
+        "factory_health.py",
+        "import_health.py",
+        "operator_units.py",
+    ):
+        shipped = payload / "delivery/scripts/release" / name
+        preflight.protected(shipped)
+        if digest(shipped) != digest(Path(__file__).parent / name):
+            raise DeliveryError("health tool differs from selected delivery")
+    receipt = read_json(
+        configuration.installed_receipt_path(selected["release_id"], selected["configuration_id"]),
+        16384,
+    )
+    configuration.validate_receipt(receipt, installed=True)
+    paths = {}
+    for name in configuration.CONFIG_FILES:
+        path = configuration.installed_path(
+            selected["release_id"], selected["configuration_id"], name
+        )
+        preflight.protected(path)
+        configuration.private(path)
+        if digest(path) != receipt["configuration_files"][str(path)]:
+            raise DeliveryError("live configuration changed since selection")
+        paths[name] = path
+    manifest_path = payload.parent / "manifest.json"
+    preflight.protected(manifest_path)
+    if digest(manifest_path) != selected["manifest_sha256"]:
+        raise DeliveryError("selected manifest changed")
+    data = read_json(manifest_path)
+    # Immutable payload bytes/images are verified by installation and every
+    # actual Sandbox request. A frequent status query is not a rebuild/certifier.
+    configuration.validate_contents(
+        paths, payload, data, attest=receipt["operator_attested_measurement"]
+    )
+    return selected, paths, payload
+
+
+def service_profile(name: str) -> tuple[int, int, int, int, int]:
+    if name in services.PROFILES:
+        user, memory, pids, cpu, _, _ = services.PROFILES[name]
+        return preflight.ACCOUNTS[user], memory, pids, cpu, 64
+    if name in {"postgres", "rabbitmq"}:
+        return (1203 if name == "postgres" else 1204), 512, 64, 100, 256
+    if name == "sandbox-factory":
+        return 0, 256, 32, 50, 64
+    raise DeliveryError("unknown health service")
+
+
+def inspect_unit(name: str, selected: dict) -> dict:
+    uid, memory, pids, cpu, file_mib = service_profile(name)
+    unit = "kf-" + name + ".service"
+    link = service_install.SYSTEM_UNITS / unit
+    if name in services.PROFILES:
+        directory = service_install.known_link(link)
+        expected = service_install.UNITS / (
+            selected["release_id"] + "." + selected["configuration_id"]
+        )
+    elif name == "sandbox-factory":
+        directory = factory_service.known_link(link)
+        expected = factory_service.UNITS / (
+            selected["release_id"] + "." + selected["configuration_id"]
+        )
+    else:
+        directory = dependency_install.known_link(link)
+        expected = dependency_install.ROOT / (
+            selected["release_id"] + "." + selected["configuration_id"]
+        )
+    if directory != expected:
+        raise DeliveryError("service snapshot is not selected")
+    value = dict(
+        line.split("=", 1)
+        for line in service_install.systemctl(
+            "show",
+            unit,
+            "--property=LoadState,ActiveState,SubState,MainPID,FragmentPath,ControlGroup",
+        ).splitlines()
+        if "=" in line
+    )
+    if (
+        value.get("LoadState") != "loaded"
+        or value.get("ActiveState") != "active"
+        or value.get("SubState") != "running"
+        or value.get("FragmentPath") not in {str(link), str(directory / unit)}
+        or value.get("ControlGroup") != "/kfonlinejudge.slice/" + unit
+    ):
+        raise DeliveryError("selected service is not running")
+    pid = int(value.get("MainPID", "0"))
+    if not 1 < pid < 2**31:
+        raise DeliveryError("running service PID differs")
+    info = dict(
+        line.split(":", 1)
+        for line in Path(f"/proc/{pid}/status").read_text().splitlines()
+        if ":" in line
+    )
+    if (
+        info.get("Uid", "").split() != [str(uid)] * 4
+        or info.get("Gid", "").split() != [str(uid)] * 4
+    ):
+        raise DeliveryError("effective service identity differs")
+    if not set(info.get("Groups", "").split()) <= {str(uid)}:
+        raise DeliveryError("unexpected service supplementary authority")
+    if name != "sandbox-factory" and (
+        info.get("NoNewPrivs", "").strip() != "1"
+        or any(
+            not info.get(k, "").strip() or int(info[k].strip(), 16) != 0
+            for k in ("CapInh", "CapPrm", "CapEff", "CapBnd", "CapAmb")
+        )
+    ):
+        raise DeliveryError("effective ordinary privilege boundary differs")
+    if name == "sandbox-factory" and (
+        info.get("NoNewPrivs", "").strip() != "0" or info.get("Seccomp", "").strip() != "0"
+    ):
+        raise DeliveryError("factory inherited incompatible hardening")
+    if name in {"api", "outbox", "problem-import"} and info.get("NSpid", "").split()[-1:] != ["1"]:
+        raise DeliveryError("Python service private PID namespace missing")
+    group = Path("/sys/fs/cgroup" + value["ControlGroup"])
+    actual = {
+        k: (group / k).read_text().strip()
+        for k in ("memory.max", "memory.swap.max", "pids.max", "cpu.max")
+    }
+    if actual != {
+        "memory.max": str(memory * 1024**2),
+        "memory.swap.max": "0",
+        "pids.max": str(pids),
+        "cpu.max": f"{cpu * 1000} 100000",
+    }:
+        raise DeliveryError("effective service limits differ")
+    if Path(f"/proc/{pid}/cgroup").read_text().strip() != "0::" + value["ControlGroup"]:
+        raise DeliveryError("effective service cgroup differs")
+    expected_file = str(file_mib * 1024**2)
+    # /proc format has names with spaces; fixed relevant prefixes are unambiguous.
+    limits = Path(f"/proc/{pid}/limits").read_text().splitlines()
+    expected = {
+        "Max core file size": "0",
+        "Max file size": expected_file,
+        "Max open files": "64" if name == "problem-import" else "256",
+        "Max stack size": str(16 * 1024**2),
+    }
+    for prefix, bound in expected.items():
+        rows = [line[len(prefix) :].split() for line in limits if line.startswith(prefix)]
+        if len(rows) != 1 or rows[0][:2] != [bound, bound]:
+            raise DeliveryError("effective service process limits differ")
+    return {
+        "active": True,
+        "uid": uid,
+        "limits": actual,
+        "memory_current": int((group / "memory.current").read_text()),
+        "pids_current": int((group / "pids.current").read_text()),
+    }
+
+
+def validate_role(role: str, observed: dict) -> None:
+    expected = {
+        "login": configuration.PROFILES[role],
+        "database": database.DATABASE,
+        "canlogin": True,
+        "superuser": False,
+        "createdb": False,
+        "createrole": False,
+        "replication": False,
+        "bypassrls": False,
+        "member": True,
+        "inherited": True,
+        "set_role": False,
+        "connection_limit": 8,
+        "privileges": PRIVILEGES[role],
+    }
+    if canonical(observed) != canonical(expected):
+        raise DeliveryError("database role authority differs")
+
+
+async def roles(endpoints: dict) -> dict:
+    import asyncpg
+
+    result = {}
+    for role, requirements in PRIVILEGES.items():
+        connection = await asyncpg.connect(
+            endpoints["database"][role],
+            timeout=2,
+            command_timeout=2,
+            server_settings={
+                "application_name": "kf-node-health",
+                "default_transaction_read_only": "on",
+                "statement_timeout": "2000",
+                "lock_timeout": "500",
+                "idle_in_transaction_session_timeout": "3000",
+            },
+        )
+        try:
+            row = await connection.fetchrow(
+                "SELECT session_user login,current_database() database,rolcanlogin canlogin,"
+                "rolsuper superuser,rolcreatedb createdb,rolcreaterole createrole,"
+                "rolreplication replication,"
+                "rolbypassrls bypassrls,rolconnlimit connection_limit,"
+                "pg_has_role(session_user,$1,'MEMBER') member,"
+                "pg_has_role(session_user,$1,'USAGE') inherited,"
+                "pg_has_role(session_user,$1,'SET') set_role "
+                "FROM pg_roles WHERE rolname=session_user",
+                database.MEMBERSHIP[configuration.PROFILES[role]],
+            )
+            observed = dict(row)
+            observed["privileges"] = {}
+            for key in requirements:
+                parts = key.split(":")
+                observed["privileges"][key] = await connection.fetchval(
+                    "SELECT has_table_privilege(current_user,$1,$2)"
+                    if len(parts) == 2
+                    else "SELECT has_column_privilege(current_user,$1,$2,$3)",
+                    *parts,
+                )
+            validate_role(role, observed)
+            result[role] = {"authority_verified": True}
+        finally:
+            await connection.close(timeout=1)
+    return result
+
+
+def http(url: str, *, token: str | None = None) -> dict:
+    # Fixed numeric loopback paths only; inherited proxy settings and redirects
+    # may not turn an operator query into a request to another service/host.
+    if url not in {
+        "http://127.0.0.1:8000/api/v1/health/live",
+        "http://127.0.0.1:8100/health/ready",
+        "http://127.0.0.1:8100/internal/operations/status",
+        "http://127.0.0.1:8100/internal/operations/probe-id",
+    }:
+        raise DeliveryError("unknown health HTTP target")
+
+    class NoRedirect(urllib.request.HTTPRedirectHandler):
+        def redirect_request(self, *_):
+            return None
+
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect())
+    request = urllib.request.Request(
+        url, headers={"Authorization": "Bearer " + token} if token else {}
+    )
+    with opener.open(request, timeout=5) as response:
+        if response.status != 200:
+            raise DeliveryError("health HTTP unavailable")
+        result = operator_units.decode(response.read(32769))
+    if not isinstance(result, dict):
+        raise DeliveryError("health HTTP object required")
+    return result
+
+
+def validate_factory(value: dict, expected: str) -> None:
+    fields(
+        value,
+        {
+            "schema_version",
+            "configuration_sha256",
+            "supervisor_authorized",
+            "execution_gate",
+            "isolation_verified",
+            "max_concurrent_runs",
+            "active_helpers",
+            "cleanup_complete",
+            "remaining_requests",
+            "remaining_lifetime_seconds",
+        },
+        "factory health",
+    )
+    if (
+        value["schema_version"] != 1
+        or type(value["schema_version"]) is not int
+        or value["configuration_sha256"] != expected
+        or value["supervisor_authorized"] is not True
+        or value["execution_gate"] != "required_per_execution"
+        or value["isolation_verified"] is not False
+        or type(value["max_concurrent_runs"]) is not int
+        or value["max_concurrent_runs"] != 1
+        or type(value["active_helpers"]) is not int
+        or not 0 <= value["active_helpers"] <= 1
+        or type(value["cleanup_complete"]) is not bool
+        or (value["cleanup_complete"] and value["active_helpers"])
+        or type(value["remaining_requests"]) is not int
+        or not 0 <= value["remaining_requests"] <= 4096
+        or type(value["remaining_lifetime_seconds"]) is not int
+        or not 0 <= value["remaining_lifetime_seconds"] <= 86400
+    ):
+        raise DeliveryError("factory authorization/binding status differs")
+
+
+def validate_execution(value: dict, admission: dict) -> dict:
+    fields(
+        value,
+        {"schema_version", "admission", "configured_workers", "lease_seconds", "counters"},
+        "execution snapshot",
+    )
+    if (
+        type(value["schema_version"]) is not int
+        or value["schema_version"] != 1
+        or type(value["configured_workers"]) is not int
+        or value["configured_workers"] != 1
+        or value["lease_seconds"] != 120
+        or type(value["lease_seconds"]) is not int
+        or canonical(value["admission"]) != canonical(admission)
+    ):
+        raise DeliveryError("live coordinator admission differs")
+    counters = fields(
+        value["counters"],
+        {"queued", "in_flight", "cleanup_unconfirmed", "database_clock_unix_ms"},
+        "execution counters",
+    )
+    for name in ("queued", "in_flight", "cleanup_unconfirmed"):
+        item = fields(counters[name], {"count_capped", "truncated"}, "bounded execution counter")
+        if (
+            type(item["count_capped"]) is not int
+            or not 0 <= item["count_capped"] <= 128
+            or type(item["truncated"]) is not bool
+            or (item["truncated"] and item["count_capped"] != 128)
+        ):
+            raise DeliveryError("execution counter bound differs")
+    if (
+        type(counters["database_clock_unix_ms"]) is not int
+        or counters["database_clock_unix_ms"] <= 0
+    ):
+        raise DeliveryError("execution counter clock unavailable")
+    return counters
+
+
+def admission_ready(root: dict, counters: dict, broker: dict, admission: dict) -> bool:
+    reserve_seconds = (
+        max(
+            admission["legacy"]["max_job_wall_time_us"],
+            admission["multi_runtime"]["max_job_wall_time_us"],
+        )
+        + 999999
+    ) // 1000000 + 60
+    return bool(
+        root["cleanup_complete"]
+        and root["active_helpers"] == 0
+        and root["remaining_requests"] >= 257
+        and root["remaining_lifetime_seconds"] >= reserve_seconds
+        and counters["in_flight"] == {"count_capped": 0, "truncated": False}
+        and counters["cleanup_unconfirmed"] == {"count_capped": 0, "truncated": False}
+        and broker["consumers"] == 1
+    )
+
+
+async def notification_probe(url: str, identifier: str) -> None:
+    import aio_pika
+    from pamqp.commands import Basic
+
+    for name in ("aio_pika", "aiormq", "pamqp"):
+        logging.getLogger(name).setLevel(logging.CRITICAL)
+    connection = await aio_pika.connect(url, timeout=2, heartbeat=30)
+    try:
+        channel = await connection.channel(publisher_confirms=True, on_return_raises=True)
+        await channel.declare_queue(dependencies.QUEUE, passive=True, timeout=2)
+        # A synthetically absent evaluation notification cannot create a grade or
+        # execute a program: Coordinator must find and lease an existing row.
+        body = canonical(
+            {"schema_version": 1, "kind": "judge.evaluation", "evaluation_id": identifier}
+        ).strip()
+        reply = await channel.default_exchange.publish(
+            aio_pika.Message(
+                body,
+                content_type="application/json",
+                delivery_mode=aio_pika.DeliveryMode.PERSISTENT,
+            ),
+            routing_key=dependencies.QUEUE,
+            mandatory=True,
+            timeout=2,
+        )
+        if not isinstance(reply, Basic.Ack):
+            raise DeliveryError("broker delivery not confirmed")
+    finally:
+        await asyncio.wait_for(connection.close(), timeout=1)
+
+
+def observe(*, probe: bool) -> dict:
+    operator_units.guarded("node-health")
+    selected, paths, payload = binding()
+    result = {
+        "schema_version": 1,
+        "source_sha": read_json(payload.parent / "manifest.json")["source_sha"],
+        "configuration_id": selected["configuration_id"],
+        "ready_for_admission": False,
+        "isolation_verified": False,
+        "components": {},
+        "errors": [],
+    }
+    components, errors = result["components"], result["errors"]
+    for name in [*services.PROFILES, "postgres", "rabbitmq", "sandbox-factory"]:
+        try:
+            components[name] = inspect_unit(name, selected)
+        except (OSError, ValueError):
+            errors.append(name + "_unavailable")
+    endpoints = dependency_install.installed_endpoints(selected)
+    try:
+        components["db_authority"] = asyncio.run(roles(endpoints))
+    except Exception:
+        errors.append("db_authority_unavailable")
+    try:
+        live = http("http://127.0.0.1:8000/api/v1/health/live")
+        if live.get("status") != "ok" or live.get("service") != "platform":
+            raise DeliveryError("API live shape differs")
+        components["api_live"] = True
+    except (OSError, ValueError):
+        errors.append("api_live_unavailable")
+    try:
+        coordinator = read_json(paths["coordinator.json"], 32768)
+        counters = validate_execution(
+            http(
+                "http://127.0.0.1:8100/internal/operations/status",
+                token=coordinator["platform_token"],
+            ),
+            coordinator["admission"],
+        )
+        components["execution"] = counters
+    except (OSError, ValueError):
+        errors.append("execution_owner_unavailable")
+    try:
+        components["broker"] = operator_units.run("broker-health", payload)
+    except (OSError, ValueError, subprocess.SubprocessError):
+        errors.append("broker_metadata_unavailable")
+    try:
+        state = operator_units.run("import-health", payload)
+        fields(
+            state,
+            {
+                "schema_version",
+                "worker_ready",
+                "reserved_bytes",
+                "uploaded_capped",
+                "validating_capped",
+                "truncated",
+            },
+            "import status",
+        )
+        if (
+            type(state["schema_version"]) is not int
+            or state["schema_version"] != 1
+            or state["worker_ready"] is not True
+            or state["truncated"] is not False
+        ):
+            raise DeliveryError("import heartbeat or counter unavailable")
+        for key, maximum in (
+            ("reserved_bytes", 268435456),
+            ("uploaded_capped", 128),
+            ("validating_capped", 1),
+        ):
+            if type(state[key]) is not int or not 0 <= state[key] <= maximum:
+                raise DeliveryError("import status bound differs")
+        components["import_pipeline"] = state
+    except (OSError, ValueError, subprocess.SubprocessError):
+        errors.append("import_pipeline_unavailable")
+    try:
+        status = operator_units.run("factory-health", payload)
+        validate_factory(status, digest(paths["runtime.json"]))
+        components["factory_status"] = status
+    except (OSError, ValueError, subprocess.SubprocessError):
+        errors.append("factory_status_unavailable")
+    if probe:
+        try:
+            value = fields(
+                http(
+                    "http://127.0.0.1:8100/internal/operations/probe-id",
+                    token=coordinator["platform_token"],
+                ),
+                {"schema_version", "evaluation_id", "absent"},
+                "notification probe identity",
+            )
+            identifier = UUID(value["evaluation_id"])
+            if (
+                type(value["schema_version"]) is not int
+                or value["schema_version"] != 1
+                or value["absent"] is not True
+                or identifier.int == 0
+                or str(identifier) != value["evaluation_id"]
+            ):
+                raise DeliveryError("notification probe authority unavailable")
+            asyncio.run(
+                asyncio.wait_for(
+                    notification_probe(endpoints["amqp"]["outbox"], str(identifier)), timeout=6
+                )
+            )
+            components["notification_delivery_confirmed"] = True
+        except Exception:
+            errors.append("notification_delivery_unconfirmed")
+    if install.selection(check_configuration=False) != selected:
+        errors.append("selected_binding_changed")
+    if not errors:
+        root = components["factory_status"]
+        counters = components["execution"]
+        result["ready_for_admission"] = admission_ready(
+            root, counters, components["broker"], coordinator["admission"]
+        )
+    return result
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Bounded installed service readiness")
+    parser.add_argument("--inside", action="store_true")
+    parser.add_argument("--probe-notification", action="store_true")
+    args = parser.parse_args()
+    try:
+        if args.inside:
+            result = observe(probe=args.probe_notification)
+        else:
+            if os.getuid() != 0 or os.geteuid() != 0:
+                raise DeliveryError("root operator required")
+            selected = install.selection(check_configuration=False)
+            result = operator_units.run(
+                "node-health",
+                install.RELEASES / selected["release_id"] / "payload",
+                notification_probe=args.probe_notification,
+            )
+        print(json.dumps(result, sort_keys=True))
+        if not args.inside and not result["ready_for_admission"]:
+            raise SystemExit(2)
+    except (OSError, ValueError, subprocess.SubprocessError) as error:
+        parser.exit(
+            1,
+            "Node health refused: "
+            + (str(error) if isinstance(error, DeliveryError) else type(error).__name__)
+            + "\n",
+        )
+
+
+if __name__ == "__main__":
+    main()

scripts/release/operator_units.py

@@ -0,0 +1,231 @@
+"""Finite fixed-purpose operator helpers; no caller-controlled executable or identity."""
+
+import json
+import os
+import resource
+import subprocess
+from pathlib import Path
+
+from . import configuration, install, preflight, service_install
+from .policy import DeliveryError, invalid_constant, unique_object
+from .services import python_environment
+
+PROFILES = {
+    "node-health": ("root", 0, 128, 16, 50, 90),
+    "broker-health": ("rabbitmq", 1204, 256, 64, 50, 45),
+    "factory-health": ("kf-worker", 1202, 32, 4, 10, 10),
+    "import-health": ("kf-oj", 1200, 128, 16, 25, 10),
+}
+
+
+def guarded(name: str) -> None:
+    _, uid, memory, pids, cpu, _ = PROFILES[name]
+    if (
+        os.getuid() != uid
+        or os.geteuid() != uid
+        or os.getgid() != uid
+        or not set(os.getgroups()) <= {uid}
+    ):
+        raise DeliveryError("fixed operator helper identity required")
+    observed = dict(
+        line.split(":", 1)
+        for line in Path("/proc/self/status").read_text().splitlines()
+        if ":" in line
+    )
+    if observed.get("NoNewPrivs", "").strip() != "1" or any(
+        not observed.get(k, "").strip() or int(observed[k].strip(), 16) != 0
+        for k in ("CapInh", "CapPrm", "CapEff", "CapBnd", "CapAmb")
+    ):
+        raise DeliveryError("operator helper privilege boundary differs")
+    group = preflight.current_cgroup()
+    if str(group) != "/sys/fs/cgroup/kfonlinejudge.slice/kf-" + name + ".service":
+        raise DeliveryError("operator helper cgroup differs")
+    for directory, expected in (
+        (group, [str(memory * 1024**2), "0", str(pids), f"{cpu * 1000} 100000"]),
+        (
+            Path("/sys/fs/cgroup/kfonlinejudge.slice"),
+            [str(3840 * 1024**2), "0", "384", "200000 100000"],
+        ),
+    ):
+        actual = [
+            (directory / k).read_text().strip()
+            for k in ("memory.max", "memory.swap.max", "pids.max", "cpu.max")
+        ]
+        if actual != expected:
+            raise DeliveryError("operator helper/node budgets differ")
+    for key, maximum in (
+        (resource.RLIMIT_CORE, 0),
+        (resource.RLIMIT_FSIZE, 16 * 1024**2),
+        (resource.RLIMIT_NOFILE, 256),
+        (resource.RLIMIT_STACK, 16 * 1024**2),
+    ):
+        if resource.getrlimit(key) != (maximum, maximum):
+            raise DeliveryError("operator helper process budgets differ")
+
+
+def decode(contents: bytes, maximum: int = 32768):
+    if len(contents) > maximum:
+        raise DeliveryError("operator output exceeds bound")
+    return json.loads(
+        contents,
+        object_pairs_hook=unique_object,
+        parse_constant=invalid_constant,
+    )
+
+
+def run(name: str, payload: Path, *, notification_probe: bool = False) -> dict:
+    if os.getuid() != 0 or os.geteuid() != 0 or name not in PROFILES:
+        raise DeliveryError("fixed root operator helper required")
+    user, _, memory, pids, cpu, seconds = PROFILES[name]
+    unit = "kf-" + name + ".service"
+    current = dict(
+        line.split("=", 1)
+        for line in service_install.systemctl(
+            "show", unit, "--property=LoadState,ActiveState,MainPID"
+        ).splitlines()
+        if "=" in line
+    )
+    if (
+        current.get("LoadState") != "not-found"
+        or current.get("ActiveState") != "inactive"
+        or current.get("MainPID") != "0"
+    ):
+        raise DeliveryError("operator helper already active or unknown")
+    release_id = payload.parent.name
+    selected = install.selection(check_configuration=False)
+    if (
+        payload != install.RELEASES / selected["release_id"] / "payload"
+        or release_id != selected["release_id"]
+    ):
+        raise DeliveryError("operator helper is not the selected payload")
+    description = "KF fixed " + name + " " + release_id + "." + selected["configuration_id"]
+    command = [
+        "/usr/bin/systemd-run",
+        "--quiet",
+        "--wait",
+        "--pipe",
+        "--collect",
+        "--unit=" + unit,
+        "--description=" + description,
+        "--slice=kfonlinejudge.slice",
+    ]
+    properties = {
+        "Type": "exec",
+        "User": user,
+        "Group": user,
+        "UMask": "0077",
+        "MemoryMax": str(memory) + "M",
+        "MemorySwapMax": "0",
+        "TasksMax": str(pids),
+        "CPUQuota": str(cpu) + "%",
+        "LimitCORE": "0",
+        "LimitFSIZE": "16M",
+        "LimitNOFILE": "256",
+        "LimitSTACK": "16M",
+        "RuntimeMaxSec": str(seconds),
+        "TimeoutStartSec": "10",
+        "TimeoutStopSec": "10",
+        "Restart": "no",
+        "KillMode": "control-group",
+        "NoNewPrivileges": "yes",
+        "CapabilityBoundingSet": "",
+        "AmbientCapabilities": "",
+        "ProtectSystem": "strict",
+        "ProtectHome": "yes",
+        "PrivateTmp": "yes",
+        "PrivateDevices": "yes",
+        "ProtectControlGroups": "yes",
+        "RestrictAddressFamilies": "AF_UNIX AF_INET",
+        "WorkingDirectory": str(payload),
+    }
+    for key, value in properties.items():
+        command.append("--property=" + key + "=" + value)
+    if name == "import-health":
+        path = configuration.installed_path(
+            selected["release_id"], selected["configuration_id"], "import.env"
+        )
+        command += [
+            "--property=EnvironmentFile=" + str(path),
+            "--property=PrivatePIDs=yes",
+            "--property=PrivateNetwork=yes",
+            "--property=RestrictAddressFamilies=AF_UNIX",
+        ]
+    if name == "import-health":
+        if notification_probe:
+            raise DeliveryError("unexpected import probe override")
+        command += [
+            "--property=Environment=PATH=/usr/bin:/bin LC_ALL=C.UTF-8 "
+            + python_environment(payload)
+        ]
+        command += [
+            str(payload / "python/bin/python3.13"),
+            "-s",
+            "-P",
+            "-m",
+            "scripts.release.import_health",
+            "--inside",
+        ]
+    else:
+        command += ["/usr/bin/env", "-i", "PATH=/usr/bin:/bin", "LC_ALL=C.UTF-8"]
+    if name == "factory-health":
+        if notification_probe:
+            raise DeliveryError("unexpected factory probe override")
+        command += [
+            *python_environment(payload).split(),
+            str(payload / "python/bin/python3.13"),
+            "-s",
+            "-P",
+            "-m",
+            "scripts.release.factory_health",
+            "--inside",
+        ]
+    elif name != "import-health":
+        command += [
+            *python_environment(payload).split(),
+            str(payload / "python/bin/python3.13"),
+            "-s",
+            "-P",
+            "-m",
+            "scripts.release." + name.replace("-", "_"),
+            "--inside",
+        ]
+        if notification_probe and name == "node-health":
+            command.append("--probe-notification")
+        elif notification_probe:
+            raise DeliveryError("unexpected broker probe override")
+    try:
+        # Pipe output is bounded by the fixed helper and the service file/memory
+        # ceilings; helper tools below cap each pipe before serializing it.
+        result = subprocess.run(
+            command,
+            capture_output=True,
+            timeout=seconds + 15,
+            env={"PATH": "/usr/bin:/bin", "LC_ALL": "C"},
+        )
+        if result.returncode or len(result.stderr) > 4096:
+            raise DeliveryError("operator helper unavailable")
+        value = decode(result.stdout)
+        if not isinstance(value, dict):
+            raise DeliveryError("operator helper object required")
+    finally:
+        actual = dict(
+            line.split("=", 1)
+            for line in service_install.systemctl(
+                "show", unit, "--property=LoadState,ActiveState,MainPID,Transient,Description"
+            ).splitlines()
+            if "=" in line
+        )
+        if actual.get("LoadState") != "not-found":
+            if actual.get("Transient") != "yes" or actual.get("Description") != description:
+                raise DeliveryError("foreign operator helper; refused cleanup")
+            service_install.systemctl("stop", unit)
+            actual = dict(
+                line.split("=", 1)
+                for line in service_install.systemctl(
+                    "show", unit, "--property=ActiveState,MainPID"
+                ).splitlines()
+                if "=" in line
+            )
+            if actual.get("ActiveState") != "inactive" or actual.get("MainPID") != "0":
+                raise DeliveryError("operator helper cleanup unconfirmed")
+    return value