feat(ops): add bounded owner and kernel observations

backend/tests/test_delivery.py

@@ -37,6 +37,38 @@ restore = importlib.import_module("scripts.release.restore")
 restore_policy = importlib.import_module("scripts.release.restore_policy")
 recovery_identity = importlib.import_module("scripts.release.recovery_identity")
 recovery_database = importlib.import_module("scripts.release.recovery_database")
+node_observation = importlib.import_module("scripts.release.node_observation")
+
+
+def test_kernel_observation_keeps_cumulative_counters_and_rejects_duplicate_rows():
+    raw = (
+        "usage_usec 17\nuser_usec 11\nsystem_usec 6\nnr_periods 9\nnr_throttled 2\nthrottled_usec 3"
+    )
+    observed = node_observation.counters(raw, node_observation.CPU_FIELDS)
+    assert observed["usage_usec"] == 17 and observed["nr_throttled"] == 2
+    with pytest.raises(policy.DeliveryError):
+        node_observation.counters(raw + "\nusage_usec 18", node_observation.CPU_FIELDS)
+    with pytest.raises(policy.DeliveryError):
+        node_observation.counters("usage_usec 17", node_observation.CPU_FIELDS)
+
+
+def test_kernel_io_observation_aggregates_devices_without_returning_device_identities():
+    actual = node_observation.io_totals(
+        "8:0 rbytes=5 wbytes=7 rios=1 wios=2 dbytes=0 dios=0\n"
+        "8:1 rbytes=11 wbytes=13 rios=3 wios=4 dbytes=0 dios=0"
+    )
+    assert actual == {"devices_observed": 2, "rbytes": 16, "wbytes": 20, "rios": 4, "wios": 6}
+    assert node_observation.io_totals("")["devices_observed"] == 0
+    with pytest.raises(policy.DeliveryError):
+        node_observation.io_totals(
+            "8:0 rbytes=5 wbytes=7 rios=1 wios=2\n8:0 rbytes=0 wbytes=0 rios=0 wios=0"
+        )
+
+
+@pytest.mark.parametrize("value", ["-1", "max", "True", str(2**63)])
+def test_kernel_observation_does_not_replace_unavailable_or_invalid_counters_with_zero(value):
+    with pytest.raises(policy.DeliveryError):
+        node_observation.number(value)
 
 
 def test_postgres_recovery_retains_narrow_existing_role_authorities():

scripts/release/README.md

@@ -503,3 +503,23 @@ readiness and actual four-language/history checks in the second VM. Neither
 data nor credential receipts assert execution safety or healthy public service.
 These system steps and the 30-minute end-to-end recovery target remain pending
 independent VM acceptance.
+
+`python3.13 -m scripts.release.node_observation` dispatches one finite,
+read-only local observation through its fixed Root cap0/no-new-privileges helper
+(128 MiB, zero swap, 16 tasks, 25% CPU, 120 seconds, single named unit).
+It reads kernel memory/PIDs current and peak, CPU throttling, memory events,
+aggregate device IO and filesystem bytes/inodes without enumerating private
+source, hidden cases or credential files. Current selected service identities,
+ACLs, execution leases/cleanup, broker/outbox/import backlog and finite factory
+capacity come from the existing status owners. Notification probing stays off;
+it does not publish, claim, acknowledge, close admission or repair state.
+
+The result includes source/configuration binding, a sample time and observation
+duration. Independent readings are a bounded time window, not an atomic backup
+cut. Missing kernel fields, stopped services or failed owner queries are explicit
+incomplete observations; there is no invented zero/healthy fallback. Busy
+admission can remain false in a complete observation. `isolation_verified`
+remains false: every real execution still needs its own Sandbox gate. Private
+operational output is for the local operator and must stay outside public API,
+Git and CI artifacts. Actual resource/queue observations and long-run evidence
+still require the independent VM.

scripts/release/node_observation.py

