refactor: remove old pod crates

crates/worker/src/runtime/mod.rs

@@ -1,2 +1,2 @@
 pub mod dir;
-pub use ::pod_registry;
+pub mod worker_allocation;

crates/worker/src/runtime/worker_allocation.rs

@@ -0,0 +1,29 @@
+//! Process-local Worker allocation table used only for scope ownership checks.
+//!
+//! This module is intentionally not a runtime identity store. Runtime Worker
+//! identity, creation and durable persistence remain owned by worker-runtime
+//! fs-store plus its execution backend mapping; this table coordinates
+//! in-process scope delegation while a Worker is running.
+
+mod conflict;
+mod error;
+mod lifecycle;
+mod mutate;
+mod table;
+
+#[cfg(test)]
+mod test_util;
+
+pub use conflict::{
+    ConflictOwner, find_conflict_owner, find_conflict_owners, is_within_effective_write,
+};
+pub use error::ScopeLockError;
+pub use lifecycle::{
+    ScopeAllocationGuard, SegmentLockInfo, adopt_allocation, install_top_level,
+    install_top_level_with_deny, lookup_segment, update_segment,
+};
+pub use mutate::{
+    delegate_scope, reclaim_delegated_scope, reclaim_stale, reclaim_stale_with, register_worker,
+    register_worker_with_deny, release_worker,
+};
+pub use table::{Allocation, LockFile, LockFileGuard, default_allocation_path};

crates/worker/src/runtime/worker_allocation/conflict.rs

@@ -0,0 +1,299 @@
+//! Pure functions that decide whether scope rules collide.
+//!
+//! These helpers are read-only over [`LockFile`]; they never touch the
+//! file or the lock itself. The mutating operations in [`crate::mutate`]
+//! call them under the [`crate::LockFileGuard`].
+
+use manifest::{Permission, ScopeRule};
+
+use super::table::{Allocation, LockFile};
+
+/// Whether `a` and `b` claim any overlapping concrete path.
+///
+/// Recursive rules cover `target/**`; non-recursive rules cover the
+/// target itself and its direct children. The four cases enumerate
+/// when those coverage sets intersect.
+pub(crate) fn rules_overlap(a: &ScopeRule, b: &ScopeRule) -> bool {
+    match (a.recursive, b.recursive) {
+        (true, true) => a.target.starts_with(&b.target) || b.target.starts_with(&a.target),
+        (true, false) => {
+            // a covers a.target/**; b covers {b.target, b.target/*}.
+            b.target.starts_with(&a.target) || a.target.parent() == Some(b.target.as_path())
+        }
+        (false, true) => {
+            a.target.starts_with(&b.target) || b.target.parent() == Some(a.target.as_path())
+        }
+        (false, false) => {
+            a.target == b.target
+                || a.target.parent() == Some(b.target.as_path())
+                || b.target.parent() == Some(a.target.as_path())
+        }
+    }
+}
+
+/// Does `cover` fully contain `inner`'s claimed paths?
+pub(crate) fn covers_fully(cover: &ScopeRule, inner: &ScopeRule) -> bool {
+    if cover.permission < inner.permission {
+        return false;
+    }
+    if cover.recursive {
+        inner.target.starts_with(&cover.target)
+    } else {
+        inner.target == cover.target && !inner.recursive
+    }
+}
+
+/// Check whether `rule` is contained in `parent`'s effective write
+/// scope: its allow set covers `rule`, no deny rule caps it, and no
+/// child of `parent` has already taken a piece that would overlap
+/// `rule`.
+pub fn is_within_effective_write(lock: &LockFile, parent: &str, rule: &ScopeRule) -> bool {
+    let Some(alloc) = lock.find(parent) else {
+        return false;
+    };
+    if rule.permission != Permission::Write {
+        return alloc.scope_allow.iter().any(|r| covers_fully(r, rule));
+    }
+    let covered = alloc
+        .scope_allow
+        .iter()
+        .filter(|r| r.permission == Permission::Write)
+        .any(|r| covers_fully(r, rule));
+    if !covered {
+        return false;
+    }
+    let denied = alloc
+        .scope_deny
+        .iter()
+        .filter(|r| r.permission == Permission::Write)
+        .any(|r| rules_overlap(r, rule));
+    if denied {
+        return false;
+    }
+    let child_conflict = lock
+        .allocations
+        .iter()
+        .filter(|a| a.delegated_from.as_deref() == Some(parent))
+        .flat_map(|a| a.scope_allow.iter())
+        .filter(|r| r.permission == Permission::Write)
+        .any(|r| rules_overlap(r, rule));
+    !child_conflict
+}
+
+/// The Worker and rule that actually own a conflicting write scope.
+#[derive(Debug, Clone)]
+pub struct ConflictOwner {
+    pub worker_name: String,
+    pub rule: ScopeRule,
+}
+
+/// Find the Worker/rule that actually owns a write scope overlapping `rule`.
+///
+/// Walks the delegation tree: if an allocation overlaps `rule`, we
+/// descend into its children and return the deepest overlapping node
+/// as the true owner. `exempt` names a Worker whose ownership is
+/// permitted (used during delegation: the spawner itself is allowed
+/// to still own the rule's region because it is handing it down).
+pub fn find_conflict_owner(
+    lock: &LockFile,
+    rule: &ScopeRule,
+    exempt: Option<&str>,
+) -> Option<ConflictOwner> {
+    find_conflict_owners(lock, rule, exempt).into_iter().next()
+}
+
+/// Find every top-level delegation tree owner that conflicts with `rule`.
+pub fn find_conflict_owners(
+    lock: &LockFile,
+    rule: &ScopeRule,
+    exempt: Option<&str>,
+) -> Vec<ConflictOwner> {
+    if rule.permission != Permission::Write {
+        return Vec::new();
+    }
+    lock.allocations
+        .iter()
+        .filter(|a| a.delegated_from.is_none())
+        .filter_map(|alloc| find_conflict_in_subtree(lock, alloc, rule))
+        .filter(|owner| Some(owner.worker_name.as_str()) != exempt)
+        .collect()
+}
+
+fn find_conflict_in_subtree(
+    lock: &LockFile,
+    alloc: &Allocation,
+    rule: &ScopeRule,
+) -> Option<ConflictOwner> {
+    let overlapping_rule = alloc
+        .scope_allow
+        .iter()
+        .filter(|r| r.permission == Permission::Write)
+        .find(|r| rules_overlap(r, rule))?;
+
+    let fully_denied_here = alloc
+        .scope_deny
+        .iter()
+        .filter(|r| r.permission == Permission::Write)
+        .any(|r| covers_fully(r, rule));
+    if fully_denied_here {
+        return None;
+    }
+
+    for child in lock
+        .allocations
+        .iter()
+        .filter(|a| a.delegated_from.as_deref() == Some(alloc.worker_name.as_str()))
+    {
+        if let Some(owner) = find_conflict_in_subtree(lock, child, rule) {
+            return Some(owner);
+        }
+    }
+    Some(ConflictOwner {
+        worker_name: alloc.worker_name.clone(),
+        rule: overlapping_rule.clone(),
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::test_util::*;
+    use super::super::{
+        ScopeLockError, delegate_scope, register_worker, register_worker_with_deny,
+    };
+    use super::*;
+    use tempfile::TempDir;
+
+    #[test]
+    fn rules_overlap_prefix_relation() {
+        assert!(rules_overlap(
+            &write_rule("/src", true),
+            &write_rule("/src/core", true)
+        ));
+        assert!(rules_overlap(
+            &write_rule("/src/core", true),
+            &write_rule("/src", true),
+        ));
+        assert!(!rules_overlap(
+            &write_rule("/src", true),
+            &write_rule("/docs", true),
+        ));
+    }
+
+    #[test]
+    fn rules_overlap_non_recursive() {
+        assert!(!rules_overlap(
+            &write_rule("/src", false),
+            &write_rule("/src/a/b", true),
+        ));
+        assert!(rules_overlap(
+            &write_rule("/src", false),
+            &write_rule("/src/child", false),
+        ));
+    }
+
+    #[test]
+    fn conflict_detection_descends_to_real_owner() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        register_worker(
+            &mut g,
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/src", true)],
+            sid(),
+        )
+        .unwrap();
+        delegate_scope(
+            &mut g,
+            "a",
+            "b".into(),
+            std::process::id(),
+            sock("b"),
+            vec![write_rule("/src/core", true)],
+            &delegation_scope(vec![write_rule("/src", true)]),
+        )
+        .unwrap();
+        // A different top-level Worker trying to register /src/core/x
+        // should be blamed on B (deepest owner), not A.
+        let err = register_worker(
+            &mut g,
+            "x".into(),
+            std::process::id(),
+            sock("x"),
+            vec![write_rule("/src/core/x", true)],
+            sid(),
+        )
+        .unwrap_err();
+        match err {
+            ScopeLockError::WriteConflict { competitor, .. } => assert_eq!(competitor, "b"),
+            other => panic!("expected WriteConflict, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn denied_write_region_is_not_claimed_by_restored_parent() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        register_worker_with_deny(
+            &mut g,
+            "parent".into(),
+            std::process::id(),
+            sock("parent"),
+            vec![write_rule("/src", true)],
+            vec![write_rule("/src/core", true)],
+            sid(),
+        )
+        .unwrap();
+        register_worker(
+            &mut g,
+            "child".into(),
+            std::process::id(),
+            sock("child"),
+            vec![write_rule("/src/core", true)],
+            sid(),
+        )
+        .unwrap();
+    }
+
+    #[test]
+    fn partial_deny_does_not_hide_parent_conflict() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        register_worker_with_deny(
+            &mut g,
+            "parent".into(),
+            std::process::id(),
+            sock("parent"),
+            vec![write_rule("/src", true)],
+            vec![write_rule("/src/core", true)],
+            sid(),
+        )
+        .unwrap();
+
+        let err = register_worker(
+            &mut g,
+            "other".into(),
+            std::process::id(),
+            sock("other"),
+            vec![write_rule("/src", true)],
+            sid(),
+        )
+        .unwrap_err();
+
+        match err {
+            ScopeLockError::WriteConflict {
+                competitor,
+                competitor_rule,
+                ..
+            } => {
+                assert_eq!(competitor, "parent");
+                assert_eq!(competitor_rule.target, std::path::PathBuf::from("/src"));
+            }
+            other => panic!("expected WriteConflict, got {other:?}"),
+        }
+    }
+}

crates/worker/src/runtime/worker_allocation/error.rs

@@ -0,0 +1,40 @@
+//! Error type for mutating pod-worker allocation operations.
+
+use std::io;
+use std::path::PathBuf;
+
+use manifest::{ScopeError, ScopeRule};
+use session_store::SegmentId;
+
+/// Errors raised by the mutating pod-worker allocation operations.
+#[derive(Debug, thiserror::Error)]
+pub enum ScopeLockError {
+    #[error("I/O error on workers.json: {0}")]
+    Io(#[from] io::Error),
+    #[error("pod name `{0}` is already registered")]
+    DuplicateWorkerName(String),
+    #[error("requested scope `{}` conflicts with pod `{competitor}` rule `{}`", .rule.target.display(), .competitor_rule.target.display())]
+    WriteConflict {
+        competitor: String,
+        rule: ScopeRule,
+        competitor_rule: ScopeRule,
+    },
+    #[error(
+        "requested scope `{}` is not within spawner `{spawner}`'s delegation scope",
+        .rule.target.display()
+    )]
+    NotSubset { spawner: String, rule: ScopeRule },
+    #[error("invalid delegation scope: {source}")]
+    InvalidScope { source: ScopeError },
+    #[error("pod `{0}` is not registered")]
+    UnknownWorker(String),
+    #[error(
+        "session {segment_id} is already held by pod `{worker_name}` at {}",
+        .socket.display()
+    )]
+    SegmentConflict {
+        segment_id: SegmentId,
+        worker_name: String,
+        socket: PathBuf,
+    },
+}

