fix: resolve #3

app/Services/Wire/AgentResult.php

@@ -0,0 +1,16 @@
+<?php
+
+namespace App\Services\Wire;
+
+/**
+ * Outcome of running a single command on an agent over the Coolify Wire
+ * Protocol: the captured stdout/stderr streams and the remote exit code.
+ */
+final class AgentResult
+{
+    public function __construct(
+        public readonly int $exitCode,
+        public readonly string $stdout,
+        public readonly string $stderr,
+    ) {}
+}

app/Services/Wire/CoolifyAgentClient.php

@@ -0,0 +1,197 @@
+<?php
+
+namespace App\Services\Wire;
+
+use App\Models\Server;
+
+/**
+ * Control-plane side of the Coolify Wire Protocol (CWP).
+ *
+ * Connects to a server's agent daemon over TCP and runs a command by exchanging
+ * CWP frames, replacing the per-command `ssh` process used by the SSH transport.
+ * The frame-level {@see self::exchange()} is split out from connection handling
+ * so it can be exercised over an in-memory socket pair without a live daemon.
+ */
+class CoolifyAgentClient
+{
+    /** Bytes requested per socket read while reassembling frames. */
+    private const READ_CHUNK = 65536;
+
+    /**
+     * Should this server's commands be routed through the agent transport?
+     *
+     * Opt-in and defaulting to false: unless the transport is enabled globally
+     * and on the server, callers keep using the SSH transport unchanged.
+     */
+    public static function shouldHandle(Server $server): bool
+    {
+        if (! config('constants.agent.enabled')) {
+            return false;
+        }
+
+        return (bool) data_get($server, 'settings.use_agent', false);
+    }
+
+    /**
+     * Run a command on the server's agent and return the full result.
+     *
+     * @throws WireProtocolException on connection failure or a protocol error
+     */
+    public static function execute(Server $server, string $command, int $timeout): AgentResult
+    {
+        $host = $server->ip;
+        $port = (int) config('constants.agent.port', 9000);
+        $token = (string) data_get($server, 'settings.agent_token', config('constants.agent.token', ''));
+
+        $address = str_contains($host, ':') ? "tcp://[{$host}]:{$port}" : "tcp://{$host}:{$port}";
+
+        $stream = @stream_socket_client(
+            $address,
+            $errno,
+            $errstr,
+            (float) config('constants.agent.connection_timeout', 10),
+            STREAM_CLIENT_CONNECT,
+        );
+
+        if ($stream === false) {
+            throw new WireProtocolException("Unable to reach agent at {$address}: {$errstr} ({$errno}).");
+        }
+
+        try {
+            return self::exchange($stream, $token, $command, $timeout);
+        } finally {
+            fclose($stream);
+        }
+    }
+
+    /**
+     * Perform the handshake-then-exec exchange over an already-open stream.
+     *
+     * @param  resource  $stream  A connected, blocking stream resource.
+     *
+     * @throws WireProtocolException on a malformed/unexpected frame or timeout
+     */
+    public static function exchange($stream, string $token, string $command, int $timeout): AgentResult
+    {
+        $streamId = 1;
+        $deadline = microtime(true) + max(1, $timeout);
+        $buffer = '';
+
+        self::write($stream, WireProtocol::encodeJson(WireProtocol::HELLO_TYPE, $streamId, [
+            'version' => WireProtocol::VERSION,
+            'token' => $token,
+            'features' => ['exec'],
+        ]));
+
+        $welcome = self::read($stream, $buffer, $deadline);
+
+        if ($welcome->isType(WireProtocol::ERROR_TYPE)) {
+            throw new WireProtocolException('Agent rejected handshake: '.self::errorMessage($welcome));
+        }
+
+        if (! $welcome->isType(WireProtocol::WELCOME_TYPE)) {
+            throw new WireProtocolException("Expected WELCOME frame, received type {$welcome->type}.");
+        }
+
+        self::write($stream, WireProtocol::encodeJson(WireProtocol::EXEC_TYPE, $streamId, [
+            'command' => $command,
+            'timeout' => $timeout,
+        ]));
+
+        $stdout = '';
+        $stderr = '';
+        $exitCode = null;
+
+        while ($exitCode === null) {
+            $frame = self::read($stream, $buffer, $deadline);
+
+            switch ($frame->type) {
+                case WireProtocol::STDOUT_TYPE:
+                    $stdout .= $frame->payload;
+                    break;
+                case WireProtocol::STDERR_TYPE:
+                    $stderr .= $frame->payload;
+                    break;
+                case WireProtocol::EXIT_TYPE:
+                    $exitCode = (int) ($frame->json()['code'] ?? 0);
+                    break;
+                case WireProtocol::PING_TYPE:
+                    self::write($stream, WireProtocol::encode(WireProtocol::PONG_TYPE, $frame->streamId));
+                    break;
+                case WireProtocol::ERROR_TYPE:
+                    throw new WireProtocolException('Agent error: '.self::errorMessage($frame));
+                default:
+                    throw new WireProtocolException("Unexpected frame type {$frame->type} during execution.");
+            }
+        }
+
+        return new AgentResult($exitCode, $stdout, $stderr);
+    }
+
+    /**
+     * Read bytes until one complete frame can be decoded, then consume it.
+     *
+     * @param  resource  $stream
+     *
+     * @throws WireProtocolException on timeout or a closed connection
+     */
+    private static function read($stream, string &$buffer, float $deadline): WireFrame
+    {
+        while (true) {
+            $decoded = WireProtocol::tryDecode($buffer);
+
+            if ($decoded !== null) {
+                $buffer = substr($buffer, $decoded['length']);
+
+                return $decoded['frame'];
+            }
+
+            $remaining = $deadline - microtime(true);
+
+            if ($remaining <= 0) {
+                throw new WireProtocolException('Timed out waiting for an agent frame.');
+            }
+
+            stream_set_timeout($stream, (int) ceil($remaining), (int) (fmod($remaining, 1) * 1_000_000));
+            $chunk = fread($stream, self::READ_CHUNK);
+
+            if ($chunk === false || ($chunk === '' && feof($stream))) {
+                throw new WireProtocolException('Agent connection closed before a complete frame arrived.');
+            }
+
+            $buffer .= $chunk;
+        }
+    }
+
+    /**
+     * Write a full frame, looping until every byte is flushed.
+     *
+     * @param  resource  $stream
+     *
+     * @throws WireProtocolException when the connection cannot accept the bytes
+     */
+    private static function write($stream, string $frame): void
+    {
+        $offset = 0;
+        $total = strlen($frame);
+
+        while ($offset < $total) {
+            $written = fwrite($stream, substr($frame, $offset));
+
+            if ($written === false || $written === 0) {
+                throw new WireProtocolException('Failed to write frame to agent connection.');
+            }
+
+            $offset += $written;
+        }
+    }
+
+    private static function errorMessage(WireFrame $frame): string
+    {
+        try {
+            return (string) ($frame->json()['message'] ?? 'unknown error');
+        } catch (WireProtocolException) {
+            return 'unknown error';
+        }
+    }
+}

app/Services/Wire/WireFrame.php

@@ -0,0 +1,43 @@
+<?php
+
+namespace App\Services\Wire;
+
+/**
+ * A single decoded Coolify Wire Protocol (CWP) frame.
+ *
+ * Frames are immutable value objects. Structured frames (HELLO, WELCOME, EXEC,
+ * EXIT, ERROR) carry a UTF-8 JSON payload; stream frames (STDOUT, STDERR) carry
+ * raw bytes so binary output is forwarded without base64 inflation.
+ */
+final class WireFrame
+{
+    public function __construct(
+        public readonly int $type,
+        public readonly int $streamId,
+        public readonly string $payload = '',
+        public readonly int $flags = 0,
+    ) {}
+
+    public function isType(int $type): bool
+    {
+        return $this->type === $type;
+    }
+
+    /**
+     * Decode the payload as JSON.
+     *
+     * @return array<string, mixed>
+     *
+     * @throws WireProtocolException when the payload is not a valid JSON object
+     */
+    public function json(): array
+    {
+        $decoded = json_decode($this->payload, true);
+
+        if (! is_array($decoded)) {
+            throw new WireProtocolException('Frame payload is not a JSON object.');
+        }
+
+        return $decoded;
+    }
+}