drive: cloud run ef088191

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<StepCompletedPayload>
         .collect()
 }
 
-struct PausingObserver {
-    hub: Arc<ProtocolHub>,
-    committed: Arc<(Mutex<bool>, Condvar)>,
-    resume: Arc<(Mutex<bool>, 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<bool>, Condvar)>) {
-    let (ready, condition) = &**signal;
-    let mut ready = ready.lock().unwrap();
-    while !*ready {
-        ready = condition.wait(ready).unwrap();
-    }
-}
-
-fn send_signal(signal: &Arc<(Mutex<bool>, 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<dyn JournalObserver> = 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:?}");
 

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 <test_name>`
+   - **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 <test_name> || 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.