From b32a98a1ae8c623877f242212d1b9cabd0f597f3 Mon Sep 17 00:00:00 2001 From: Edvane Cristina Santos Daltro Date: Fri, 30 Oct 2026 22:29:52 -0400 Subject: [PATCH 1/2] Bound pending file observer replacement ownership --- .../workflows/file-observation-regression.yml | 32 +++++ crates/cyber-tools/src/file_changes/owner.rs | 120 +++++++++++++++++- .../file-observation-pending-owner.md | 37 ++++++ .../complete-plugin-hook-runtime/tasks.md | 5 + 4 files changed, 187 insertions(+), 7 deletions(-) create mode 100644 openspec/changes/complete-plugin-hook-runtime/file-observation-pending-owner.md diff --git a/.github/workflows/file-observation-regression.yml b/.github/workflows/file-observation-regression.yml index 271836fad..58eb5db21 100644 --- a/.github/workflows/file-observation-regression.yml +++ b/.github/workflows/file-observation-regression.yml @@ -152,6 +152,38 @@ jobs: path: ${{ runner.temp }}/observation-membership-base.log if-no-files-found: error retention-days: 14 + - name: Verify pending observer ownership without the fix + if: runner.os == 'Linux' + shell: bash + run: | + set -euo pipefail + python3 - <<'PYCONTROL' + import os,re,subprocess + from pathlib import Path + owner=Path('crates/cyber-tools/src/file_changes/owner.rs') + source=owner.read_text() + needle='self.sessions.is_empty() && !self.waiting_for_previous' + assert source.count(needle)==1 + try: + owner.write_text(source.replace(needle,'self.sessions.is_empty()')) + name='file_changes::owner::tests::readmission_reuses_one_pending_replacement_until_previous_owner_settles' + result=subprocess.run(['cargo','test','--locked','-p','cyber-tools','--lib',name,'--','--exact'],capture_output=True,text=True) + log=result.stdout+result.stderr + Path(os.environ['RUNNER_TEMP'],'observation-pending-owner-base.log').write_text(log) + assert result.returncode!=0, log[-8000:] + assert re.search(r'^test '+re.escape(name)+r' \.\.\. FAILED$',log,re.M), log[-8000:] + assert 'readmission must not grow a task chain behind a settling owner' in log, log[-8000:] + finally: + owner.write_text(source) + PYCONTROL + - name: Preserve pending observer ownership regression evidence + if: always() && runner.os == 'Linux' + uses: actions/upload-artifact@v4 + with: + name: pending-observer-owner-baseline + path: ${{ runner.temp }}/observation-pending-owner-base.log + if-no-files-found: error + retention-days: 14 - name: Verify known-private retirement without the fix if: runner.os == 'Linux' shell: bash diff --git a/crates/cyber-tools/src/file_changes/owner.rs b/crates/cyber-tools/src/file_changes/owner.rs index 4c0ce6716..b80ba91cd 100644 --- a/crates/cyber-tools/src/file_changes/owner.rs +++ b/crates/cyber-tools/src/file_changes/owner.rs @@ -28,6 +28,9 @@ const RECONCILE: Duration = Duration::from_secs(3); #[derive(Default)] struct Membership { sessions: BTreeMap, + /// Readmission reuses this owner while the previous incarnation settles. + /// Cancelling an empty pending owner would grow an unbounded join chain. + waiting_for_previous: bool, /// Keep only the newest committed acknowledgement per current Session. /// Both maps share the 256-Session bound; input bursts cannot grow a queue. renewals: BTreeMap, @@ -43,8 +46,8 @@ impl Membership { self.renewals.remove(session); self.scopes.remove(session); } - fn is_empty(&self) -> bool { - self.sessions.is_empty() + fn can_stop(&self) -> bool { + self.sessions.is_empty() && !self.waiting_for_previous } } @@ -69,15 +72,20 @@ impl Registry { } let mut members = entry.members.lock().unwrap_or_else(PoisonError::into_inner); members.remove(session); - if members.is_empty() { + if members.can_stop() { entry.cancel.cancel(); } } } } impl Entry { - fn add(&self, host: &BuiltinHost, session: &str) { + fn add_if_live(&self, host: &BuiltinHost, session: &str) -> bool { let mut members = self.members.lock().unwrap_or_else(PoisonError::into_inner); + // Activation may cancel an empty pending owner. Serialize admission with + // that decision rather than adding an already-cancelled child token. + if self.cancel.is_cancelled() { + return false; + } if members.sessions.contains_key(session) || members.sessions.len() < MAX_SESSIONS { members .sessions @@ -91,6 +99,7 @@ impl Entry { "FileChanged coverage unavailable: Session limit reached", ); } + true } } @@ -115,9 +124,8 @@ impl Observers { .retain(|_, entry| !entry.task.is_finished()); registry.remove_other_memberships(&path, &info.id); if let Some(entry) = registry.entries.get(&path) - && !entry.cancel.is_cancelled() + && entry.add_if_live(host, &info.id) { - entry.add(host, &info.id); return; } let previous = registry.entries.remove(&path); @@ -132,6 +140,7 @@ impl Observers { let cancel = CancellationToken::new(); let members = Arc::new(Mutex::new(Membership { sessions: BTreeMap::from([(info.id.clone(), cancel.child_token())]), + waiting_for_previous: previous.is_some(), renewals: BTreeMap::new(), scopes: BTreeSet::new(), explicit: Hints::default(), @@ -145,6 +154,13 @@ impl Observers { if let Some(previous) = previous { let _ = previous.task.await; } + { + let mut members = membership.lock().unwrap_or_else(PoisonError::into_inner); + members.waiting_for_previous = false; + if members.sessions.is_empty() { + owner.cancel(); + } + } run(weak, root, directory, membership, owner).await; }); registry.entries.insert( @@ -210,7 +226,7 @@ impl Observers { for entry in registry.entries.values() { let mut members = entry.members.lock().unwrap_or_else(PoisonError::into_inner); members.remove(session); - if members.is_empty() { + if members.can_stop() { entry.cancel.cancel(); } } @@ -859,6 +875,96 @@ async fn start_observation( mod tests { use super::*; + #[tokio::test] + async fn readmission_reuses_one_pending_replacement_until_previous_owner_settles() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().canonicalize().unwrap(); + let store = Arc::new( + cyber_store::Store::open(cyber_store::StoreOptions::new( + cyber_core::paths::DatabaseLocation::Memory, + cyber_server::runtime::Runtime::registry(), + )) + .unwrap(), + ); + let host = BuiltinHost::new(crate::HostOptions { + store, + tool_output_dir: root.join("tool-output"), + allowed_dirs: Vec::new(), + home: root.join("home"), + shell: "bash".into(), + config: Arc::new(|_| Ok((serde_json::json!({}), BTreeMap::new()))), + global_config_dir: root.join("config"), + env: Arc::new(std::collections::HashMap::::new()), + models: None, + temp_dir: root.join("tmp"), + sandbox_policy: None, + sandbox_helper: None, + isolation: crate::IsolationOptions { + dangerously_bypass: false, + container: false, + }, + }); + let info: SessionInfo = serde_json::from_value(serde_json::json!({ + "id": "session", "title": "fixture", "directory": root, + "parent_id": null, "agent": "default", "model": "fixture/model", + "mode": "default", "created_ms": 0 + })) + .unwrap(); + let gate = Arc::new(tokio::sync::Notify::new()); + let held = Arc::clone(&gate); + let previous = tokio::spawn(async move { held.notified().await }); + let cancel = CancellationToken::new(); + cancel.cancel(); + let observers = Observers(Mutex::new(Registry { + entries: BTreeMap::from([( + root.clone(), + Entry { + members: Arc::new(Mutex::new(Membership::default())), + cancel, + task: previous, + }, + )]), + closed: false, + })); + let mut replacements = std::collections::HashSet::new(); + let mut removed_tokens_cancelled = true; + for _ in 0..256 { + observers.refresh(&host, &info); + let token = { + let registry = observers.0.lock().unwrap(); + let entry = ®istry.entries[&root]; + replacements.insert(entry.task.id()); + member_cancellation(&entry.members, &info.id).unwrap() + }; + observers.remove(&info.id); + removed_tokens_cancelled &= token.is_cancelled(); + } + observers.refresh(&host, &info); + let final_token = { + let registry = observers.0.lock().unwrap(); + let entry = ®istry.entries[&root]; + replacements.insert(entry.task.id()); + member_cancellation(&entry.members, &info.id).unwrap() + }; + let final_membership_live = !final_token.is_cancelled(); + gate.notify_one(); + observers.shutdown().await; + assert!(removed_tokens_cancelled); + assert!( + final_membership_live, + "readmission needs live callback authority" + ); + assert!( + final_token.is_cancelled(), + "shutdown must revoke pending authority" + ); + assert_eq!( + replacements.len(), + 1, + "readmission must not grow a task chain behind a settling owner" + ); + } + #[test] fn removing_membership_cancels_held_delivery_without_stopping_other_members() { let owner = CancellationToken::new(); diff --git a/openspec/changes/complete-plugin-hook-runtime/file-observation-pending-owner.md b/openspec/changes/complete-plugin-hook-runtime/file-observation-pending-owner.md new file mode 100644 index 000000000..0389ab297 --- /dev/null +++ b/openspec/changes/complete-plugin-hook-runtime/file-observation-pending-owner.md @@ -0,0 +1,37 @@ +# Bounded pending FileChanged owner replacement + +Review reproduced a resource-lifetime defect in the existing observer registry. +A cancelled owner could still be settling. Each remove/readmit cycle then created +another Tokio task awaiting the previous task. The registry held one entry while +its nested chain grew: 256 cycles created 257 replacement tasks in the regression. +The new regression executed against unchanged production code at main +`3ecf0755` and failed with 257 replacement tasks. Only the test was added for +that baseline run. + +A pending replacement now retains its owner token until its predecessor settles. +Removing a Session still cancels that Session's callback token immediately and +removes its renewal/scope admission. Readmission joins the single pending owner +with a fresh callback token. Once the predecessor settles, activation clears the +pending flag under the membership lock and cancels an empty owner. Admission +checks liveness under that same lock, so it cannot attach an already-cancelled +child token during activation. Shutdown independently cancels and joins owners, +including their retained predecessors. No uncertain resources are discarded. + +The regression uses the real registry refresh/remove paths and a controlled held +predecessor. It requires one replacement across 256 cycles, cancellation of every +removed token, fresh live authority after final readmission and revocation on +joined shutdown. The original test executed and failed at the task-chain assertion +with 257 replacements. The corrected file-observation library suite passes 58 +tests with one subprocess helper intentionally ignored by the parent test runner. +All 24 actual HTTP callback lifecycle regressions also pass locally. These +component results do not prove native confinement or current-source platform +acceptance. Scoped `cyber-tools --lib` Clippy with warnings denied and Rust +formatting pass. Strict OpenSpec validates all 113 items; spec lint reports zero +errors and 25 warnings. + +The focused workflow removes only the pending-owner cancellation guard as a +counterexample. It must execute the regression and fail its exact task-chain +assertion; compilation failure cannot satisfy the control. The exact control +was executed locally and failed at that assertion. Platform CI and composed-feature +acceptance remain required. Task 5.2 +stays open, as do the separate plugin native admission and protocol gates. diff --git a/openspec/changes/complete-plugin-hook-runtime/tasks.md b/openspec/changes/complete-plugin-hook-runtime/tasks.md index 826a7af7d..1cefbb691 100644 --- a/openspec/changes/complete-plugin-hook-runtime/tasks.md +++ b/openspec/changes/complete-plugin-hook-runtime/tasks.md @@ -86,3 +86,8 @@ then review reproduced false deletion dispatch when a known private inode/subtre replaces a public baseline name. The new silent-retirement correction has two real HTTP regressions and a missing-fix behavioral control. Task 5.2 remains open for current-source platform acceptance; see the integration evidence above. + +The [pending observer owner correction](file-observation-pending-owner.md) bounds +remove/readmit replacement behind a settling owner while retaining cancellation +and joined shutdown. Focused and platform acceptance remain required; task 5.2 +stays open. From a7b76884e01dbfb7287f78bde111aea10f92d3b5 Mon Sep 17 00:00:00 2001 From: Edvane Cristina Santos Daltro Date: Fri, 30 Oct 2026 22:37:03 -0400 Subject: [PATCH 2/2] Record observer verification against merged feature changes --- .../file-observation-pending-owner.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/openspec/changes/complete-plugin-hook-runtime/file-observation-pending-owner.md b/openspec/changes/complete-plugin-hook-runtime/file-observation-pending-owner.md index 0389ab297..d2701a5d2 100644 --- a/openspec/changes/complete-plugin-hook-runtime/file-observation-pending-owner.md +++ b/openspec/changes/complete-plugin-hook-runtime/file-observation-pending-owner.md @@ -26,12 +26,12 @@ tests with one subprocess helper intentionally ignored by the parent test runner All 24 actual HTTP callback lifecycle regressions also pass locally. These component results do not prove native confinement or current-source platform acceptance. Scoped `cyber-tools --lib` Clippy with warnings denied and Rust -formatting pass. Strict OpenSpec validates all 113 items; spec lint reports zero +formatting pass. Strict OpenSpec validates all 114 items; spec lint reports zero errors and 25 warnings. The focused workflow removes only the pending-owner cancellation guard as a counterexample. It must execute the regression and fail its exact task-chain assertion; compilation failure cannot satisfy the control. The exact control was executed locally and failed at that assertion. Platform CI and composed-feature -acceptance remain required. Task 5.2 -stays open, as do the separate plugin native admission and protocol gates. +acceptance remain required. Task 5.2 stays open, as do the separate plugin native +admission and protocol gates.