crates/worker/src/runtime/worker_allocation/lifecycle.rs

@@ -0,0 +1,341 @@
+//! Owned-allocation guards and the high-level entry points that open
+//! the default worker allocation path, mutate it, and return a guard that cleans
+//! up on drop.
+
+use std::path::{Path, PathBuf};
+
+use manifest::ScopeRule;
+use session_store::SegmentId;
+
+use super::error::ScopeLockError;
+use super::mutate::release_worker;
+use super::table::{LockFileGuard, default_allocation_path};
+
+/// Owned allocation: on drop, opens the lock file and releases this
+/// Worker's entry. The guard keeps only the name + lock-file path; it
+/// does not hold the `flock` for the Worker's lifetime.
+#[derive(Debug)]
+pub struct ScopeAllocationGuard {
+    worker_name: String,
+    lock_path: PathBuf,
+}
+
+impl ScopeAllocationGuard {
+    pub fn worker_name(&self) -> &str {
+        &self.worker_name
+    }
+
+    pub fn lock_path(&self) -> &Path {
+        &self.lock_path
+    }
+}
+
+impl Drop for ScopeAllocationGuard {
+    fn drop(&mut self) {
+        if let Ok(mut guard) = LockFileGuard::open(&self.lock_path) {
+            let _ = release_worker(&mut guard, &self.worker_name);
+        }
+    }
+}
+
+/// Open the default lock file, register a top-level Worker, and return a
+/// guard that will release the allocation on drop.
+pub fn install_top_level(
+    worker_name: String,
+    pid: u32,
+    socket: PathBuf,
+    scope_allow: Vec<ScopeRule>,
+    segment_id: SegmentId,
+) -> Result<ScopeAllocationGuard, ScopeLockError> {
+    install_top_level_with_deny(
+        worker_name,
+        pid,
+        socket,
+        scope_allow,
+        Vec::new(),
+        segment_id,
+    )
+}
+
+/// Open the default lock file, register a top-level Worker with explicit
+/// deny rules, and return a guard that will release the allocation on
+/// drop.
+pub fn install_top_level_with_deny(
+    worker_name: String,
+    pid: u32,
+    socket: PathBuf,
+    scope_allow: Vec<ScopeRule>,
+    scope_deny: Vec<ScopeRule>,
+    segment_id: SegmentId,
+) -> Result<ScopeAllocationGuard, ScopeLockError> {
+    let lock_path = default_allocation_path()?;
+    let mut guard = LockFileGuard::open(&lock_path)?;
+    super::mutate::register_worker_with_deny(
+        &mut guard,
+        worker_name.clone(),
+        pid,
+        socket,
+        scope_allow,
+        scope_deny,
+        segment_id,
+    )?;
+    Ok(ScopeAllocationGuard {
+        worker_name,
+        lock_path,
+    })
+}
+
+/// Take ownership of an existing allocation that was pre-registered by
+/// a spawning Worker.
+///
+/// The spawning flow is two-stage: the spawner calls
+/// [`crate::delegate_scope`] (with its own pid as a live placeholder,
+/// `segment_id = None`), then exec's the child; the child, once
+/// running, calls this function to rewrite the allocation's pid +
+/// segment_id to its own and claim the [`ScopeAllocationGuard`] so
+/// the entry is released when the child exits.
+pub fn adopt_allocation(
+    worker_name: String,
+    new_pid: u32,
+    segment_id: SegmentId,
+) -> Result<ScopeAllocationGuard, ScopeLockError> {
+    let lock_path = default_allocation_path()?;
+    let mut guard = LockFileGuard::open(&lock_path)?;
+    let alloc = guard
+        .data_mut()
+        .find_mut(&worker_name)
+        .ok_or_else(|| ScopeLockError::UnknownWorker(worker_name.clone()))?;
+    alloc.pid = new_pid;
+    alloc.segment_id = Some(segment_id);
+    guard.save()?;
+    Ok(ScopeAllocationGuard {
+        worker_name,
+        lock_path,
+    })
+}
+
+/// Rewrite the `segment_id` recorded for `worker_name` to
+/// `new_segment_id`.
+///
+/// The Worker's in-memory `segment_id` can change underneath the
+/// allocation in two normal places:
+///
+/// - `Worker::compact` mints a fresh session and swaps it in.
+/// - `session_store::ensure_head_or_fork` auto-forks when another
+///   writer has advanced the store head behind our back.
+///
+/// Both paths must call this so subsequent [`lookup_segment`] queries
+/// find the live session id, not the old one. Without this update a
+/// concurrent `restore_from_manifest(new_id)` would see "no live
+/// writer" and proceed to register a competing allocation on the
+/// session this Worker just moved into.
+///
+/// The lock is opened once and the allocation is rewritten inside the
+/// guard, so the segment_id collision check is atomic with the
+/// rewrite.
+pub fn update_segment(worker_name: &str, new_segment_id: SegmentId) -> Result<(), ScopeLockError> {
+    let lock_path = default_allocation_path()?;
+    let mut guard = LockFileGuard::open(&lock_path)?;
+    if let Some(other) = guard.data().find_by_segment(new_segment_id) {
+        if other.worker_name != worker_name {
+            return Err(ScopeLockError::SegmentConflict {
+                segment_id: new_segment_id,
+                worker_name: other.worker_name.clone(),
+                socket: other.socket.clone(),
+            });
+        }
+    }
+    let alloc = guard
+        .data_mut()
+        .find_mut(worker_name)
+        .ok_or_else(|| ScopeLockError::UnknownWorker(worker_name.into()))?;
+    alloc.segment_id = Some(new_segment_id);
+    guard.save()?;
+    Ok(())
+}
+
+/// Information about a Worker that currently holds an allocation for a
+/// given session.
+#[derive(Debug, Clone)]
+pub struct SegmentLockInfo {
+    pub worker_name: String,
+    pub socket: PathBuf,
+    pub pid: u32,
+}
+
+/// Open the default lock file, reclaim stale entries, and return the
+/// allocation currently writing to `segment_id`, if any.
+///
+/// Used by `Worker::restore_from_manifest` to refuse a resume that would
+/// race a live writer on the same source session.
+pub fn lookup_segment(segment_id: SegmentId) -> Result<Option<SegmentLockInfo>, ScopeLockError> {
+    let lock_path = default_allocation_path()?;
+    let mut guard = LockFileGuard::open(&lock_path)?;
+    super::mutate::reclaim_stale(&mut guard);
+    Ok(guard
+        .data()
+        .find_by_segment(segment_id)
+        .map(|a| SegmentLockInfo {
+            worker_name: a.worker_name.clone(),
+            socket: a.socket.clone(),
+            pid: a.pid,
+        }))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::table::Allocation;
+    use super::super::test_util::*;
+    use super::*;
+    use tempfile::TempDir;
+
+    /// Mimic what the spawner does before the child comes up: push an
+    /// allocation for the child carrying the spawner's (live) pid as a
+    /// placeholder. Exists only in tests.
+    fn delegate_placeholder(g: &mut LockFileGuard, worker_name: &str, placeholder_pid: u32) {
+        g.data_mut().allocations.push(Allocation {
+            worker_name: worker_name.to_string(),
+            pid: placeholder_pid,
+            socket: sock(worker_name),
+            scope_allow: vec![write_rule("/tmp/child", true)],
+            scope_deny: Vec::new(),
+            delegated_from: None,
+            segment_id: None,
+        });
+        g.save().unwrap();
+    }
+
+    #[test]
+    fn scope_allocation_guard_releases_on_drop() {
+        let dir = TempDir::new().unwrap();
+        let _sandbox = RuntimeDirSandbox::new(dir.path());
+        let lock_path = dir.path().join("workers.json");
+        let guard = install_top_level(
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/src", true)],
+            sid(),
+        )
+        .unwrap();
+        {
+            let g = LockFileGuard::open(&lock_path).unwrap();
+            assert!(g.data().find("a").is_some());
+        }
+        drop(guard);
+        {
+            let g = LockFileGuard::open(&lock_path).unwrap();
+            assert!(g.data().find("a").is_none());
+        }
+    }
+
+    #[test]
+    fn adopt_allocation_rewrites_pid_and_releases_on_drop() {
+        let dir = TempDir::new().unwrap();
+        let _sandbox = RuntimeDirSandbox::new(dir.path());
+        let lock_path = dir.path().join("workers.json");
+        // Pre-register an allocation under spawner's pid, as delegate_scope would.
+        {
+            let mut g = LockFileGuard::open(&lock_path).unwrap();
+            delegate_placeholder(&mut g, "child", std::process::id());
+        }
+        let child_pid = std::process::id().wrapping_add(1);
+        let guard = adopt_allocation("child".into(), child_pid, sid()).unwrap();
+        {
+            let g = LockFileGuard::open(&lock_path).unwrap();
+            let alloc = g.data().find("child").unwrap();
+            assert_eq!(alloc.pid, child_pid);
+        }
+        drop(guard);
+        {
+            let g = LockFileGuard::open(&lock_path).unwrap();
+            assert!(g.data().find("child").is_none());
+        }
+    }
+
+    #[test]
+    fn adopt_allocation_errors_on_unknown_pod() {
+        let dir = TempDir::new().unwrap();
+        let _sandbox = RuntimeDirSandbox::new(dir.path());
+        let err = adopt_allocation("ghost".into(), 42, sid()).unwrap_err();
+        assert!(matches!(err, ScopeLockError::UnknownWorker(ref n) if n == "ghost"));
+    }
+
+    #[test]
+    fn lookup_session_returns_live_writer_info() {
+        let dir = TempDir::new().unwrap();
+        let _sandbox = RuntimeDirSandbox::new(dir.path());
+        let s = sid();
+        let guard = install_top_level(
+            "live".into(),
+            std::process::id(),
+            sock("live"),
+            vec![write_rule("/work", true)],
+            s,
+        )
+        .unwrap();
+        let info = lookup_segment(s).unwrap().expect("expected live writer");
+        assert_eq!(info.worker_name, "live");
+        assert_eq!(info.socket, sock("live"));
+        drop(guard);
+        // After the guard's release, the lookup goes back to None.
+        assert!(lookup_segment(s).unwrap().is_none());
+    }
+
+    #[test]
+    fn update_session_rewrites_allocation_session_id() {
+        let dir = TempDir::new().unwrap();
+        let _sandbox = RuntimeDirSandbox::new(dir.path());
+        let original = sid();
+        let updated = sid();
+        let _guard = install_top_level(
+            "p".into(),
+            std::process::id(),
+            sock("p"),
+            vec![write_rule("/work", true)],
+            original,
+        )
+        .unwrap();
+        update_segment("p", updated).unwrap();
+        // lookup against the original is now empty, the updated id wins.
+        assert!(lookup_segment(original).unwrap().is_none());
+        assert_eq!(lookup_segment(updated).unwrap().unwrap().worker_name, "p");
+    }
+
+    #[test]
+    fn update_session_rejects_when_target_already_held() {
+        let dir = TempDir::new().unwrap();
+        let _sandbox = RuntimeDirSandbox::new(dir.path());
+        let s_a = sid();
+        let s_b = sid();
+        let _g_a = install_top_level(
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/work/a", true)],
+            s_a,
+        )
+        .unwrap();
+        let _g_b = install_top_level(
+            "b".into(),
+            std::process::id(),
+            sock("b"),
+            vec![write_rule("/work/b", true)],
+            s_b,
+        )
+        .unwrap();
+        // `a` cannot adopt b's live session id.
+        let err = update_segment("a", s_b).unwrap_err();
+        match err {
+            ScopeLockError::SegmentConflict {
+                worker_name,
+                segment_id,
+                ..
+            } => {
+                assert_eq!(worker_name, "b");
+                assert_eq!(segment_id, s_b);
+            }
+            other => panic!("expected SegmentConflict, got {other:?}"),
+        }
+    }
+}

