#!/usr/bin/env php
<?php

if (PHP_SAPI !== 'cli') {
    fwrite(STDERR, "This agent must run from the CLI.\n");
    exit(1);
}

final class AgentConfig
{
    public function __construct(public array $data, public string $path)
    {
    }

    public static function load(string $path): self
    {
        if (!is_file($path)) {
            throw new RuntimeException('Missing config file: ' . $path);
        }

        $decoded = json_decode((string)file_get_contents($path), true);
        if (!is_array($decoded)) {
            throw new RuntimeException('Invalid config file: ' . $path);
        }

        return new self($decoded, $path);
    }

    public function get(string $key, mixed $default = null): mixed
    {
        return array_key_exists($key, $this->data) ? $this->data[$key] : $default;
    }

    public function set(string $key, mixed $value): void
    {
        $this->data[$key] = $value;
    }

    public function save(): void
    {
        file_put_contents($this->path, json_encode($this->data, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);
    }
}

final class HttpRequestTimeoutException extends RuntimeException
{
}

final class OperationAbortException extends RuntimeException
{
    public function __construct(string $message, private readonly bool $cancelled = false)
    {
        parent::__construct($message);
    }

    public function isCancellation(): bool
    {
        return $this->cancelled;
    }
}

final class HttpJsonClient
{
    public function __construct(private readonly string $baseUrl, private readonly array $defaultHeaders = [])
    {
    }

    public function post(string $path, array $payload, int $timeoutSeconds = 30): ?array
    {
        $response = $this->requestJson(
            'POST',
            rtrim($this->baseUrl, '/') . '/' . ltrim($path, '/'),
            $payload,
            $timeoutSeconds
        );

        return $response['decoded'];
    }

    public function getJson(string $url, int $timeoutSeconds = 10): array
    {
        $response = $this->requestJson('GET', $url, null, $timeoutSeconds);
        return $response['decoded'] ?? [];
    }

    public function download(string $url, int $timeoutSeconds = 60): string
    {
        $response = $this->requestRaw('GET', $url, null, $timeoutSeconds, ['Accept: */*']);
        return $response['body'];
    }

    private function requestJson(string $method, string $url, ?array $payload, int $timeoutSeconds): array
    {
        $headers = array_values(array_merge(['Accept: application/json'], $this->defaultHeaders));
        if ($payload !== null) {
            $headers[] = 'Content-Type: application/json';
        }

        $response = $this->requestRaw($method, $url, $payload, $timeoutSeconds, $headers);
        $decoded = json_decode($response['body'], true);
        if (!is_array($decoded)) {
            throw new RuntimeException('Invalid JSON response from ' . $method . ' ' . $url);
        }

        $response['decoded'] = $decoded;
        return $response;
    }

    private function requestRaw(
        string $method,
        string $url,
        ?array $payload,
        int $timeoutSeconds,
        array $headers
    ): array {
        $responseHeaders = [];
        $ch = curl_init($url);
        curl_setopt_array($ch, [
            CURLOPT_CUSTOMREQUEST => $method,
            CURLOPT_RETURNTRANSFER => true,
            CURLOPT_TIMEOUT => $timeoutSeconds,
            CURLOPT_CONNECTTIMEOUT => min(10, $timeoutSeconds),
            CURLOPT_FOLLOWLOCATION => true,
            CURLOPT_HTTPHEADER => $headers,
            CURLOPT_HEADERFUNCTION => static function ($curl, string $headerLine) use (&$responseHeaders): int {
                $trimmed = trim($headerLine);
                if ($trimmed !== '') {
                    $responseHeaders[] = $trimmed;
                }
                return strlen($headerLine);
            },
        ]);

        if ($payload !== null) {
            curl_setopt($ch, CURLOPT_POSTFIELDS, json_encode($payload, JSON_UNESCAPED_SLASHES));
        }

        $raw = curl_exec($ch);
        $status = (int)curl_getinfo($ch, CURLINFO_RESPONSE_CODE);
        $errno = curl_errno($ch);
        $error = curl_error($ch);
        curl_close($ch);

        if ($raw === false) {
            $message = sprintf('%s %s failed: %s', $method, $url, $error !== '' ? $error : 'unknown curl error');
            if ($errno === CURLE_OPERATION_TIMEDOUT) {
                throw new HttpRequestTimeoutException($message);
            }

            throw new RuntimeException($message);
        }

        if ($status >= 400) {
            $preview = substr(preg_replace('/\s+/', ' ', (string)$raw) ?? '', 0, 240);
            $headerPreview = implode('; ', array_slice($responseHeaders, 0, 8));
            throw new RuntimeException(sprintf(
                '%s %s returned HTTP %d. Headers: %s. Body preview: %s',
                $method,
                $url,
                $status,
                $headerPreview !== '' ? $headerPreview : 'none',
                $preview !== '' ? $preview : 'empty'
            ));
        }

        return [
            'status' => $status,
            'body' => (string)$raw,
            'headers' => $responseHeaders,
        ];
    }
}

final class Logger
{
    /** @var callable|null */
    private $sink = null;

    public function __construct(private readonly string $logFile)
    {
    }

    public function setSink(?callable $sink): void
    {
        $this->sink = $sink;
    }

    public function info(string $message): void
    {
        $this->write('INFO', $message);
    }

    public function warning(string $message): void
    {
        $this->write('WARNING', $message);
    }

    public function error(string $message): void
    {
        $this->write('ERROR', $message);
    }

    private function write(string $level, string $message): void
    {
        $line = '[' . date('c') . "] {$level} {$message}\n";
        file_put_contents($this->logFile, $line, FILE_APPEND);
        fwrite($level === 'ERROR' ? STDERR : STDOUT, $line);
        if ($this->sink !== null) {
            try {
                call_user_func($this->sink, $level, $message);
            } catch (Throwable) {
                // Logging must never break the agent loop.
            }
        }
    }
}

final class LocalStateStore
{
    private SQLite3 $db;

    public function __construct(string $path)
    {
        if (!class_exists('SQLite3')) {
            throw new RuntimeException('The edge-agent container requires sqlite3 support.');
        }

        $directory = dirname($path);
        if (!is_dir($directory)) {
            @mkdir($directory, 0777, true);
        }

        $this->db = new SQLite3($path);
        $this->db->busyTimeout(5000);
        $this->db->exec('PRAGMA journal_mode = WAL;');
        $this->db->exec('PRAGMA synchronous = NORMAL;');
        $this->db->exec(
            'CREATE TABLE IF NOT EXISTS kv (
                key TEXT PRIMARY KEY,
                value_json TEXT NOT NULL,
                updated_at TEXT NOT NULL
            )'
        );
        $this->db->exec(
            'CREATE TABLE IF NOT EXISTS outbox (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                item_type TEXT NOT NULL,
                endpoint TEXT NOT NULL,
                payload_json TEXT NOT NULL,
                created_at TEXT NOT NULL
            )'
        );
    }

    public function getJson(string $key, mixed $default = null): mixed
    {
        $statement = $this->db->prepare('SELECT value_json FROM kv WHERE key = :key LIMIT 1');
        $statement->bindValue(':key', $key, SQLITE3_TEXT);
        $result = $statement->execute();
        $row = $result instanceof SQLite3Result ? $result->fetchArray(SQLITE3_ASSOC) : false;
        if (!is_array($row) || !array_key_exists('value_json', $row)) {
            return $default;
        }

        $decoded = json_decode((string)$row['value_json'], true);
        return json_last_error() === JSON_ERROR_NONE ? $decoded : $default;
    }

    public function setJson(string $key, mixed $value): void
    {
        $statement = $this->db->prepare(
            'INSERT INTO kv (key, value_json, updated_at)
             VALUES (:key, :value_json, :updated_at)
             ON CONFLICT(key) DO UPDATE SET value_json = excluded.value_json, updated_at = excluded.updated_at'
        );
        $statement->bindValue(':key', $key, SQLITE3_TEXT);
        $statement->bindValue(':value_json', json_encode($value, JSON_UNESCAPED_SLASHES), SQLITE3_TEXT);
        $statement->bindValue(':updated_at', date('c'), SQLITE3_TEXT);
        $statement->execute();
    }

    public function enqueue(string $type, string $endpoint, array $payload): void
    {
        $statement = $this->db->prepare(
            'INSERT INTO outbox (item_type, endpoint, payload_json, created_at)
             VALUES (:item_type, :endpoint, :payload_json, :created_at)'
        );
        $statement->bindValue(':item_type', $type, SQLITE3_TEXT);
        $statement->bindValue(':endpoint', $endpoint, SQLITE3_TEXT);
        $statement->bindValue(':payload_json', json_encode($payload, JSON_UNESCAPED_SLASHES), SQLITE3_TEXT);
        $statement->bindValue(':created_at', date('c'), SQLITE3_TEXT);
        $statement->execute();
    }

    /**
     * @return array<int,array<string,mixed>>
     */
    public function queuedItems(int $limit = 25): array
    {
        $statement = $this->db->prepare(
            'SELECT id, item_type, endpoint, payload_json, created_at
             FROM outbox
             ORDER BY id ASC
             LIMIT :limit'
        );
        $statement->bindValue(':limit', max(1, $limit), SQLITE3_INTEGER);
        $result = $statement->execute();
        $items = [];
        if (!$result instanceof SQLite3Result) {
            return $items;
        }

        while (($row = $result->fetchArray(SQLITE3_ASSOC)) !== false) {
            $items[] = [
                'id' => (int)$row['id'],
                'type' => (string)$row['item_type'],
                'endpoint' => (string)$row['endpoint'],
                'payload' => json_decode((string)$row['payload_json'], true) ?: [],
                'created_at' => (string)$row['created_at'],
            ];
        }

        return $items;
    }

    public function removeOutboxItem(int $id): void
    {
        $statement = $this->db->prepare('DELETE FROM outbox WHERE id = :id');
        $statement->bindValue(':id', $id, SQLITE3_INTEGER);
        $statement->execute();
    }

    public function outboxSummary(): array
    {
        $countResult = $this->db->querySingle('SELECT COUNT(*) FROM outbox');
        $oldestResult = $this->db->querySingle('SELECT created_at FROM outbox ORDER BY id ASC LIMIT 1');
        $lastSyncAt = $this->getJson('last_sync_at');

        return [
            'queued' => (int)$countResult,
            'oldest_queued_at' => is_string($oldestResult) && $oldestResult !== '' ? $oldestResult : null,
            'last_replayed_at' => is_string($lastSyncAt) && $lastSyncAt !== '' ? $lastSyncAt : null,
        ];
    }
}

final class BrokerWebSocketClient
{
    private const CONNECT_TIMEOUT_SECONDS = 15;
    private const RECONNECT_DELAY_SECONDS = 2;

    /** @var resource|null */
    private $socket = null;
    private ?string $brokerUrl = null;
    private ?int $gatewayId = null;
    private ?string $agentToken = null;
    private ?string $agentInstanceId = null;
    private string $readBuffer = '';
    private int $reconnectAfterEpoch = 0;
    private array $state = [
        'url' => null,
        'connected' => false,
        'lastError' => null,
        'disconnectReason' => null,
        'lastConnectedAt' => null,
        'lastDisconnectedAt' => null,
    ];

    public function __construct(private readonly Logger $logger)
    {
    }

    public function configure(?string $brokerUrl, ?int $gatewayId, ?string $agentToken, ?string $agentInstanceId): void
    {
        $normalizedUrl = $this->normalizeBrokerBaseUrl($brokerUrl);
        $gatewayId = $gatewayId !== null && $gatewayId > 0 ? $gatewayId : null;
        $agentToken = trim((string)($agentToken ?? '')) !== '' ? trim((string)$agentToken) : null;
        $agentInstanceId = trim((string)($agentInstanceId ?? '')) !== '' ? trim((string)$agentInstanceId) : null;

        $didChange = $this->brokerUrl !== $normalizedUrl
            || $this->gatewayId !== $gatewayId
            || $this->agentToken !== $agentToken
            || $this->agentInstanceId !== $agentInstanceId;

        $this->brokerUrl = $normalizedUrl;
        $this->gatewayId = $gatewayId;
        $this->agentToken = $agentToken;
        $this->agentInstanceId = $agentInstanceId;
        $this->state['url'] = $normalizedUrl;

        if ($didChange && $this->socket !== null) {
            $this->disconnect('config_changed');
        }
    }

    public function isConnected(): bool
    {
        return is_resource($this->socket) && ($this->state['connected'] ?? false) === true;
    }

    public function state(): array
    {
        return $this->state;
    }

