Remove external mode, fix server/client roles

- SDK now connects as client (hegeld is the server)
- Removed HegelMode enum and mode-related functions
- Renamed set_embedded_connection to set_connection
- SDK waits for hegeld to start and initiates version negotiation

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: claude

.claude/CLAUDE.md

@@ -55,16 +55,16 @@ hegel-rust/
 
 ### How It Works
 
-The SDK spawns the `hegel` CLI as a subprocess with `--client-mode`. The test binary creates a Unix socket server, and hegel connects for each test case. The build script (`build.rs`) automatically installs Python and hegel into cargo's `OUT_DIR/hegel` via uv if not found on PATH.
+The SDK creates a socket path and spawns the `hegel` CLI as a subprocess. Hegeld binds to the socket and listens for connections. The SDK then connects as a client, and a single persistent connection is maintained for the program run. Multiple tests can be executed over this connection. The build script (`build.rs`) automatically installs Python and hegel into cargo's `OUT_DIR/hegel` via uv if not found on PATH.
 
 ### Protocol
 
-Each test case follows this handshake:
-1. Hegel connects to the SDK's socket
-2. Hegel sends: `{"is_last_run": bool}` (is_last_run=true on final replay for output)
-3. SDK responds: `{"type": "handshake_ack"}`
-4. SDK runs test, communicating via `generate`/`start_span`/`stop_span` commands
-5. SDK sends result: `{"type": "test_result", "result": "pass"|"fail"|"reject", ...}`
+The protocol uses CBOR encoding over multiplexed channels. For each test:
+1. SDK sends `run_test` request on control channel
+2. Hegeld sends `test_case` events with channel IDs for each test case
+3. SDK runs test function, sending `generate`/`start_span`/`stop_span` requests on the test channel
+4. SDK sends `mark_complete` with status (VALID, INVALID, or INTERESTING)
+5. After all test cases, hegeld sends `test_done` with results`
 
 ### Thread-Local State
 

src/embedded.rs

@@ -1,6 +1,5 @@
 use crate::gen::{
-    clear_embedded_connection, set_embedded_connection, set_is_last_run, set_mode,
-    take_generated_values, HegelMode,
+    clear_connection, set_connection, set_is_last_run, take_generated_values,
 };
 use crate::protocol::{
     cbor_to_json, json_to_cbor, Channel, Connection,
@@ -9,11 +8,12 @@ use crate::protocol::{
 use ciborium::Value as CborValue;
 use serde_json::{json, Value};
 use std::cell::RefCell;
-use std::os::unix::net::UnixListener;
+use std::os::unix::net::UnixStream;
 use std::panic::{self, catch_unwind, AssertUnwindSafe};
 use std::process::{Command, Stdio};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Once};
+use std::time::Duration;
 use tempfile::TempDir;
 
 static PANIC_HOOK_INIT: Once = Once::new();
@@ -79,7 +79,7 @@ impl Verbosity {
 /// Special marker used to identify assume(false) panics.
 const REJECT_MARKER: &str = "HEGEL_REJECT";
 
-/// Run property-based tests using Hegel in embedded mode with default options.
+/// Run property-based tests using Hegel with default options.
 ///
 /// This is a convenience function for simple cases. For configuration options,
 /// use [`Hegel::new`] with the builder pattern.
@@ -178,18 +178,14 @@ where
     pub fn run(self) {
         init_panic_hook();
 
-        // Create temp directory with socket
+        // Create temp directory with socket path
         let temp_dir = TempDir::new().expect("Failed to create temp directory");
         let socket_path = temp_dir.path().join("hegel.sock");
 
-        // Create Unix socket server
-        let listener = UnixListener::bind(&socket_path).expect("Failed to bind socket");
-
-        // Build hegel command
+        // Build hegel command - hegeld will bind to the socket and listen
         let hegel_path = self.hegel_path.as_deref().unwrap_or(HEGEL_BINARY_PATH);
         let mut cmd = Command::new(hegel_path);
-        cmd.arg("--client-mode")
-            .arg(&socket_path)
+        cmd.arg(&socket_path)
             .arg("--verbosity")
             .arg(self.verbosity.as_str());
 
@@ -204,29 +200,47 @@ where
 
         let mut child = cmd.spawn().expect("Failed to spawn hegel");
 
-        // Accept the single connection from hegeld
-        let stream = match listener.accept() {
-            Ok((stream, _)) => stream,
-            Err(e) => {
+        // Wait for hegeld to create the socket and start listening
+        let mut attempts = 0;
+        let stream = loop {
+            if socket_path.exists() {
+                match UnixStream::connect(&socket_path) {
+                    Ok(stream) => break stream,
+                    Err(e) if attempts < 50 => {
+                        // Socket exists but not yet listening
+                        std::thread::sleep(Duration::from_millis(100));
+                        attempts += 1;
+                        continue;
+                    }
+                    Err(e) => {
+                        let _ = child.kill();
+                        panic!("Failed to connect to hegeld socket: {}", e);
+                    }
+                }
+            }
+            if attempts >= 50 {
                 let _ = child.kill();
-                panic!("Failed to accept connection: {}", e);
+                panic!("Timeout waiting for hegeld to create socket");
             }
+            std::thread::sleep(Duration::from_millis(100));
+            attempts += 1;
         };
 
         // Create connection and perform version negotiation
         let connection = Connection::new(stream);
 
-        // Handle version negotiation (as server accepting from hegeld)
+        // Initiate version negotiation (SDK is the client)
         let control = connection.control_channel();
-        let (req_id, payload) = control.receive_request().expect("Failed to receive handshake");
-
-        if payload != VERSION_NEGOTIATION_MESSAGE {
-            panic!("Invalid version negotiation message: {:?}", String::from_utf8_lossy(&payload));
+        let req_id = control.send_request(VERSION_NEGOTIATION_MESSAGE.to_vec())
+            .expect("Failed to send version negotiation");
+        let response = control.receive_response(req_id)
+            .expect("Failed to receive version response");
+
+        if response != VERSION_NEGOTIATION_OK {
+            let _ = child.kill();
+            panic!("Version negotiation failed: {:?}", String::from_utf8_lossy(&response));
         }
 
-        control.send_response(req_id, VERSION_NEGOTIATION_OK.to_vec())
-            .expect("Failed to send version ack");
-
         if self.verbosity == Verbosity::Debug {
             eprintln!("Version negotiation complete");
         }
@@ -346,17 +360,15 @@ fn run_test_case<F: FnMut()>(
     _verbosity: Verbosity,
     got_interesting: &Arc<AtomicBool>,
 ) -> (String, Option<Value>) {
-    // Set thread-local state
-    set_mode(HegelMode::Embedded);
+    // Set thread-local state for this test case
     set_is_last_run(is_final);
-    set_embedded_connection(Arc::clone(connection), test_channel.clone_for_embedded());
+    set_connection(Arc::clone(connection), test_channel.clone_for_embedded());
 
     // Run test in catch_unwind
     let result = catch_unwind(AssertUnwindSafe(test_fn));
 
     // Clear connection before returning (test is done generating)
-    clear_embedded_connection();
-    set_mode(HegelMode::External);
+    clear_connection();
 
     match result {
         Ok(()) => ("VALID".to_string(), None),

src/gen/mod.rs

@@ -43,38 +43,9 @@ pub(crate) mod exit_codes {
 }
 use std::cell::{Cell, RefCell};
 use std::marker::PhantomData;
-use std::os::unix::net::UnixStream;
 use std::sync::Arc;
 
-use crate::protocol::{cbor_to_json, json_to_cbor, negotiate_version, Channel, Connection};
-
-// ============================================================================
-// Mode Management
-// ============================================================================
-
-/// Operating mode for the SDK.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
-pub enum HegelMode {
-    /// SDK connects to a running hegel server (external process runs tests)
-    #[default]
-    External,
-    /// SDK accepts connections from hegel (embedded in test binary)
-    Embedded,
-}
-
-thread_local! {
-    static MODE: Cell<HegelMode> = const { Cell::new(HegelMode::External) };
-}
-
-/// Get the current operating mode.
-pub fn current_mode() -> HegelMode {
-    MODE.with(|m| m.get())
-}
-
-/// Set the operating mode (used by embedded module).
-pub(crate) fn set_mode(mode: HegelMode) {
-    MODE.with(|m| m.set(mode));
-}
+use crate::protocol::{cbor_to_json, json_to_cbor, Channel, Connection};
 
 // ============================================================================
 // State Management (Thread-Local)
@@ -92,7 +63,7 @@ pub(crate) fn is_last_run() -> bool {
     IS_LAST_RUN.with(|r| r.get())
 }
 
-/// Set the is_last_run flag (used by embedded module).
+/// Set the is_last_run flag.
 pub(crate) fn set_is_last_run(is_last: bool) {
     IS_LAST_RUN.with(|r| r.set(is_last));
 }
@@ -137,90 +108,14 @@ fn is_debug() -> bool {
     std::env::var("HEGEL_DEBUG").is_ok()
 }
 
-fn get_socket_path() -> String {
-    std::env::var("HEGEL_SOCKET").expect("HEGEL_SOCKET environment variable not set")
-}
-
-/// Check if we have an active connection.
-fn is_connected() -> bool {
-    CONNECTION.with(|conn| conn.borrow().is_some())
-}
-
-/// Get the current span depth.
-fn get_span_depth() -> usize {
-    CONNECTION.with(|conn| {
-        conn.borrow()
-            .as_ref()
-            .map(|s| s.span_depth)
-            .unwrap_or(0)
-    })
-}
-
-/// Open a connection. Panics if already connected.
-pub(crate) fn open_connection() {
-    CONNECTION.with(|conn| {
-        let mut conn = conn.borrow_mut();
-        assert!(
-            conn.is_none(),
-            "open_connection called while already connected"
-        );
-
-        let socket_path = get_socket_path();
-        let stream = match UnixStream::connect(&socket_path) {
-            Ok(s) => s,
-            Err(e) => {
-                eprintln!(
-                    "Failed to connect to Hegel socket at {}: {}",
-                    socket_path, e
-                );
-                std::process::exit(exit_codes::SOCKET_ERROR);
-            }
-        };
-
-        let connection = Connection::new(stream);
-
-        // Perform version negotiation
-        if let Err(e) = negotiate_version(&connection) {
-            eprintln!("Version negotiation failed: {}", e);
-            std::process::exit(exit_codes::SOCKET_ERROR);
-        }
-
-        // Get the control channel for sending requests
-        let channel = connection.control_channel();
-
-        *conn = Some(ConnectionState {
-            connection,
-            channel,
-            span_depth: 0,
-        });
-    });
-}
-
-/// Close the connection. Panics if not connected or if spans are still open.
-pub(crate) fn close_connection() {
-    CONNECTION.with(|conn| {
-        let mut conn = conn.borrow_mut();
-        let state = conn
-            .as_ref()
-            .expect("close_connection called while not connected");
-        assert_eq!(
-            state.span_depth, 0,
-            "close_connection called with {} unclosed span(s)",
-            state.span_depth
-        );
-        *conn = None;
-    });
-}
-
-/// Set the connection from an already-connected stream (used by embedded module).
-/// This is used when the SDK creates a server and accepts connections from hegel.
+/// Set the connection for the current test case.
 /// The channel parameter is the test case channel assigned by the server.
-pub(crate) fn set_embedded_connection(connection: Arc<Connection>, channel: Channel) {
+pub(crate) fn set_connection(connection: Arc<Connection>, channel: Channel) {
     CONNECTION.with(|conn| {
         let mut conn = conn.borrow_mut();
         assert!(
             conn.is_none(),
-            "set_embedded_connection called while already connected"
+            "set_connection called while already connected"
         );
 
         *conn = Some(ConnectionState {
@@ -231,8 +126,8 @@ pub(crate) fn set_embedded_connection(connection: Arc<Connection>, channel: Chan
     });
 }
 
-/// Clear the embedded connection (used by embedded module).
-pub(crate) fn clear_embedded_connection() {
+/// Clear the connection after a test case completes.
+pub(crate) fn clear_connection() {
     CONNECTION.with(|conn| {
         *conn.borrow_mut() = None;
     });
@@ -332,19 +227,10 @@ pub(crate) fn request_from_schema(schema: &Value) -> Result<Value, StopTestError
 
 /// Generate a value from a schema.
 pub fn generate_from_schema<T: serde::de::DeserializeOwned>(schema: &Value) -> T {
-    // In embedded mode, connection is already set - don't try to open/close
-    let need_connection = !is_connected() && current_mode() == HegelMode::External;
-    if need_connection {
-        open_connection();
-    }
-
     let result = match request_from_schema(schema) {
         Ok(v) => v,
         Err(StopTestError) => {
             // Server ran out of data - reject this test case
-            if need_connection {
-                close_connection();
-            }
             crate::assume(false);
             unreachable!("assume(false) should not return")
         }
@@ -372,9 +258,6 @@ pub fn start_span(label: u64) {
     increment_span_depth();
     if let Err(StopTestError) = send_request("start_span", &json!({"label": label})) {
         decrement_span_depth();
-        if get_span_depth() == 0 && current_mode() == HegelMode::External {
-            close_connection();
-        }
         crate::assume(false);
     }
 }
@@ -387,11 +270,6 @@ pub fn stop_span(discard: bool) {
     decrement_span_depth();
     // Ignore StopTest errors from stop_span - we're already closing
     let _ = send_request("stop_span", &json!({"discard": discard}));
-    // Only close connection in external mode - in embedded mode, the
-    // connection is managed by the embedded module
-    if get_span_depth() == 0 && current_mode() == HegelMode::External {
-        close_connection();
-    }
 }
 
 // ============================================================================