crates/worker/src/runtime/worker_allocation/mutate.rs

@@ -0,0 +1,773 @@
+//! Mutating operations over the allocation table. All of these expect
+//! the caller to hold a [`LockFileGuard`] for the worker allocation's lock file.
+
+use std::io;
+use std::path::PathBuf;
+
+use manifest::{DelegationScope, Permission, ScopeRule};
+use session_store::SegmentId;
+
+use super::conflict::{find_conflict_owner, find_conflict_owners};
+use super::error::ScopeLockError;
+use super::table::{Allocation, LockFileGuard};
+
+/// Register a top-level Worker (started directly by a human, no
+/// delegation parent). Reclaims stale entries before checking
+/// conflicts so a crashed Worker's allocation doesn't block the new one.
+///
+/// Rejects when another live allocation is already writing to
+/// `segment_id`, so two `restore_from_manifest` calls under different
+/// `worker_name`s cannot both grab the same session log.
+pub fn register_worker(
+    guard: &mut LockFileGuard,
+    worker_name: String,
+    pid: u32,
+    socket: PathBuf,
+    scope_allow: Vec<ScopeRule>,
+    segment_id: SegmentId,
+) -> Result<(), ScopeLockError> {
+    register_worker_with_deny(
+        guard,
+        worker_name,
+        pid,
+        socket,
+        scope_allow,
+        Vec::new(),
+        segment_id,
+    )
+}
+
+/// Register a top-level Worker with explicit deny rules that reduce the
+/// claimed effective write scope.
+///
+/// Conflict semantics: if every Worker overlapping a requested allow rule
+/// is fully covered by one of `scope_deny`, the conflict is suppressed
+/// and the registration proceeds. The check is structural (deny ⊇
+/// competitor.rule), not relational — it does not verify that the
+/// competitor actually descends from this Worker's prior delegations.
+/// In practice this is safe because the canonical restore caller derives
+/// `scope_deny` from outstanding child worker metadata delegations, so any
+/// covered competitor is expected to be a descendant of the original
+/// allocation. Direct callers must uphold the same invariant.
+pub fn register_worker_with_deny(
+    guard: &mut LockFileGuard,
+    worker_name: String,
+    pid: u32,
+    socket: PathBuf,
+    scope_allow: Vec<ScopeRule>,
+    scope_deny: Vec<ScopeRule>,
+    segment_id: SegmentId,
+) -> Result<(), ScopeLockError> {
+    reclaim_stale(guard);
+    if guard.data().find(&worker_name).is_some() {
+        return Err(ScopeLockError::DuplicateWorkerName(worker_name));
+    }
+    if let Some(existing) = guard.data().find_by_segment(segment_id) {
+        return Err(ScopeLockError::SegmentConflict {
+            segment_id,
+            worker_name: existing.worker_name.clone(),
+            socket: existing.socket.clone(),
+        });
+    }
+    for rule in scope_allow
+        .iter()
+        .filter(|r| r.permission == Permission::Write)
+    {
+        let conflicts = find_conflict_owners(guard.data(), rule, None);
+        let all_denied = !conflicts.is_empty()
+            && conflicts.iter().all(|owner| {
+                scope_deny
+                    .iter()
+                    .filter(|r| r.permission == Permission::Write)
+                    .any(|deny| super::conflict::covers_fully(deny, &owner.rule))
+            });
+        if all_denied {
+            continue;
+        }
+        if let Some(competitor) = conflicts.into_iter().next() {
+            return Err(ScopeLockError::WriteConflict {
+                competitor: competitor.worker_name,
+                rule: rule.clone(),
+                competitor_rule: competitor.rule,
+            });
+        }
+    }
+    guard.data_mut().allocations.push(Allocation {
+        worker_name,
+        pid,
+        socket,
+        scope_allow,
+        scope_deny,
+        delegated_from: None,
+        segment_id: Some(segment_id),
+    });
+    guard.save()?;
+    Ok(())
+}
+
+/// Register a spawned Worker whose scope is delegated from `spawner`.
+/// The requested scope must be within the spawner's delegation authority;
+/// overlap with any Worker other than `spawner` is a conflict.
+pub fn delegate_scope(
+    guard: &mut LockFileGuard,
+    spawner: &str,
+    spawned: String,
+    pid: u32,
+    socket: PathBuf,
+    scope_allow: Vec<ScopeRule>,
+    delegation_scope: &DelegationScope,
+) -> Result<(), ScopeLockError> {
+    reclaim_stale(guard);
+    if guard.data().find(&spawned).is_some() {
+        return Err(ScopeLockError::DuplicateWorkerName(spawned));
+    }
+    if guard.data().find(spawner).is_none() {
+        return Err(ScopeLockError::UnknownWorker(spawner.into()));
+    }
+    for rule in &scope_allow {
+        let allowed = delegation_scope
+            .allows_rule(rule)
+            .map_err(|source| ScopeLockError::InvalidScope { source })?;
+        if !allowed {
+            return Err(ScopeLockError::NotSubset {
+                spawner: spawner.into(),
+                rule: rule.clone(),
+            });
+        }
+        if rule.permission == Permission::Write {
+            if let Some(competitor) = find_conflict_owner(guard.data(), rule, Some(spawner)) {
+                return Err(ScopeLockError::WriteConflict {
+                    competitor: competitor.worker_name,
+                    rule: rule.clone(),
+                    competitor_rule: competitor.rule,
+                });
+            }
+        }
+    }
+    guard.data_mut().allocations.push(Allocation {
+        worker_name: spawned,
+        pid,
+        socket,
+        scope_allow,
+        scope_deny: Vec::new(),
+        delegated_from: Some(spawner.into()),
+        // Pre-reservation. The child fills in its own segment_id when
+        // it calls `adopt_allocation` after the worker is built.
+        segment_id: None,
+    });
+    guard.save()?;
+    Ok(())
+}
+
+/// Remove a Worker's allocation. Surviving children are reparented to
+/// the removed Worker's own `delegated_from`, so the delegation tree
+/// stays connected.
+pub fn release_worker(guard: &mut LockFileGuard, worker_name: &str) -> Result<(), ScopeLockError> {
+    let idx = guard
+        .data()
+        .allocations
+        .iter()
+        .position(|a| a.worker_name == worker_name);
+    let Some(idx) = idx else {
+        return Err(ScopeLockError::UnknownWorker(worker_name.into()));
+    };
+    let removed = guard.data().allocations[idx].clone();
+    for alloc in guard.data_mut().allocations.iter_mut() {
+        if alloc.delegated_from.as_deref() == Some(worker_name) {
+            alloc.delegated_from.clone_from(&removed.delegated_from);
+        }
+    }
+    guard.data_mut().allocations.remove(idx);
+    guard.save()?;
+    Ok(())
+}
+
+/// Reclaim a child delegation back into its parent allocation.
+///
+/// This is idempotent for missing deny entries. For each delegated Write rule,
+/// at most one exact matching deny rule is removed from the parent's `scope_deny`
+/// even when the child allocation is already absent; restore reconciliation uses
+/// that case when durable Worker-state still records an outstanding delegation but
+/// the live lock file no longer has a child allocation.
+pub fn reclaim_delegated_scope(
+    guard: &mut LockFileGuard,
+    parent: &str,
+    child: &str,
+    delegated_scope: &[ScopeRule],
+) -> Result<(), ScopeLockError> {
+    let child_idx = guard
+        .data()
+        .allocations
+        .iter()
+        .position(|a| a.worker_name == child);
+    let removed_child_parent = child_idx
+        .map(|idx| guard.data().allocations[idx].delegated_from.clone())
+        .unwrap_or(None);
+
+    if let Some(parent_alloc) = guard.data_mut().find_mut(parent) {
+        for rule in delegated_scope
+            .iter()
+            .filter(|rule| rule.permission == Permission::Write)
+        {
+            if let Some(idx) = parent_alloc.scope_deny.iter().position(|deny| deny == rule) {
+                parent_alloc.scope_deny.remove(idx);
+            }
+        }
+    }
+
+    if let Some(idx) = child_idx {
+        for alloc in guard.data_mut().allocations.iter_mut() {
+            if alloc.delegated_from.as_deref() == Some(child) {
+                alloc.delegated_from.clone_from(&removed_child_parent);
+            }
+        }
+        guard.data_mut().allocations.remove(idx);
+    }
+
+    guard.save()?;
+    Ok(())
+}
+
+/// Remove allocations whose PID is dead, reparenting children to the
+/// dead Worker's `delegated_from`. Idempotent and best-effort — I/O
+/// errors on save are swallowed so a crashed Worker's entry never blocks
+/// forward progress.
+pub fn reclaim_stale(guard: &mut LockFileGuard) {
+    reclaim_stale_with(guard, pid_alive);
+}
+
+/// Test seam: stale reclaim with a caller-supplied liveness probe.
+pub fn reclaim_stale_with(guard: &mut LockFileGuard, mut is_alive: impl FnMut(u32) -> bool) {
+    let dead: Vec<String> = guard
+        .data()
+        .allocations
+        .iter()
+        .filter(|a| !is_alive(a.pid))
+        .map(|a| a.worker_name.clone())
+        .collect();
+    if dead.is_empty() {
+        return;
+    }
+    for name in &dead {
+        let Some(idx) = guard
+            .data()
+            .allocations
+            .iter()
+            .position(|a| a.worker_name == *name)
+        else {
+            continue;
+        };
+        let removed = guard.data().allocations[idx].clone();
+        for alloc in guard.data_mut().allocations.iter_mut() {
+            if alloc.delegated_from.as_deref() == Some(name.as_str()) {
+                alloc.delegated_from.clone_from(&removed.delegated_from);
+            }
+        }
+        guard.data_mut().allocations.remove(idx);
+    }
+    let _ = guard.save();
+}
+
+/// `kill(pid, 0)` — returns true if the process exists (even when we
+/// don't own it), false only on ESRCH.
+fn pid_alive(pid: u32) -> bool {
+    if pid == 0 {
+        return false;
+    }
+    let ret = unsafe { libc::kill(pid as libc::pid_t, 0) };
+    if ret == 0 {
+        return true;
+    }
+    io::Error::last_os_error()
+        .raw_os_error()
+        .map(|e| e != libc::ESRCH)
+        .unwrap_or(false)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::is_within_effective_write;
+    use super::super::test_util::*;
+    use super::*;
+    use tempfile::TempDir;
+
+    #[test]
+    fn register_detects_write_conflict() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        register_worker(
+            &mut g,
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/src", true)],
+            sid(),
+        )
+        .unwrap();
+        let err = register_worker(
+            &mut g,
+            "b".into(),
+            std::process::id(),
+            sock("b"),
+            vec![write_rule("/src/core", true)],
+            sid(),
+        )
+        .unwrap_err();
+        match err {
+            ScopeLockError::WriteConflict { competitor, .. } => assert_eq!(competitor, "a"),
+            other => panic!("expected WriteConflict, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn duplicate_worker_name_rejected() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        register_worker(
+            &mut g,
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/src", true)],
+            sid(),
+        )
+        .unwrap();
+        let err = register_worker(
+            &mut g,
+            "a".into(),
+            std::process::id(),
+            sock("a2"),
+            vec![write_rule("/docs", true)],
+            sid(),
+        )
+        .unwrap_err();
+        assert!(matches!(err, ScopeLockError::DuplicateWorkerName(ref n) if n == "a"));
+    }
+
+    #[test]
+    fn delegate_must_be_subset() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        register_worker(
+            &mut g,
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/src", true)],
+            sid(),
+        )
+        .unwrap();
+        let err = delegate_scope(
+            &mut g,
+            "a",
+            "b".into(),
+            std::process::id(),
+            sock("b"),
+            vec![write_rule("/docs", true)],
+            &delegation_scope(vec![write_rule("/src", true)]),
+        )
+        .unwrap_err();
+        assert!(matches!(err, ScopeLockError::NotSubset { .. }));
+    }
+
+    #[test]
+    fn delegate_uses_delegation_scope_not_direct_effective_write() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        register_worker(
+            &mut g,
+            "orchestrator".into(),
+            std::process::id(),
+            sock("orchestrator"),
+            vec![read_rule("/workspace", true)],
+            sid(),
+        )
+        .unwrap();
+
+        delegate_scope(
+            &mut g,
+            "orchestrator",
+            "coder".into(),
+            std::process::id(),
+            sock("coder"),
+            vec![write_rule("/workspace/.worktree/task", true)],
+            &delegation_scope(vec![write_rule("/workspace", true)]),
+        )
+        .unwrap();
+
+        let coder = g.data().find("coder").expect("coder allocation");
+        assert_eq!(coder.delegated_from.as_deref(), Some("orchestrator"));
+    }
+
+    #[test]
+    fn delegate_succeeds_within_parent_scope() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        register_worker(
+            &mut g,
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/src", true)],
+            sid(),
+        )
+        .unwrap();
+        delegate_scope(
+            &mut g,
+            "a",
+            "b".into(),
+            std::process::id(),
+            sock("b"),
+            vec![write_rule("/src/core", true)],
+            &delegation_scope(vec![write_rule("/src", true)]),
+        )
+        .unwrap();
+        assert_eq!(g.data().allocations.len(), 2);
+        // A's effective write no longer covers /src/core because B has it.
+        assert!(!is_within_effective_write(
+            g.data(),
+            "a",
+            &write_rule("/src/core", true)
+        ));
+        // A still covers its own uninvolved areas.
+        assert!(is_within_effective_write(
+            g.data(),
+            "a",
+            &write_rule("/src/other", true)
+        ));
+    }
+
+    #[test]
+    fn delegate_rejects_sibling_overlap() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        register_worker(
+            &mut g,
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/src", true)],
+            sid(),
+        )
+        .unwrap();
+        delegate_scope(
+            &mut g,
+            "a",
+            "b".into(),
+            std::process::id(),
+            sock("b"),
+            vec![write_rule("/src/core", true)],
+            &delegation_scope(vec![write_rule("/src", true)]),
+        )
+        .unwrap();
+        // Sibling C from A tries to take /src/core/sub — already under B's scope.
+        let err = delegate_scope(
+            &mut g,
+            "a",
+            "c".into(),
+            std::process::id(),
+            sock("c"),
+            vec![write_rule("/src/core/sub", true)],
+            &delegation_scope(vec![write_rule("/src", true)]),
+        )
+        .unwrap_err();
+        match err {
+            ScopeLockError::WriteConflict { competitor, .. } => assert_eq!(competitor, "b"),
+            other => panic!("expected WriteConflict, got {other:?}"),
+        }
+    }
+
+    #[test]
+    fn release_reparents_children() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        register_worker(
+            &mut g,
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/src", true)],
+            sid(),
+        )
+        .unwrap();
+        delegate_scope(
+            &mut g,
+            "a",
+            "b".into(),
+            std::process::id(),
+            sock("b"),
+            vec![write_rule("/src/core", true)],
+            &delegation_scope(vec![write_rule("/src", true)]),
+        )
+        .unwrap();
+        delegate_scope(
+            &mut g,
+            "b",
+            "d".into(),
+            std::process::id(),
+            sock("d"),
+            vec![write_rule("/src/core/x", true)],
+            &delegation_scope(vec![write_rule("/src/core", true)]),
+        )
+        .unwrap();
+        release_worker(&mut g, "b").unwrap();
+        // D should now list A as its delegated_from.
+        let d = g.data().find("d").unwrap();
+        assert_eq!(d.delegated_from.as_deref(), Some("a"));
+        assert!(g.data().find("b").is_none());
+    }
+
+    #[test]
+    fn reclaim_delegated_scope_removes_child_and_one_parent_deny_layer() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        let delegated_rule = write_rule("/src/core", true);
+        register_worker_with_deny(
+            &mut g,
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/src", true)],
+            vec![delegated_rule.clone(), delegated_rule.clone()],
+            sid(),
+        )
+        .unwrap();
+        register_worker(
+            &mut g,
+            "b".into(),
+            std::process::id(),
+            sock("b"),
+            vec![delegated_rule.clone()],
+            sid(),
+        )
+        .unwrap();
+
+        reclaim_delegated_scope(&mut g, "a", "b", std::slice::from_ref(&delegated_rule)).unwrap();
+        let a = g.data().find("a").unwrap();
+        assert_eq!(a.scope_deny, vec![delegated_rule.clone()]);
+        assert!(g.data().find("b").is_none());
+
+        reclaim_delegated_scope(&mut g, "a", "b", std::slice::from_ref(&delegated_rule)).unwrap();
+        let a = g.data().find("a").unwrap();
+        assert!(
+            a.scope_deny.is_empty(),
+            "a missing child allocation still reclaims one matching parent deny"
+        );
+    }
+
+    #[test]
+    fn reclaim_delegated_scope_removes_parent_deny_when_child_allocation_missing() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        let delegated_rule = write_rule("/src/core", true);
+        register_worker_with_deny(
+            &mut g,
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/src", true)],
+            vec![delegated_rule.clone()],
+            sid(),
+        )
+        .unwrap();
+
+        reclaim_delegated_scope(
+            &mut g,
+            "a",
+            "missing",
+            std::slice::from_ref(&delegated_rule),
+        )
+        .unwrap();
+
+        let a = g.data().find("a").unwrap();
+        assert!(a.scope_deny.is_empty());
+    }
+
+    #[test]
+    fn reclaim_stale_reparents_and_removes_dead_entries() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        register_worker(
+            &mut g,
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/src", true)],
+            sid(),
+        )
+        .unwrap();
+        delegate_scope(
+            &mut g,
+            "a",
+            "b".into(),
+            std::process::id(),
+            sock("b"),
+            vec![write_rule("/src/core", true)],
+            &delegation_scope(vec![write_rule("/src", true)]),
+        )
+        .unwrap();
+        delegate_scope(
+            &mut g,
+            "b",
+            "d".into(),
+            std::process::id(),
+            sock("d"),
+            vec![write_rule("/src/core/x", true)],
+            &delegation_scope(vec![write_rule("/src/core", true)]),
+        )
+        .unwrap();
+        // Simulate B crashing by rewriting its pid to one the probe
+        // will treat as dead.
+        let fake_dead_pid: u32 = 0xffff_fff0;
+        for alloc in g.data_mut().allocations.iter_mut() {
+            if alloc.worker_name == "b" {
+                alloc.pid = fake_dead_pid;
+            }
+        }
+        reclaim_stale_with(&mut g, |pid| pid != fake_dead_pid);
+        assert!(g.data().find("b").is_none());
+        let d = g.data().find("d").unwrap();
+        assert_eq!(d.delegated_from.as_deref(), Some("a"));
+    }
+
+    #[test]
+    fn read_rules_do_not_conflict_with_write() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        register_worker(
+            &mut g,
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/src", true)],
+            sid(),
+        )
+        .unwrap();
+        // B only reads under the same tree — allowed.
+        register_worker(
+            &mut g,
+            "b".into(),
+            std::process::id(),
+            sock("b"),
+            vec![read_rule("/src", true)],
+            sid(),
+        )
+        .unwrap();
+        assert_eq!(g.data().allocations.len(), 2);
+    }
+
+    #[test]
+    fn releasing_pod_reopens_scope_for_fresh_registration() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        register_worker(
+            &mut g,
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/src", true)],
+            sid(),
+        )
+        .unwrap();
+        release_worker(&mut g, "a").unwrap();
+        register_worker(
+            &mut g,
+            "b".into(),
+            std::process::id(),
+            sock("b"),
+            vec![write_rule("/src", true)],
+            sid(),
+        )
+        .unwrap();
+    }
+
+    #[test]
+    fn delegated_scope_returns_to_parent_on_release() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        register_worker(
+            &mut g,
+            "a".into(),
+            std::process::id(),
+            sock("a"),
+            vec![write_rule("/src", true)],
+            sid(),
+        )
+        .unwrap();
+        delegate_scope(
+            &mut g,
+            "a",
+            "b".into(),
+            std::process::id(),
+            sock("b"),
+            vec![write_rule("/src/core", true)],
+            &delegation_scope(vec![write_rule("/src", true)]),
+        )
+        .unwrap();
+        assert!(!is_within_effective_write(
+            g.data(),
+            "a",
+            &write_rule("/src/core", true)
+        ));
+        release_worker(&mut g, "b").unwrap();
+        // /src/core is back in A's effective write scope.
+        assert!(is_within_effective_write(
+            g.data(),
+            "a",
+            &write_rule("/src/core", true)
+        ));
+    }
+
+    #[test]
+    fn register_pod_rejects_session_id_collision() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        let shared_session = sid();
+        register_worker(
+            &mut g,
+            "first".into(),
+            std::process::id(),
+            sock("first"),
+            vec![write_rule("/work/a", true)],
+            shared_session,
+        )
+        .unwrap();
+        // Second registration tries to grab the same segment_id under
+        // a different worker_name. Without the SegmentConflict check both
+        // would succeed and race on the same jsonl.
+        let err = register_worker(
+            &mut g,
+            "second".into(),
+            std::process::id(),
+            sock("second"),
+            vec![write_rule("/work/b", true)],
+            shared_session,
+        )
+        .unwrap_err();
+        match err {
+            ScopeLockError::SegmentConflict {
+                segment_id,
+                worker_name,
+                ..
+            } => {
+                assert_eq!(segment_id, shared_session);
+                assert_eq!(worker_name, "first");
+            }
+            other => panic!("expected SegmentConflict, got {other:?}"),
+        }
+    }
+}