    public function ensureConnected(): bool
    {
        if ($this->isConnected()) {
            return true;
        }

        if ($this->brokerUrl === null || $this->gatewayId === null || $this->agentToken === null) {
            return false;
        }

        if (time() < $this->reconnectAfterEpoch) {
            return false;
        }

        try {
            $this->connect();
            return true;
        } catch (Throwable $throwable) {
            $this->state['connected'] = false;
            $this->state['lastError'] = $throwable->getMessage();
            $this->state['disconnectReason'] = 'connect_failed';
            $this->state['lastDisconnectedAt'] = date('c');
            $this->reconnectAfterEpoch = time() + self::RECONNECT_DELAY_SECONDS;
            return false;
        }
    }

    public function disconnect(string $reason = 'closed'): void
    {
        if (is_resource($this->socket)) {
            @fclose($this->socket);
        }

        $this->socket = null;
        $this->readBuffer = '';
        $this->state['connected'] = false;
        $this->state['disconnectReason'] = $reason;
        $this->state['lastDisconnectedAt'] = date('c');
        $this->reconnectAfterEpoch = time() + self::RECONNECT_DELAY_SECONDS;
    }

    public function send(array $message): bool
    {
        if (!$this->ensureConnected()) {
            return false;
        }

        try {
            $this->writeFrame(json_encode($message, JSON_UNESCAPED_SLASHES));
            return true;
        } catch (Throwable $throwable) {
            $this->state['lastError'] = $throwable->getMessage();
            $this->disconnect('send_failed');
            return false;
        }
    }

    /**
     * @return array<int,array<string,mixed>>
     */
    public function readMessages(int $maxMessages = 16): array
    {
        if (!$this->ensureConnected()) {
            return [];
        }

        try {
            $this->readAvailableBytes();
        } catch (Throwable $throwable) {
            $this->state['lastError'] = $throwable->getMessage();
            $this->disconnect('read_failed');
            return [];
        }

        $messages = [];
        for ($index = 0; $index < max(1, $maxMessages); $index++) {
            $frame = $this->extractFrame();
            if ($frame === null) {
                break;
            }

            $opcode = (int)($frame['opcode'] ?? 0x1);
            $payload = (string)($frame['payload'] ?? '');

            if ($opcode === 0x8) {
                $this->disconnect('server_closed');
                break;
            }

            if ($opcode === 0x9) {
                try {
                    $this->writeFrame($payload, 0xA);
                } catch (Throwable) {
                    $this->disconnect('pong_failed');
                }
                continue;
            }

            if ($opcode !== 0x1 || trim($payload) === '') {
                continue;
            }

            $decoded = json_decode($payload, true);
            if (is_array($decoded)) {
                $messages[] = $decoded;
            }
        }

        return $messages;
    }

    private function connect(): void
    {
        $parts = $this->buildSocketParts();
        $transport = ($parts['scheme'] ?? 'ws') === 'wss' ? 'ssl' : 'tcp';
        $host = (string)$parts['host'];
        $port = (int)$parts['port'];
        $path = (string)$parts['path'];

        $socket = @stream_socket_client(
            sprintf('%s://%s:%d', $transport, $host, $port),
            $errno,
            $error,
            self::CONNECT_TIMEOUT_SECONDS,
            STREAM_CLIENT_CONNECT
        );
        if (!is_resource($socket)) {
            throw new RuntimeException(sprintf(
                'Unable to connect to broker %s:%d: %s',
                $host,
                $port,
                trim((string)$error) !== '' ? trim((string)$error) : 'connection refused'
            ));
        }

        stream_set_timeout($socket, self::CONNECT_TIMEOUT_SECONDS);
        stream_set_blocking($socket, true);

        $key = base64_encode(random_bytes(16));
        $request = implode("\r\n", [
            'GET ' . $path . ' HTTP/1.1',
            'Host: ' . $host . ($this->isDefaultPort($parts) ? '' : ':' . $port),
            'Upgrade: websocket',
            'Connection: Upgrade',
            'Sec-WebSocket-Version: 13',
            'Sec-WebSocket-Key: ' . $key,
            "\r\n",
        ]);

        $written = fwrite($socket, $request);
        if ($written === false || $written < strlen($request)) {
            fclose($socket);
            throw new RuntimeException('Unable to write broker handshake request');
        }

        $response = '';
        $startedAt = microtime(true);
        while (!str_contains($response, "\r\n\r\n")) {
            $chunk = fread($socket, 2048);
            if ($chunk === false || $chunk === '') {
                if ((microtime(true) - $startedAt) >= self::CONNECT_TIMEOUT_SECONDS) {
                    fclose($socket);
                    throw new RuntimeException('Timed out while reading broker handshake response');
                }
                usleep(25000);
                continue;
            }

            $response .= $chunk;
        }

        if (!preg_match('/\AHTTP\/1\.[01]\s+101\b/i', $response)) {
            fclose($socket);
            throw new RuntimeException('Broker rejected websocket upgrade: ' . trim(strtok($response, "\r\n")));
        }

        $expectedAccept = base64_encode(
            sha1($key . '258EAFA5-E914-47DA-95CA-C5AB0DC85B11', true)
        );
        if (!preg_match('/Sec-WebSocket-Accept:\s*(.+)\r\n/i', $response, $matches)) {
            fclose($socket);
            throw new RuntimeException('Broker handshake missing Sec-WebSocket-Accept header');
        }

        if (!hash_equals($expectedAccept, trim((string)$matches[1]))) {
            fclose($socket);
            throw new RuntimeException('Broker handshake validation failed');
        }

        stream_set_blocking($socket, false);
        stream_set_timeout($socket, 0);
        $this->socket = $socket;
        $this->readBuffer = '';
        $this->reconnectAfterEpoch = 0;
        $this->state['connected'] = true;
        $this->state['lastError'] = null;
        $this->state['disconnectReason'] = null;
        $this->state['lastConnectedAt'] = date('c');
    }

    private function readAvailableBytes(): void
    {
        if (!is_resource($this->socket)) {
            return;
        }

        while (true) {
            $chunk = @fread($this->socket, 8192);
            if ($chunk === false) {
                throw new RuntimeException('Unable to read from broker socket');
            }

            if ($chunk === '') {
                if (feof($this->socket)) {
                    $this->disconnect('server_eof');
                }
                break;
            }

            $this->readBuffer .= $chunk;
            if (strlen($chunk) < 8192) {
                break;
            }
        }
    }

    private function extractFrame(): ?array
    {
        $length = strlen($this->readBuffer);
        if ($length < 2) {
            return null;
        }

        $first = ord($this->readBuffer[0]);
        $second = ord($this->readBuffer[1]);
        $opcode = $first & 0x0F;
        $masked = ($second & 0x80) === 0x80;
        $payloadLength = $second & 0x7F;
        $offset = 2;

        if ($payloadLength === 126) {
            if ($length < 4) {
                return null;
            }
            $payloadLength = unpack('n', substr($this->readBuffer, $offset, 2))[1];
            $offset += 2;
        } elseif ($payloadLength === 127) {
            if ($length < 10) {
                return null;
            }
            $parts = unpack('Nhigh/Nlow', substr($this->readBuffer, $offset, 8));
            $payloadLength = ((int)$parts['high'] << 32) | (int)$parts['low'];
            $offset += 8;
        }

        $maskingKey = '';
        if ($masked) {
            if ($length < ($offset + 4)) {
                return null;
            }
            $maskingKey = substr($this->readBuffer, $offset, 4);
            $offset += 4;
        }

        if ($length < ($offset + $payloadLength)) {
            return null;
        }

        $payload = substr($this->readBuffer, $offset, $payloadLength);
        $this->readBuffer = (string)substr($this->readBuffer, $offset + $payloadLength);

        if ($masked) {
            $payload = $this->unmaskPayload($payload, $maskingKey);
        }

        return [
            'opcode' => $opcode,
            'payload' => $payload,
        ];
    }

    private function writeFrame(string $payload, int $opcode = 0x1): void
    {
        if (!is_resource($this->socket)) {
            throw new RuntimeException('Broker socket is not connected');
        }

        $frame = chr(0x80 | ($opcode & 0x0F));
        $length = strlen($payload);
        if ($length < 126) {
            $frame .= chr(0x80 | $length);
        } elseif ($length <= 0xFFFF) {
            $frame .= chr(0x80 | 126) . pack('n', $length);
        } else {
            $frame .= chr(0x80 | 127) . pack('NN', 0, $length);
        }

        $mask = random_bytes(4);
        $frame .= $mask . $this->maskPayload($payload, $mask);

        $written = @fwrite($this->socket, $frame);
        if ($written === false || $written < strlen($frame)) {
            throw new RuntimeException('Unable to write websocket frame to broker');
        }
    }

    private function buildSocketParts(): array
    {
        $url = $this->brokerUrl;
        if ($url === null || $this->gatewayId === null || $this->agentToken === null) {
            throw new RuntimeException('Broker websocket is not configured');
        }

        $query = http_build_query([
            'gatewayId' => $this->gatewayId,
            'token' => $this->agentToken,
            'agentInstanceId' => $this->agentInstanceId,
        ]);
        $socketUrl = rtrim($url, '/') . '/ws/agent?' . $query;
        $parts = parse_url($socketUrl);
        if (!is_array($parts) || empty($parts['host'])) {
            throw new RuntimeException('Invalid broker websocket URL: ' . $socketUrl);
        }

        return [
            'scheme' => strtolower((string)($parts['scheme'] ?? 'ws')),
            'host' => (string)$parts['host'],
            'port' => (int)($parts['port'] ?? (((string)($parts['scheme'] ?? 'ws')) === 'wss' ? 443 : 80)),
            'path' => (string)($parts['path'] ?? '/')
                . (isset($parts['query']) && trim((string)$parts['query']) !== '' ? '?' . $parts['query'] : ''),
        ];
    }

    private function isDefaultPort(array $parts): bool
    {
        $scheme = (string)($parts['scheme'] ?? 'ws');
        $port = (int)($parts['port'] ?? 0);
        return ($scheme === 'wss' && $port === 443) || ($scheme !== 'wss' && $port === 80);
    }

    private function normalizeBrokerBaseUrl(?string $value): ?string
    {
        $trimmed = rtrim(trim((string)($value ?? '')), '/');
        if ($trimmed === '') {
            return null;
        }

        if (str_starts_with($trimmed, 'ws://') || str_starts_with($trimmed, 'wss://')) {
            return $trimmed;
        }

        if (str_starts_with($trimmed, 'https://')) {
            return 'wss://' . substr($trimmed, 8);
        }

        if (str_starts_with($trimmed, 'http://')) {
            return 'ws://' . substr($trimmed, 7);
        }

        return 'ws://' . $trimmed;
    }

    private function maskPayload(string $payload, string $mask): string
    {
        $length = strlen($payload);
        $result = '';
        for ($index = 0; $index < $length; $index++) {
            $result .= chr(ord($payload[$index]) ^ ord($mask[$index % 4]));
        }
        return $result;
    }

    private function unmaskPayload(string $payload, string $mask): string
    {
        return $this->maskPayload($payload, $mask);
    }
}

final class AgentShellBridge
{
    /** @var array<string,array<string,mixed>> */
    private array $sessions = [];

    public function __construct(private readonly string $defaultCwd)
    {
    }

    public function open(array $payload, callable $send): void
    {
        $sessionId = trim((string)($payload['sessionId'] ?? ''));
        if ($sessionId === '') {
            return;
        }

        $this->close(['sessionId' => $sessionId], $send);

        $cwd = trim((string)($payload['cwd'] ?? $this->defaultCwd));
        if ($cwd === '' || !is_dir($cwd)) {
            $cwd = $this->defaultCwd;
        }

        [$command, $args] = $this->defaultShellCommand(
            trim((string)($payload['shellCommand'] ?? '')),
            isset($payload['shellArgs']) && is_array($payload['shellArgs']) ? (array)$payload['shellArgs'] : []
        );

        $descriptorSpec = [
            0 => ['pipe', 'r'],
            1 => ['pipe', 'w'],
            2 => ['pipe', 'w'],
        ];

        $process = @proc_open(
            array_merge([$command], $args),
            $descriptorSpec,
            $pipes,
            $cwd,
            array_merge($_ENV, ['TERM' => 'xterm-256color'])
        );

        if (!is_resource($process)) {
            $send([
                'type' => 'SHELL_OUTPUT',
                'sessionId' => $sessionId,
                'data' => "Failed to start root shell.\r\n",
            ]);
            $send([
                'type' => 'SHELL_EXIT',
                'sessionId' => $sessionId,
                'code' => 1,
            ]);
            return;
        }

        foreach ($pipes as $pipe) {
            stream_set_blocking($pipe, false);
        }

        $this->sessions[$sessionId] = [
            'process' => $process,
            'pipes' => $pipes,
        ];

        $send([
            'type' => 'SHELL_OPENED',
            'sessionId' => $sessionId,
        ]);
        $send([
            'type' => 'SHELL_OUTPUT',
            'sessionId' => $sessionId,
            'data' => sprintf(
                "Connected to gateway host shell.\r\nWorking directory: %s\r\nShortcuts: docker ps | systemctl status truckwash-edge-gateway-stack.service | ./gateway-launcher.sh reconcile\r\n\r\n",
                $cwd
            ),
        ]);
    }

