diff --git a/docs/active-work.md b/docs/active-work.md index 95b6e9b..2525cc3 100644 --- a/docs/active-work.md +++ b/docs/active-work.md @@ -171,7 +171,7 @@ If it does not, stop and repair the remote/fetch configuration. to recur; the next occurrence carries its own evidence under whoever's PR, and a Stage B framing follows then. -## Journey/GPU directory-target ratchet — PR #183 GATED AFTER REVIEW ROUND 1 +## Journey/GPU directory-target ratchet — PR #183 REGATING PUBLIC PATH - **Approved correction, not new product behavior.** GPU initial-target framing Q#GT6 / acceptance 10 and Journey Stage 1a N2/N5 already make @@ -186,12 +186,15 @@ If it does not, stop and repair the remote/fetch configuration. . It is intentionally open and unmerged pending user review. - **Scope is one acceptance ratchet:** remove `"."` from the four - genuinely invalid cases and add a transport-level positive which - requires snapshot-first + `InitialTargetResult::Opened` for `"."`, - then consumes the post-quiescence replacement snapshot and requires - dired's canonical header plus a known directory entry before proving - the same daemon can open a following file target. No production - source, protocol, framing decision, or coherence grade changes. + genuinely invalid cases and drive the public `pmacs --gpu .` root + broker through the real managed GPU connector. The positive requires + snapshot-first + `InitialTargetResult::Opened`, then consumes the + post-quiescence replacement snapshot and requires dired's canonical + header plus a known directory entry before proving the same daemon can + open a following file target. The private display-less acceptance + probe now reports its snapshot count and final materialized text so + that public path is observable. No normal frontend/daemon behavior, + protocol, framing decision, or coherence grade changes. - **Review round 1: three findings, all real and corrected.** The first test stopped at the deliberately pre-existing bootstrap document, so it did not pin the resolver's later dired commit. The @@ -200,6 +203,14 @@ If it does not, stop and repair the remote/fetch configuration. had advanced. The first is now a post-quiescence transport assertion; the latter two are corrected in this revision. The complete matrix below is green on the corrected tree. +- **Live public-path check tightened that correction further.** A user + report that `pmacs --gpu .` differed from `pmacs .` did not reproduce: + the live default daemon delivered both snapshots, and a traced + windowed invocation applied both and displayed dired. It nevertheless + exposed that the corrected acceptance still attached a raw protocol + client rather than invoking the public root broker and real GPU + connector. The test now covers those surfaces and passes **15/15**; + its full matrix is being rerun before this revision is pushed. - **Post-review full gate matrix is green at `ec4191f`:** - `cargo fmt --check`; - strict workspace clippy; diff --git a/pmacs-gpu/src/main.rs b/pmacs-gpu/src/main.rs index a26f340..86132fc 100644 --- a/pmacs-gpu/src/main.rs +++ b/pmacs-gpu/src/main.rs @@ -936,10 +936,20 @@ fn run_headless_managed_probe( } }; let mut client = managed.client; + let initial_message = client.take_initial_message(); let initial_target_ready = matches!( - client.take_initial_message(), + initial_message.as_ref(), Some(InstanceMessage::BufferSnapshot { .. }) ); + let mut buffer_facts = ManagedProbeBufferFacts::default(); + if let Some(message) = initial_message.as_ref() + && let Err(error) = buffer_facts.observe(message) + { + let contents = format!("phase=error\nerror={error}\n"); + let _ = write_probe_report(report, &contents); + eprintln!("pmacs-gpu managed probe: {error}"); + return 7; + } let daemon = managed.daemon; let protocol = client.server_protocol_version(); @@ -961,8 +971,14 @@ fn run_headless_managed_probe( let mut last_wait_result = None; let mut last_disconnect = String::new(); if ready - && let Err(error) = - write_managed_probe_report(report, "ready", protocol, &daemon, &disconnect) + && let Err(error) = write_managed_probe_report( + report, + "ready", + protocol, + &daemon, + &buffer_facts, + &disconnect, + ) { eprintln!( "pmacs-gpu managed probe: writing {} failed: {error}", @@ -976,11 +992,23 @@ fn run_headless_managed_probe( } match event_rx.recv_timeout(Duration::from_millis(50)) { Ok(AttachEvent::Message(message)) => { - if matches!(*message, InstanceMessage::BufferSnapshot { .. }) && !ready { + let is_snapshot = matches!(*message, InstanceMessage::BufferSnapshot { .. }); + if let Err(error) = buffer_facts.observe(&message) { + let contents = format!("phase=error\nerror={error}\n"); + let _ = write_probe_report(report, &contents); + eprintln!("pmacs-gpu managed probe: {error}"); + return 7; + } + if is_snapshot { ready = true; - if let Err(error) = - write_managed_probe_report(report, "ready", protocol, &daemon, &disconnect) - { + if let Err(error) = write_managed_probe_report( + report, + "ready", + protocol, + &daemon, + &buffer_facts, + &disconnect, + ) { eprintln!( "pmacs-gpu managed probe: writing {} failed: {error}", report.display() @@ -1006,9 +1034,14 @@ fn run_headless_managed_probe( || wait_result != last_wait_result || disconnect != last_disconnect) { - if let Err(error) = - write_managed_probe_report(report, "ready", protocol, &daemon, &disconnect) - { + if let Err(error) = write_managed_probe_report( + report, + "ready", + protocol, + &daemon, + &buffer_facts, + &disconnect, + ) { eprintln!( "pmacs-gpu managed probe: writing {} failed: {error}", report.display() @@ -1021,9 +1054,14 @@ fn run_headless_managed_probe( } if ready && stdin_closed { - if let Err(error) = - write_managed_probe_report(report, "complete", protocol, &daemon, &disconnect) - { + if let Err(error) = write_managed_probe_report( + report, + "complete", + protocol, + &daemon, + &buffer_facts, + &disconnect, + ) { eprintln!( "pmacs-gpu managed probe: writing {} failed: {error}", report.display() @@ -1043,11 +1081,42 @@ fn run_headless_managed_probe( } } +#[derive(Default)] +struct ManagedProbeBufferFacts { + snapshots: u32, + last_snapshot_text: String, +} + +impl ManagedProbeBufferFacts { + fn observe(&mut self, message: &InstanceMessage) -> Result<(), String> { + let InstanceMessage::BufferSnapshot { crdt_snapshot, .. } = message else { + return Ok(()); + }; + let doc = loro::LoroDoc::new(); + doc.import(crdt_snapshot) + .map_err(|error| format!("BufferSnapshot import failed: {error:?}"))?; + self.snapshots += 1; + self.last_snapshot_text = doc.get_text(LORO_TEXT_CONTAINER).to_string(); + Ok(()) + } +} + +fn hex_bytes(bytes: &[u8]) -> String { + use std::fmt::Write as _; + + let mut encoded = String::with_capacity(bytes.len() * 2); + for byte in bytes { + let _ = write!(encoded, "{byte:02x}"); + } + encoded +} + fn write_managed_probe_report( report: &Path, phase: &str, protocol: u32, daemon: &attach::ManagedDaemonFacts, + buffer_facts: &ManagedProbeBufferFacts, disconnect: &str, ) -> std::io::Result<()> { use std::fmt::Write as _; @@ -1056,6 +1125,12 @@ fn write_managed_probe_report( let _ = writeln!(out, "phase={phase}"); let _ = writeln!(out, "server_protocol_version={protocol}"); let _ = writeln!(out, "buffer_snapshot=true"); + let _ = writeln!(out, "buffer_snapshots={}", buffer_facts.snapshots); + let _ = writeln!( + out, + "last_snapshot_hex={}", + hex_bytes(buffer_facts.last_snapshot_text.as_bytes()) + ); let _ = writeln!(out, "spawned_daemon={}", daemon.spawned_daemon()); let _ = writeln!( out, diff --git a/tests/gpu_invocation_acceptance.rs b/tests/gpu_invocation_acceptance.rs index 13dc020..ec6be89 100644 --- a/tests/gpu_invocation_acceptance.rs +++ b/tests/gpu_invocation_acceptance.rs @@ -121,6 +121,19 @@ mod crdt { .collect() } + fn decode_hex(encoded: &str) -> String { + assert_eq!(encoded.len() % 2, 0, "hex payload must have even length"); + let bytes = encoded + .as_bytes() + .chunks_exact(2) + .map(|pair| { + let pair = std::str::from_utf8(pair).expect("hex pair is UTF-8"); + u8::from_str_radix(pair, 16).expect("decode hex pair") + }) + .collect::>(); + String::from_utf8(bytes).expect("snapshot text is UTF-8") + } + fn wait_for_fact( report: &Path, key: &str, @@ -223,34 +236,6 @@ mod crdt { stream: UnixStream, } - impl TargetSession { - fn wait_for_replacement_snapshot(&mut self) -> (pmacs::buffer::BufferId, String) { - let deadline = Instant::now() + Duration::from_secs(10); - loop { - assert!( - Instant::now() < deadline, - "target frontend did not receive a replacement buffer snapshot" - ); - match read_message::(&mut self.stream) - .expect("read target frontend after bootstrap") - { - InstanceMessage::BufferSnapshot { - buffer_id, - crdt_snapshot, - } if buffer_id != self.buffer_id => { - let replica = - CrdtState::new(self.frontend_id.0).expect("replacement buffer replica"); - replica - .import_snapshot(&crdt_snapshot) - .expect("import replacement buffer snapshot"); - return (buffer_id, replica.materialize_string()); - } - _ => {} - } - } - } - } - fn attach_target(socket: &Path, cwd: &Path, path: &Path) -> TargetSession { use std::os::unix::ffi::OsStrExt; @@ -384,6 +369,16 @@ mod crdt { } impl ManagedProbe { + fn from_child(mut child: Child, report: &Path) -> Self { + let stdin = child.stdin.take().expect("probe stdin"); + Self { + child, + stdin: Some(stdin), + report: report.to_owned(), + daemon_pid: None, + } + } + fn spawn(socket: &Path, report: &Path, daemon_executable: &Path, home: &Path) -> Self { Self::spawn_with_env(socket, report, daemon_executable, home, &[]) } @@ -446,18 +441,11 @@ mod crdt { for (key, value) in envs { command.env(key, value); } - let mut child = command.spawn().expect("spawn managed probe"); - let stdin = child.stdin.take().expect("probe stdin"); - Self { - child, - stdin: Some(stdin), - report: report.to_owned(), - daemon_pid: None, - } + Self::from_child(command.spawn().expect("spawn managed probe"), report) } - fn wait_ready(&mut self) -> HashMap { - let facts = wait_for_fact(&self.report, "phase", "ready", Duration::from_secs(10)); + fn wait_for(&mut self, key: &str, expected: &str) -> HashMap { + let facts = wait_for_fact(&self.report, key, expected, Duration::from_secs(10)); if facts .get("spawned_daemon") .is_some_and(|value| value == "true") @@ -467,6 +455,10 @@ mod crdt { facts } + fn wait_ready(&mut self) -> HashMap { + self.wait_for("phase", "ready") + } + fn close(mut self) -> std::process::ExitStatus { self.stdin.take(); wait_for_fact(&self.report, "phase", "complete", Duration::from_secs(5)); @@ -691,25 +683,46 @@ mod crdt { } #[test] - fn directory_target_reaches_ready_and_leaves_the_daemon_usable() { + fn public_gpu_directory_target_reaches_dired_and_leaves_the_daemon_usable() { let temp = secure_tempdir(); let socket = temp.path().join("directory-target.sock"); + let report = temp.path().join("directory-target-report"); + let wrapper = temp.path().join("headless-gpu"); let listed_name = "listed-before-bootstrap.txt"; fs::write(temp.path().join(listed_name), "listed\n").expect("write listed file"); - let mut daemon = spawn_daemon(&socket, &[]); - - // Journey Stage 1a superseded the old IsADirectory failure: - // `attach_target` requires the production snapshot-first sequence - // followed by `InitialTargetResult::Opened`. The synchronous - // snapshot is deliberately the pre-existing document; dired's - // post-await commit replaces it on a later daemon tick. - let mut directory = attach_target(&socket, temp.path(), Path::new(".")); - let bootstrap_buffer = directory.buffer_id; - let (dired_buffer, listing) = directory.wait_for_replacement_snapshot(); - assert_ne!( - dired_buffer, bootstrap_buffer, - "the asynchronous resolver must replace the bootstrap document" + write_script( + &wrapper, + "test \"$1\" = \"--managed-attach\"\n\ + socket=$2\n\ + daemon=$3\n\ + shift 3\n\ + exec \"$PMACS_REAL_GPU\" --headless-managed-probe \ + \"$socket\" \"$PMACS_TEST_REPORT\" \"$daemon\" \"$@\"", ); + + let mut command = Command::new(pmacs_binary()); + command + .args(["--gpu", "--socket"]) + .arg(&socket) + .arg(".") + .current_dir(temp.path()) + .env(TEST_GPU_OVERRIDE, &wrapper) + .env("PMACS_REAL_GPU", gpu_binary()) + .env("PMACS_TEST_REPORT", &report) + .env("HOME", temp.path()) + .env("XDG_CONFIG_HOME", temp.path()) + .stdin(Stdio::piped()) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + let mut directory = + ManagedProbe::from_child(command.spawn().expect("spawn public GPU command"), &report); + + // The public root broker and real managed GPU connector must stay + // alive through Journey N2's asynchronous dired commit. Snapshot + // one is the deliberately pre-existing bootstrap document; snapshot + // two is the post-quiescence directory surface. + let facts = directory.wait_for("buffer_snapshots", "2"); + let listing = decode_hex(&facts["last_snapshot_hex"]); let canonical = fs::canonicalize(temp.path()).expect("canonical directory"); let mut lines = listing.lines(); let expected_header = format!("{}:", canonical.display()); @@ -727,10 +740,8 @@ mod crdt { let survivor = attach_target(&socket, temp.path(), Path::new("still-alive.txt")); assert_eq!(survivor.replica.materialize_string(), "alive\n"); - drop(directory); drop(survivor); - signal_pid(daemon.id(), Signal::SIGTERM); - assert!(wait_for_exit(&mut daemon, Duration::from_secs(5)).success()); + assert!(directory.close().success()); } #[test]