Skip to content

fix: reconnect agent sandbox status stream #119

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
biersoeckli merged 2 commits into from
Aug 31, 2026
Merged

fix: reconnect agent sandbox status stream #119

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5e07241
fix: reconnect agent sandbox status stream
biersoeckli Aug 31, 2026
adbdd3a
test: wait for deployment details api
biersoeckli Aug 31, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 7 additions & 0 deletions src/__tests__/integration/server/api/v1/api.integration.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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),
);
Expand Down
146 changes: 92 additions & 54 deletions src/app/project/agent/[agentId]/sandboxes/agent-sandboxes-card.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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";
Expand Down Expand Up @@ -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,
Expand All @@ -49,83 +52,118 @@ export default function AgentSandboxesCard({
const [sandboxes, setSandboxes] = useState<SandboxInfo[]>([]);
const [loading, setLoading] = useState(false);
const [isConnected, setIsConnected] = useState(false);
const readerRef = useRef<ReadableStreamDefaultReader<string> | null>(null);

// SSE stream for live sandbox updates
useEffect(() => {
const controller = new AbortController();
let stopped = false;
let reader: ReadableStreamDefaultReader<string> | null = null;
let retryTimeout: ReturnType<typeof setTimeout> | null = null;
let resolveRetry: (() => void) | null = null;
let retryAttempt = 0;

const waitForRetry = (delayMs: number) => new Promise<void>(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]);

Expand Down
8 changes: 7 additions & 1 deletion src/app/project/app/[appId]/overview/deployment-status-badge.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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':
Expand All @@ -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':
Expand All @@ -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':
Expand All @@ -73,4 +79,4 @@ function getTextColorForStatus(status: DeploymentStatus) {
default:
return 'text-slate-800';
}
}
}
25 changes: 21 additions & 4 deletions src/server/services/agent-runtime.service.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) =>
Expand All @@ -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';
}

Expand Down
32 changes: 31 additions & 1 deletion src/server/services/agent-runtime.service.unit.spec.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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',
Expand All @@ -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);
Expand Down
Loading