    public function input(array $payload): void
    {
        $sessionId = trim((string)($payload['sessionId'] ?? ''));
        if ($sessionId === '' || !isset($this->sessions[$sessionId])) {
            return;
        }

        $stdin = $this->sessions[$sessionId]['pipes'][0] ?? null;
        if (is_resource($stdin)) {
            @fwrite($stdin, (string)($payload['data'] ?? ''));
        }
    }

    public function resize(array $payload): void
    {
        $sessionId = trim((string)($payload['sessionId'] ?? ''));
        if ($sessionId === '' || !isset($this->sessions[$sessionId])) {
            return;
        }

        $this->sessions[$sessionId]['cols'] = (int)($payload['cols'] ?? 0);
        $this->sessions[$sessionId]['rows'] = (int)($payload['rows'] ?? 0);
    }

    public function close(array $payload, callable $send): void
    {
        $sessionId = trim((string)($payload['sessionId'] ?? ''));
        if ($sessionId === '' || !isset($this->sessions[$sessionId])) {
            return;
        }

        $session = $this->sessions[$sessionId];
        $process = $session['process'] ?? null;
        if (is_resource($process)) {
            @proc_terminate($process);
        }

        $this->finalizeSession($sessionId, $send);
    }

    public function pump(callable $send): void
    {
        foreach (array_keys($this->sessions) as $sessionId) {
            $session = $this->sessions[$sessionId];
            $stdout = $session['pipes'][1] ?? null;
            $stderr = $session['pipes'][2] ?? null;

            foreach ([$stdout, $stderr] as $pipe) {
                if (!is_resource($pipe)) {
                    continue;
                }

                $chunk = stream_get_contents($pipe);
                if ($chunk !== false && $chunk !== '') {
                    $send([
                        'type' => 'SHELL_OUTPUT',
                        'sessionId' => $sessionId,
                        'data' => $chunk,
                    ]);
                }
            }

            $process = $session['process'] ?? null;
            if (!is_resource($process)) {
                $this->finalizeSession($sessionId, $send);
                continue;
            }

            $status = proc_get_status($process);
            if (!is_array($status) || ($status['running'] ?? false) === true) {
                continue;
            }

            $this->finalizeSession($sessionId, $send, (int)($status['exitcode'] ?? 0));
        }
    }

    public function dispose(callable $send): void
    {
        foreach (array_keys($this->sessions) as $sessionId) {
            $this->close(['sessionId' => $sessionId], $send);
        }
    }

    /**
     * @return array{0:string,1:array<int,string>}
     */
    private function defaultShellCommand(string $command, array $args): array
    {
        if ($command !== '') {
            return [$command, array_values(array_map('strval', $args))];
        }

        if (DIRECTORY_SEPARATOR === '\\') {
            return ['cmd.exe', ['/Q']];
        }

        return ['/bin/bash', ['-l']];
    }

    private function finalizeSession(string $sessionId, callable $send, int $exitCode = 0): void
    {
        if (!isset($this->sessions[$sessionId])) {
            return;
        }

        $session = $this->sessions[$sessionId];
        foreach ((array)($session['pipes'] ?? []) as $pipe) {
            if (is_resource($pipe)) {
                @fclose($pipe);
            }
        }

        $process = $session['process'] ?? null;
        if (is_resource($process)) {
            $status = proc_get_status($process);
            $exitCode = is_array($status) && isset($status['exitcode']) ? (int)$status['exitcode'] : $exitCode;
            @proc_close($process);
        }

        unset($this->sessions[$sessionId]);
        $send([
            'type' => 'SHELL_EXIT',
            'sessionId' => $sessionId,
            'code' => $exitCode >= 0 ? $exitCode : 0,
        ]);
    }
}

final class TruckwashEdgeAgent
{
    private const DEFAULT_INSTALL_DIR = '/opt/truckwash-edge-agent';
    private const DEFAULT_RUNTIME_DIR = '/opt/truckwash-edge-agent/runtime';
    private const DEFAULT_STATE_DATABASE = '/opt/truckwash-edge-agent/runtime/gateway-state.sqlite';
    private const DEFAULT_WORKER_BASE_URL = 'http://lan-worker:8090';
    private const DEFAULT_UPDATE_WINDOW = '02:00-04:00';
    private const OPERATION_COMPLETE_TIMEOUT_SECONDS = 120;
    private const BROKER_MESSAGE_PUMP_LIMIT = 12;
    private const LOOP_STALE_AFTER_SECONDS = 30;
    private const CONTROL_PLANE_SYNC_STALE_AFTER_SECONDS = 90;

    private AgentConfig $config;
    private HttpJsonClient $http;
    private HttpJsonClient $workerHttp;
    private Logger $logger;
    private LocalStateStore $stateStore;
    private BrokerWebSocketClient $brokerClient;
    private AgentShellBridge $shellBridge;
    private string $installDir;
    private string $runtimeDir;
    private string $statePath;
    private string $lastOperationSnapshotPath;
    private string $lastHeartbeatMarkerPath;
    private string $controlPlaneStatusPath;
    private string $stagedUpdatePath;
    private int $lastHeartbeatAt = 0;
    private int $lastMachineSignalPollAt = 0;
    private int $lastMachineSignalMonitorRefreshAt = 0;
    private ?array $lastControlPlaneResponse = null;
    private string $agentInstanceId;

    public function __construct(string $configPath)
    {
        $this->config = AgentConfig::load($configPath);
        $this->installDir = rtrim((string)$this->config->get('installDir', self::DEFAULT_INSTALL_DIR), DIRECTORY_SEPARATOR);
        $this->runtimeDir = rtrim((string)$this->config->get('runtimeDir', self::DEFAULT_RUNTIME_DIR), DIRECTORY_SEPARATOR);
        if (!is_dir($this->runtimeDir)) {
            @mkdir($this->runtimeDir, 0777, true);
        }

        $this->http = new HttpJsonClient((string)$this->config->get('apiUrl'));
        $this->workerHttp = new HttpJsonClient(
            (string)$this->config->get('workerBaseUrl', self::DEFAULT_WORKER_BASE_URL),
            $this->workerAuthorizationHeaders()
        );
        $this->logger = new Logger($this->runtimeDir . DIRECTORY_SEPARATOR . 'agent.log');
        $this->stateStore = new LocalStateStore((string)$this->config->get('stateDatabasePath', self::DEFAULT_STATE_DATABASE));
        $this->statePath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'current-operation.json';
        $this->lastOperationSnapshotPath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'last-operation.json';
        $this->lastHeartbeatMarkerPath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'last-heartbeat-ok.txt';
        $this->controlPlaneStatusPath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'control-plane-status.json';
        $this->stagedUpdatePath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'staged-update.json';
        $this->agentInstanceId = $this->ensureAgentInstanceId();
        $this->brokerClient = new BrokerWebSocketClient($this->logger);
        $this->shellBridge = new AgentShellBridge($this->installDir);
        $this->logger->setSink(fn(string $level, string $message): bool => $this->emitLogFrame($level, $message));
        $this->configureBrokerClient();
        $this->initializeControlPlaneStatus();
    }

    public function run(): void
    {
        $this->logger->info('Truckwash compose edge-agent starting.');
        $this->recoverPreviousOperationState();

        while (true) {
            try {
                $this->touchLoopHeartbeat();
                $this->reloadConfigFromDisk();
                $this->ensureClaimed();
                $this->configureBrokerClient();
                $this->flushOutbox();
                $this->pumpBrokerTransport();
                $this->heartbeat();
                $this->pollMachineStartSignals();
                if ($this->resumePendingOperationCompletion()) {
                    $this->pumpBrokerTransport();
                    continue;
                }
                $processedManagementOperation = false;
                if (!$this->isBrokerConnected()) {
                    $processedManagementOperation = $this->processManagementOperation();
                }
                if (!$processedManagementOperation && !$this->isBrokerConnected()) {
                    $this->processCommandQueue();
                }
                $this->pumpBrokerTransport();
                $this->flushOutbox();
            } catch (Throwable $throwable) {
                $this->logger->error($throwable->getMessage());
                sleep(2);
            }

            usleep(250000);
        }
    }

    private function ensureClaimed(): void
    {
        if ($this->config->get('gatewayId') && $this->config->get('agentToken')) {
            return;
        }

        $installedVersion = (string)$this->config->get('installedVersion', 'compose-php-agent-v2');
        try {
            $response = $this->http->post('/edge-agent/claim', [
                'token' => (string)$this->config->get('installToken'),
                'hostname' => gethostname() ?: 'truckwash-edge',
                'installed_version' => $installedVersion,
                'metadata' => [
                    'runtime' => 'compose-php',
                    'runtime_mode' => 'compose',
                    'php_version' => PHP_VERSION,
                    'agent_instance_id' => $this->agentInstanceId,
                    'update_window' => (string)$this->config->get('updateWindow', self::DEFAULT_UPDATE_WINDOW),
                    'container_health' => $this->buildContainerHealth(),
                    'outbox_status' => $this->buildOutboxStatus(),
                    'last_sync_at' => $this->currentLastSuccessfulSyncAt(),
                    'control_plane_status' => $this->buildControlPlaneStatusPayload(),
                    'rollback_status' => $this->readRollbackStatus(),
                    'staged_version' => $this->currentStagedUpdate(),
                ],
            ]);
        } catch (Throwable $throwable) {
            $this->recordTransportFailure('Gateway claim failed', $throwable);
            throw $throwable;
        }

        $payload = $response['data'] ?? [];
        $gateway = is_array($payload['gateway'] ?? null) ? (array)$payload['gateway'] : [];
        $this->config->set('gatewayId', $gateway['id'] ?? null);
        $this->config->set('agentToken', $payload['agent_token'] ?? null);
        if (!empty($payload['broker_url'])) {
            $this->config->set('brokerUrl', (string)$payload['broker_url']);
        }
        $this->config->set('installedVersion', $installedVersion);
        $this->config->set('targetVersion', (string)($gateway['target_version'] ?? $installedVersion));
        $this->config->set('agentInstanceId', $this->agentInstanceId);
        $this->config->save();
        $this->recordSuccessfulSync();
        $this->logger->info('Claimed gateway ' . (string)($gateway['id'] ?? 'unknown') . '.');
    }

    private function configureBrokerClient(): void
    {
        $this->brokerClient->configure(
            $this->config->get('brokerUrl'),
            (int)$this->config->get('gatewayId', 0),
            $this->config->get('agentToken'),
            $this->agentInstanceId
        );
    }

    private function isBrokerConnected(): bool
    {
        return $this->brokerClient->isConnected();
    }

    private function pumpBrokerTransport(): void
    {
        $this->configureBrokerClient();
        $this->brokerClient->ensureConnected();
        $messages = $this->brokerClient->readMessages(self::BROKER_MESSAGE_PUMP_LIMIT);
        foreach ($messages as $message) {
            $this->handleBrokerMessage($message);
        }

        $this->shellBridge->pump(fn(array $message): bool => $this->sendBrokerMessage($message));
    }

    private function sendBrokerMessage(array $message): bool
    {
        $sent = $this->brokerClient->send($message);
        if ($sent) {
            $this->recordSuccessfulSync();
        }
        return $sent;
    }

