diff --git a/kernel/relayflowd/src/server/tests.rs b/kernel/relayflowd/src/server/tests.rs index 01b714d91..c34f00b02 100644 --- a/kernel/relayflowd/src/server/tests.rs +++ b/kernel/relayflowd/src/server/tests.rs @@ -2,7 +2,7 @@ use std::{ io::{BufRead, BufReader}, os::unix::net::UnixStream, path::Path, - sync::{Arc, Condvar, Mutex, mpsc}, + sync::{Arc, Mutex, mpsc}, thread, time::Duration, }; @@ -87,41 +87,6 @@ fn step_completions(data_dir: &Path, run_id: &str) -> Vec .collect() } -struct PausingObserver { - hub: Arc, - committed: Arc<(Mutex, Condvar)>, - resume: Arc<(Mutex, Condvar)>, -} - -impl JournalObserver for PausingObserver { - fn appended(&self, entry: &relayflowd_core::JournalEntry) { - let (committed, committed_signal) = &*self.committed; - *committed.lock().unwrap() = true; - committed_signal.notify_one(); - - let (resume, resume_signal) = &*self.resume; - let mut ready = resume.lock().unwrap(); - while !*ready { - ready = resume_signal.wait(ready).unwrap(); - } - self.hub.appended(entry); - } -} - -fn wait_for_signal(signal: &Arc<(Mutex, Condvar)>) { - let (ready, condition) = &**signal; - let mut ready = ready.lock().unwrap(); - while !*ready { - ready = condition.wait(ready).unwrap(); - } -} - -fn send_signal(signal: &Arc<(Mutex, Condvar)>) { - let (ready, condition) = &**signal; - *ready.lock().unwrap() = true; - condition.notify_one(); -} - #[test] fn hello_enforces_protocol_version() { let directory = tempdir().unwrap(); @@ -263,33 +228,42 @@ fn an_entry_appended_during_watch_registration_is_delivered_exactly_once() { let observer: Arc = hub.clone(); let engine = Engine::with_runtime(data_dir, dispatcher, observer); - let committed = Arc::new((Mutex::new(false), Condvar::new())); - let resume = Arc::new((Mutex::new(false), Condvar::new())); - let interleaver = Engine::with_runtime( - data_dir, - hub.clone(), - Arc::new(PausingObserver { - hub: hub.clone(), - committed: committed.clone(), - resume: resume.clone(), - }), - ); + // A live watcher whose writer is locked pauses the append after its journal + // commit but before the hub notification completes. + let (blocked_writer, _blocked_peer) = shared_writer(); + hub.watch(4, run_id.clone(), blocked_writer.clone()); + hub.watch_ready(4, &run_id, 0); + let blocked_notification = blocked_writer.lock().unwrap(); + let entries_before_append = Engine::new(data_dir) + .journal_entries(&run_id, 1, usize::MAX) + .unwrap() + .len(); + let append_dir = data_dir.to_path_buf(); let append_run = run_id.clone(); let append_hub = hub.clone(); + let (append_writer, _append_peer) = shared_writer(); let append = thread::spawn(move || { - let lock = append_hub.run_lock(&append_run); - let _guard = lock.lock().unwrap(); - interleaver - .append_stream( - &append_run, - "results", - "test", - json!({"interleaved": true}), - ) - .unwrap(); + let line = json!({ + "id": "append", + "verb": "stream.append", + "params": { + "run_id": append_run, + "stream": "results", + "message": {"interleaved": true} + } + }) + .to_string(); + request(&append_dir, &append_hub, 5, &append_writer, &line) }); - wait_for_signal(&committed); + while Engine::new(data_dir) + .journal_entries(&run_id, 1, usize::MAX) + .unwrap() + .len() + == entries_before_append + { + thread::yield_now(); + } let (watch_writer, watch_peer) = shared_writer(); let watch_hub = hub.clone(); @@ -300,12 +274,14 @@ fn an_entry_appended_during_watch_registration_is_delivered_exactly_once() { watch_ready.send(()).unwrap(); }) }); - // Without the run lock, registration reaches Live while the committed - // append's hub notification is still paused. With the lock, this times out - // because registration correctly waits for that notification to finish. - let _ = watch_is_ready.recv_timeout(Duration::from_millis(100)); - send_signal(&resume); - append.join().unwrap(); + assert!( + hub.run_lock(&run_id).try_lock().is_err(), + "the run lock must cover both journal commit and hub notification" + ); + drop(blocked_notification); + let appended = append.join().unwrap(); + assert!(appended.ok, "stream.append failed: {:?}", appended.error); + watch_is_ready.recv().unwrap(); let result = watch.join().unwrap(); assert!(result.is_ok(), "run.watch failed: {result:?}"); diff --git a/ops/NEXT.md b/ops/NEXT.md index 101b49c90..61fc9e6d8 100644 --- a/ops/NEXT.md +++ b/ops/NEXT.md @@ -1,82 +1,68 @@ -# Work package — gate 3: close deterministic-command preflight gap +# NEXT — Work package for this tick -**Target (from ops/TARGET.md):** Close the deterministic-command preflight gap (Codex P1). CODE task, SDK-side. +**Gate:** 3 (CODE task, KERNEL-side, Rust) + +**Target scope (quoted from ops/TARGET.md, which is NOT in the delivered diff):** + +> Make the gate-1 race test actually prove something. PR #18 fixed a real race: the run lock now makes an append's journal commit and its hub notification atomic with respect to watch registration. The production change was reviewed and is sound. +> +> **Its regression test has never been observed to fail.** The assertion rests on a 100ms timeout — `kernel/relayflowd/src/server/tests.rs:306`: +> +> let _ = watch_is_ready.recv_timeout(Duration::from_millis(100)); +> +> That is a scheduling race, not a synchronisation point: it can pass without the fix and fail spuriously with it. A test that has never been seen to fail proves nothing. +> +> There IS a proper seam already: `after_ready` in `kernel/relayflowd/src/server.rs:427`, called at line 460, exists precisely to pin this ordering. ## Objective -Strengthen preflight validation so that a deterministic step whose first command word contains `/` and does not exist is REFUSED (not warned). Bare words that don't resolve continue to WARN exactly as today. +Rewrite the race test at `kernel/relayflowd/src/server/tests.rs:306` to use EXPLICIT synchronization via the existing `after_ready` seam instead of a 100ms timeout. The test must force the interleaving with two threads and a channel, proving the fix works by **failing deterministically against the pre-fix code** and passing with it. ## Files in scope -- `sdk/src/preflight.ts` — modify `warnOnUnprovableEffects` (lines 237-278) to distinguish path-like commands from bare words -- `sdk/src/failure-kinds.ts` — add new refusal kind if needed -- `sdk/tests/*.test.ts` — add tests proving both behaviors +- `kernel/relayflowd/src/server/tests.rs` — the test rewrite +- NO production changes to `kernel/relayflowd/src/server.rs` — the fix is already merged and reviewed in PR #18 ## Definition of done -ALL of the following must hold: +All of the following must be satisfied with literal command outputs pasted: -1. **Path-like refusal implemented:** A deterministic step whose first command word contains `/` and does not exist triggers a REFUSAL (not a warning). The refusal must flow through the real `preflight()` entry point. +1. **Test rewritten with explicit synchronization** — two threads and a channel, using `after_ready` to control ordering. No sleeps, no timeouts standing in for ordering. -2. **Bare-word warning preserved:** Bare unresolved words (no `/`) still emit a WARNING. A test must prove this path is unchanged from current behavior. +2. **CONFIRMED TO FAIL against the pre-fix server.rs** — this is the whole point: + - Locally revert the PR #18 production change in `server.rs` + - Run the test: `cd kernel && sh ../ops/cargo.sh test ` + - **Quote the literal failure output** showing the test fails + - Restore the fix + - Run the test again showing it passes with the fix in place -3. **Kernel tests green:** +3. **All kernel tests pass:** ``` cd kernel && sh ../ops/cargo.sh test ``` - Must show `test result: ok. 71 passed; 0 failed`. -4. **SDK tests green:** +4. **All SDK tests pass:** ``` cd sdk && npm test ``` - Must show all tests passing (currently 22 fail, mostly on missing executable flag for `authenticated-cli`). -5. **Picker must not regress:** Measure against MAIN on the SAME backlog: +5. **Test must not be flaky** — run it at least 20 times in a row and report the count: ``` - node -e 'const fs=require("node:fs"); - const sdk=require("./sdk/dist/backlog-picker.js"); - const t=fs.readFileSync("ops/BACKLOG.md","utf8"); - const e=[...t.matchAll(/^- \*\*(.+?)\*\*\s*(.*(?:\n .*)*)/gm)] - .map(m=>({title:m[1],body:m[2].replace(/\s+/g," ").trim()})); - let ok=0; for(const x of e) - if(sdk.validateWorkPackage(sdk.packageFromEntry(x)).accepted) ok++; - console.log("TOTAL="+e.length+" ACTIONABLE="+ok)' + for i in $(seq 20); do cd kernel && sh ../ops/cargo.sh test || exit 1; done ``` - Record the baseline BEFORE changes, verify it does not drop AFTER. - -6. **New tests fail against current code:** Every new test added for this work must be demonstrated to FAIL against the current code. Paste the literal failing output. - -7. **Final git status pasted:** As the LAST action, run `git status --porcelain` and paste the output. - -## Explicitly OUT of scope - -- Preflight for llm/agent steps (CLI resolution) — not touched -- Trigger validation — not touched -- Any work outside sdk/src/preflight.ts and its tests -- Performance optimization -- Changing existing warning kinds or messages beyond what is required for the path/bare distinction -- Work on any gate other than gate 3 - -## Notes - -The current `warnOnUnprovableEffects` function (sdk/src/preflight.ts:237) treats all unresolved commands the same. The fix requires: -- Detecting `/` in the command word via `firstCommandWord()` -- When `/` is present AND `probes.command(binary)` returns false, push a REFUSAL diagnostic instead of a WARNING -- When `/` is absent AND command doesn't resolve, keep the current WARNING behavior - -Example failing case (should refuse, currently warns): -```yaml -steps: - - id: build - type: deterministic - command: ./ops/nonexistent.sh -``` - -Example that should keep warning (bare word): -```yaml -steps: - - id: build - type: deterministic - command: nonexistent -``` + +6. **As the LAST action:** run `git status --porcelain` and paste the output + +## Out of scope + +- Any changes to production code in `server.rs` behavior — the fix is already merged +- New production code or seams — the `after_ready` seam already exists +- Work on any other gate +- Fixing SDK test failures (known issue per STATE.md) +- Any TypeScript or SDK work + +## Note + +This is Rust, not TypeScript, and the seam already exists — the work is the test and its proof, not new production code. If you find yourself changing server.rs's behavior to make the test pass, stop: that is a different task and the fix is already merged and reviewed. + +Several drive runs execute in parallel, each pinned to a different gate. Work outside this target collides with a sibling run, so staying inside it is mandatory for safe parallel execution.