crates/worker/src/runtime/worker_allocation/table.rs

@@ -0,0 +1,301 @@
+//! On-disk allocation table and the `flock`-protected guard.
+
+use std::fs::{DirBuilder, File, OpenOptions};
+use std::io::{self, Read, Seek, SeekFrom, Write};
+use std::os::unix::fs::{DirBuilderExt, OpenOptionsExt};
+use std::path::{Path, PathBuf};
+use std::thread;
+use std::time::{Duration, Instant};
+
+use fs4::fs_std::FileExt;
+use manifest::{ScopeRule, paths};
+use serde::{Deserialize, Serialize};
+use session_store::SegmentId;
+
+const LOCK_WAIT_TIMEOUT: Duration = Duration::from_secs(10);
+const LOCK_WAIT_POLL_INTERVAL: Duration = Duration::from_millis(25);
+
+/// On-disk representation of the allocation table.
+#[derive(Debug, Clone, Default, Serialize, Deserialize)]
+pub struct LockFile {
+    #[serde(default)]
+    pub allocations: Vec<Allocation>,
+}
+
+/// One Worker's scope allocation.
+///
+/// `scope_allow` is the full set of allow rules the Worker was granted.
+/// Portions delegated out to child Workers are **not** subtracted in
+/// storage — the effective write scope is derived on the fly by
+/// removing rules owned by any Worker whose `delegated_from` points to
+/// this one. Keeping the raw allow set makes reparenting (stale
+/// reclaim) trivial.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Allocation {
+    /// Worker name — also the identity used throughout orchestration.
+    pub worker_name: String,
+    /// Owning process. Checked with `kill(pid, 0)` for stale detection.
+    pub pid: u32,
+    /// Worker's Unix socket path.
+    pub socket: PathBuf,
+    /// Allow rules granted to this Worker (write + read).
+    pub scope_allow: Vec<ScopeRule>,
+    /// Deny rules that cap this Worker's effective scope. Normally empty for
+    /// fresh allocations; restored Workers use this to avoid reclaiming
+    /// previously delegated write regions.
+    #[serde(default)]
+    pub scope_deny: Vec<ScopeRule>,
+    /// Name of the Worker that delegated scope to this one, or `None` for
+    /// a top-level Worker started directly by a human.
+    pub delegated_from: Option<String>,
+    /// Segment ID this Worker is currently writing to. `None` means this
+    /// is a pre-reservation made by a spawner via [`super::super::delegate_scope`]
+    /// before the child has come up; the child fills it in at
+    /// [`crate::adopt_allocation`] time.
+    #[serde(default)]
+    pub segment_id: Option<SegmentId>,
+}
+
+impl LockFile {
+    pub fn find(&self, worker_name: &str) -> Option<&Allocation> {
+        self.allocations
+            .iter()
+            .find(|a| a.worker_name == worker_name)
+    }
+
+    pub fn find_mut(&mut self, worker_name: &str) -> Option<&mut Allocation> {
+        self.allocations
+            .iter_mut()
+            .find(|a| a.worker_name == worker_name)
+    }
+
+    /// Find the allocation currently writing to `segment_id`. Skips
+    /// pre-reservations whose `segment_id` is still `None`.
+    pub fn find_by_segment(&self, segment_id: SegmentId) -> Option<&Allocation> {
+        self.allocations
+            .iter()
+            .find(|a| a.segment_id == Some(segment_id))
+    }
+}
+
+/// Default on-disk path: `<runtime_dir>/workers.json` resolved via
+/// [`manifest::paths::worker_allocation_path`]. Tests should point this
+/// elsewhere by setting `YOI_HOME` or `YOI_RUNTIME_DIR` to a
+/// tempdir.
+pub fn default_allocation_path() -> io::Result<PathBuf> {
+    paths::worker_allocation_path().ok_or_else(|| {
+        io::Error::new(
+            io::ErrorKind::NotFound,
+            "could not resolve workers.json path (no YOI_HOME / \
+             YOI_RUNTIME_DIR / XDG_RUNTIME_DIR / HOME)",
+        )
+    })
+}
+
+/// RAII guard over an exclusively-locked lock file.
+///
+/// The file is kept open for the lifetime of the guard; `flock(LOCK_EX)`
+/// is released automatically on drop. Mutations go through
+/// [`LockFileGuard::data_mut`] and are committed with
+/// [`LockFileGuard::save`] before dropping — callers who mutate but
+/// never call `save` leave the table unchanged, which is the right
+/// behaviour for error paths.
+pub struct LockFileGuard {
+    file: File,
+    data: LockFile,
+}
+
+impl LockFileGuard {
+    /// Open the lock file at `path` (creating it + parent dirs if
+    /// needed), acquire an exclusive `flock`, then parse the contents.
+    ///
+    /// An empty file is treated as an empty allocation table.
+    ///
+    /// File is created with mode `0600` and its parent directory with
+    /// mode `0700` so no other user on the machine can read the
+    /// allocation table. Existing files/directories are left alone.
+    pub fn open(path: &Path) -> io::Result<Self> {
+        if let Some(parent) = path.parent() {
+            DirBuilder::new()
+                .recursive(true)
+                .mode(0o700)
+                .create(parent)?;
+        }
+        let file = OpenOptions::new()
+            .read(true)
+            .write(true)
+            .create(true)
+            .truncate(false)
+            .mode(0o600)
+            .open(path)?;
+        let started = Instant::now();
+        loop {
+            match FileExt::try_lock_exclusive(&file) {
+                Ok(true) => break,
+                Ok(false) => {
+                    if started.elapsed() >= LOCK_WAIT_TIMEOUT {
+                        return Err(io::Error::new(
+                            io::ErrorKind::TimedOut,
+                            format!(
+                                "timed out waiting for worker allocation lock `{}`",
+                                path.display()
+                            ),
+                        ));
+                    }
+                    thread::sleep(LOCK_WAIT_POLL_INTERVAL);
+                }
+                Err(error) if error.kind() == io::ErrorKind::WouldBlock => {
+                    if started.elapsed() >= LOCK_WAIT_TIMEOUT {
+                        return Err(io::Error::new(
+                            io::ErrorKind::TimedOut,
+                            format!(
+                                "timed out waiting for worker allocation lock `{}`",
+                                path.display()
+                            ),
+                        ));
+                    }
+                    thread::sleep(LOCK_WAIT_POLL_INTERVAL);
+                }
+                Err(error) => return Err(error),
+            }
+        }
+        let mut this = Self {
+            file,
+            data: LockFile::default(),
+        };
+        this.reload()?;
+        Ok(this)
+    }
+
+    fn reload(&mut self) -> io::Result<()> {
+        self.file.seek(SeekFrom::Start(0))?;
+        let mut buf = String::new();
+        self.file.read_to_string(&mut buf)?;
+        self.data = if buf.trim().is_empty() {
+            LockFile::default()
+        } else {
+            serde_json::from_str(&buf).map_err(|e| {
+                io::Error::new(
+                    io::ErrorKind::InvalidData,
+                    format!("workers.json parse error: {e}"),
+                )
+            })?
+        };
+        Ok(())
+    }
+
+    pub fn data(&self) -> &LockFile {
+        &self.data
+    }
+
+    pub fn data_mut(&mut self) -> &mut LockFile {
+        &mut self.data
+    }
+
+    /// Serialise `self.data` back to the file (truncate + rewrite).
+    pub fn save(&mut self) -> io::Result<()> {
+        let json = serde_json::to_vec_pretty(&self.data).map_err(io::Error::other)?;
+        self.file.seek(SeekFrom::Start(0))?;
+        self.file.set_len(0)?;
+        self.file.write_all(&json)?;
+        self.file.sync_data()?;
+        Ok(())
+    }
+}
+
+impl Drop for LockFileGuard {
+    fn drop(&mut self) {
+        let _ = FileExt::unlock(&self.file);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::register_worker;
+    use super::super::test_util::*;
+    use super::*;
+    use tempfile::TempDir;
+
+    #[test]
+    fn open_creates_empty_lock_file() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let guard = LockFileGuard::open(&path).unwrap();
+        assert!(guard.data().allocations.is_empty());
+        assert!(path.exists());
+    }
+
+    #[test]
+    fn open_creates_file_with_owner_only_permissions() {
+        use std::os::unix::fs::PermissionsExt;
+        let dir = TempDir::new().unwrap();
+        let parent = dir.path().join("yoi");
+        let path = parent.join("workers.json");
+        let _guard = LockFileGuard::open(&path).unwrap();
+        let file_mode = std::fs::metadata(&path).unwrap().permissions().mode() & 0o777;
+        assert_eq!(file_mode, 0o600, "file mode = {file_mode:o}");
+        let dir_mode = std::fs::metadata(&parent).unwrap().permissions().mode() & 0o777;
+        assert_eq!(dir_mode, 0o700, "dir mode = {dir_mode:o}");
+    }
+
+    #[test]
+    fn save_and_reopen_roundtrip() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        {
+            let mut g = open_empty(&path);
+            register_worker(
+                &mut g,
+                "a".into(),
+                std::process::id(),
+                sock("a"),
+                vec![write_rule("/src", true)],
+                sid(),
+            )
+            .unwrap();
+        }
+        let guard = LockFileGuard::open(&path).unwrap();
+        assert_eq!(guard.data().allocations.len(), 1);
+        assert_eq!(guard.data().allocations[0].worker_name, "a");
+    }
+
+    #[test]
+    fn find_by_session_skips_none_placeholders() {
+        let dir = TempDir::new().unwrap();
+        let path = dir.path().join("workers.json");
+        let mut g = open_empty(&path);
+        // Pre-reservation: delegate_scope leaves segment_id = None
+        // until adopt_allocation rewrites it. find_by_segment must not
+        // match those placeholders, otherwise a freshly-spawning child
+        // would shadow itself before it has even chosen a session.
+        register_worker(
+            &mut g,
+            "parent".into(),
+            std::process::id(),
+            sock("parent"),
+            vec![write_rule("/p", true)],
+            sid(),
+        )
+        .unwrap();
+        super::super::delegate_scope(
+            &mut g,
+            "parent",
+            "child".into(),
+            std::process::id(),
+            sock("child"),
+            vec![write_rule("/p/sub", true)],
+            &delegation_scope(vec![write_rule("/p", true)]),
+        )
+        .unwrap();
+
+        let target_session = sid();
+        // The placeholder allocation has segment_id = None and must
+        // not be returned for any lookup.
+        assert!(g.data().find_by_segment(target_session).is_none());
+
+        // After adopt-style rewrite, the same allocation is now found.
+        g.data_mut().find_mut("child").unwrap().segment_id = Some(target_session);
+        let found = g.data().find_by_segment(target_session).unwrap();
+        assert_eq!(found.worker_name, "child");
+    }
+}