    private function handleBrokerMessage(array $message): void
    {
        $type = strtoupper(trim((string)($message['type'] ?? '')));
        if ($type === '') {
            return;
        }

        switch ($type) {
            case 'COMMAND':
                $this->processBrokerCommand($message);
                return;

            case 'TASK_DISPATCH':
                $taskType = strtoupper(trim((string)($message['taskType'] ?? '')));
                if ($taskType === 'OPERATION' && isset($message['operation']) && is_array($message['operation'])) {
                    $this->executeManagementOperation((array)$message['operation']);
                }
                return;

            case 'TASK_CANCEL':
                $taskType = strtoupper(trim((string)($message['taskType'] ?? '')));
                if ($taskType === 'OPERATION') {
                    $operation = isset($message['operation']) && is_array($message['operation']) ? (array)$message['operation'] : [];
                    $operationId = (int)($operation['id'] ?? 0);
                    if ($operationId > 0) {
                        $this->markOperationCancellationRequested($operationId);
                        $this->logger->warning('Cancellation requested for operation ' . $operationId . ' via broker.');
                    }
                }
                return;

            case 'OPEN_ROOT_SHELL':
                $this->shellBridge->open(
                    isset($message['payload']) && is_array($message['payload']) ? (array)$message['payload'] : [],
                    fn(array $payload): bool => $this->sendBrokerMessage($payload)
                );
                return;

            case 'SHELL_INPUT':
                $this->shellBridge->input(
                    isset($message['payload']) && is_array($message['payload']) ? (array)$message['payload'] : []
                );
                return;

            case 'RESIZE_ROOT_SHELL':
                $this->shellBridge->resize(
                    isset($message['payload']) && is_array($message['payload']) ? (array)$message['payload'] : []
                );
                return;

            case 'CLOSE_ROOT_SHELL':
                $this->shellBridge->close(
                    isset($message['payload']) && is_array($message['payload']) ? (array)$message['payload'] : [],
                    fn(array $payload): bool => $this->sendBrokerMessage($payload)
                );
                return;

            default:
                return;
        }
    }

    private function heartbeat(bool $force = false, array $metadata = []): void
    {
        $interval = (int)$this->config->get('heartbeatIntervalSeconds', 15);
        if (!$force && (time() - $this->lastHeartbeatAt) < max(5, $interval)) {
            return;
        }

        $gatewayId = (int)$this->config->get('gatewayId');
        if ($gatewayId <= 0) {
            return;
        }

        $this->recordHeartbeatAttempt();
        $operationState = $this->readOperationState();
        $payload = [
            'agent_token' => (string)$this->config->get('agentToken'),
            'status' => 'ONLINE',
            'hostname' => gethostname() ?: 'truckwash-edge',
            'installed_version' => (string)$this->config->get('installedVersion', 'compose-php-agent-v2'),
            'target_version' => (string)$this->config->get(
                'targetVersion',
                $this->config->get('installedVersion', 'compose-php-agent-v2')
            ),
            'metadata' => array_merge([
                'agent_instance_id' => $this->agentInstanceId,
                'runtime' => 'compose-php',
                'runtime_mode' => 'compose',
                'current_operation' => $operationState,
                'system_metrics' => $this->buildSystemMetrics(),
                'container_health' => $this->buildContainerHealth(),
                'outbox_status' => $this->buildOutboxStatus(),
                'last_sync_at' => $this->currentLastSuccessfulSyncAt(),
                'control_plane_status' => $this->buildControlPlaneStatusPayload(),
                'update_window' => (string)$this->config->get('updateWindow', self::DEFAULT_UPDATE_WINDOW),
                'staged_version' => $this->currentStagedUpdate(),
                'rollback_status' => $this->readRollbackStatus(),
                'command_transport' => $this->isBrokerConnected() ? 'BROKER_WS' : 'API_POLLING',
                'broker_connected' => $this->isBrokerConnected(),
                'broker_url' => $this->brokerClient->state()['url'] ?? $this->config->get('brokerUrl'),
                'broker_last_error' => $this->brokerClient->state()['lastError'] ?? null,
                'broker_last_connected_at' => $this->brokerClient->state()['lastConnectedAt'] ?? null,
                'broker_last_disconnected_at' => $this->brokerClient->state()['lastDisconnectedAt'] ?? null,
                'broker_disconnect_reason' => $this->brokerClient->state()['disconnectReason'] ?? null,
            ], $metadata),
        ];

        $posted = $this->sendControlPlaneEvent(
            '/edge-agent/gateways/' . $gatewayId . '/heartbeat',
            $payload,
            'heartbeat'
        );
        if (!$posted) {
            return;
        }

        $heartbeatSucceededAt = date('c');
        $this->applyBrokerUrlFromControlPlaneResponse($this->lastControlPlaneResponse);
        $this->lastHeartbeatAt = time();
        $this->recordSuccessfulSync($heartbeatSucceededAt, [
            'last_heartbeat_success_at' => $heartbeatSucceededAt,
        ]);
        file_put_contents($this->lastHeartbeatMarkerPath, json_encode([
            'gateway_id' => $gatewayId,
            'at' => $heartbeatSucceededAt,
            'agent_instance_id' => $this->agentInstanceId,
        ], JSON_UNESCAPED_SLASHES) . PHP_EOL);
    }

    private function pollMachineStartSignals(): void
    {
        $interval = max(1, (int)$this->config->get('machineSignalPollIntervalSeconds', 2));
        if ((time() - $this->lastMachineSignalPollAt) < $interval) {
            return;
        }
        $this->lastMachineSignalPollAt = time();

        $gatewayId = (int)$this->config->get('gatewayId');
        $agentToken = (string)$this->config->get('agentToken');
        if ($gatewayId <= 0 || trim($agentToken) === '') {
            return;
        }

        foreach ($this->machineSignalMonitors($gatewayId, $agentToken) as $monitor) {
            $localIp = trim((string)($monitor['local_ip'] ?? ''));
            if ($localIp === '') {
                continue;
            }

            try {
                $component = strtolower(trim((string)($monitor['component'] ?? 'input'))) === 'switch' ? 'switch' : 'input';
                $channel = (int)($monitor['channel'] ?? 0);
                $status = $this->machineSignalMonitorStatus($localIp, $channel, $component);
                $on = $this->machineSignalOnState($status, $component);
                if ($on === null) {
                    continue;
                }

                $stateKey = sprintf(
                    'selfserve_machine_signal:%s:%s:%d',
                    preg_replace('/[^A-Za-z0-9_\-:.]+/', '_', (string)($monitor['relay_id'] ?? 'relay')),
                    $component,
                    $channel
                );
                $previous = $this->stateStore->getJson($stateKey, null);
                $previousOn = is_array($previous) && array_key_exists('on', $previous) ? (bool)$previous['on'] : false;

                if ($on && !$previousOn) {
                    $event = $component === 'switch' ? 'switch.on' : 'input.toggle_on';
                    $payload = [
                        'agent_token' => $agentToken,
                        'agent_instance_id' => $this->agentInstanceId,
                        'lane_id' => (int)($monitor['lane_id'] ?? 0),
                        'relay_id' => (string)($monitor['relay_id'] ?? ''),
                        'device_id' => (string)($monitor['device_id'] ?? ''),
                        'component' => $component,
                        'channel' => $channel,
                        'event' => $event,
                        'source' => 'edge_gateway_poll',
                        'status' => $status,
                    ];
                    $payload[$component === 'switch' ? 'output' : 'state'] = true;

                    $this->sendControlPlaneEvent(
                        '/edge-agent/gateways/' . $gatewayId . '/selfserve/machine-signal',
                        $payload,
                        'machine_signal'
                    );
                }

                $this->stateStore->setJson($stateKey, [
                    'on' => $on,
                    'component' => $component,
                    'channel' => $channel,
                    'relay_id' => (string)($monitor['relay_id'] ?? ''),
                    'updated_at' => date('c'),
                ]);
            } catch (Throwable $throwable) {
                $this->logger->warning('Machine start signal poll failed: ' . $throwable->getMessage());
            }
        }
    }

    /**
     * @return array<int,array<string,mixed>>
     */
    private function machineSignalMonitors(int $gatewayId, string $agentToken): array
    {
        $cache = $this->stateStore->getJson('selfserve_machine_signal_monitors', []);
        if (
            is_array($cache)
            && isset($cache['monitors'], $cache['refreshed_at'])
            && is_array($cache['monitors'])
            && (time() - (int)$cache['refreshed_at']) < 60
        ) {
            return array_values(array_filter($cache['monitors'], 'is_array'));
        }

        try {
            $response = $this->http->post('/edge-agent/gateways/' . $gatewayId . '/selfserve/machine-signal-bindings', [
                'agent_token' => $agentToken,
                'agent_instance_id' => $this->agentInstanceId,
            ], 10);
            $payload = is_array($response['data'] ?? null) ? (array)$response['data'] : (is_array($response) ? $response : []);
            $monitors = is_array($payload['monitors'] ?? null) ? array_values(array_filter((array)$payload['monitors'], 'is_array')) : [];
            $this->lastMachineSignalMonitorRefreshAt = time();
            $this->stateStore->setJson('selfserve_machine_signal_monitors', [
                'refreshed_at' => $this->lastMachineSignalMonitorRefreshAt,
                'monitors' => $monitors,
            ]);

            return $monitors;
        } catch (Throwable $throwable) {
            if (is_array($cache) && isset($cache['monitors']) && is_array($cache['monitors'])) {
                return array_values(array_filter($cache['monitors'], 'is_array'));
            }
            $this->logger->warning('Machine start signal monitor refresh failed: ' . $throwable->getMessage());
            return [];
        }
    }

    private function machineSignalMonitorStatus(string $localIp, int $channel, string $component): array
    {
        if ($component === 'input') {
            try {
                $input = $this->workerHttp->post('/relay/input-status', [
                    'local_ip' => $localIp,
                    'channel' => $channel,
                ], 5) ?? [];
                return [
                    'input_state' => $input['state'] ?? null,
                    'input' => $input,
                ];
            } catch (Throwable) {
                // Fall back to the combined status endpoint below.
            }
        }

        return $this->workerHttp->post('/relay/status', [
            'local_ip' => $localIp,
            'channel' => $channel,
            'include_input' => $component === 'input',
        ], 8) ?? [];
    }

    private function machineSignalOnState(array $status, string $component): ?bool
    {
        if ($component === 'switch') {
            if (array_key_exists('output', $status)) {
                return (bool)$status['output'];
            }
            if (array_key_exists('on', $status)) {
                return (bool)$status['on'];
            }
            if (isset($status['raw']) && is_array($status['raw']) && array_key_exists('output', $status['raw'])) {
                return (bool)$status['raw']['output'];
            }

            return null;
        }

        if (array_key_exists('input_state', $status)) {
            return $status['input_state'] === null ? null : (bool)$status['input_state'];
        }
        if (isset($status['input']) && is_array($status['input']) && array_key_exists('state', $status['input'])) {
            return (bool)$status['input']['state'];
        }
        if (array_key_exists('state', $status)) {
            return (bool)$status['state'];
        }

        return null;
    }

    private function processCommandQueue(): void
    {
        $gatewayId = (int)$this->config->get('gatewayId');
        if ($gatewayId <= 0) {
            return;
        }

        $waitSeconds = (int)$this->config->get('operationPollTimeoutSeconds', 20);
        try {
            $response = $this->http->post('/edge-agent/gateways/' . $gatewayId . '/commands/poll', [
                'agent_token' => (string)$this->config->get('agentToken'),
                'wait_seconds' => $waitSeconds,
            ], $this->pollRequestTimeoutSeconds($waitSeconds));
        } catch (HttpRequestTimeoutException) {
            return;
        } catch (Throwable $throwable) {
            $this->logger->warning('Command polling failed: ' . $throwable->getMessage());
            return;
        }

        $command = $response['data'] ?? null;
        if (!is_array($command) || empty($command['id'])) {
            return;
        }

        $jobId = (int)$command['id'];
        $type = (string)($command['command_type'] ?? $command['commandType'] ?? '');
        $request = is_array($command['payload'] ?? null) ? (array)$command['payload'] : [];

        try {
            $result = $this->executeEdgeCommand($type, $request);

            $this->sendControlPlaneEvent(
                '/edge-agent/gateways/' . $gatewayId . '/commands/' . $jobId . '/result',
                [
                    'agent_token' => (string)$this->config->get('agentToken'),
                    'ok' => true,
                    'result' => $result,
                ],
                'command_result'
            );
        } catch (Throwable $throwable) {
            $this->sendControlPlaneEvent(
                '/edge-agent/gateways/' . $gatewayId . '/commands/' . $jobId . '/result',
                [
                    'agent_token' => (string)$this->config->get('agentToken'),
                    'ok' => false,
                    'error' => $throwable->getMessage(),
                ],
                'command_result'
            );
        }
    }

