diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml
index bd22d21..c467ec1 100644
--- a/.github/workflows/rust-test.yml
+++ b/.github/workflows/rust-test.yml
@@ -68,7 +68,18 @@ jobs:
           cargo test -p lance-context-wal --features placement,admission,ipc,record,arrow-contract --lib
           cargo test -p lance-context-wal --features arrow-contract,serde_json/preserve_order --lib ingest_contract::arrow
       - name: WAL experiment process lifecycle tests
-        run: cargo test -p lance-context-wal --features runner --test runner
+        run: cargo test -p lance-context-wal --features runner --test runner --bin context-wal-resource
+      - name: WAL resource observation in a container cgroup namespace
+        run: |
+          cargo test -p lance-context-wal --features runner --test resource --no-run --message-format=json > "$RUNNER_TEMP/wal-resource-build.jsonl"
+          wal_resource_test=$(jq -rs '[.[] | select(.reason == "compiler-artifact" and .target.name == "resource" and (.target.kind | index("test")) and .executable != null) | .executable] | unique | if length == 1 then .[0] else error("expected one resource test executable") end' "$RUNNER_TEMP/wal-resource-build.jsonl")
+          # Hosted runners use a host cgroup path. Remount cgroup2 inside private
+          # mount/cgroup namespaces, then drop privileges before running tests.
+          sudo unshare --mount --cgroup --propagation private sh -eu -c '
+            umount /sys/fs/cgroup
+            mount -t cgroup2 cgroup2 /sys/fs/cgroup
+            exec setpriv --reuid "$1" --regid "$2" --clear-groups --no-new-privs "$3"
+          ' sh "$(id -u)" "$(id -g)" "$wal_resource_test"
       - name: Start etcd for HA scheduler test
         run: |
           ETCD_VERSION=3.7.0
@@ -139,7 +150,20 @@ jobs:
       - name: Clean stale coverage profile data
         run: cargo llvm-cov clean --workspace
       - name: Collect coverage (unit and integration tests)
-        run: cargo llvm-cov --no-report --workspace --all-features
+        run: |
+          # All features include the resource integration test. Give coverage
+          # the same cgroup scope as the ordinary test job, without running
+          # Cargo or test executables as root. Preserve the build configuration.
+          sudo --preserve-env=CARGO_HOME,RUSTUP_HOME,RUSTUP_TOOLCHAIN,RUSTFLAGS,RUST_MIN_STACK,LANCE_MEM_POOL_SIZE,RUST_BACKTRACE,CARGO_INCREMENTAL \
+            unshare --mount --cgroup --propagation private sh -eu -c '
+              umount /sys/fs/cgroup
+              mount -t cgroup2 cgroup2 /sys/fs/cgroup
+              wal_test_uid=$1
+              wal_test_gid=$2
+              shift 2
+              exec setpriv --reuid "$wal_test_uid" --regid "$wal_test_gid" --clear-groups --no-new-privs "$@"
+            ' sh "$(id -u)" "$(id -g)" env "HOME=$HOME" "PATH=$PATH" \
+            cargo llvm-cov --no-report --workspace --all-features
       - name: Collect coverage (etcd-backed ignored tests)
         env:
           ETCD_TEST_ENDPOINTS: http://127.0.0.1:2379
diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml
index 0d332f7..5552ed3 100644
--- a/crates/lance-context-wal/Cargo.toml
+++ b/crates/lance-context-wal/Cargo.toml
@@ -43,6 +43,11 @@ name = "context-wal-runner"
 path = "src/bin/runner.rs"
 required-features = ["runner"]
 
+[[bin]]
+name = "context-wal-resource"
+path = "src/bin/resource.rs"
+required-features = ["runner"]
+
 [[bin]]
 name = "context-wal-lab"
 path = "src/bin/lab.rs"
