test: give smoke agent a distinct name

packaging/test/e2e_smoke.sh

@@ -4,6 +4,7 @@ set -euo pipefail
 SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
 python3 "$SCRIPT_DIR/packaging/test/test_e2e_sse.py"
 python3 "$SCRIPT_DIR/packaging/test/test_e2e_fixture.py"
+python3 "$SCRIPT_DIR/packaging/test/test_e2e_network.py"
 tmp="$(mktemp -d)"
 COLLECTOR_PID=""
 ADVISOR_STUB_PID=""
@@ -70,7 +71,7 @@ fi
 # Create a test overlay config with fast sampling interval for smoke test
 cat > "$tmp/test_config.yaml" <<EOF
 agents:
-  - { name: cursor, match: ["fake-cursor"] }
+  - { name: cursor, match: ["sa-e2e-agent"] }
 net_sample_interval_ms: 200
 socket_path: "$SOCKET_PATH"
 db_path: "$tmp/events.db"
@@ -137,7 +138,8 @@ if [ "$TRANSCRIPT_READY" != true ]; then
   exit 1
 fi
 
-# Create fake-cursor Go test agent binary to guarantee process name matching in sysctl proc list
+# Use a fixture-specific executable name, distinct from installed harnesses,
+# while retaining process name matching in the native process list.
 cat > "$tmp/fake_cursor_main.go" <<EOF
 package main
 
@@ -225,7 +227,6 @@ func main() {
 		}
 		probe.Close()
 		l = cand
-		fmt.Fprintf(os.Stderr, "network fixture listening on %s\n", l.Addr())
 		break
 	}
 	if l == nil {
@@ -245,17 +246,18 @@ func main() {
 	if err != nil {
 		panic(fmt.Errorf("establish network fixture: %w", err))
 	}
+	fmt.Fprintf(os.Stderr, "network fixture connected to %s\n", l.Addr())
 	time.Sleep(30 * time.Second)
 	close(done)
 	clientConn.Close()
 	runtime.KeepAlive(memory)
 }
 EOF
 
-go build -o "$tmp/fake-cursor" "$tmp/fake_cursor_main.go"
+go build -o "$tmp/sa-e2e-agent" "$tmp/fake_cursor_main.go"
 
 # Execute fake agent
-"$tmp/fake-cursor" "$tmp" > "$tmp/fake-agent.log" 2>&1 &
+"$tmp/sa-e2e-agent" "$tmp" > "$tmp/fake-agent.log" 2>&1 &
 AGENT_PID=$!
 FIXTURE_AGENT_PID=$AGENT_PID
 

packaging/test/test_e2e_network.py

@@ -0,0 +1,63 @@
+#!/usr/bin/env python3
+"""Build and run the smoke agent against a real local non-loopback socket."""
+import ipaddress
+import os
+import pathlib
+import re
+import select
+import subprocess
+import tempfile
+import time
+import unittest
+
+
+class NetworkFixtureTests(unittest.TestCase):
+    def test_smoke_agent_establishes_non_loopback_connection(self):
+        source = pathlib.Path(__file__).with_name("e2e_smoke.sh").read_text()
+        agent_source = source.split('cat > "$tmp/fake_cursor_main.go" <<EOF\n', 1)[1].split('\nEOF', 1)[0]
+        build = re.search(r'^go build -o "\$tmp/([^"/]+)" "\$tmp/fake_cursor_main.go"$', source, re.M)
+        self.assertIsNotNone(build, "smoke fixture build command not found")
+        binary_name = build.group(1)
+        match = re.search(r'name: cursor, match: \["([^"]+)"\]', source)
+        self.assertIsNotNone(match, "smoke fixture process matcher not found")
+        self.assertEqual(binary_name, match.group(1))
+        self.assertIn(f'"$tmp/{binary_name}" "$tmp" > "$tmp/fake-agent.log"', source)
+        with tempfile.TemporaryDirectory() as directory:
+            root = pathlib.Path(directory)
+            (root / "fake_cursor_main.go").write_text(agent_source)
+            subprocess.run(
+                ["bash", "-c", 'tmp=$1\n' + build.group(0), "network-fixture", directory],
+                check=True, capture_output=True, text=True, timeout=60,
+            )
+            process = subprocess.Popen(
+                [str(root / binary_name), directory],
+                env={**os.environ, "HOME": directory},
+                stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
+            )
+            output = b""
+            try:
+                deadline = time.monotonic() + 10
+                while time.monotonic() < deadline:
+                    readable, _, _ = select.select([process.stderr], [], [], max(0, deadline - time.monotonic()))
+                    if not readable:
+                        break
+                    chunk = os.read(process.stderr.fileno(), 8192)
+                    if not chunk:
+                        break
+                    output += chunk
+                    if b"network fixture connected to " in output:
+                        break
+                self.assertIn(b"network fixture connected to ", output, output.decode())
+                endpoint = output.decode().split("network fixture connected to ", 1)[1].splitlines()[0]
+                host = ipaddress.ip_address(endpoint.rsplit(":", 1)[0])
+                self.assertFalse(host.is_loopback, endpoint)
+                self.assertFalse(host.is_unspecified, endpoint)
+                self.assertIsNone(process.poll(), output.decode())
+            finally:
+                if process.poll() is None:
+                    process.terminate()
+                process.communicate(timeout=5)
+
+
+if __name__ == "__main__":
+    unittest.main()