@@ -0,0 +1,211 @@
+"""Finite read-only owner/kernel observation; no execution or admission mutation."""
+
+import argparse
+import json
+import os
+import re
+import subprocess
+import time
+from pathlib import Path
+
+from . import install, journal, node_health, operator_units, preflight, services
+from .policy import DeliveryError, canonical, digest, integer, read_json
+
+NODE = Path("/sys/fs/cgroup/kfonlinejudge.slice")
+UNITS = (*services.PROFILES, "postgres", "rabbitmq", "sandbox-factory", "proxy")
+CPU_FIELDS = {
+    "usage_usec",
+    "user_usec",
+    "system_usec",
+    "nr_periods",
+    "nr_throttled",
+    "throttled_usec",
+}
+MEMORY_FIELDS = {"low", "high", "max", "oom", "oom_kill", "oom_group_kill"}
+IO_FIELDS = {"rbytes", "wbytes", "rios", "wios"}
+
+
+def text(path: Path, maximum: int = 4096) -> str:
+    with path.open() as source:
+        value = source.read(maximum + 1)
+    if len(value) > maximum:
+        raise DeliveryError("kernel observation byte limit")
+    return value.strip()
+
+
+def number(value: str) -> int:
+    if not isinstance(value, str) or re.fullmatch(r"[0-9]{1,19}", value) is None:
+        raise DeliveryError("invalid kernel counter")
+    return integer(int(value), 0, 2**63 - 1, "kernel counter")
+
+
+def counters(value: str, selected: set[str]) -> dict:
+    result = {}
+    lines = value.splitlines()
+    if not 1 <= len(lines) <= 32:
+        raise DeliveryError("kernel counter inventory limit")
+    for line in lines:
+        parts = line.split()
+        if len(parts) != 2 or parts[0] in result:
+            raise DeliveryError("invalid kernel counter row")
+        result[parts[0]] = number(parts[1])
+    if not selected <= result.keys():
+        raise DeliveryError("required kernel counter absent")
+    return {key: result[key] for key in sorted(selected)}
+
+
+def io_totals(value: str) -> dict:
+    """Bound the device inventory and omit device identities from the output."""
+    lines = value.splitlines()
+    if len(lines) > 16:
+        raise DeliveryError("kernel IO device inventory limit")
+    totals = dict.fromkeys(sorted(IO_FIELDS), 0)
+    devices = set()
+    for line in lines:
+        parts = line.split()
+        if not parts or re.fullmatch(r"[0-9]{1,10}:[0-9]{1,10}", parts[0]) is None:
+            raise DeliveryError("invalid kernel IO device row")
+        if parts[0] in devices or len(parts) > 16:
+            raise DeliveryError("duplicate or excessive kernel IO row")
+        devices.add(parts[0])
+        values = {}
+        for member in parts[1:]:
+            key, separator, raw = member.partition("=")
+            if not separator or key in values:
+                raise DeliveryError("invalid kernel IO counter")
+            values[key] = number(raw)
+        if not IO_FIELDS <= values.keys():
+            raise DeliveryError("required kernel IO counter absent")
+        for key in totals:
+            totals[key] = integer(totals[key] + values[key], 0, 2**63 - 1, "total kernel IO")
+    return {"devices_observed": len(devices), **totals}
+
+
+def kernel(path: Path) -> dict:
+    preflight.protected(path, directory=True)
+    return {
+        "memory_current_bytes": number(text(path / "memory.current")),
+        "memory_peak_bytes": number(text(path / "memory.peak")),
+        "pids_current": number(text(path / "pids.current")),
+        "pids_peak": number(text(path / "pids.peak")),
+        "memory_events": counters(text(path / "memory.events"), MEMORY_FIELDS),
+        "cpu": counters(text(path / "cpu.stat"), CPU_FIELDS),
+        "io": io_totals(text(path / "io.stat", 8192)),
+        "limits": {
+            key: text(path / key, 64)
+            for key in ("memory.max", "memory.swap.max", "pids.max", "cpu.max")
+        },
+    }
+
+
+def filesystem(path: Path) -> dict:
+    # Aggregate filesystem figures never enumerate source/hidden/credential files.
+    if path.is_symlink() or not path.is_dir():
+        raise DeliveryError("ordinary observation storage directory required")
+    value = os.statvfs(path)
+    return {
+        "capacity_bytes": value.f_frsize * value.f_blocks,
+        "available_bytes": value.f_frsize * value.f_bavail,
+        "used_bytes": value.f_frsize * (value.f_blocks - value.f_bfree),
+        "inodes_total": value.f_files,
+        "inodes_free": value.f_favail,
+        "read_only": bool(value.f_flag & os.ST_RDONLY),
+    }
+
+
+def observe() -> dict:
+    operator_units.guarded("node-observation")
+    selected, _, payload = node_health.binding()
+    for name in ("node_observation.py", "operator_units.py"):
+        shipped = payload / "delivery/scripts/release" / name
+        preflight.protected(shipped)
+        if digest(shipped) != digest(Path(__file__).parent / name):
+            raise DeliveryError("observation program differs from selected delivery")
+    started = time.monotonic()
+    result = {
+        "schema_version": 1,
+        "kind": "kf.node.read-only-observation",
+        "source_sha": read_json(payload.parent / "manifest.json")["source_sha"],
+        "configuration_id": selected["configuration_id"],
+        "observed_unix_ms": time.time_ns() // 1_000_000,
+        "read_only": True,
+        "isolation_verified": False,
+        "complete": False,
+        "kernel": {},
+        "filesystems": {},
+        "owner_health": None,
+        "errors": [],
+    }
+    for label, path in [
+        ("node", NODE),
+        *[(name, NODE / ("kf-" + name + ".service")) for name in UNITS],
+        ("tasks", NODE / "kf-judge-tasks"),
+    ]:
+        try:
+            result["kernel"][label] = kernel(path)
+        except (OSError, ValueError):
+            result["errors"].append("kernel_" + label + "_unavailable")
+    paths = {
+        "root": Path("/var/lib/kf-onlinejudge"),
+        "artifacts": Path("/var/lib/kf-onlinejudge/artifacts"),
+        "imports": Path("/var/lib/kf-onlinejudge/imports"),
+        "postgres": Path("/var/lib/kf-onlinejudge/postgres"),
+        "rabbitmq": Path("/var/lib/kf-onlinejudge/rabbitmq"),
+        "proxy": Path("/var/lib/kf-onlinejudge/proxy"),
+        "backups": Path("/var/lib/kf-onlinejudge/backups"),
+    }
+    try:
+        paths["journal"] = Path("/var/log/journal") / (
+            journal.machine_id() + "." + journal.NAMESPACE
+        )
+    except (OSError, ValueError):
+        result["errors"].append("filesystem_journal_unavailable")
+    for label, path in paths.items():
+        try:
+            result["filesystems"][label] = filesystem(path)
+        except (OSError, ValueError):
+            result["errors"].append("filesystem_" + label + "_unavailable")
+    try:
+        # Owner status/ACL/MQ/factory observers run under their existing fixed
+        # identities. Notification probing is off: no publish/ack/claim/cleanup.
+        health = operator_units.run("node-health", payload)
+        if (
+            health.get("source_sha") != result["source_sha"]
+            or health.get("configuration_id") != selected["configuration_id"]
+        ):
+            raise DeliveryError("owner health observation binding differs")
+        result["owner_health"] = health
+        if health.get("errors"):
+            result["errors"].append("owner_observation_incomplete")
+    except (OSError, ValueError, subprocess.SubprocessError):
+        result["errors"].append("owner_observation_unavailable")
+    if install.selection(check_configuration=False) != selected:
+        result["errors"].append("selected_binding_changed")
+    result["sample_duration_ms"] = int((time.monotonic() - started) * 1000)
+    result["complete"] = not result["errors"]
+    if len(canonical(result)) > 32768:
+        raise DeliveryError("read-only observation result exceeds output bound")
+    return result
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Finite read-only owner and kernel observation")
+    parser.add_argument("--inside", action="store_true", help=argparse.SUPPRESS)
+    args = parser.parse_args()
+    try:
+        if args.inside:
+            value = observe()
+        else:
+            selected = install.selection()
+            value = operator_units.run(
+                "node-observation", install.RELEASES / selected["release_id"] / "payload"
+            )
+        print(json.dumps(value, sort_keys=True))
+        if not args.inside and not value["complete"]:
+            raise SystemExit(2)
+    except (OSError, ValueError, subprocess.SubprocessError) as error:
+        parser.exit(1, "Read-only node observation unavailable: " + type(error).__name__ + "\n")
+
+
+if __name__ == "__main__":
+    main()

scripts/release/operator_units.py

@@ -17,6 +17,7 @@ PROFILES = {
     "import-health": ("kf-oj", 1200, 128, 16, 25, 10),
     "outbox-health": ("kf-oj", 1200, 128, 16, 25, 10),
     "node-maintenance": ("root", 0, 256, 32, 50, 900),
+    "node-observation": ("root", 0, 128, 16, 25, 120),
 }