    private function processBrokerCommand(array $message): void
    {
        $commandId = trim((string)($message['commandId'] ?? ''));
        $type = (string)($message['commandType'] ?? '');
        $request = isset($message['payload']) && is_array($message['payload']) ? (array)$message['payload'] : [];
        if ($commandId === '') {
            return;
        }

        try {
            $result = $this->executeEdgeCommand($type, $request);
            $this->sendBrokerMessage([
                'type' => 'COMMAND_RESULT',
                'commandId' => $commandId,
                'ok' => true,
                'payload' => $result,
            ]);
        } catch (Throwable $throwable) {
            $this->sendBrokerMessage([
                'type' => 'COMMAND_RESULT',
                'commandId' => $commandId,
                'ok' => false,
                'error' => $throwable->getMessage(),
            ]);
        }
    }

    private function executeEdgeCommand(string $type, array $request): array
    {
        return match (strtoupper(trim($type))) {
            'DISCOVER_SHELLY' => $this->runDiscovery(),
            'GET_RELAY_STATUS' => $this->readRelayStatus($request),
            'SET_RELAY_STATE' => $this->switchRelay($request),
            default => throw new RuntimeException('Unsupported command type: ' . $type),
        };
    }

    private function processManagementOperation(): bool
    {
        $gatewayId = (int)$this->config->get('gatewayId');
        if ($gatewayId <= 0) {
            return false;
        }

        $waitSeconds = (int)$this->config->get('operationPollTimeoutSeconds', 20);
        try {
            $response = $this->http->post('/edge-agent/gateways/' . $gatewayId . '/operations/next', [
                'agent_token' => (string)$this->config->get('agentToken'),
                'wait_seconds' => $waitSeconds,
                'agent_instance_id' => $this->agentInstanceId,
            ], $this->pollRequestTimeoutSeconds($waitSeconds));
        } catch (HttpRequestTimeoutException) {
            return false;
        } catch (Throwable $throwable) {
            $this->logger->warning('Operation polling failed: ' . $throwable->getMessage());
            return false;
        }

        $operation = $response['data'] ?? null;
        if (!is_array($operation) || empty($operation['id'])) {
            return false;
        }

        return $this->executeManagementOperation((array)$operation);
    }

    private function executeManagementOperation(array $operation): bool
    {
        $gatewayId = (int)$this->config->get('gatewayId');
        if ($gatewayId <= 0) {
            return false;
        }

        $operationId = (int)($operation['id'] ?? 0);
        $type = (string)($operation['type'] ?? '');
        $request = is_array($operation['request'] ?? null) ? (array)$operation['request'] : [];
        if ($operationId <= 0 || trim($type) === '') {
            return false;
        }

        $this->persistOperationState([
            'gateway_id' => $gatewayId,
            'operation_id' => $operationId,
            'type' => $type,
            'status' => 'IN_PROGRESS',
            'stage' => 'claimed',
            'progress' => 5,
            'agent_instance_id' => $this->agentInstanceId,
            'started_at' => date('c'),
            'request' => $request,
        ]);

        $this->emitOperationProgress(
            $gatewayId,
            $operationId,
            'OPERATION_AGENT_STARTED',
            'Compose edge-agent started processing the operation',
            10,
            ['type' => $type, 'agent_instance_id' => $this->agentInstanceId],
            'starting'
        );

        try {
            $result = match ($type) {
                'DISCOVERY' => $this->runManagementDiscovery($gatewayId, $operationId, $request),
                'UPDATE' => $this->runUpdate($gatewayId, $operationId, $request),
                'UNINSTALL' => $this->runUninstall($gatewayId, $operationId, $request),
                default => throw new RuntimeException('Unsupported operation type: ' . $type),
            };
            $this->abortIfOperationCancelled($operationId);

            $this->finalizeOperationCompletion(
                $gatewayId,
                $operationId,
                [
                    'agent_token' => (string)$this->config->get('agentToken'),
                    'ok' => true,
                    'result' => $result,
                ],
                'COMPLETED',
                ['result' => $result]
            );
        } catch (Throwable $throwable) {
            $errorCode = $throwable instanceof OperationAbortException && $throwable->isCancellation()
                ? 'EDGE_GATEWAY_CANCELLED'
                : $this->classifyManagementError($throwable, $type);
            if (!($throwable instanceof OperationAbortException)) {
                $this->postOperationEvent($gatewayId, $operationId, [
                    'level' => 'ERROR',
                    'code' => $errorCode,
                    'message' => $throwable->getMessage(),
                    'context' => ['type' => $type, 'progress' => 100, 'agent_instance_id' => $this->agentInstanceId],
                ]);
            }

            $this->finalizeOperationCompletion(
                $gatewayId,
                $operationId,
                [
                    'agent_token' => (string)$this->config->get('agentToken'),
                    'ok' => false,
                    'error_code' => $errorCode,
                    'error_message' => $throwable->getMessage(),
                ],
                'FAILED',
                [
                    'error_code' => $errorCode,
                    'error_message' => $throwable->getMessage(),
                ]
            );
        }

        $this->clearOperationCancellationRequested($operationId);
        return true;
    }

    private function postOperationEvent(int $gatewayId, int $operationId, array $payload): ?array
    {
        return $this->requestControlPlaneEvent(
            '/edge-agent/gateways/' . $gatewayId . '/operations/' . $operationId . '/events',
            [
                'agent_token' => (string)$this->config->get('agentToken'),
                'level' => (string)($payload['level'] ?? 'INFO'),
                'code' => $payload['code'] ?? null,
                'message' => (string)($payload['message'] ?? 'Operation event'),
                'context' => is_array($payload['context'] ?? null) ? (array)$payload['context'] : [],
            ],
            'operation_event'
        );
    }

    private function emitOperationProgress(
        int $gatewayId,
        int $operationId,
        string $code,
        string $message,
        int $progress,
        array $context = [],
        ?string $stage = null
    ): void {
        $this->abortIfOperationCancelled($operationId);

        $operationState = $this->readOperationState();
        if ($operationState !== null) {
            $operationState['progress'] = max(0, min(100, $progress));
            $operationState['stage'] = $stage ?? ($operationState['stage'] ?? 'running');
            $operationState['updated_at'] = date('c');
            $this->persistOperationState($operationState);
        }

        $this->heartbeat(true, [
            'current_operation' => $this->readOperationState(),
        ]);

        $response = $this->postOperationEvent($gatewayId, $operationId, [
            'level' => 'INFO',
            'code' => $code,
            'message' => $message,
            'context' => array_merge($context, [
                'progress' => max(0, min(100, $progress)),
                'agent_instance_id' => $this->agentInstanceId,
            ]),
        ]);
        $this->guardOperationEventResponse($response);
    }

    private function runDiscovery(): array
    {
        try {
            $response = $this->workerHttp->post('/discover', [
                'hostname' => gethostname() ?: 'truckwash-edge',
                'gateway_id' => (int)$this->config->get('gatewayId', 0),
            ], 20);
            $inventory = isset($response['inventory']) && is_array($response['inventory']) ? (array)$response['inventory'] : [];
            return ['inventory' => $inventory];
        } catch (Throwable) {
            $hostname = gethostname() ?: 'truckwash-edge';
            return ['inventory' => [[
                'device_id' => 'gateway-runtime-' . substr(sha1($hostname), 0, 10),
                'local_ip' => gethostbyname($hostname),
                'model' => 'TruckWash Edge Gateway',
                'channel_count' => 1,
                'online' => true,
                'capabilities' => [
                    'local_discovery' => true,
                    'relay_commands' => true,
                    'gateway_management_v2' => true,
                ],
                'metadata' => [
                    'hostname' => $hostname,
                    'runtime' => 'compose-php',
                    'php_version' => PHP_VERSION,
                ],
            ]]];
        }
    }

    private function runManagementDiscovery(int $gatewayId, int $operationId, array $request): array
    {
        $this->emitOperationProgress(
            $gatewayId,
            $operationId,
            'DISCOVERY_COLLECTING',
            'Collecting local gateway inventory from lan-worker',
            35,
            [],
            'discovering'
        );
        $inventory = isset($request['inventory']) && is_array($request['inventory']) && $request['inventory'] !== []
            ? (array)$request['inventory']
            : (array)($this->runDiscovery()['inventory'] ?? []);

        $this->emitOperationProgress($gatewayId, $operationId, 'DISCOVERY_COMPLETED', 'Inventory collected', 90, [
            'device_count' => count($inventory),
        ], 'finishing');

        return ['inventory' => $inventory];
    }