crates/worker/src/runtime/worker_allocation/test_util.rs

@@ -0,0 +1,105 @@
+//! Shared test helpers for the pod-worker allocation crate.
+//!
+//! Visible to all `#[cfg(test)]` modules under `super::test_util::*`.
+
+use std::path::{Path, PathBuf};
+use std::sync::{LazyLock, Mutex, MutexGuard};
+
+use manifest::{DelegationScope, Permission, ScopeConfig, ScopeRule};
+use session_store::SegmentId;
+
+use super::table::LockFileGuard;
+
+pub(crate) fn sid() -> SegmentId {
+    session_store::new_segment_id()
+}
+
+/// Serialises tests that mutate runtime-dir env vars. The test
+/// harness runs tests on multiple threads inside a single process,
+/// so env-var writes from one test would otherwise leak into a
+/// parallel test's `default_allocation_path()` lookup.
+pub(crate) static ENV_LOCK: LazyLock<Mutex<()>> = LazyLock::new(|| Mutex::new(()));
+
+/// Sandbox `YOI_RUNTIME_DIR` to a tempdir for the duration of
+/// a test; restore the previous value (and any `YOI_HOME` /
+/// `XDG_RUNTIME_DIR` that would otherwise outrank it) on drop.
+pub(crate) struct RuntimeDirSandbox {
+    prev_runtime: Option<String>,
+    prev_home: Option<String>,
+    prev_xdg: Option<String>,
+    _guard: MutexGuard<'static, ()>,
+}
+
+impl RuntimeDirSandbox {
+    pub(crate) fn new(dir: &Path) -> Self {
+        let guard = ENV_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        let prev_runtime = std::env::var("YOI_RUNTIME_DIR").ok();
+        let prev_home = std::env::var("YOI_HOME").ok();
+        let prev_xdg = std::env::var("XDG_RUNTIME_DIR").ok();
+        // SAFETY: ENV_LOCK serialises env writes across this test
+        // module; other modules that touch env vars rely on their
+        // own lock or `serial_test`.
+        unsafe {
+            std::env::remove_var("YOI_HOME");
+            std::env::remove_var("XDG_RUNTIME_DIR");
+            std::env::set_var("YOI_RUNTIME_DIR", dir);
+        }
+        Self {
+            prev_runtime,
+            prev_home,
+            prev_xdg,
+            _guard: guard,
+        }
+    }
+}
+
+impl Drop for RuntimeDirSandbox {
+    fn drop(&mut self) {
+        unsafe {
+            match &self.prev_runtime {
+                Some(v) => std::env::set_var("YOI_RUNTIME_DIR", v),
+                None => std::env::remove_var("YOI_RUNTIME_DIR"),
+            }
+            match &self.prev_home {
+                Some(v) => std::env::set_var("YOI_HOME", v),
+                None => std::env::remove_var("YOI_HOME"),
+            }
+            match &self.prev_xdg {
+                Some(v) => std::env::set_var("XDG_RUNTIME_DIR", v),
+                None => std::env::remove_var("XDG_RUNTIME_DIR"),
+            }
+        }
+    }
+}
+
+pub(crate) fn write_rule(path: &str, recursive: bool) -> ScopeRule {
+    ScopeRule {
+        target: PathBuf::from(path),
+        permission: Permission::Write,
+        recursive,
+    }
+}
+
+pub(crate) fn read_rule(path: &str, recursive: bool) -> ScopeRule {
+    ScopeRule {
+        target: PathBuf::from(path),
+        permission: Permission::Read,
+        recursive,
+    }
+}
+
+pub(crate) fn delegation_scope(rules: Vec<ScopeRule>) -> DelegationScope {
+    DelegationScope::from_config(&ScopeConfig {
+        allow: rules,
+        deny: Vec::new(),
+    })
+    .expect("test delegation scope")
+}
+
+pub(crate) fn sock(name: &str) -> PathBuf {
+    PathBuf::from(format!("/tmp/{name}.sock"))
+}
+
+pub(crate) fn open_empty(path: &Path) -> LockFileGuard {
+    LockFileGuard::open(path).unwrap()
+}