diff --git a/crates/lance-context-wal/src/bin/resource.rs b/crates/lance-context-wal/src/bin/resource.rs
new file mode 100644
index 0000000..5bbce6d
--- /dev/null
+++ b/crates/lance-context-wal/src/bin/resource.rs
@@ -0,0 +1,363 @@
+//! Read-only Linux experiment sampling of one recorded process incarnation.
+//! Same-container cgroup v2 counters are shared, never additive with process RSS.
+use anyhow::{ensure, Context, Result};
+use clap::Parser;
+use serde::{Deserialize, Serialize};
+use std::{
+    collections::{BTreeMap, BTreeSet},
+    fs,
+    io::Write,
+    os::unix::fs::MetadataExt,
+    path::{Path, PathBuf},
+    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
+};
+
+#[derive(Parser)]
+struct Args {
+    /// Existing runner operation; this command never launches or stops it.
+    #[arg(long)]
+    operation: PathBuf,
+    #[arg(long, default_value_t = 500)]
+    interval_ms: u64,
+    #[arg(long)]
+    disk_directory: Option<PathBuf>,
+}
+#[derive(Deserialize)]
+struct Identity {
+    pid: u32,
+    start_ticks: u64,
+    boot: String,
+}
+#[derive(Serialize)]
+struct Process {
+    state: char,
+    start_ticks: u64,
+    cpu_ticks: u64,
+    rss_bytes: u64,
+}
+fn stat(text: &str, page_bytes: u64) -> Result<Process> {
+    let (_, rest) = text.rsplit_once(") ").context("invalid process stat")?;
+    let fields: Vec<_> = rest.split_whitespace().collect();
+    ensure!(fields.len() >= 22, "truncated process stat");
+    Ok(Process {
+        state: fields[0].chars().next().context("empty process state")?,
+        start_ticks: fields[19].parse()?,
+        cpu_ticks: fields[11]
+            .parse::<u64>()?
+            .checked_add(fields[12].parse()?)
+            .context("CPU count overflow")?,
+        rss_bytes: fields[21]
+            .parse::<u64>()?
+            .checked_mul(page_bytes)
+            .context("RSS overflow")?,
+    })
+}
+fn counters(text: &str) -> Result<BTreeMap<String, u64>> {
+    text.lines()
+        .map(|line| {
+            let mut fields = line.split_whitespace();
+            let key = fields
+                .next()
+                .context("missing counter name")?
+                .trim_end_matches(':')
+                .to_owned();
+            let value = fields.next().context("missing counter value")?.parse()?;
+            ensure!(fields.next().is_none(), "unexpected counter fields");
+            Ok((key, value))
+        })
+        .collect()
+}
+fn gone(error: &anyhow::Error) -> bool {
+    error.downcast_ref::<std::io::Error>().is_some_and(|error| {
+        error.kind() == std::io::ErrorKind::NotFound || error.raw_os_error() == Some(libc::ESRCH)
+    })
+}
+fn process(identity: &Identity, page_bytes: u64) -> Result<Option<Process>> {
+    let read =
+        fs::read_to_string(format!("/proc/{}/stat", identity.pid)).map_err(anyhow::Error::from);
+    let body = match read {
+        Ok(body) => body,
+        Err(error) if gone(&error) => return Ok(None),
+        Err(error) => return Err(error),
+    };
+    let process = stat(&body, page_bytes)?;
+    if process.start_ticks != identity.start_ticks || matches!(process.state, 'Z' | 'X') {
+        return Ok(None);
+    }
+    Ok(Some(process))
+}
+fn finish_sample<T>(sample: Result<T>, identity: &Identity, page_bytes: u64) -> Result<Option<T>> {
+    // A dying task can lose access to /proc/<pid>/io before stat reports it as
+    // a zombie. Wait briefly for positive evidence that this incarnation ended;
+    // permission denial alone must never be reported as successful completion.
+    let permission_denied = sample.as_ref().err().is_some_and(|error| {
+        error
+            .downcast_ref::<std::io::Error>()
+            .is_some_and(|error| error.kind() == std::io::ErrorKind::PermissionDenied)
+    });
+    let deadline = Instant::now() + Duration::from_millis(100);
+    loop {
+        if process(identity, page_bytes)?.is_none() {
+            return Ok(None);
+        }
+        if !permission_denied || Instant::now() >= deadline {
+            return sample.map(Some);
+        }
+        std::thread::sleep(Duration::from_millis(2));
+    }
+}
+fn allocated(directory: &Path) -> Result<u64> {
+    let mut total = 0u64;
+    let mut inodes = BTreeSet::new();
+    for entry in fs::read_dir(directory)? {
+        let entry = entry?;
+        let metadata = match fs::symlink_metadata(entry.path()) {
+            Ok(metadata) => metadata,
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue, // collection race
+            Err(error) => return Err(error.into()),
+        };
+        ensure!(
+            metadata.is_file(),
+            "disk sampler requires a flat directory of regular files"
+        );
+        if !inodes.insert((metadata.dev(), metadata.ino())) {
+            continue;
+        }
+        total = total
+            .checked_add(
+                metadata
+                    .blocks()
+                    .checked_mul(512)
+                    .context("disk block overflow")?,
+            )
+            .context("disk total overflow")?;
+    }
+    Ok(total)
+}
+fn main() -> Result<()> {
+    let args = Args::parse();
+    ensure!(
+        (100..=10_000).contains(&args.interval_ms),
+        "invalid sample interval"
+    );
+    let identity: Identity = serde_json::from_slice(&fs::read(args.operation.join("child.json"))?)?;
+    ensure!(identity.pid > 0, "invalid PID");
+    ensure!(
+        fs::read_to_string("/proc/sys/kernel/random/boot_id")?.trim() == identity.boot,
+        "boot identity changed"
+    );
+    ensure!(
+        fs::read_to_string("/proc/self/cgroup")?.trim() == "0::/",
+        "sampler requires a cgroup v2 namespace rooted at its container"
+    );
+    // A cgroup namespace without a remount can still expose an ancestor's
+    // filesystem. Confirm the mounted root contains this process directly.
+    let own_pid = std::process::id().to_string();
+    ensure!(
+        fs::read_to_string("/sys/fs/cgroup/cgroup.procs")
+            .context("read mounted cgroup membership")?
+            .lines()
+            .any(|line| line.trim() == own_pid),
+        "cgroup filesystem is not rooted at the sampler's cgroup"
+    );
+    let page = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
+    let ticks = unsafe { libc::sysconf(libc::_SC_CLK_TCK) };
+    ensure!(
+        page > 0 && ticks > 0,
+        "invalid kernel clock/page configuration"
+    );
+    let started = Instant::now();
+    let interval = Duration::from_millis(args.interval_ms);
+    let mut out = std::io::BufWriter::new(std::io::stdout().lock());
+    let mut samples = 0u64;
+    loop {
+        let begin = Instant::now();
+        let Some(initial) = process(&identity, page as u64)? else {
+            break;
+        };
+        let sample = (|| -> Result<serde_json::Value> {
+            // Do not accidentally attribute the sampler's container to a process
+            // in another cgroup. All experiment samplers run in their target Pod.
+            ensure!(
+                fs::read_to_string(format!("/proc/{}/cgroup", identity.pid))
+                    .context("read target cgroup")?
+                    == fs::read_to_string("/proc/self/cgroup").context("read sampler cgroup")?,
+                "target and sampler cgroups differ"
+            );
+            let io = counters(
+                &fs::read_to_string(format!("/proc/{}/io", identity.pid))
+                    .context("read target I/O counters")?,
+            )?;
+            let cpu = counters(
+                &fs::read_to_string("/sys/fs/cgroup/cpu.stat")
+                    .context("read cgroup CPU counters")?,
+            )?;
+            let memory_events = counters(
+                &fs::read_to_string("/sys/fs/cgroup/memory.events")
+                    .context("read cgroup memory events")?,
+            )?;
+            let memory = fs::read_to_string("/sys/fs/cgroup/memory.current")?
+                .trim()
+                .parse::<u64>()?;
+            let disk = args.disk_directory.as_deref().map(allocated).transpose()?;
+            Ok(
+                serde_json::json!({"kind":"resource_sample", "pid":identity.pid,"start_ticks":identity.start_ticks,
+                "timestamp_ms":SystemTime::now().duration_since(UNIX_EPOCH)?.as_millis(),
+                "seconds":started.elapsed().as_secs_f64(),"process":initial,"ticks_per_second":ticks,
+                "io":io,"cgroup_cpu":cpu,"cgroup_memory_bytes":memory,"cgroup_memory_events":memory_events,
+                "allocated_directory_bytes":disk,"sample_wall_seconds":begin.elapsed().as_secs_f64()}),
+            )
+        })();
+        // Ignore incomplete samples only when this exact process disappeared.
+        let Some(sample) = finish_sample(sample, &identity, page as u64)? else {
+            break;
+        };
+        serde_json::to_writer(&mut out, &sample)?;
+        out.write_all(b"\n")?;
+        out.flush()?;
+        samples += 1;
+        std::thread::sleep(interval.saturating_sub(begin.elapsed()));
+    }
+    serde_json::to_writer(
+        &mut out,
+        &serde_json::json!({"kind":"resource_sampling_complete","samples":samples,
+        "seconds":started.elapsed().as_secs_f64(),"reason":"recorded_process_gone"}),
+    )?;
+    out.write_all(b"\n")?;
+    out.flush()?;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::{
+        io::Read,
+        process::{Child, Command, Stdio},
+    };
+
+    struct OwnedChild(Child);
+    impl Drop for OwnedChild {
+        fn drop(&mut self) {
+            let _ = self.0.kill();
+            let _ = self.0.wait();
+        }
+    }
+
+    #[test]
+    #[ignore = "subprocess fixture for real procfs permission transitions"]
+    fn permission_fixture() -> Result<()> {
+        let ready = std::env::var_os("CONTEXT_RESOURCE_PERMISSION_FIXTURE")
+            .context("fixture path missing")?;
+        ensure!(unsafe { libc::prctl(libc::PR_SET_DUMPABLE, 0) } == 0);
+        fs::write(ready, b"ready")?;
+        std::io::stdin().read_exact(&mut [0u8])?;
+        Ok(())
+    }
+
+    fn denied_target() -> Result<(OwnedChild, Identity, anyhow::Error)> {
+        let directory = tempfile::tempdir()?;
+        let ready = directory.path().join("ready");
+        let target = OwnedChild(
+            Command::new(std::env::current_exe()?)
+                .args(["--exact", "tests::permission_fixture", "--ignored"])
+                .env("CONTEXT_RESOURCE_PERMISSION_FIXTURE", &ready)
+                .stdin(Stdio::piped())
+                .stdout(Stdio::null())
+                .spawn()?,
+        );
+        let deadline = Instant::now() + Duration::from_secs(5);
+        while !ready.exists() {
+            ensure!(
+                Instant::now() < deadline,
+                "permission fixture did not start"
+            );
+            std::thread::sleep(Duration::from_millis(2));
+        }
+        let pid = target.0.id();
+        let p = stat(&fs::read_to_string(format!("/proc/{pid}/stat"))?, 4096)?;
+        let error = fs::read_to_string(format!("/proc/{pid}/io")).unwrap_err();
+        assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied);
+        Ok((
+            target,
+            Identity {
+                pid,
+                start_ticks: p.start_ticks,
+                boot: String::new(),
+            },
+            anyhow::Error::from(error).context("read target I/O counters"),
+        ))
+    }
+
+    #[test]
+    fn permission_denial_during_exit_requires_observed_incarnation_end() -> Result<()> {
+        let (mut target, identity, error) = denied_target()?;
+        let mut release = target.0.stdin.take().unwrap();
+        let exit = std::thread::spawn(move || {
+            std::thread::sleep(Duration::from_millis(10));
+            release.write_all(b"x").unwrap();
+        });
+        assert!(finish_sample::<()>(Err(error), &identity, 4096)?.is_none());
+        exit.join().unwrap();
+        ensure!(target.0.wait()?.success(), "fixture failed");
+        Ok(())
+    }
+
+    #[test]
+    fn permission_denial_for_a_live_process_is_an_error() -> Result<()> {
+        let (_target, identity, error) = denied_target()?;
+        let error = finish_sample::<()>(Err(error), &identity, 4096).unwrap_err();
+        assert_eq!(
+            error.downcast_ref::<std::io::Error>().unwrap().kind(),
+            std::io::ErrorKind::PermissionDenied
+        );
+        assert!(process(&identity, 4096)?.is_some());
+        Ok(())
+    }
+    #[test]
+    fn process_stat_handles_spaces_and_parentheses_and_rejects_truncation() -> Result<()> {
+        let text = format!(
+            "42 (worker ) x) S {} 11 12 0 0 0 0 0 0 99 1024 7",
+            ["0"; 10].join(" ")
+        );
+        let p = stat(&text, 4096)?;
+        assert_eq!(
+            (p.start_ticks, p.cpu_ticks, p.rss_bytes),
+            (99, 23, 7 * 4096)
+        );
+        assert!(stat("42 (truncated) S 1", 4096).is_err());
+        assert_eq!(
+            counters("read_bytes: 4\nwrite_bytes: 8\n")?["write_bytes"],
+            8
+        );
+        assert!(counters("bad counter fields").is_err());
+        Ok(())
+    }
+    #[test]
+    fn directory_accounting_does_not_follow_links_or_hide_nested_data() -> Result<()> {
+        let root = tempfile::tempdir()?;
+        let file = root.path().join("data");
+        fs::write(&file, vec![1; 8192])?;
+        assert_eq!(allocated(root.path())?, fs::metadata(&file)?.blocks() * 512);
+        fs::hard_link(&file, root.path().join("hard-link"))?;
+        assert_eq!(allocated(root.path())?, fs::metadata(&file)?.blocks() * 512);
+        std::os::unix::fs::symlink(&file, root.path().join("link"))?;
+        assert!(allocated(root.path()).is_err());
+        fs::remove_file(root.path().join("link"))?;
+        fs::create_dir(root.path().join("nested"))?;
+        assert!(allocated(root.path()).is_err());
+        Ok(())
+    }
+    #[test]
+    fn a_reused_pid_is_not_the_recorded_process() -> Result<()> {
+        let pid = std::process::id();
+        let own = stat(&fs::read_to_string(format!("/proc/{pid}/stat"))?, 4096)?;
+        let identity = Identity {
+            pid,
+            start_ticks: own.start_ticks + 1,
+            boot: String::new(),
+        };
+        assert!(process(&identity, 4096)?.is_none());
+        Ok(())
+    }
+}
diff --git a/crates/lance-context-wal/tests/resource.rs b/crates/lance-context-wal/tests/resource.rs
new file mode 100644
index 0000000..ce22768
--- /dev/null
+++ b/crates/lance-context-wal/tests/resource.rs
@@ -0,0 +1,58 @@
+#![cfg(all(target_os = "linux", feature = "runner"))]
+use anyhow::{ensure, Result};
+use std::{
+    fs,
+    io::{BufRead, BufReader},
+    process::{Child, Command, Stdio},
+};
+struct OwnedChild(Child);
+impl Drop for OwnedChild {
+    fn drop(&mut self) {
+        let _ = self.0.kill();
+        let _ = self.0.wait();
+    }
+}
+#[test]
+fn samples_the_original_process_and_finishes_when_it_exits() -> Result<()> {
+    let directory = tempfile::tempdir()?;
+    let mut target = OwnedChild(Command::new("sleep").arg("10").spawn()?);
+    let stat = fs::read_to_string(format!("/proc/{}/stat", target.0.id()))?;
+    let fields: Vec<_> = stat
+        .rsplit_once(") ")
+        .unwrap()
+        .1
+        .split_whitespace()
+        .collect();
+    fs::write(
+        directory.path().join("child.json"),
+        serde_json::to_vec(&serde_json::json!({
+            "pid":target.0.id(),"start_ticks":fields[19].parse::<u64>()?,
+            "boot":fs::read_to_string("/proc/sys/kernel/random/boot_id")?.trim(),
+        }))?,
+    )?;
+    let mut sampler = OwnedChild(
+        Command::new(env!("CARGO_BIN_EXE_context-wal-resource"))
+            .arg("--operation")
+            .arg(directory.path())
+            .args(["--interval-ms", "100"])
+            .stdout(Stdio::piped())
+            .spawn()?,
+    );
+    let mut out = BufReader::new(sampler.0.stdout.take().unwrap());
+    for _ in 0..2 {
+        let mut line = String::new();
+        ensure!(out.read_line(&mut line)? > 0, "missing sample");
+        let sample: serde_json::Value = serde_json::from_str(&line)?;
+        assert_eq!(sample["kind"], "resource_sample");
+        assert_eq!(sample["pid"], target.0.id());
+        assert!(sample["cgroup_memory_bytes"].as_u64().unwrap() > 0);
+    }
+    target.0.kill()?;
+    target.0.wait()?;
+    let lines: Vec<_> = out.lines().collect::<std::io::Result<_>>()?;
+    let last: serde_json::Value = serde_json::from_str(lines.last().unwrap())?;
+    assert_eq!(last["kind"], "resource_sampling_complete");
+    assert!(last["samples"].as_u64().unwrap() >= 2);
+    ensure!(sampler.0.wait()?.success(), "sampler failed");
+    Ok(())
+}
diff --git a/docs/design/wal-resource-observation.md b/docs/design/wal-resource-observation.md
new file mode 100644
index 0000000..12b6fa3
--- /dev/null
+++ b/docs/design/wal-resource-observation.md
@@ -0,0 +1,49 @@
+# Observe one Rust experiment process
+
+Tracking: https://github.com/lance-format/lance-context/issues/378
+
+The optional Linux resource sampler observes an existing operation from
+`context-wal-runner`. It never launches, restarts or stops the target:
+
+```sh
+cargo build --release -p lance-context-wal --features runner --bin context-wal-resource
+context-wal-resource --operation /tmp/example-operation --interval-ms 500
+```
+
+The operation must already have a recorded child identity. The sampler validates
+the boot ID, PID and process start time. It checks that identity before and after
+each sample and emits a completion record when the recorded incarnation ends.
+PID reuse cannot silently switch the measured process. Run the sampler in the
+same container as the target: it requires cgroup v2 with a namespace rooted at
+`0::/` and validates that target and observer belong to the same cgroup.
+It also verifies that the mounted cgroup root directly contains the sampler,
+so entering a cgroup namespace alone cannot attribute an ancestor's counters.
+
+JSON Lines output records process CPU ticks, RSS and procfs I/O; cgroup CPU usage,
+throttling, current memory and memory events; actual elapsed times and collection
+duration. An optional `--disk-directory /tmp/example-wal` counts allocated blocks
+in a flat directory, deduplicates hard links and rejects symlinks or subdirectories.
+Files removed concurrently by collection may disappear during the scan.
+
+The target can lose procfs I/O access during exit before its process state shows
+a zombie. On a permission-denied sample, the observer waits at most 100 ms for
+positive evidence that the exact incarnation ended. Only then may it discard the
+incomplete sample and finish. A permission error for a still-live target remains
+an error. Read failures include the location being observed.
+
+These are sampled observations, not continuous peaks, quotas or physical device
+write amplification. Directory scans and counter reads are not atomic. Process
+and cgroup counters have different scopes and must not be added: other processes
+and the sampler itself can contribute to the container counters. Collection starts
+after the target identity is available and can miss startup or short-lived peaks.
+Record the actual observation interval; its CPU total is not an ingestion-only
+cost unless that interval has independently been aligned to ingestion. Observer
+overhead is included and needs a separate comparison to quantify.
+
+Five unit tests cover stat parsing, PID reuse, directory accounting and real
+procfs permission transitions for exiting versus still-live targets. A separate
+process test verifies sampling and completion after target exit. CI explicitly
+enables these optional-feature tests; the integration test runs unprivileged in
+private mount/cgroup namespaces with a namespace-rooted cgroup2 mount. The
+ignored test fixture is invoked by the
+permission regressions. The tool does not change production ingestion semantics.