neuwirth
/
pmacs
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

test(journey): ratchet the public GPU directory path 

Drive pmacs --gpu . through the root broker and real managed GPU
connector, keep the session alive through the asynchronous dired
replacement, and assert its canonical listing before daemon reuse.

Expose snapshot count and materialized text through the private
display-less acceptance probe so the public path is observable.
This commit is contained in:
Levi Neuwirth 2026-07-28 10:27:22 -04:00
parent 3be4285f7c
commit 34b8f28cf2
3 changed files with 172 additions and 75 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
docs/active-work.md
View File

@ -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.
<https://github.com/levineuwirth/pmacs/pull/183>. 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;

101
pmacs-gpu/src/main.rs
View File

@ -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,

121
tests/gpu_invocation_acceptance.rs
View File

@ -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::<Vec<_>>();
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::<InstanceMessage>(&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<String, String> {
let facts = wait_for_fact(&self.report, "phase", "ready", Duration::from_secs(10));
fn wait_for(&mut self, key: &str, expected: &str) -> HashMap<String, String> {
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<String, String> {
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]