    private function runUpdate(int $gatewayId, int $operationId, array $request): array
    {
        $targetVersion = trim((string)($request['target_version'] ?? $request['targetVersion'] ?? ''));
        if ($targetVersion === '') {
            throw new RuntimeException('Update request is missing target version');
        }

        $updateWindow = trim((string)($request['updateWindow'] ?? $this->config->get('updateWindow', self::DEFAULT_UPDATE_WINDOW)));
        if ($updateWindow === '') {
            $updateWindow = self::DEFAULT_UPDATE_WINDOW;
        }

        $requiredArtifacts = [
            [
                'url' => (string)($request['artifactUrl'] ?? ''),
                'sha256' => (string)($request['artifactSha256'] ?? ''),
                'path' => $this->installDir . DIRECTORY_SEPARATOR . 'agent.php',
                'label' => 'edge-agent runtime',
                'progress' => 35,
                'code' => 'UPDATE_DOWNLOAD_AGENT',
            ],
            [
                'url' => (string)($request['lanWorkerArtifactUrl'] ?? ''),
                'sha256' => (string)($request['lanWorkerArtifactSha256'] ?? ''),
                'path' => $this->installDir . DIRECTORY_SEPARATOR . 'lan-worker.php',
                'label' => 'lan-worker runtime',
                'progress' => 45,
                'code' => 'UPDATE_DOWNLOAD_WORKER',
            ],
            [
                'url' => (string)($request['autoUpdaterArtifactUrl'] ?? ''),
                'sha256' => (string)($request['autoUpdaterArtifactSha256'] ?? ''),
                'path' => $this->installDir . DIRECTORY_SEPARATOR . 'auto-updater.php',
                'label' => 'auto-updater runtime',
                'progress' => 50,
                'code' => 'UPDATE_DOWNLOAD_AUTO_UPDATER',
            ],
            [
                'url' => (string)($request['composeFileUrl'] ?? ''),
                'sha256' => (string)($request['composeFileSha256'] ?? ''),
                'path' => $this->installDir . DIRECTORY_SEPARATOR . 'docker-compose.gateway.yml',
                'label' => 'compose stack',
                'progress' => 55,
                'code' => 'UPDATE_DOWNLOAD_COMPOSE',
            ],
            [
                'url' => (string)($request['edgeAgentDockerfileUrl'] ?? ''),
                'sha256' => (string)($request['edgeAgentDockerfileSha256'] ?? ''),
                'path' => $this->installDir . DIRECTORY_SEPARATOR . 'Dockerfile.edge-agent',
                'label' => 'edge-agent Dockerfile',
                'progress' => 60,
                'code' => 'UPDATE_DOWNLOAD_EDGE_DOCKERFILE',
            ],
            [
                'url' => (string)($request['lanWorkerDockerfileUrl'] ?? ''),
                'sha256' => (string)($request['lanWorkerDockerfileSha256'] ?? ''),
                'path' => $this->installDir . DIRECTORY_SEPARATOR . 'Dockerfile.lan-worker',
                'label' => 'lan-worker Dockerfile',
                'progress' => 65,
                'code' => 'UPDATE_DOWNLOAD_WORKER_DOCKERFILE',
            ],
            [
                'url' => (string)($request['autoUpdaterDockerfileUrl'] ?? ''),
                'sha256' => (string)($request['autoUpdaterDockerfileSha256'] ?? ''),
                'path' => $this->installDir . DIRECTORY_SEPARATOR . 'Dockerfile.auto-updater',
                'label' => 'auto-updater Dockerfile',
                'progress' => 68,
                'code' => 'UPDATE_DOWNLOAD_AUTO_UPDATER_DOCKERFILE',
            ],
            [
                'url' => (string)($request['launcherScriptUrl'] ?? ''),
                'sha256' => (string)($request['launcherScriptSha256'] ?? ''),
                'path' => $this->installDir . DIRECTORY_SEPARATOR . 'gateway-launcher.sh',
                'label' => 'gateway launcher',
                'progress' => 72,
                'code' => 'UPDATE_DOWNLOAD_LAUNCHER',
            ],
            [
                'url' => (string)($request['stackServiceUnitUrl'] ?? ''),
                'sha256' => (string)($request['stackServiceUnitSha256'] ?? ''),
                'path' => $this->installDir . DIRECTORY_SEPARATOR . 'truckwash-edge-gateway-stack.service',
                'label' => 'compose systemd unit',
                'progress' => 80,
                'code' => 'UPDATE_DOWNLOAD_STACK_SERVICE',
            ],
            [
                'url' => (string)($request['serviceUnitUrl'] ?? ''),
                'sha256' => (string)($request['serviceUnitSha256'] ?? ''),
                'path' => $this->installDir . DIRECTORY_SEPARATOR . 'truckwash-edge-agent.service',
                'label' => 'compatibility systemd unit',
                'progress' => 84,
                'code' => 'UPDATE_DOWNLOAD_LEGACY_SERVICE',
            ],
        ];

        foreach ([
            'composeFileUrl',
            'launcherScriptUrl',
            'stackServiceUnitUrl',
            'autoUpdaterArtifactUrl',
            'autoUpdaterDockerfileUrl',
        ] as $requiredKey) {
            if (trim((string)($request[$requiredKey] ?? '')) === '') {
                throw new RuntimeException('Update request is missing compose artifact ' . $requiredKey);
            }
        }

        $this->emitOperationProgress($gatewayId, $operationId, 'UPDATE_VALIDATED', 'Validated compose update request', 20, [
            'target_version' => $targetVersion,
            'update_window' => $updateWindow,
        ], 'validating');

        $changedFiles = [];
        foreach ($requiredArtifacts as $artifact) {
            if ($artifact['url'] === '') {
                continue;
            }

            $this->emitOperationProgress(
                $gatewayId,
                $operationId,
                (string)$artifact['code'],
                'Downloading ' . (string)$artifact['label'],
                (int)$artifact['progress'],
                ['url' => (string)$artifact['url']],
                'downloading'
            );
            $changedFiles[] = $this->downloadToFileAtomic(
                (string)$artifact['url'],
                (string)$artifact['path'],
                (string)$artifact['sha256']
            );
        }

        $this->ensureExecutable($this->installDir . DIRECTORY_SEPARATOR . 'agent.php');
        $this->ensureExecutable($this->installDir . DIRECTORY_SEPARATOR . 'lan-worker.php');
        $this->ensureExecutable($this->installDir . DIRECTORY_SEPARATOR . 'auto-updater.php');
        $this->ensureExecutable($this->installDir . DIRECTORY_SEPARATOR . 'gateway-launcher.sh');

        $stagedAt = date('c');
        $stagedUpdate = [
            'target_version' => $targetVersion,
            'staged_at' => $stagedAt,
            'apply_after' => $this->nextUpdateWindowStartIso($updateWindow),
            'status' => 'STAGED',
            'update_window' => $updateWindow,
            'compose_project_name' => (string)($request['composeProjectName'] ?? $this->config->get('composeProjectName', 'truckwash-edge-gateway')),
            'stack_service_name' => (string)($request['stackServiceName'] ?? $this->config->get('stackServiceName', 'truckwash-edge-gateway-stack.service')),
            'changed_files' => $changedFiles,
        ];

        $this->emitOperationProgress($gatewayId, $operationId, 'UPDATE_STAGE_METADATA', 'Writing staged update metadata', 90, [], 'writing-config');
        $this->config->set('targetVersion', $targetVersion);
        $this->config->set('lastStagedUpdate', $stagedUpdate);
        $this->config->save();
        $this->stateStore->setJson('staged_update', $stagedUpdate);
        $this->stateStore->setJson('rollback_status', [
            'state' => 'IDLE',
            'reason' => null,
            'rolled_back_to' => null,
            'at' => null,
        ]);
        file_put_contents($this->stagedUpdatePath, json_encode($stagedUpdate, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);

        $this->emitOperationProgress($gatewayId, $operationId, 'UPDATE_STAGED', 'Compose rollout staged for the maintenance window', 96, [
            'target_version' => $targetVersion,
            'apply_after' => $stagedUpdate['apply_after'],
        ], 'staged');

        return [
            'applied' => false,
            'installed_version' => (string)$this->config->get('installedVersion', 'compose-php-agent-v2'),
            'staged_version' => $targetVersion,
            'target_version' => $targetVersion,
            'staged_at' => $stagedAt,
            'apply_after' => $stagedUpdate['apply_after'],
            'update_window' => $updateWindow,
            'restart_required' => true,
            'service_name' => (string)($request['stackServiceName'] ?? $this->config->get('stackServiceName', 'truckwash-edge-gateway-stack.service')),
            'changed_files' => $changedFiles,
            'agent_instance_id' => $this->agentInstanceId,
            'rollback_status' => $this->readRollbackStatus(),
        ];
    }

    private function runUninstall(int $gatewayId, int $operationId, array $request): array
    {
        $this->emitOperationProgress($gatewayId, $operationId, 'UNINSTALL_PREPARE', 'Preparing uninstall manifest', 40, [], 'preparing-uninstall');
        $manifestPath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'uninstall-plan.json';
        $manifest = [
            'service_name' => (string)($request['service_name'] ?? $this->config->get('stackServiceName', 'truckwash-edge-gateway-stack.service')),
            'install_dir' => (string)($request['install_dir'] ?? $this->installDir),
            'generated_at' => date('c'),
            'agent_instance_id' => $this->agentInstanceId,
        ];
        file_put_contents($manifestPath, json_encode($manifest, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);
        file_put_contents($this->runtimeDir . DIRECTORY_SEPARATOR . 'uninstall.flag', date('c') . PHP_EOL);
        $this->config->set('pendingUninstall', $manifest);
        $this->config->save();

        $this->emitOperationProgress($gatewayId, $operationId, 'UNINSTALL_READY', 'Gateway marked for controlled uninstall', 90, [
            'manifest_path' => $manifestPath,
        ], 'uninstall-ready');

        return [
            'uninstalled' => true,
            'service_name' => $manifest['service_name'],
            'install_dir' => $manifest['install_dir'],
            'manual_cleanup_required' => true,
            'cleanup_manifest_path' => $manifestPath,
        ];
    }

    private function readRelayStatus(array $request): array
    {
        $localIp = (string)($request['localIp'] ?? $request['local_ip'] ?? '');
        $channel = (int)($request['channel'] ?? 0);

        try {
            return $this->workerHttp->post('/relay/status', [
                'local_ip' => $localIp,
                'channel' => $channel,
            ], 8) ?? [];
        } catch (Throwable) {
            return $this->fetchShellyState($localIp, $channel);
        }
    }

    private function switchRelay(array $request): array
    {
        $localIp = (string)($request['localIp'] ?? $request['local_ip'] ?? '');
        $channel = (int)($request['channel'] ?? 0);
        $on = (bool)($request['on'] ?? false);

        try {
            return $this->workerHttp->post('/relay/switch', [
                'local_ip' => $localIp,
                'channel' => $channel,
                'on' => $on,
            ], 8) ?? [];
        } catch (Throwable) {
            $rpcUrl = sprintf('http://%s/rpc/Switch.Set?id=%d&on=%s', $localIp, $channel, $on ? 'true' : 'false');
            try {
                $this->http->getJson($rpcUrl, 8);
            } catch (Throwable) {
                $legacyUrl = sprintf('http://%s/relay/%d?turn=%s', $localIp, $channel, $on ? 'on' : 'off');
                $this->http->getJson($legacyUrl, 8);
            }

            return $this->fetchShellyState($localIp, $channel);
        }
    }

    private function downloadToFileAtomic(string $url, string $path, string $expectedSha256 = ''): array
    {
        $directory = dirname($path);
        if (!is_dir($directory)) {
            @mkdir($directory, 0777, true);
        }

        $temporaryPath = $path . '.download';
        $backupPath = $path . '.bak';
        $raw = $this->http->download($url, 60);
        file_put_contents($temporaryPath, $raw);

        if ($expectedSha256 !== '') {
            $actualSha = hash_file('sha256', $temporaryPath);
            if (!is_string($actualSha) || !hash_equals(strtolower($expectedSha256), strtolower($actualSha))) {
                @unlink($temporaryPath);
                throw new RuntimeException('Artifact checksum mismatch for ' . basename($path));
            }
        }

        if (is_file($backupPath)) {
            @unlink($backupPath);
        }

        if (is_file($path) && !@rename($path, $backupPath)) {
            @unlink($temporaryPath);
            throw new RuntimeException('Could not create backup for ' . $path);
        }

        if (!@rename($temporaryPath, $path)) {
            if (is_file($backupPath)) {
                @rename($backupPath, $path);
            }
            @unlink($temporaryPath);
            throw new RuntimeException('Could not promote downloaded artifact for ' . $path);
        }

        return [
            'path' => $path,
            'url' => $url,
            'backup_path' => is_file($backupPath) ? $backupPath : null,
            'sha256' => hash_file('sha256', $path),
        ];
    }

    private function classifyManagementError(Throwable $throwable, string $type): string
    {
        $message = strtolower(trim($throwable->getMessage()));
        if (str_contains($message, 'token') || str_contains($message, 'credential')) {
            return 'EDGE_GATEWAY_INVALID_TOKEN';
        }
        if (str_contains($message, 'timeout') || str_contains($message, 'http 5')) {
            return 'EDGE_GATEWAY_OPERATION_TIMEOUT';
        }
        if ($type === 'UPDATE' && (str_contains($message, 'version') || str_contains($message, 'checksum'))) {
            return 'EDGE_GATEWAY_UNSUPPORTED_VERSION';
        }

        return 'EDGE_GATEWAY_VALIDATION_FAILED';
    }

    private function fetchShellyState(string $localIp, int $channel): array
    {
        if ($localIp === '') {
            throw new RuntimeException('Missing Shelly IP address');
        }

        try {
            $payload = $this->http->getJson(sprintf('http://%s/rpc/Switch.GetStatus?id=%d', $localIp, $channel), 8);
            return [
                'online' => true,
                'on' => (bool)($payload['output'] ?? false),
                'output' => (bool)($payload['output'] ?? false),
                'raw' => $payload,
            ];
        } catch (Throwable) {
            $payload = $this->http->getJson(sprintf('http://%s/relay/%d', $localIp, $channel), 8);
            return [
                'online' => true,
                'on' => (bool)($payload['ison'] ?? false),
                'output' => (bool)($payload['ison'] ?? false),
                'raw' => $payload,
            ];
        }
    }

    private function buildSystemMetrics(): array
    {
        $diskTotal = @disk_total_space($this->installDir);
        $diskFree = @disk_free_space($this->installDir);
        $diskUsed = (is_numeric($diskTotal) && is_numeric($diskFree)) ? ((float)$diskTotal - (float)$diskFree) : null;
        $diskUsagePct = ($diskTotal && $diskUsed !== null && $diskTotal > 0)
            ? (int)round(($diskUsed / (float)$diskTotal) * 100)
            : null;
        $memoryLimitBytes = $this->iniBytes((string)ini_get('memory_limit'));
        $memoryUsage = memory_get_usage(true);
        $memoryUsagePct = ($memoryLimitBytes !== null && $memoryLimitBytes > 0)
            ? (int)round(($memoryUsage / $memoryLimitBytes) * 100)
            : null;
        $loadAverage = function_exists('sys_getloadavg') ? sys_getloadavg() : [];

        return [
            'memory_usage_bytes' => $memoryUsage,
            'memory_peak_bytes' => memory_get_peak_usage(true),
            'memory_usage_pct' => $memoryUsagePct,
            'cpu_usage_pct' => is_array($loadAverage) && isset($loadAverage[0]) ? max(0, (int)round((float)$loadAverage[0] * 100)) : null,
            'load_average' => $loadAverage,
            'disk_usage_pct' => $diskUsagePct,
            'disk_used_bytes' => $diskUsed,
            'disk_total_bytes' => is_numeric($diskTotal) ? (int)$diskTotal : null,
            'disk_mount' => $this->installDir,
            'latency_ms' => null,
        ];
    }

    private function buildContainerHealth(): array
    {
        $services = [[
            'name' => 'edge-agent',
            'status' => 'healthy',
            'updated_at' => date('c'),
        ]];

        $services[] = $this->probeWorkerHealth();
        $services[] = $this->probeTcpService('redis', 'redis', 6379);
        $services[] = $this->probeTcpService('mariadb', 'mariadb', 3306);
        $services[] = $this->probeHttpService('minio', 'http://minio:9000/minio/health/live');
        $services[] = $this->probeAutoUpdaterHealth();

        $healthyCount = count(array_filter($services, static fn(array $service): bool => (string)($service['status'] ?? '') === 'healthy'));
        $state = $healthyCount === count($services) ? 'ONLINE' : 'DEGRADED';

        return [
            'state' => $state,
            'summary' => sprintf('%d/%d containers healthy', $healthyCount, count($services)),
            'services' => $services,
        ];
    }

    private function buildOutboxStatus(): array
    {
        $summary = $this->stateStore->outboxSummary();
        $queued = (int)($summary['queued'] ?? 0);
        return [
            'state' => $queued > 0 ? 'QUEUED' : 'IN_SYNC',
            'queued' => $queued,
            'oldest_queued_at' => $summary['oldest_queued_at'] ?? null,
            'last_replayed_at' => $summary['last_replayed_at'] ?? null,
            'summary' => $queued > 0 ? sprintf('%d outbound items queued', $queued) : 'Outbox is empty',
        ];
    }

    private function initializeControlPlaneStatus(): void
    {
        $this->writeControlPlaneStatus();
    }

    private function touchLoopHeartbeat(): void
    {
        $this->writeControlPlaneStatus([
            'last_loop_at' => date('c'),
        ]);
    }

    private function currentLastSuccessfulSyncAt(): ?string
    {
        $current = $this->readControlPlaneStatus()['last_successful_sync_at'] ?? $this->stateStore->getJson('last_sync_at');
        return $this->normalizeControlPlaneStatusTimestamp($current);
    }

    private function recordHeartbeatAttempt(): string
    {
        $attemptedAt = date('c');
        $this->writeControlPlaneStatus([
            'last_heartbeat_attempt_at' => $attemptedAt,
        ]);
        return $attemptedAt;
    }

    private function recordSuccessfulSync(?string $at = null, array $statusOverrides = []): void
    {
        $syncedAt = $this->normalizeControlPlaneStatusTimestamp($at ?? date('c')) ?? date('c');
        $this->stateStore->setJson('last_sync_at', $syncedAt);
        $statusOverrides['last_successful_sync_at'] = $syncedAt;
        $this->writeControlPlaneStatus($statusOverrides);
    }

    private function recordTransportFailure(string $context, Throwable $throwable): void
    {
        $message = $this->normalizeControlPlaneStatusString($throwable->getMessage()) ?? $throwable::class;
        $this->writeControlPlaneStatus([
            'last_transport_failure_at' => date('c'),
            'last_transport_error' => $message,
        ]);
        $this->logger->warning($context . ': ' . $message);
    }

    /**
     * @return array<string,mixed>
     */
    private function buildControlPlaneStatusPayload(): array
    {
        return $this->writeControlPlaneStatus();
    }

    /**
     * @return array<string,mixed>
     */
    private function readControlPlaneStatus(): array
    {
        if (!is_file($this->controlPlaneStatusPath)) {
            return [];
        }

        $decoded = json_decode((string)file_get_contents($this->controlPlaneStatusPath), true);
        return is_array($decoded) ? $decoded : [];
    }

    /**
     * @param array<string,mixed> $overrides
     * @return array<string,mixed>
     */
    private function writeControlPlaneStatus(array $overrides = []): array
    {
        $current = $this->readControlPlaneStatus();
        $outboxSummary = $this->stateStore->outboxSummary();
        $lastSuccessfulSyncAt = array_key_exists('last_successful_sync_at', $overrides)
            ? $overrides['last_successful_sync_at']
            : ($current['last_successful_sync_at'] ?? $this->stateStore->getJson('last_sync_at'));
        $lastTransportError = array_key_exists('last_transport_error', $overrides)
            ? $overrides['last_transport_error']
            : ($current['last_transport_error'] ?? null);
        $lastTransportFailureAt = array_key_exists('last_transport_failure_at', $overrides)
            ? $overrides['last_transport_failure_at']
            : ($current['last_transport_failure_at'] ?? null);
        $lastHeartbeatAttemptAt = array_key_exists('last_heartbeat_attempt_at', $overrides)
            ? $overrides['last_heartbeat_attempt_at']
            : ($current['last_heartbeat_attempt_at'] ?? null);
        $lastHeartbeatSuccessAt = array_key_exists('last_heartbeat_success_at', $overrides)
            ? $overrides['last_heartbeat_success_at']
            : ($current['last_heartbeat_success_at'] ?? null);
        $lastLoopAt = array_key_exists('last_loop_at', $overrides)
            ? $overrides['last_loop_at']
            : ($current['last_loop_at'] ?? null);

        $status = [
            'started_at' => $this->normalizeControlPlaneStatusTimestamp($current['started_at'] ?? null) ?? date('c'),
            'agent_instance_id' => $this->agentInstanceId,
            'last_loop_at' => $this->normalizeControlPlaneStatusTimestamp($lastLoopAt),
            'last_heartbeat_attempt_at' => $this->normalizeControlPlaneStatusTimestamp($lastHeartbeatAttemptAt),
            'last_heartbeat_success_at' => $this->normalizeControlPlaneStatusTimestamp($lastHeartbeatSuccessAt),
            'last_successful_sync_at' => $this->normalizeControlPlaneStatusTimestamp($lastSuccessfulSyncAt),
            'outbox_queued' => array_key_exists('outbox_queued', $overrides)
                ? max(0, (int)$overrides['outbox_queued'])
                : max(0, (int)($outboxSummary['queued'] ?? 0)),
            'broker_connected' => array_key_exists('broker_connected', $overrides)
                ? (bool)$overrides['broker_connected']
                : $this->isBrokerConnected(),
            'last_transport_error' => $this->normalizeControlPlaneStatusString($lastTransportError),
            'last_transport_failure_at' => $this->normalizeControlPlaneStatusTimestamp($lastTransportFailureAt),
        ];

        file_put_contents(
            $this->controlPlaneStatusPath,
            json_encode($status, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL
        );

        return $status;
    }

    private function normalizeControlPlaneStatusTimestamp(mixed $value): ?string
    {
        return is_string($value) && trim($value) !== '' ? trim($value) : null;
    }

    private function normalizeControlPlaneStatusString(mixed $value): ?string
    {
        if (is_string($value)) {
            $normalized = trim($value);
            return $normalized !== '' ? $normalized : null;
        }
        if (is_scalar($value)) {
            $normalized = trim((string)$value);
            return $normalized !== '' ? $normalized : null;
        }

        return null;
    }

    private function flushOutbox(): void
    {
        $items = $this->stateStore->queuedItems(25);
        foreach ($items as $item) {
            try {
                $timeoutSeconds = (string)($item['type'] ?? '') === 'operation_complete'
                    ? self::OPERATION_COMPLETE_TIMEOUT_SECONDS
                    : 20;
                $endpoint = (string)$item['endpoint'];
                $payload = is_array($item['payload'] ?? null) ? (array)$item['payload'] : [];
                $brokerDispatch = $this->dispatchBrokerControlPlaneEvent($endpoint, $payload);
                if ($brokerDispatch !== true || $this->shouldPersistControlPlaneEventOverHttp($endpoint)) {
                    $this->http->post($endpoint, $payload, $timeoutSeconds);
                }
                $this->stateStore->removeOutboxItem((int)$item['id']);
                $this->recordSuccessfulSync();
            } catch (Throwable $throwable) {
                $this->recordTransportFailure(
                    'Outbox replay blocked on ' . (string)$item['type'] . ' for ' . (string)$item['endpoint'],
                    $throwable
                );
                break;
            }
        }
    }

    private function probeWorkerHealth(): array
    {
        try {
            $workerHealth = $this->workerHttp->getJson(
                rtrim((string)$this->config->get('workerBaseUrl', self::DEFAULT_WORKER_BASE_URL), '/') . '/health',
                5
            );
            return [
                'name' => 'lan-worker',
                'status' => (string)($workerHealth['status'] ?? 'healthy'),
                'updated_at' => (string)($workerHealth['timestamp'] ?? date('c')),
            ];
        } catch (Throwable $throwable) {
            return [
                'name' => 'lan-worker',
                'status' => 'degraded',
                'updated_at' => date('c'),
                'error' => $throwable->getMessage(),
            ];
        }
    }

    private function probeHttpService(string $name, string $url): array
    {
        try {
            $this->http->download($url, 5);
            return [
                'name' => $name,
                'status' => 'healthy',
                'updated_at' => date('c'),
            ];
        } catch (Throwable $throwable) {
            return [
                'name' => $name,
                'status' => 'degraded',
                'updated_at' => date('c'),
                'error' => $throwable->getMessage(),
            ];
        }
    }

    private function probeTcpService(string $name, string $host, int $port): array
    {
        $socket = @fsockopen($host, $port, $errno, $error, 3);
        if (is_resource($socket)) {
            fclose($socket);
            return [
                'name' => $name,
                'status' => 'healthy',
                'updated_at' => date('c'),
            ];
        }

        return [
            'name' => $name,
            'status' => 'degraded',
            'updated_at' => date('c'),
            'error' => trim((string)$error) !== '' ? trim((string)$error) : 'TCP connection failed',
            'code' => $errno > 0 ? $errno : null,
        ];
    }

    private function probeAutoUpdaterHealth(): array
    {
        $path = $this->runtimeDir . DIRECTORY_SEPARATOR . 'auto-updater-heartbeat.json';
        if (!is_file($path)) {
            return [
                'name' => 'auto-updater',
                'status' => 'degraded',
                'updated_at' => date('c'),
                'error' => 'No auto-updater heartbeat recorded',
            ];
        }

        $decoded = json_decode((string)file_get_contents($path), true);
        $status = trim((string)($decoded['status'] ?? 'idle'));
        $ageSeconds = max(0, time() - filemtime($path));
        $healthy = $ageSeconds <= 90 && $status !== 'error';

        return [
            'name' => 'auto-updater',
            'status' => $healthy ? 'healthy' : 'degraded',
            'updated_at' => (string)($decoded['updated_at'] ?? date('c')),
            'mode' => $status,
            'error' => $healthy ? null : (string)($decoded['last_output'] ?? 'Auto-updater heartbeat is stale'),
        ];
    }

    private function sendControlPlaneEvent(string $endpoint, array $payload, string $type): bool
    {
        $this->lastControlPlaneResponse = null;
        $brokerDispatch = $this->dispatchBrokerControlPlaneEvent($endpoint, $payload);
        if ($brokerDispatch === true && !$this->shouldPersistControlPlaneEventOverHttp($endpoint)) {
            return true;
        }

        try {
            $response = $this->http->post($endpoint, $payload, 20);
            $this->lastControlPlaneResponse = is_array($response) ? $response : null;
            $this->recordSuccessfulSync();
            return true;
        } catch (Throwable $throwable) {
            if ($brokerDispatch === true && !$this->shouldPersistControlPlaneEventOverHttp($endpoint)) {
                $this->recordTransportFailure(
                    'Broker dispatched ' . $type . ' but HTTP persistence failed on ' . $endpoint,
                    $throwable
                );
                return true;
            }

            $this->stateStore->enqueue($type, $endpoint, $payload);
            $this->recordTransportFailure(
                'Queued ' . $type . ' to local outbox after transport failure on ' . $endpoint,
                $throwable
            );
            return false;
        }
    }

    private function applyBrokerUrlFromControlPlaneResponse(?array $response): void
    {
        if (!is_array($response)) {
            return;
        }

        $payload = is_array($response['data'] ?? null) ? (array)$response['data'] : $response;
        $brokerUrl = trim((string)($payload['broker_url'] ?? $payload['gateway']['broker_url'] ?? ''));
        if ($brokerUrl === '') {
            return;
        }

        $current = trim((string)$this->config->get('brokerUrl'));
        if (rtrim($current, '/') === rtrim($brokerUrl, '/')) {
            return;
        }

        $this->config->set('brokerUrl', rtrim($brokerUrl, '/'));
        $this->config->save();
        $this->configureBrokerClient();
        $this->logger->info('Updated broker URL from control plane heartbeat response.');
    }

    private function shouldPersistControlPlaneEventOverHttp(string $endpoint): bool
    {
        return preg_match('#/edge-agent/gateways/\d+/heartbeat$#', $endpoint) === 1;
    }

    private function requestControlPlaneEvent(string $endpoint, array $payload, string $type, int $timeoutSeconds = 20): ?array
    {
        $brokerDispatch = $this->dispatchBrokerControlPlaneEvent($endpoint, $payload);
        if ($brokerDispatch === true) {
            return ['data' => ['status' => 'ACKNOWLEDGED']];
        }

        try {
            $response = $this->http->post($endpoint, $payload, $timeoutSeconds);
            $this->recordSuccessfulSync();
            return is_array($response) ? $response : null;
        } catch (Throwable $throwable) {
            $this->stateStore->enqueue($type, $endpoint, $payload);
            $this->recordTransportFailure(
                'Queued ' . $type . ' to local outbox after transport failure on ' . $endpoint,
                $throwable
            );
            return null;
        }
    }

    private function guardOperationEventResponse(?array $response): void
    {
        $operation = is_array($response['data'] ?? null) ? (array)$response['data'] : null;
        if ($operation === null) {
            return;
        }

        $status = strtoupper(trim((string)($operation['status'] ?? '')));
        if ($status === 'CANCEL_REQUESTED' || $status === 'CANCELLED') {
            throw new OperationAbortException('Operation cancelled by operator', true);
        }

        if ($status === 'FAILED') {
            throw new OperationAbortException(
                trim((string)($operation['error_message'] ?? '')) !== ''
                    ? (string)$operation['error_message']
                    : 'Gateway operation failed'
            );
        }
    }

    private function markOperationCancellationRequested(int $operationId): void
    {
        $ids = $this->operationCancellationRequests();
        if (!in_array($operationId, $ids, true)) {
            $ids[] = $operationId;
            $this->stateStore->setJson('cancel_requested_operations', array_values($ids));
        }
    }

    private function clearOperationCancellationRequested(int $operationId): void
    {
        $ids = array_values(array_filter(
            $this->operationCancellationRequests(),
            static fn(int $currentId): bool => $currentId !== $operationId
        ));
        $this->stateStore->setJson('cancel_requested_operations', $ids);
    }

    /**
     * @return array<int,int>
     */
    private function operationCancellationRequests(): array
    {
        $stored = $this->stateStore->getJson('cancel_requested_operations', []);
        if (!is_array($stored)) {
            return [];
        }

        return array_values(array_filter(
            array_map(static fn(mixed $value): int => (int)$value, $stored),
            static fn(int $value): bool => $value > 0
        ));
    }

    private function abortIfOperationCancelled(int $operationId): void
    {
        if (in_array($operationId, $this->operationCancellationRequests(), true)) {
            throw new OperationAbortException('Operation cancelled by operator', true);
        }
    }

    private function readRollbackStatus(): array
    {
        $rollbackPath = $this->runtimeDir . DIRECTORY_SEPARATOR . 'rollback-status.json';
        if (is_file($rollbackPath)) {
            $decoded = json_decode((string)file_get_contents($rollbackPath), true);
            if (is_array($decoded)) {
                return $decoded;
            }
        }

        $rollback = $this->stateStore->getJson('rollback_status', null);
        if (is_array($rollback)) {
            return $rollback;
        }

        return [
            'state' => 'IDLE',
            'reason' => null,
            'rolled_back_to' => null,
            'at' => null,
        ];
    }

    private function currentStagedUpdate(): ?array
    {
        if (is_file($this->stagedUpdatePath)) {
            $decoded = json_decode((string)file_get_contents($this->stagedUpdatePath), true);
            if (is_array($decoded)) {
                return $decoded;
            }
        }

        $staged = $this->stateStore->getJson('staged_update', null);
        return is_array($staged) ? $staged : null;
    }

    private function ensureExecutable(string $path): void
    {
        if (is_file($path)) {
            @chmod($path, 0755);
        }
    }

    private function workerAuthorizationHeaders(): array
    {
        $agentToken = trim((string)$this->config->get('agentToken', ''));
        return $agentToken !== '' ? ['X-Truckwash-Worker-Token: ' . $agentToken] : [];
    }

    private function ensureAgentInstanceId(): string
    {
        $configured = trim((string)$this->config->get('agentInstanceId', ''));
        if ($configured !== '') {
            return substr($configured, 0, 128);
        }

        $generated = sprintf(
            '%s-%s',
            preg_replace('/[^A-Za-z0-9\-]+/', '-', gethostname() ?: 'truckwash-edge') ?: 'truckwash-edge',
            substr(bin2hex(random_bytes(6)), 0, 12)
        );
        $this->config->set('agentInstanceId', $generated);
        $this->config->save();
        return $generated;
    }

    private function recoverPreviousOperationState(): void
    {
        $state = $this->readOperationState();
        if ($state === null) {
            return;
        }

        $completion = is_array($state['completion'] ?? null) ? (array)$state['completion'] : null;
        $message = $completion !== null
            ? 'Recovered previous operation awaiting backend completion acknowledgement: operation %s (%s) last stage %s.'
            : 'Recovered previous unfinished operation snapshot: operation %s (%s) last stage %s.';

        $this->logger->warning(sprintf(
            $message,
            (string)($state['operation_id'] ?? 'unknown'),
            (string)($state['type'] ?? 'unknown'),
            (string)($state['stage'] ?? 'unknown')
        ));
    }

    private function finalizeOperationCompletion(
        int $gatewayId,
        int $operationId,
        array $payload,
        string $snapshotStatus,
        array $snapshotExtra = []
    ): void {
        $state = $this->readOperationState() ?? [];
        $completion = [
            'endpoint' => '/edge-agent/gateways/' . $gatewayId . '/operations/' . $operationId . '/complete',
            'payload' => $payload,
            'snapshot_status' => $snapshotStatus,
            'snapshot_extra' => $snapshotExtra,
        ];

        $state['status'] = 'COMPLETION_PENDING';
        $state['stage'] = 'awaiting_completion_ack';
        $state['updated_at'] = date('c');
        $state['completion'] = $completion;
        $this->persistOperationState($state);

        if (!$this->dispatchOperationCompletion($completion, true)) {
            return;
        }

        $this->snapshotOperationState($snapshotStatus, $snapshotExtra);
        $this->clearOperationState();
    }

    private function resumePendingOperationCompletion(): bool
    {
        $state = $this->readOperationState();
        if ($state === null) {
            return false;
        }

        $completion = is_array($state['completion'] ?? null) ? (array)$state['completion'] : null;
        if ($completion === null) {
            return false;
        }

        if (!$this->dispatchOperationCompletion($completion, false)) {
            return true;
        }

        $snapshotStatus = trim((string)($completion['snapshot_status'] ?? $state['status'] ?? 'FAILED'));
        if ($snapshotStatus === '') {
            $snapshotStatus = 'FAILED';
        }

        $snapshotExtra = is_array($completion['snapshot_extra'] ?? null)
            ? (array)$completion['snapshot_extra']
            : [];
        $this->snapshotOperationState($snapshotStatus, $snapshotExtra);
        $this->clearOperationState();
        $this->logger->info(sprintf(
            'Acknowledged completion for recovered operation %s.',
            (string)($state['operation_id'] ?? 'unknown')
        ));

        return true;
    }

    private function dispatchOperationCompletion(array $completion, bool $queueOnFailure): bool
    {
        $endpoint = trim((string)($completion['endpoint'] ?? ''));
        $payload = is_array($completion['payload'] ?? null) ? (array)$completion['payload'] : [];
        if ($endpoint === '') {
            $this->logger->error('Unable to dispatch operation completion: missing endpoint.');
            return false;
        }

        $brokerDispatch = $this->dispatchBrokerControlPlaneEvent($endpoint, $payload);
        if ($brokerDispatch === true) {
            return true;
        }

        try {
            $this->http->post($endpoint, $payload, self::OPERATION_COMPLETE_TIMEOUT_SECONDS);
            $this->recordSuccessfulSync();
            return true;
        } catch (Throwable $throwable) {
            if ($queueOnFailure) {
                $this->stateStore->enqueue('operation_complete', $endpoint, $payload);
                $this->recordTransportFailure(
                    'Queued operation_complete to local outbox after completion acknowledgement failure on ' . $endpoint,
                    $throwable
                );
                return false;
            }

            $this->recordTransportFailure(
                'Retrying backend completion acknowledgement later for ' . $endpoint,
                $throwable
            );
            return false;
        }
    }

    private function dispatchBrokerControlPlaneEvent(string $endpoint, array $payload): ?bool
    {
        if (!$this->isBrokerConnected()) {
            return null;
        }

        if (preg_match('#/edge-agent/gateways/(\d+)/heartbeat$#', $endpoint)) {
            return $this->sendBrokerMessage([
                'type' => 'TELEMETRY',
                'payload' => $this->stripAgentAuthentication($payload),
            ]);
        }

        if (preg_match('#/edge-agent/gateways/\d+/operations/(\d+)/events$#', $endpoint, $matches)) {
            return $this->sendBrokerMessage([
                'type' => 'TASK_EVENT',
                'operationId' => (int)$matches[1],
                'payload' => $this->stripAgentAuthentication($payload),
            ]);
        }

        if (preg_match('#/edge-agent/gateways/\d+/operations/(\d+)/complete$#', $endpoint, $matches)) {
            return $this->sendBrokerMessage([
                'type' => 'TASK_RESULT',
                'operationId' => (int)$matches[1],
                'payload' => $this->stripAgentAuthentication($payload),
            ]);
        }

        if (preg_match('#/edge-agent/gateways/\d+/logs$#', $endpoint)) {
            return $this->sendBrokerMessage([
                'type' => 'LOG_FRAME',
                'payload' => $this->stripAgentAuthentication($payload),
            ]);
        }

        return null;
    }

    private function stripAgentAuthentication(array $payload): array
    {
        unset($payload['agent_token']);
        return $payload;
    }

    private function emitLogFrame(string $level, string $message): bool
    {
        $gatewayId = (int)$this->config->get('gatewayId', 0);
        if ($gatewayId <= 0 || trim($message) === '') {
            return false;
        }

        $payload = [
            'level' => strtoupper(trim($level)) ?: 'INFO',
            'message' => $message,
            'stream' => 'agent',
            'source' => 'EDGE_AGENT',
            'context' => [
                'agent_instance_id' => $this->agentInstanceId,
            ],
        ];

        $endpoint = '/edge-agent/gateways/' . $gatewayId . '/logs';
        $brokerDispatch = $this->dispatchBrokerControlPlaneEvent($endpoint, $payload);
        if ($brokerDispatch === true) {
            return true;
        }

        try {
            $this->http->post($endpoint, $payload, 10);
            return true;
        } catch (Throwable) {
            return false;
        }
    }

    private function persistOperationState(array $state): void
    {
        file_put_contents($this->statePath, json_encode($state, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);
    }

    private function snapshotOperationState(string $status, array $extra = []): void
    {
        $state = $this->readOperationState() ?? [];
        unset($state['completion']);
        $state['status'] = $status;
        $state['finished_at'] = date('c');
        $state = array_merge($state, $extra);
        file_put_contents($this->lastOperationSnapshotPath, json_encode($state, JSON_PRETTY_PRINT | JSON_UNESCAPED_SLASHES) . PHP_EOL);
    }

    private function clearOperationState(): void
    {
        if (is_file($this->statePath)) {
            @unlink($this->statePath);
        }
    }

    private function readOperationState(): ?array
    {
        if (!is_file($this->statePath)) {
            return null;
        }

        $decoded = json_decode((string)file_get_contents($this->statePath), true);
        return is_array($decoded) ? $decoded : null;
    }

    private function nextUpdateWindowStartIso(string $window): string
    {
        $parts = explode('-', $window, 2);
        $start = trim((string)($parts[0] ?? '02:00'));
        if (!preg_match('/^\d{2}:\d{2}$/', $start)) {
            $start = '02:00';
        }

        $now = new DateTimeImmutable('now');
        [$hour, $minute] = array_map('intval', explode(':', $start));
        $candidate = $now->setTime($hour, $minute, 0);
        if ($candidate <= $now) {
            $candidate = $candidate->modify('+1 day');
        }

        return $candidate->format(DateTimeInterface::ATOM);
    }

    private function reloadConfigFromDisk(): void
    {
        $reloaded = AgentConfig::load($this->config->path);
        $this->config = $reloaded;
        $this->workerHttp = new HttpJsonClient((string)$this->config->get('workerBaseUrl', self::DEFAULT_WORKER_BASE_URL));
        $this->configureBrokerClient();
    }

    private function pollRequestTimeoutSeconds(int $waitSeconds): int
    {
        return max(5, $waitSeconds + 2);
    }

    private function iniBytes(string $value): ?int
    {
        $normalized = trim(strtolower($value));
        if ($normalized === '' || $normalized === '-1') {
            return null;
        }

        $unit = substr($normalized, -1);
        $number = (float)$normalized;
        return match ($unit) {
            'g' => (int)round($number * 1024 * 1024 * 1024),
            'm' => (int)round($number * 1024 * 1024),
            'k' => (int)round($number * 1024),
            default => is_numeric($normalized) ? (int)$normalized : null,
        };
    }
}

$configPath = null;
foreach ($argv as $index => $argument) {
    if ($argument === '--config' && isset($argv[$index + 1])) {
        $configPath = $argv[$index + 1];
    }
}

if ($configPath === null) {
    fwrite(STDERR, "Usage: php agent.php --config /path/to/config.json\n");
    exit(1);
}

$agent = new TruckwashEdgeAgent($configPath);
$agent->run();