From 5e072412dcb13c91a0a626fdba0330ad5e5c700a Mon Sep 17 00:00:00 2001 From: biersoeckli Date: Mon, 31 Aug 2026 10:50:24 +0000 Subject: [PATCH 1/2] fix: reconnect agent sandbox status stream --- .../sandboxes/agent-sandboxes-card.tsx | 146 +++++++++++------- .../overview/deployment-status-badge.tsx | 8 +- src/server/services/agent-runtime.service.ts | 25 ++- .../agent-runtime.service.unit.spec.ts | 32 +++- 4 files changed, 151 insertions(+), 60 deletions(-) diff --git a/src/app/project/agent/[agentId]/sandboxes/agent-sandboxes-card.tsx b/src/app/project/agent/[agentId]/sandboxes/agent-sandboxes-card.tsx index ae04d7ad..f92feaf6 100644 --- a/src/app/project/agent/[agentId]/sandboxes/agent-sandboxes-card.tsx +++ b/src/app/project/agent/[agentId]/sandboxes/agent-sandboxes-card.tsx @@ -1,6 +1,6 @@ 'use client'; -import { useEffect, useRef, useState } from "react"; +import { useEffect, useState } from "react"; import { Card, CardContent, CardDescription, CardHeader, CardTitle } from "@/components/ui/card"; import { Button } from "@/components/ui/button"; import { SimpleDataTable } from "@/components/custom/simple-data-table"; @@ -34,6 +34,9 @@ interface SandboxInfo { createdAt: string | null; } +const SSE_RETRY_BASE_DELAY_MS = 1_000; +const SSE_RETRY_MAX_DELAY_MS = 30_000; + export default function AgentSandboxesCard({ agentId, readonly, @@ -49,83 +52,118 @@ export default function AgentSandboxesCard({ const [sandboxes, setSandboxes] = useState([]); const [loading, setLoading] = useState(false); const [isConnected, setIsConnected] = useState(false); - const readerRef = useRef | null>(null); // SSE stream for live sandbox updates useEffect(() => { const controller = new AbortController(); + let stopped = false; + let reader: ReadableStreamDefaultReader | null = null; + let retryTimeout: ReturnType | null = null; + let resolveRetry: (() => void) | null = null; + let retryAttempt = 0; + + const waitForRetry = (delayMs: number) => new Promise(resolve => { + resolveRetry = resolve; + retryTimeout = setTimeout(() => { + retryTimeout = null; + resolveRetry = null; + resolve(); + }, delayMs); + }); const connectSse = async () => { - try { - const response = await fetch('/api/agent-sandboxes', { - method: 'POST', - headers: { 'Content-Type': 'text/event-stream' }, - body: JSON.stringify({ agentId }), - signal: controller.signal, - }); + while (!stopped) { + try { + const response = await fetch('/api/agent-sandboxes', { + method: 'POST', + headers: { 'Content-Type': 'text/event-stream' }, + body: JSON.stringify({ agentId }), + signal: controller.signal, + }); - if (!response.ok || !response.body) return; + if (!response.ok || !response.body) { + throw new Error(`SSE request failed with status ${response.status}`); + } - setIsConnected(true); - const reader = response.body - .pipeThrough(new TextDecoderStream()) - .getReader(); - readerRef.current = reader; + setIsConnected(true); + reader = response.body + .pipeThrough(new TextDecoderStream()) + .getReader(); - let buffer = ''; - while (true) { - const { value, done } = await reader.read(); - if (done) break; + let buffer = ''; + while (!stopped) { + const { value, done } = await reader.read(); + if (done) break; - buffer += value; + buffer += value; - // Parse SSE frames: split by double newline - const frames = buffer.split('\n\n'); - buffer = frames.pop() || ''; // keep incomplete frame in buffer + // Parse SSE frames: split by double newline + const frames = buffer.split('\n\n'); + buffer = frames.pop() || ''; // keep incomplete frame in buffer - for (const frame of frames) { - const lines = frame.split('\n'); - for (const line of lines) { - if (line.startsWith('data: ')) { - try { - const msg = JSON.parse(line.slice(6)); - if (msg.type === 'FULL' && Array.isArray(msg.data)) { - setSandboxes(ListUtils.dedupByName(msg.data, 'name')); - } else if (msg.type === 'ADDED' && msg.sandbox) { - setSandboxes(prev => { - if (prev.some(i => i.name === msg.sandbox.name)) return prev; - return [...prev, msg.sandbox]; - }); - } else if (msg.type === 'MODIFIED' && msg.sandbox) { - setSandboxes(prev => prev.map(i => - i.name === msg.sandbox.name ? msg.sandbox : i - )); - } else if (msg.type === 'DELETED' && msg.sandbox?.name) { - setSandboxes(prev => prev.filter(i => - i.name !== msg.sandbox.name - )); + for (const frame of frames) { + const lines = frame.split('\n'); + for (const line of lines) { + if (line.startsWith('data: ')) { + try { + const msg = JSON.parse(line.slice(6)); + if (msg.type === 'FULL' && Array.isArray(msg.data)) { + setSandboxes(ListUtils.dedupByName(msg.data, 'name')); + } else if (msg.type === 'ADDED' && msg.sandbox) { + setSandboxes(prev => { + if (prev.some(i => i.name === msg.sandbox.name)) return prev; + return [...prev, msg.sandbox]; + }); + } else if (msg.type === 'MODIFIED' && msg.sandbox) { + setSandboxes(prev => prev.map(i => + i.name === msg.sandbox.name ? msg.sandbox : i + )); + } else if (msg.type === 'DELETED' && msg.sandbox?.name) { + setSandboxes(prev => prev.filter(i => + i.name !== msg.sandbox.name + )); + } + } catch { + // Ignore malformed SSE payloads and keep the stream alive. } - } catch { - // ignore parse errors on partial chunks } } } } + } catch (err: any) { + if (err?.name !== 'AbortError' && !stopped) { + console.error('Agent sandboxes SSE error:', err); + } + } finally { + reader = null; + if (!stopped) { + setIsConnected(false); + } } - } catch (err: any) { - if (err?.name !== 'AbortError') { - console.error('Agent sandboxes SSE error:', err); - } - } finally { - setIsConnected(false); + + if (stopped) break; + + const exponentialDelay = Math.min( + SSE_RETRY_BASE_DELAY_MS * 2 ** retryAttempt, + SSE_RETRY_MAX_DELAY_MS, + ); + const jitteredDelay = exponentialDelay * (0.8 + Math.random() * 0.4); + retryAttempt += 1; + await waitForRetry(jitteredDelay); } }; - connectSse(); + void connectSse(); return () => { + stopped = true; controller.abort(); - readerRef.current?.cancel(); + if (retryTimeout) { + clearTimeout(retryTimeout); + retryTimeout = null; + } + resolveRetry?.(); + void reader?.cancel(); }; }, [agentId]); diff --git a/src/app/project/app/[appId]/overview/deployment-status-badge.tsx b/src/app/project/app/[appId]/overview/deployment-status-badge.tsx index e864cc60..5628be1a 100644 --- a/src/app/project/app/[appId]/overview/deployment-status-badge.tsx +++ b/src/app/project/app/[appId]/overview/deployment-status-badge.tsx @@ -20,6 +20,8 @@ function getTextForStatus(status: DeploymentStatus) { switch (status) { case 'SHUTDOWN': return 'Shutdown'; + case 'SHUTTING_DOWN': + return 'Stopping'; case 'BUILDING': return 'Building'; case 'ERROR': @@ -40,6 +42,8 @@ function getBackgroundColorForStatus(status: DeploymentStatus) { case 'SHUTDOWN': return 'bg-slate-100'; + case 'SHUTTING_DOWN': + return 'bg-yellow-100'; case 'ERROR': return 'bg-red-100'; case 'BUILDING': @@ -60,6 +64,8 @@ function getTextColorForStatus(status: DeploymentStatus) { case 'SHUTDOWN': return 'text-slate-800'; + case 'SHUTTING_DOWN': + return 'text-yellow-800'; case 'ERROR': return 'text-red-800'; case 'BUILDING': @@ -73,4 +79,4 @@ function getTextColorForStatus(status: DeploymentStatus) { default: return 'text-slate-800'; } -} \ No newline at end of file +} diff --git a/src/server/services/agent-runtime.service.ts b/src/server/services/agent-runtime.service.ts index 056b9081..5642d977 100644 --- a/src/server/services/agent-runtime.service.ts +++ b/src/server/services/agent-runtime.service.ts @@ -143,7 +143,11 @@ class AgentRuntimeService { } private resolveClaimStatus(claim: any): DeploymentStatus { - const conditions: Array<{ type: string; status: string; message?: string }> = + if (claim?.metadata?.deletionTimestamp) { + return 'SHUTTING_DOWN'; + } + + const conditions: Array<{ type: string; status: string; reason?: string; message?: string }> = claim?.status?.conditions || []; const ready = conditions.find((c) => @@ -153,13 +157,26 @@ class AgentRuntimeService { return 'DEPLOYED'; } - const failed = conditions.find((c) => - (c.type === 'Ready' || c.type === 'Available') && c.status === 'False', + const readinessCondition = conditions.find((c) => + c.type === 'Ready' || c.type === 'Available', ); - if (failed) { + if (readinessCondition?.reason === 'ClaimExpired' || readinessCondition?.reason === 'Expired') { + return 'SHUTTING_DOWN'; + } + + const terminalFailureReasons = new Set([ + 'TemplateNotFound', + 'WarmPoolNotFound', + 'InvalidMetadata', + 'EnvVarsInjectionRejected', + 'VolumeClaimTemplatesError', + 'ReconcilerError', + ]); + if (terminalFailureReasons.has(readinessCondition?.reason ?? '')) { return 'ERROR'; } + // Ready=False is the controller's normal state while a claim is being fulfilled. return 'DEPLOYING'; } diff --git a/src/server/services/agent-runtime.service.unit.spec.ts b/src/server/services/agent-runtime.service.unit.spec.ts index 200d16ff..913b47f7 100644 --- a/src/server/services/agent-runtime.service.unit.spec.ts +++ b/src/server/services/agent-runtime.service.unit.spec.ts @@ -399,7 +399,7 @@ describe('agent-runtime.service', () => { expect(status).toBe('DEPLOYED'); }); - it('returns ERROR when claim has failed condition', async () => { + it('returns DEPLOYING when the claim is not ready yet', async () => { vi.mocked(dataAccess.client.agent.findUnique).mockResolvedValue(mockAgent() as any); vi.mocked(agentSandboxAdapter.getSandboxClaim).mockResolvedValue({ apiVersion: 'extensions.agents.x-k8s.io/v1beta1', @@ -411,9 +411,39 @@ describe('agent-runtime.service', () => { const status = await agentRuntimeService.getAgentStatus(AGENT_ID); + expect(status).toBe('DEPLOYING'); + }); + + it('returns ERROR for a terminal SandboxClaim reconciliation failure', async () => { + vi.mocked(dataAccess.client.agent.findUnique).mockResolvedValue(mockAgent() as any); + vi.mocked(agentSandboxAdapter.getSandboxClaim).mockResolvedValue({ + apiVersion: 'extensions.agents.x-k8s.io/v1beta1', + kind: 'SandboxClaim', + metadata: { name: AGENT_ID }, + spec: { warmPoolRef: { name: AGENT_ID } }, + status: { conditions: [{ type: 'Ready', status: 'False', reason: 'WarmPoolNotFound' }] }, + } as any); + + const status = await agentRuntimeService.getAgentStatus(AGENT_ID); + expect(status).toBe('ERROR'); }); + it('returns SHUTTING_DOWN while the claim is terminating', async () => { + vi.mocked(dataAccess.client.agent.findUnique).mockResolvedValue(mockAgent() as any); + vi.mocked(agentSandboxAdapter.getSandboxClaim).mockResolvedValue({ + apiVersion: 'extensions.agents.x-k8s.io/v1beta1', + kind: 'SandboxClaim', + metadata: { name: AGENT_ID, deletionTimestamp: '2026-08-31T12:00:00Z' }, + spec: { warmPoolRef: { name: AGENT_ID } }, + status: { conditions: [{ type: 'Ready', status: 'True' }] }, + } as any); + + const status = await agentRuntimeService.getAgentStatus(AGENT_ID); + + expect(status).toBe('SHUTTING_DOWN'); + }); + it('compares status text for deployed to Running', async () => { vi.mocked(dataAccess.client.agent.findUnique).mockResolvedValue(mockAgent() as any); vi.mocked(agentSandboxAdapter.getSandboxClaim).mockResolvedValue(mockClaim(true) as any); From adbdd3a9b290d7c248e45b413232346e10849eca Mon Sep 17 00:00:00 2001 From: biersoeckli Date: Mon, 31 Aug 2026 11:03:48 +0000 Subject: [PATCH 2/2] test: wait for deployment details api --- .../integration/server/api/v1/api.integration.spec.ts | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/__tests__/integration/server/api/v1/api.integration.spec.ts b/src/__tests__/integration/server/api/v1/api.integration.spec.ts index 956ea692..39dccd99 100644 --- a/src/__tests__/integration/server/api/v1/api.integration.spec.ts +++ b/src/__tests__/integration/server/api/v1/api.integration.spec.ts @@ -335,6 +335,13 @@ describe('REST API v1 integration', () => { expect(deployResponse.deploymentId).toEqual(expect.any(String)); // retrieve deployment details + await expect.poll(async () => ( + await apiFetch(`/api/v1/apps/${app.id}/deploy/${deployResponse.deploymentId}`, apiKey) + ).status, { + timeout: 30_000, + interval: 1_000, + }).toBe(200); + const deploymentDetails = await expectApiJson( await apiFetch(`/api/v1/apps/${app.id}/deploy/${deployResponse.deploymentId}`, apiKey), );