diff --git a/src/Cli/WorkflowMode.php b/src/Cli/WorkflowMode.php index a5aa20d..52a1524 100644 --- a/src/Cli/WorkflowMode.php +++ b/src/Cli/WorkflowMode.php @@ -12,12 +12,14 @@ use Claw\Project\Issue; use Claw\Project\IssueStatus; use Claw\Project\ProjectStore; +use Claw\Project\RunStatus; use Claw\Project\Strategy; use Claw\Run\Triage; use Claw\Server; use Claw\ServerClient; use Claw\ServerLocator; use Claw\Trace\Level; +use Claw\Trace\TraceFormat; use Claw\Trace\TraceReader; /** @@ -262,23 +264,103 @@ private function runIssue(array $args, ?string $projectDir, ?Level $verbosity): // run to it rather than opening the db here. If none is running, start one and wait for it. $client = new ServerClient(new CurlHttpClient(), $this->appHome(), $this->root); + $key = $store->project()->id; + try { if ($client->running() === null) { fwrite(STDOUT, "claw run: no server running for this workspace — starting one…\n"); } $server = $client->ensure(); - $client->startRun($server, $store->project()->id, $issue->id); + $before = $client->latestRunId($server, $key, $issue->id); + $client->startRun($server, $key, $issue->id); + $runId = $client->awaitRun($server, $key, $issue->id, $before); } catch (ClawException $e) { fwrite(STDERR, 'claw run: ' . $e->getMessage() . "\n"); return 1; } - fwrite(STDOUT, "Run started for issue #{$issue->id} on the server at {$server['host']}:{$server['port']}.\n"); - fwrite(STDOUT, " watch it: claw log\n"); + if ($runId === null) { + // The run is going, we just could not attach a tail to it; the trace is still on the server. + fwrite(STDOUT, "Run started for issue #{$issue->id}; watch it with: claw log\n"); - return 0; + return 0; + } + + return $this->tailRun($client, $server, $key, $issue->id, $runId, $verbosity); + } + + /** + * Follow a server-side run to the end, printing its trace as it arrives — the terminal's live view + * of a run it no longer hosts. Polls the same trace rows an SSE subscriber gets and renders them + * exactly as an in-process run would ({@see ConsoleTraceSink}), so detaching (Ctrl-C) loses only the + * view: the run keeps going on the server, and `claw log` shows the rest. + * + * @param array{host: string, port: int} $server + */ + private function tailRun(ServerClient $client, array $server, string $key, string $issueId, string $runId, ?Level $verbosity): int + { + $threshold = ($verbosity ?? Level::Info)->value; + $color = stream_isatty(STDOUT) && getenv('NO_COLOR') === false; + $since = 0; + $flaggedGate = false; + + while (true) { + foreach ($client->trace($server, $key, $runId, $since) as $row) { + $since = max($since, \is_int($row['seq'] ?? null) ? $row['seq'] : $since); + + if ((\is_int($row['level'] ?? null) ? $row['level'] : 0) < $threshold) { + continue; + } + + $data = \is_array($row['data'] ?? null) ? $row['data'] : []; + $head = TraceFormat::line((string) ($row['phase'] ?? ''), (string) ($row['type'] ?? ''), $data); + $depth = \is_int($row['depth'] ?? null) ? $row['depth'] : 0; + fwrite(STDOUT, str_repeat(' ', $depth) . TraceFormat::paint((string) ($row['type'] ?? ''), $head, $color) . "\n"); + + if (($row['type'] ?? '') === 'question' && !$flaggedGate) { + $flaggedGate = true; + fwrite(STDOUT, " ⏸ waiting for a human — answer it in the dashboard; Ctrl-C detaches (the run continues).\n"); + } + + if (($row['type'] ?? '') === 'answer') { + $flaggedGate = false; + } + } + + $status = $this->runStatus($client->runs($server, $key, $issueId), $runId); + + if ($status === RunStatus::Done->value) { + fwrite(STDOUT, "Run #{$runId} finished issue #{$issueId}.\n"); + + return 0; + } + + if ($status === RunStatus::Failed->value) { + fwrite(STDERR, "Run #{$runId} failed; see `claw log {$runId}`.\n"); + + return 1; + } + + usleep(500_000); + } + } + + /** + * A run's current status from a runs listing, or '' when it is not there. + * + * @param list $runs + */ + private function runStatus(array $runs, string $runId): string + { + foreach ($runs as $run) { + if ($run['id'] === $runId) { + return $run['status']; + } + } + + return ''; } /** diff --git a/src/Server.php b/src/Server.php index c9e91b0..ec5fcd3 100644 --- a/src/Server.php +++ b/src/Server.php @@ -353,6 +353,12 @@ public function handle(HttpRequest $request, HttpResponse $response): void return; } + if (\preg_match('#^/api/projects/([^/]+)/issues/([^/]+)/runs$#', $path, $matches)) { + $response->json($this->store($matches[1])->runsFor($matches[2])); + + return; + } + if (\preg_match('#^/api/projects/([^/]+)/runs/([^/]+)/stream$#', $path, $matches)) { $this->stream($request, $response, $matches[1], $matches[2]); diff --git a/src/ServerClient.php b/src/ServerClient.php index 4b93f96..d961b3e 100644 --- a/src/ServerClient.php +++ b/src/ServerClient.php @@ -20,6 +20,7 @@ final readonly class ServerClient { private const int SPAWN_TIMEOUT_MS = 10_000; + private const int ATTACH_TIMEOUT_MS = 10_000; private const int POLL_MS = 100; private ServerLocator $locator; @@ -92,6 +93,125 @@ public function startRun(array $server, string $key, string $issueId): void throw new ClawException("the server refused to start the run (HTTP {$response->status})"); } + /** + * The runs recorded for an issue, oldest first — how the CLI finds the run a start produced, since + * the id is minted inside the run and not returned by the start. + * + * @param array{host: string, port: int} $server + * + * @return list + */ + public function runs(array $server, string $key, string $issueId): array + { + $url = sprintf( + 'http://%s:%d/api/projects/%s/issues/%s/runs', + $server['host'], + $server['port'], + rawurlencode($key), + rawurlencode($issueId), + ); + + $runs = []; + + foreach ($this->getList($url) as $run) { + if (\is_array($run) && isset($run['id'])) { + $runs[] = ['id' => (string) $run['id'], 'status' => (string) ($run['status'] ?? '')]; + } + } + + return $runs; + } + + /** + * A run's trace rows past $since — the poll the tail renders from, the same rows an SSE subscriber + * would receive. + * + * @param array{host: string, port: int} $server + * + * @return list> + */ + public function trace(array $server, string $key, string $runId, int $since): array + { + $url = sprintf( + 'http://%s:%d/api/projects/%s/runs/%s/trace?since=%d', + $server['host'], + $server['port'], + rawurlencode($key), + rawurlencode($runId), + $since, + ); + + $rows = []; + + foreach ($this->getList($url) as $row) { + if (\is_array($row)) { + $rows[] = $row; + } + } + + return $rows; + } + + /** + * GET a URL and decode a top-level JSON list, or [] on any error. These endpoints return bare + * arrays (a run list, trace rows), which {@see HttpResponse::json()} refuses by contract — it is for + * string-keyed objects — so the body is decoded here instead. + * + * @return list + */ + private function getList(string $url): array + { + $response = $this->http->get($url); + + if (!$response->isOk()) { + return []; + } + + $data = json_decode($response->body, true); + + return \is_array($data) && array_is_list($data) ? $data : []; + } + + /** + * The highest run id recorded for an issue, or 0 when it has none — the mark a new run must beat. + * + * @param array{host: string, port: int} $server + */ + public function latestRunId(array $server, string $key, string $issueId): int + { + $highest = 0; + + foreach ($this->runs($server, $key, $issueId) as $run) { + $highest = max($highest, (int) $run['id']); + } + + return $highest; + } + + /** + * Wait for the run a start just produced: the first run whose id beats $afterId. Null when none + * appears in time — the caller falls back to telling the user to watch elsewhere. + * + * @param array{host: string, port: int} $server + */ + public function awaitRun(array $server, string $key, string $issueId, int $afterId): ?string + { + $waited = 0; + + while ($waited < self::ATTACH_TIMEOUT_MS) { + foreach ($this->runs($server, $key, $issueId) as $run) { + if ((int) $run['id'] > $afterId) { + return $run['id']; + } + } + + usleep(self::POLL_MS * 1000); + $waited += self::POLL_MS; + } + + return null; + } + /** * The argv that launches a server, or null when one cannot be built (bin/claw is not where it * should be). Returned as a list, not a shell string, so it carries no quoting to get wrong and can diff --git a/tests/ServerClientTest.php b/tests/ServerClientTest.php index 6e638ab..2164b15 100644 --- a/tests/ServerClientTest.php +++ b/tests/ServerClientTest.php @@ -113,4 +113,48 @@ public function noSpawnCommandWhenBinClawIsAbsent(): void { Assert::same(ServerClient::spawnCommand('/nowhere/at/all', '127.0.0.1', 8787), null); } + + #[Test] + public function latestRunIdIsTheHighestNotTheLast(): void + { + $client = $this->client(new HttpResponse(200, '[{"id":"3","status":"done"},{"id":"11","status":"running"},{"id":"7","status":"failed"}]')); + + Assert::same($client->latestRunId(['host' => '127.0.0.1', 'port' => 8787], 'proj', '5'), 11); + } + + #[Test] + public function latestRunIdIsZeroWhenThereAreNoRuns(): void + { + Assert::same( + $this->client(new HttpResponse(200, '[]'))->latestRunId(['host' => '127.0.0.1', 'port' => 8787], 'proj', '5'), + 0, + ); + } + + #[Test] + public function awaitRunReturnsTheFirstRunBeyondTheMark(): void + { + $client = $this->client(new HttpResponse(200, '[{"id":"9","status":"running"}]')); + + Assert::same($client->awaitRun(['host' => '127.0.0.1', 'port' => 8787], 'proj', '5', 8), '9'); + } + + #[Test] + public function traceReturnsTheRowsVerbatim(): void + { + $client = $this->client(new HttpResponse(200, '[{"seq":1,"depth":0,"type":"info","phase":"run","level":20,"data":{}}]')); + + $rows = $client->trace(['host' => '127.0.0.1', 'port' => 8787], 'proj', '9', 0); + + Assert::same(\count($rows), 1); + Assert::same($rows[0]['seq'] ?? null, 1); + } + + #[Test] + public function traceIsEmptyOnAnErrorResponse(): void + { + $client = $this->client(new HttpResponse(404, 'nope')); + + Assert::same($client->trace(['host' => '127.0.0.1', 'port' => 8787], 'proj', '9', 0), []); + } }