From cd7109f669226a96013ecb4fc4edf9830c931a11 Mon Sep 17 00:00:00 2001 From: Shekhar Mudarapu Date: Fri, 23 Oct 2026 00:29:36 -0400 Subject: [PATCH 1/3] feat(budgets): tenant-wide daily spend cap enforced on agentless and agent calls --- services/control-plane/cmd/server/main.go | 3 + .../internal/handlers/budgets.go | 120 ++++++++++++++++++ .../internal/handlers/llm_proxy.go | 15 +++ .../internal/handlers/tenant_budget_test.go | 116 +++++++++++++++++ 4 files changed, 254 insertions(+) create mode 100644 services/control-plane/internal/handlers/tenant_budget_test.go diff --git a/services/control-plane/cmd/server/main.go b/services/control-plane/cmd/server/main.go index a83f9167..61f28dd8 100644 --- a/services/control-plane/cmd/server/main.go +++ b/services/control-plane/cmd/server/main.go @@ -634,6 +634,9 @@ func main() { httpMux.HandleFunc("GET /v1/agents/{name}/budget", budgetHandler.GetBudget) httpMux.HandleFunc("DELETE /v1/agents/{name}/budget", authHandler.WithScope(handlers.ScopeBudgetsWrite, budgetHandler.DeleteBudget)) httpMux.HandleFunc("GET /v1/budgets", budgetHandler.ListBudgets) + httpMux.HandleFunc("PUT /v1/budgets/tenant", authHandler.WithScope(handlers.ScopeBudgetsWrite, budgetHandler.UpsertTenantBudget)) + httpMux.HandleFunc("GET /v1/budgets/tenant", budgetHandler.GetTenantBudget) + httpMux.HandleFunc("DELETE /v1/budgets/tenant", authHandler.WithScope(handlers.ScopeBudgetsWrite, budgetHandler.DeleteTenantBudget)) // Marketplace — publish, fork, star agents. httpMux.HandleFunc("GET /v1/marketplace", marketplaceHandler.List) diff --git a/services/control-plane/internal/handlers/budgets.go b/services/control-plane/internal/handlers/budgets.go index de4bd176..75229437 100644 --- a/services/control-plane/internal/handlers/budgets.go +++ b/services/control-plane/internal/handlers/budgets.go @@ -17,6 +17,7 @@ package handlers import ( "context" "encoding/json" + "errors" "fmt" "net/http" "time" @@ -211,6 +212,84 @@ func (h *BudgetHandler) ListBudgets(w http.ResponseWriter, r *http.Request) { writeJSON(w, http.StatusOK, result) } +// UpsertTenantBudget handles PUT /v1/budgets/tenant. Body: {maxCostUsdPerDay, hardFail}. +func (h *BudgetHandler) UpsertTenantBudget(w http.ResponseWriter, r *http.Request) { + ctx, tenantID, err := authCtx(h.auth, r) + if err != nil { + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"}) + return + } + var body budgetDTO + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid body"}) + return + } + if body.MaxCostUsdPerDay == nil || *body.MaxCostUsdPerDay <= 0 { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "maxCostUsdPerDay must be > 0"}) + return + } + err = h.srv.WithTenant(ctx, func(tx pgx.Tx) error { + _, e := tx.Exec(ctx, ` + INSERT INTO agent_budgets (tenant_id, agent_name, max_cost_usd_per_day, hard_fail) + VALUES ($1,$2,$3,$4) + ON CONFLICT (tenant_id, agent_name) DO UPDATE SET + max_cost_usd_per_day = EXCLUDED.max_cost_usd_per_day, + hard_fail = EXCLUDED.hard_fail, + updated_at = now() + `, tenantID, TenantBudgetAgent, body.MaxCostUsdPerDay, body.HardFail) + return e + }) + if err != nil { + h.logger().Error("upsert tenant budget failed", zap.Error(err)) + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "failed to save tenant budget"}) + return + } + writeJSON(w, http.StatusOK, map[string]string{"status": "saved"}) +} + +// GetTenantBudget handles GET /v1/budgets/tenant. +func (h *BudgetHandler) GetTenantBudget(w http.ResponseWriter, r *http.Request) { + ctx, tenantID, err := authCtx(h.auth, r) + if err != nil { + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"}) + return + } + dto := budgetDTO{AgentName: TenantBudgetAgent} + err = h.srv.WithTenant(ctx, func(tx pgx.Tx) error { + return tx.QueryRow(ctx, ` + SELECT max_cost_usd_per_day, hard_fail, updated_at + FROM agent_budgets WHERE tenant_id = $1 AND agent_name = $2 + `, tenantID, TenantBudgetAgent).Scan(&dto.MaxCostUsdPerDay, &dto.HardFail, &dto.UpdatedAt) + }) + if errors.Is(err, pgx.ErrNoRows) { + writeJSON(w, http.StatusNotFound, map[string]string{"error": "no tenant budget configured"}) + return + } + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "failed"}) + return + } + writeJSON(w, http.StatusOK, dto) +} + +// DeleteTenantBudget handles DELETE /v1/budgets/tenant. +func (h *BudgetHandler) DeleteTenantBudget(w http.ResponseWriter, r *http.Request) { + ctx, tenantID, err := authCtx(h.auth, r) + if err != nil { + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"}) + return + } + err = h.srv.WithTenant(ctx, func(tx pgx.Tx) error { + _, e := tx.Exec(ctx, `DELETE FROM agent_budgets WHERE tenant_id = $1 AND agent_name = $2`, tenantID, TenantBudgetAgent) + return e + }) + if err != nil { + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "failed"}) + return + } + writeJSON(w, http.StatusOK, map[string]string{"status": "removed"}) +} + // ---------- Enforcement ---------- // BudgetCheckResult is the return value of CheckBudget. @@ -228,7 +307,48 @@ type BudgetCheckResult struct { // // If no budget is configured, Allowed=true, HardFail=false. // If configured and violating, Allowed=false and Reason explains. +// +// The tenant-wide cap (agent_budgets row with agent_name = TenantBudgetAgent) +// is checked first, against spend summed across every agent. func CheckBudget(ctx context.Context, pool *pgxpool.Pool, tenantID, agentName string, estCostUsd float64) BudgetCheckResult { + if tc := CheckTenantBudget(ctx, pool, tenantID, estCostUsd); !tc.Allowed { + return tc + } + return checkAgentBudget(ctx, pool, tenantID, agentName, estCostUsd) +} + +// TenantBudgetAgent is the reserved agent_budgets.agent_name holding the +// tenant-wide daily spend cap. Reusing agent_budgets (UNIQUE tenant_id, +// agent_name) avoids a new table, migration and RLS catalog entry; "*" cannot +// collide with a real agent name. +const TenantBudgetAgent = "*" + +// CheckTenantBudget enforces the tenant-wide max_cost_usd_per_day against +// SUM(agent_usage_daily.cost_usd) over all agents (incl. "__agentless__"). +// No row / no cap / query error -> allowed (unlimited, unchanged behavior). +func CheckTenantBudget(ctx context.Context, pool *pgxpool.Pool, tenantID string, estCostUsd float64) BudgetCheckResult { + var maxCost *float64 + var hardFail bool + if err := pool.QueryRow(ctx, ` + SELECT max_cost_usd_per_day::float8, hard_fail + FROM agent_budgets WHERE tenant_id = $1 AND agent_name = $2 + `, tenantID, TenantBudgetAgent).Scan(&maxCost, &hardFail); err != nil || maxCost == nil || *maxCost <= 0 { + return BudgetCheckResult{Allowed: true} + } + var spent float64 + _ = pool.QueryRow(ctx, ` + SELECT COALESCE(SUM(cost_usd),0)::float8 FROM agent_usage_daily + WHERE tenant_id = $1 AND usage_date = $2 + `, tenantID, usageDate(time.Now())).Scan(&spent) + res := BudgetCheckResult{Allowed: true, HardFail: hardFail, SpentToday: spent} + if spent+estCostUsd > *maxCost { + res.Allowed = false + res.Reason = fmt.Sprintf("tenant-wide daily spend would reach $%.4f exceeding tenant limit $%.4f", spent+estCostUsd, *maxCost) + } + return res +} + +func checkAgentBudget(ctx context.Context, pool *pgxpool.Pool, tenantID, agentName string, estCostUsd float64) BudgetCheckResult { var b loadedBudget var toolLimitsJSON []byte var maxCostPerDayRaw, maxCostPerRunRaw *float64 diff --git a/services/control-plane/internal/handlers/llm_proxy.go b/services/control-plane/internal/handlers/llm_proxy.go index 60c55dce..86e46711 100644 --- a/services/control-plane/internal/handlers/llm_proxy.go +++ b/services/control-plane/internal/handlers/llm_proxy.go @@ -1543,6 +1543,21 @@ func (h *LlmProxyHandler) Complete(w http.ResponseWriter, r *http.Request) { return } + // Agentless calls have no per-agent budget but still count against the + // tenant-wide cap. Checked before any provider call. (Agent-scoped calls + // get the same check inside CheckBudget below.) + if req.AgentName == "" { + if tc := CheckTenantBudget(ctx, h.srv.Pool, tenantID, 0); !tc.Allowed && tc.HardFail { + h.logger().Warn("completions: agentless call blocked by tenant budget", + zap.String("tenant_id", tenantID), zap.String("reason", tc.Reason)) + writeJSON(w, http.StatusPaymentRequired, map[string]string{ + "error": "budget limit exceeded", + "reason": tc.Reason, + }) + return + } + } + provider, model := h.resolveModelForTenant(ctx, tenantID, req.Model) // For NON-streaming, NON-agent-scoped completions ('Generate with AI' diff --git a/services/control-plane/internal/handlers/tenant_budget_test.go b/services/control-plane/internal/handlers/tenant_budget_test.go new file mode 100644 index 00000000..0d5c4b79 --- /dev/null +++ b/services/control-plane/internal/handlers/tenant_budget_test.go @@ -0,0 +1,116 @@ +package handlers + +import ( + "bytes" + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" +) + +func setTenantCap(t *testing.T, tenantID string, capUsd float64, hardFail bool) { + t.Helper() + pool := openTestPool(t) + if _, err := pool.Exec(context.Background(), ` + INSERT INTO agent_budgets (tenant_id, agent_name, max_cost_usd_per_day, hard_fail) + VALUES ($1, $2, $3, $4) + ON CONFLICT (tenant_id, agent_name) DO UPDATE SET + max_cost_usd_per_day = EXCLUDED.max_cost_usd_per_day, hard_fail = EXCLUDED.hard_fail + `, tenantID, TenantBudgetAgent, capUsd, hardFail); err != nil { + t.Fatalf("set tenant cap: %v", err) + } + t.Cleanup(func() { + _, _ = pool.Exec(context.Background(), `DELETE FROM agent_budgets WHERE tenant_id = $1 AND agent_name = $2`, tenantID, TenantBudgetAgent) + _, _ = pool.Exec(context.Background(), `DELETE FROM agent_usage_daily WHERE tenant_id = $1`, tenantID) + }) +} + +func postAgentlessCompletion(t *testing.T, h *LlmProxyHandler, tenantID string, stream bool) *httptest.ResponseRecorder { + t.Helper() + body, _ := json.Marshal(map[string]any{ + "stream": stream, + "messages": []map[string]string{{"role": "user", "content": "hi"}}, + }) + req := httptest.NewRequest(http.MethodPost, "/v1/completions", bytes.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+mintTestToken(t, tenantID, "user-"+tenantID, "owner")) + rr := httptest.NewRecorder() + h.Complete(rr, req) + return rr +} + +func TestTenantBudget_BlocksAgentlessCompletion402(t *testing.T) { + h, _ := newLlmProxyTestHandler(t) + pool := openTestPool(t) + tenant := seedCommitmentTenant(t, pool) + setTenantCap(t, tenant, 1.00, true) + if err := RecordUsage(context.Background(), pool, tenant, "__agentless__", 10, 10, 1.50, nil); err != nil { + t.Fatal(err) + } + for _, stream := range []bool{false, true} { + rr := postAgentlessCompletion(t, h, tenant, stream) + if rr.Code != http.StatusPaymentRequired { + t.Fatalf("stream=%v: want 402 before any provider call, got %d: %s", stream, rr.Code, rr.Body.String()) + } + var resp map[string]string + _ = json.Unmarshal(rr.Body.Bytes(), &resp) + if resp["reason"] == "" { + t.Errorf("stream=%v: 402 missing reason: %v", stream, resp) + } + } +} + +func TestTenantBudget_SumsAcrossAgentsAndAppliesToAgentChecks(t *testing.T) { + pool := openTestPool(t) + ctx := context.Background() + tenant := seedCommitmentTenant(t, pool) + setTenantCap(t, tenant, 1.00, true) + for _, a := range []string{"agent-a", "agent-b"} { + if err := RecordUsage(ctx, pool, tenant, a, 1, 1, 0.40, nil); err != nil { + t.Fatal(err) + } + } + // 0.80 spent: a 0.10 call fits, a 0.30 call does not. + if r := CheckBudget(ctx, pool, tenant, "agent-c", 0.10); !r.Allowed { + t.Fatalf("0.80+0.10 under cap should be allowed: %s", r.Reason) + } + r := CheckBudget(ctx, pool, tenant, "agent-c", 0.30) + if r.Allowed || !r.HardFail || r.Reason == "" { + t.Fatalf("0.80+0.30 over cap should hard-block with reason, got %+v", r) + } +} + +func TestTenantBudget_CrossTenantIsolation(t *testing.T) { + pool := openTestPool(t) + ctx := context.Background() + capped := seedCommitmentTenant(t, pool) + other := seedCommitmentTenant(t, pool) + setTenantCap(t, capped, 1.00, true) + if err := RecordUsage(ctx, pool, other, "agent-x", 1, 1, 50, nil); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM agent_usage_daily WHERE tenant_id = $1`, other) }) + if r := CheckTenantBudget(ctx, pool, capped, 0.5); !r.Allowed { + t.Fatalf("other tenant's spend must not count against capped tenant: %s", r.Reason) + } + if r := CheckTenantBudget(ctx, pool, other, 100); !r.Allowed { + t.Fatal("tenant without a cap row must be unlimited") + } +} + +func TestTenantBudget_NoRowUnlimited_SoftFailDoesNotBlock(t *testing.T) { + pool := openTestPool(t) + ctx := context.Background() + tenant := seedCommitmentTenant(t, pool) + if err := RecordUsage(ctx, pool, tenant, "agent-a", 1, 1, 1000, nil); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _, _ = pool.Exec(ctx, `DELETE FROM agent_usage_daily WHERE tenant_id = $1`, tenant) }) + if r := CheckBudget(ctx, pool, tenant, "agent-a", 5); !r.Allowed { + t.Fatalf("no cap row => unlimited: %s", r.Reason) + } + setTenantCap(t, tenant, 1.00, false) + if r := CheckTenantBudget(ctx, pool, tenant, 0); r.Allowed || r.HardFail { + t.Fatalf("soft cap should report breach with HardFail=false, got %+v", r) + } +} From a047709a6cce561e825de81ca469e68191e63355 Mon Sep 17 00:00:00 2001 From: Shekhar Mudarapu Date: Fri, 23 Oct 2026 00:31:27 -0400 Subject: [PATCH 2/3] fix(control-plane): hide the tenant-wide '*' cap from the per-agent budget list --- .../internal/handlers/budgets.go | 2 +- .../internal/handlers/tenant_budget_test.go | 23 +++++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/services/control-plane/internal/handlers/budgets.go b/services/control-plane/internal/handlers/budgets.go index 75229437..871f0de5 100644 --- a/services/control-plane/internal/handlers/budgets.go +++ b/services/control-plane/internal/handlers/budgets.go @@ -184,7 +184,7 @@ func (h *BudgetHandler) ListBudgets(w http.ResponseWriter, r *http.Request) { rows, qErr := tx.Query(ctx, ` SELECT agent_name, max_cost_usd_per_day, max_cost_usd_per_run, max_tokens_per_day, max_runs_per_day, tool_limits, hard_fail, notify_at_pct, updated_at - FROM agent_budgets WHERE tenant_id = $1 ORDER BY agent_name + FROM agent_budgets WHERE tenant_id = $1 AND agent_name <> '*' ORDER BY agent_name -- the tenant-wide cap lives at /v1/budgets/tenant `, tenantID) if qErr != nil { return qErr diff --git a/services/control-plane/internal/handlers/tenant_budget_test.go b/services/control-plane/internal/handlers/tenant_budget_test.go index 0d5c4b79..af7b9e09 100644 --- a/services/control-plane/internal/handlers/tenant_budget_test.go +++ b/services/control-plane/internal/handlers/tenant_budget_test.go @@ -7,6 +7,9 @@ import ( "net/http" "net/http/httptest" "testing" + + "github.com/dshakes/lantern/services/control-plane/internal/server" + "go.uber.org/zap" ) func setTenantCap(t *testing.T, tenantID string, capUsd float64, hardFail bool) { @@ -114,3 +117,23 @@ func TestTenantBudget_NoRowUnlimited_SoftFailDoesNotBlock(t *testing.T) { t.Fatalf("soft cap should report breach with HardFail=false, got %+v", r) } } + +// The tenant-wide "*" row is not an agent: it must not appear in the per-agent +// list the dashboard Budgets page renders. +func TestTenantBudget_HiddenFromAgentBudgetList(t *testing.T) { + pool := openTestPool(t) + tenant := seedCommitmentTenant(t, pool) + setTenantCap(t, tenant, 5.00, true) + srv := &server.Server{Pool: pool, Logger: zap.NewNop()} + h := NewBudgetHandler(srv, NewAuthHandler(srv, testJWTSecret)) + req := httptest.NewRequest(http.MethodGet, "/v1/budgets", nil) + req.Header.Set("Authorization", "Bearer "+mintTestToken(t, tenant, "user-"+tenant, "owner")) + rr := httptest.NewRecorder() + h.ListBudgets(rr, req) + if rr.Code != http.StatusOK { + t.Fatalf("list budgets: %d %s", rr.Code, rr.Body.String()) + } + if bytes.Contains(rr.Body.Bytes(), []byte(`"`+TenantBudgetAgent+`"`)) { + t.Errorf("tenant-wide row leaked into agent budget list: %s", rr.Body.String()) + } +} From a1302f918a857740aa47f20d1daa51972a0023b9 Mon Sep 17 00:00:00 2001 From: Shekhar Mudarapu Date: Fri, 23 Oct 2026 00:33:51 -0400 Subject: [PATCH 3/3] feat(sdk,web): runtime pause/resume/fork + streaming exec across TS/Python SDKs and VM page --- .../app/(dashboard)/runtime/[vmId]/page.tsx | 165 +++++++++++++++++- .../app/(dashboard)/runtime/cockpit-ui.tsx | 1 + .../app/(dashboard)/runtime/fleet-grid.tsx | 2 +- apps/web/lib/runtime-api.ts | 13 +- apps/web/lib/runtime-types.ts | 5 + packages/sdk-python/lantern/client.py | 115 +++++++++++- .../lantern/integrations/runtime_client.py | 12 ++ packages/sdk-python/lantern/types.py | 35 ++++ packages/sdk-python/tests/test_runtime.py | 92 ++++++++++ packages/sdk-ts/__tests__/runtime.test.ts | 98 +++++++++++ packages/sdk-ts/src/client.ts | 89 ++++++++++ packages/sdk-ts/src/index.ts | 8 + packages/sdk-ts/src/types.ts | 48 +++++ 13 files changed, 675 insertions(+), 8 deletions(-) create mode 100644 packages/sdk-python/tests/test_runtime.py create mode 100644 packages/sdk-ts/__tests__/runtime.test.ts diff --git a/apps/web/app/(dashboard)/runtime/[vmId]/page.tsx b/apps/web/app/(dashboard)/runtime/[vmId]/page.tsx index 0480bdc9..deecf5f6 100644 --- a/apps/web/app/(dashboard)/runtime/[vmId]/page.tsx +++ b/apps/web/app/(dashboard)/runtime/[vmId]/page.tsx @@ -10,15 +10,35 @@ import { Cpu, Activity, StopCircle, + Pause, + Play, + GitFork, } from "lucide-react"; import { formatDistanceToNow } from "date-fns"; import clsx from "clsx"; import { Button } from "@/components/button"; -import { ConfirmModal } from "@/components/modal"; +import Link from "next/link"; +import { ConfirmModal, Modal, ModalField } from "@/components/modal"; import { PageHeader } from "@/components/page-header"; import { PageSkeleton } from "@/components/skeleton"; import { useToast } from "@/components/toast"; -import { runtimeApi, UnauthorizedError } from "@/lib/runtime-api"; +import { HttpError, runtimeApi, UnauthorizedError } from "@/lib/runtime-api"; + +// Response shapes mirror POST /v1/runtime/vms/{id}/{pause,resume,fork}. +interface ForkResult { + sourceVmId: string; + snapshotId: string; + forks: Array<{ vmId: string; state: string; error?: string }>; +} + +/** Friendly copy for the contract's 402 (quota) / 409 (wrong state). */ +function lifecycleError(verb: string, err: unknown): string { + if (err instanceof HttpError) { + if (err.status === 402) return `${verb} blocked: runtime quota exceeded`; + if (err.status === 409) return `${verb} not possible in the VM's current state`; + } + return `${verb} failed: ` + (err instanceof Error ? err.message : String(err)); +} // Shapes mirror the control-plane JSON (camelCase tags on vmRow + the // events array — NOT snake_case, and the key is `events` not `audit`). @@ -34,6 +54,8 @@ interface VmDetail { createdAt: string; terminatedAt?: string | null; lastHeartbeatAt?: string | null; + snapshotId?: string | null; + parentVmId?: string | null; }; events: Array<{ id: number; action: string; attrs: Record; at: string }>; } @@ -47,6 +69,9 @@ export default function RuntimeVmPage() { const [logs, setLogs] = useState([]); const [terminating, setTerminating] = useState(false); const [confirmTerminate, setConfirmTerminate] = useState(false); + const [busy, setBusy] = useState<"pause" | "resume" | "fork" | null>(null); + const [forkOpen, setForkOpen] = useState(false); + const [forkCount, setForkCount] = useState("1"); const logBoxRef = useRef(null); // Initial + polling load of details + audit. Dedup errors so a 401 / 500 @@ -124,6 +149,66 @@ export default function RuntimeVmPage() { } }, [vmId, toast]); + const reload = useCallback(async () => { + if (!vmId) return; + try { + setDetail(await runtimeApi.get(`/v1/runtime/vms/${vmId}`)); + } catch { + /* the 5s poll will retry */ + } + }, [vmId]); + + const pause = useCallback(async () => { + if (!vmId) return; + setBusy("pause"); + try { + await runtimeApi.post(`/v1/runtime/vms/${vmId}/pause`); + toast.success("VM paused"); + await reload(); + } catch (err) { + if (!(err instanceof UnauthorizedError)) toast.error(lifecycleError("Pause", err)); + } finally { + setBusy(null); + } + }, [vmId, toast, reload]); + + const resume = useCallback(async () => { + if (!vmId) return; + setBusy("resume"); + try { + await runtimeApi.post(`/v1/runtime/vms/${vmId}/resume`); + toast.success("VM resumed"); + await reload(); + } catch (err) { + if (!(err instanceof UnauthorizedError)) toast.error(lifecycleError("Resume", err)); + } finally { + setBusy(null); + } + }, [vmId, toast, reload]); + + const forkCountNum = Number(forkCount); + const forkCountValid = Number.isInteger(forkCountNum) && forkCountNum >= 1 && forkCountNum <= 10; + + const fork = useCallback(async () => { + if (!vmId || !forkCountValid) return; + setBusy("fork"); + try { + const r = await runtimeApi.post(`/v1/runtime/vms/${vmId}/fork`, { count: forkCountNum }); + const failed = r.forks.filter((f) => f.error); + if (failed.length === 0) { + toast.success(`Forked ${r.forks.length} VM${r.forks.length === 1 ? "" : "s"}`); + } else { + toast.warning(`Forked ${r.forks.length - failed.length} of ${r.forks.length}; first error: ${failed[0].error}`); + } + setForkOpen(false); + await reload(); + } catch (err) { + if (!(err instanceof UnauthorizedError)) toast.error(lifecycleError("Fork", err)); + } finally { + setBusy(null); + } + }, [vmId, forkCountValid, forkCountNum, toast, reload]); + if (!detail) return ; const v = detail.vm; @@ -145,9 +230,26 @@ export default function RuntimeVmPage() { } action={ active ? ( - +
+ {v.state === "running" && ( + + )} + {v.state === "paused" && ( + + )} + {(v.state === "running" || v.state === "paused") && ( + + )} + +
) : null } /> @@ -163,6 +265,36 @@ export default function RuntimeVmPage() { onConfirm={terminate} /> + setForkOpen(false)} + title={`Fork ${vmId}`} + description="Snapshot this VM and start copies from the snapshot." + size="sm" + footer={ + <> + + + + } + > + + setForkCount(e.target.value)} + className="w-full rounded-lg border border-zinc-700 bg-surface-0 px-3 py-2 text-sm text-zinc-100 focus:border-lantern-500 focus:outline-none" + /> + + +
{/* Logs — wide column */}
@@ -221,6 +353,29 @@ export default function RuntimeVmPage() { )} + {(v.snapshotId || v.parentVmId) && ( +
}> +
+ {v.snapshotId && ( +
+
Snapshot
+
{v.snapshotId}
+
+ )} + {v.parentVmId && ( +
+
Parent VM
+
+ + {v.parentVmId} + +
+
+ )} +
+
+ )} +
}>
{v.isolationClass}
diff --git a/apps/web/app/(dashboard)/runtime/cockpit-ui.tsx b/apps/web/app/(dashboard)/runtime/cockpit-ui.tsx index 76774b42..e1cfaf44 100644 --- a/apps/web/app/(dashboard)/runtime/cockpit-ui.tsx +++ b/apps/web/app/(dashboard)/runtime/cockpit-ui.tsx @@ -16,6 +16,7 @@ export const STATE_STYLES: Record( if (!res.ok) { const text = await res.text().catch(() => ""); - throw new Error(`API ${res.status}: ${text || res.statusText}`); + throw new HttpError(res.status, `API ${res.status}: ${text || res.statusText}`); } if (res.status === 204) return undefined as T; return (await res.json()) as T; diff --git a/apps/web/lib/runtime-types.ts b/apps/web/lib/runtime-types.ts index ebd93c91..878711cc 100644 --- a/apps/web/lib/runtime-types.ts +++ b/apps/web/lib/runtime-types.ts @@ -10,6 +10,7 @@ export type VmState = | "pending" | "spawning" | "running" + | "paused" | "draining" | "terminated" | "failed"; @@ -25,6 +26,10 @@ export interface VmRow { terminatedAt?: string | null; lastHeartbeatAt?: string | null; spec: Record | null; + /** Set once the VM has been paused/snapshotted. */ + snapshotId?: string | null; + /** Set on VMs created by fork. */ + parentVmId?: string | null; // ---- cockpit-only, demo-seeded (see honesty contract in runtime-demo.ts) /** Friendly workload / agent name. Real rows may not carry one. */ diff --git a/packages/sdk-python/lantern/client.py b/packages/sdk-python/lantern/client.py index 89590d7c..dd0d7954 100644 --- a/packages/sdk-python/lantern/client.py +++ b/packages/sdk-python/lantern/client.py @@ -12,7 +12,19 @@ import httpx from lantern.errors import LanternApiError -from lantern.types import AgentInfo, ConnectorInfo, ConnectorResult, Run, Session, SessionMessage, StreamEvent +from lantern.types import ( + AgentInfo, + ConnectorInfo, + ConnectorResult, + ExecStreamEvent, + Run, + Session, + SessionMessage, + StreamEvent, + VmForkResult, + VmPauseResult, + VmResumeResult, +) # Bounded exponential backoff with full jitter, mirroring # packages/bridge-core/src/retry.ts. Only 429/503 and network-layer @@ -68,6 +80,7 @@ def __init__( self.receipts = self._Receipts(self) self.feedback = self._Feedback(self) self.rehearsals = self._Rehearsals(self) + self.runtime = self._Runtime(self) # ------------------------------------------------------------------ # Lifecycle @@ -969,6 +982,106 @@ async def summary_for_agent(self, agent_name: str) -> FeedbackSummary: # ------------------------------------------------------------------ # rehearsals namespace # ------------------------------------------------------------------ + # runtime namespace (microVMs) + # ------------------------------------------------------------------ + + class _Runtime: + def __init__(self, client: LanternClient) -> None: + self._c = client + + async def schedule(self, spec: dict[str, Any]) -> dict[str, Any]: + return await self._c._request("POST", "/v1/runtime/schedule", body=spec) + + async def list(self, *, state: str | None = None, limit: int | None = None) -> list[dict[str, Any]]: + data = await self._c._request("GET", "/v1/runtime/vms", params={"state": state, "limit": limit}) + return self._c._items(data, "vms") + + async def get(self, vm_id: str) -> dict[str, Any]: + """Bare VM row (the detail endpoint wraps it as ``{vm, events}``).""" + data = await self._c._request("GET", f"/v1/runtime/vms/{vm_id}") + if isinstance(data, dict) and isinstance(data.get("vm"), dict): + return data["vm"] + return data + + async def terminate(self, vm_id: str, *, grace: str | None = None) -> None: + await self._c._request("DELETE", f"/v1/runtime/vms/{vm_id}", params={"grace": grace}) + + async def exec(self, vm_id: str, command: str, argv: list[str] | None = None) -> dict[str, Any]: + """Buffered exec (execve-style: ``command`` is the executable).""" + return await self._c._request( + "POST", f"/v1/runtime/vms/{vm_id}/exec", body={"command": command, "argv": argv or []} + ) + + async def exec_stream( + self, vm_id: str, command: str, argv: list[str] | None = None + ) -> AsyncIterator[ExecStreamEvent]: + """Streaming exec over SSE: stdout/stderr chunks, then one ``exit`` event. + + Raises LanternApiError on non-2xx and on an SSE ``error`` event. + """ + client = await self._c._get_client() + headers = {**self._c._headers(), "Accept": "text/event-stream"} + body = json.dumps({"command": command, "argv": argv or []}) + async with client.stream( + "POST", f"/v1/runtime/vms/{vm_id}/exec", headers=headers, content=body + ) as resp: + if not resp.is_success: + raw = await resp.aread() + raise LanternApiError(resp.status_code, raw.decode("utf-8", errors="replace")) + + event, data = "message", "" + buffer = "" + # aiter_text decodes incrementally, so split multibyte chars are safe. + async for chunk in resp.aiter_text(): + buffer += chunk + while "\n" in buffer: + line, buffer = buffer.split("\n", 1) + line = line.rstrip("\r") + if line == "": + if data: + out = self._parse_exec_event(event, data) + if out is not None: + yield out + if out.type == "exit": + return + event, data = "message", "" + elif line.startswith("event:"): + event = line[6:].strip() + elif line.startswith("data:"): + data = line[5:].strip() + + @staticmethod + def _parse_exec_event(event: str, data: str) -> ExecStreamEvent | None: + try: + payload = json.loads(data) + except ValueError: + return None + if not isinstance(payload, dict): + return None + if event in ("stdout", "stderr"): + return ExecStreamEvent(type=event, data=str(payload.get("data", ""))) + if event == "exit": + return ExecStreamEvent(type="exit", exit_code=int(payload.get("exitCode", 0))) + if event == "error": + raise LanternApiError(500, str(payload.get("error") or payload.get("message") or "exec stream error")) + return None + + async def pause(self, vm_id: str) -> VmPauseResult: + """Snapshot + pause a running VM. 409 if not running.""" + data = await self._c._request("POST", f"/v1/runtime/vms/{vm_id}/pause") + return VmPauseResult.model_validate(data) + + async def resume(self, vm_id: str) -> VmResumeResult: + """Resume a paused VM. 402 quota exceeded, 409 if not paused.""" + data = await self._c._request("POST", f"/v1/runtime/vms/{vm_id}/resume") + return VmResumeResult.model_validate(data) + + async def fork(self, vm_id: str, *, count: int = 1) -> VmForkResult: + """Fork ``count`` (1..10) VMs from a snapshot. Per-fork failures are in ``forks[].error``.""" + data = await self._c._request("POST", f"/v1/runtime/vms/{vm_id}/fork", body={"count": count}) + return VmForkResult.model_validate(data) + + # ------------------------------------------------------------------ class _Rehearsals: def __init__(self, client: LanternClient) -> None: diff --git a/packages/sdk-python/lantern/integrations/runtime_client.py b/packages/sdk-python/lantern/integrations/runtime_client.py index 0a8c7eba..e47022fb 100644 --- a/packages/sdk-python/lantern/integrations/runtime_client.py +++ b/packages/sdk-python/lantern/integrations/runtime_client.py @@ -86,6 +86,18 @@ def exec(self, vm_id: str, command: str, argv: list[str] | None = None) -> tuple data = self._request("POST", f"/v1/runtime/vms/{vm_id}/exec", json=payload) return data.get("stdout", ""), data.get("stderr", ""), int(data.get("exitCode", 0)) + def pause(self, vm_id: str) -> dict[str, Any]: + """POST /v1/runtime/vms/{id}/pause -> {vmId, snapshotId, state:"paused"}.""" + return self._request("POST", f"/v1/runtime/vms/{vm_id}/pause") + + def resume(self, vm_id: str) -> dict[str, Any]: + """POST /v1/runtime/vms/{id}/resume -> {vmId, snapshotId, state:"running"}. 402 quota, 409 state.""" + return self._request("POST", f"/v1/runtime/vms/{vm_id}/resume") + + def fork(self, vm_id: str, *, count: int = 1) -> dict[str, Any]: + """POST /v1/runtime/vms/{id}/fork -> {sourceVmId, snapshotId, forks:[{vmId,state,error?}]}.""" + return self._request("POST", f"/v1/runtime/vms/{vm_id}/fork", json={"count": count}) + def terminate(self, vm_id: str, *, grace: str = "30s") -> None: self._request("DELETE", f"/v1/runtime/vms/{vm_id}", params={"grace": grace}) diff --git a/packages/sdk-python/lantern/types.py b/packages/sdk-python/lantern/types.py index 84f39cb7..b4aca696 100644 --- a/packages/sdk-python/lantern/types.py +++ b/packages/sdk-python/lantern/types.py @@ -727,3 +727,38 @@ async def _default_notify(opts: NotifyOptions) -> None: async def _default_subagent(agent: str, input: Any) -> Any: raise NotImplementedError("subagent() is only available in the Lantern runtime") + + +# --- Runtime (microVM) pause / resume / fork / exec-stream --------------- + + +class VmPauseResult(ApiModel): + vm_id: str + snapshot_id: str + state: str + + +class VmResumeResult(ApiModel): + vm_id: str + snapshot_id: str + state: str + + +class VmForkEntry(ApiModel): + vm_id: str + state: str + error: str | None = None + + +class VmForkResult(ApiModel): + source_vm_id: str + snapshot_id: str + forks: list[VmForkEntry] + + +class ExecStreamEvent(ApiModel): + """One SSE event from a streaming exec. ``type`` is stdout|stderr|exit.""" + + type: str + data: str = "" + exit_code: int | None = None diff --git a/packages/sdk-python/tests/test_runtime.py b/packages/sdk-python/tests/test_runtime.py new file mode 100644 index 00000000..9a9d4537 --- /dev/null +++ b/packages/sdk-python/tests/test_runtime.py @@ -0,0 +1,92 @@ +"""Tests for client.runtime (pause/resume/fork/exec_stream). httpx MockTransport, no server.""" + +from __future__ import annotations + +import json +from collections.abc import AsyncIterator +from typing import Any + +import httpx +import pytest + +from lantern.client import LanternClient +from lantern.errors import LanternApiError + + +class _Bytes(httpx.AsyncByteStream): + def __init__(self, chunks: list[bytes]) -> None: + self._chunks = chunks + + async def __aiter__(self) -> AsyncIterator[bytes]: + for c in self._chunks: + yield c + + +class _T(httpx.AsyncBaseTransport): + def __init__(self, status: int = 200, body: Any = None, sse: bytes | None = None) -> None: + self.status, self.body, self.sse = status, body, sse + self.requests: list[httpx.Request] = [] + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + self.requests.append(request) + if self.sse is not None: + # one byte per chunk: worst case for line and multibyte-char boundaries + chunks = [bytes([b]) for b in self.sse] + return httpx.Response(self.status, stream=_Bytes(chunks), headers={"content-type": "text/event-stream"}) + content = json.dumps(self.body).encode() if self.body is not None else b"" + return httpx.Response(self.status, content=content, headers={"content-type": "application/json"}) + + +def _client(t: _T) -> LanternClient: + c = LanternClient(base_url="https://api.example.com", api_key="k") + c._client = httpx.AsyncClient(base_url="https://api.example.com", transport=t, headers=c._headers()) + return c + + +async def test_pause_and_resume(): + t = _T(body={"vmId": "vm1", "snapshotId": "s1", "state": "paused"}) + r = await _client(t).runtime.pause("vm1") + assert (r.vm_id, r.snapshot_id, r.state) == ("vm1", "s1", "paused") + assert t.requests[-1].method == "POST" and t.requests[-1].url.path == "/v1/runtime/vms/vm1/pause" + + +@pytest.mark.parametrize("status", [402, 409]) +async def test_resume_errors(status: int): + t = _T(status=status, body={"error": "x"}) + with pytest.raises(LanternApiError) as ei: + await _client(t).runtime.resume("vm1") + assert ei.value.status == status + assert len(t.requests) == 1 + + +async def test_fork_body_and_partial_failure(): + t = _T(body={"sourceVmId": "vm1", "snapshotId": "s", "forks": [{"vmId": "a", "state": "running"}, {"vmId": "b", "state": "failed", "error": "boom"}]}) + r = await _client(t).runtime.fork("vm1", count=2) + assert json.loads(t.requests[-1].content) == {"count": 2} + assert r.forks[1].error == "boom" + + +async def test_exec_stream_split_chunks_and_utf8(): + sse = ( + 'event: stdout\ndata: {"data":"héllo €"}\n\n' + 'event: stderr\ndata: {"data":"warn"}\n\n' + 'event: exit\ndata: {"exitCode":3}\n\n' + ).encode() + t = _T(sse=sse) + events = [e async for e in _client(t).runtime.exec_stream("vm1", "/bin/sh", ["-c", "x"])] + assert [(e.type, e.data, e.exit_code) for e in events] == [ + ("stdout", "héllo €", None), + ("stderr", "warn", None), + ("exit", "", 3), + ] + assert t.requests[-1].headers["accept"] == "text/event-stream" + + +async def test_exec_stream_error_event_and_http_error(): + t = _T(sse=b'event: error\ndata: {"error":"vm gone"}\n\n') + with pytest.raises(LanternApiError): + [e async for e in _client(t).runtime.exec_stream("vm1", "ls")] + t2 = _T(status=409, sse=b"conflict") + with pytest.raises(LanternApiError) as ei: + [e async for e in _client(t2).runtime.exec_stream("vm1", "ls")] + assert ei.value.status == 409 diff --git a/packages/sdk-ts/__tests__/runtime.test.ts b/packages/sdk-ts/__tests__/runtime.test.ts new file mode 100644 index 00000000..abad11d3 --- /dev/null +++ b/packages/sdk-ts/__tests__/runtime.test.ts @@ -0,0 +1,98 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { LanternApiError, LanternClient, MessageStreamError } from "../src/client.js"; +import type { ExecStreamEvent } from "../src/types.js"; + +const client = new LanternClient({ baseUrl: "http://x", apiKey: "k" }); + +function json(body: unknown, status = 200): Response { + return new Response(JSON.stringify(body), { status, headers: { "Content-Type": "application/json" } }); +} + +function sse(chunks: Uint8Array[]): Response { + const stream = new ReadableStream({ + start(c) { + for (const ch of chunks) c.enqueue(ch); + c.close(); + }, + }); + return new Response(stream, { status: 200, headers: { "Content-Type": "text/event-stream" } }); +} + +async function collect(it: AsyncIterable): Promise { + const out: ExecStreamEvent[] = []; + for await (const e of it) out.push(e); + return out; +} + +afterEach(() => vi.unstubAllGlobals()); + +describe("runtime pause/resume/fork", () => { + it("pause POSTs and returns typed result", async () => { + const f = vi.fn().mockResolvedValue(json({ vmId: "vm1", snapshotId: "s1", state: "paused" })); + vi.stubGlobal("fetch", f); + const r = await client.runtime.pause("vm1"); + expect(r.state).toBe("paused"); + expect(f.mock.calls[0][0]).toBe("http://x/v1/runtime/vms/vm1/pause"); + expect(f.mock.calls[0][1].method).toBe("POST"); + }); + + it("resume surfaces 402 and 409 as LanternApiError without retry", async () => { + for (const status of [402, 409]) { + const f = vi.fn().mockResolvedValue(new Response("nope", { status })); + vi.stubGlobal("fetch", f); + await expect(client.runtime.resume("vm1")).rejects.toMatchObject({ status }); + expect(f).toHaveBeenCalledTimes(1); + } + }); + + it("fork defaults count to 1 and sends the given count", async () => { + const f = vi.fn().mockImplementation(async () => + json({ sourceVmId: "vm1", snapshotId: "s1", forks: [{ vmId: "a", state: "running" }, { vmId: "b", state: "failed", error: "boom" }] }), + ); + vi.stubGlobal("fetch", f); + const r = await client.runtime.fork("vm1"); + expect(JSON.parse(f.mock.calls[0][1].body)).toEqual({ count: 1 }); + await client.runtime.fork("vm1", { count: 3 }); + expect(JSON.parse(f.mock.calls[1][1].body)).toEqual({ count: 3 }); + expect(r.forks[1].error).toBe("boom"); + }); +}); + +describe("runtime.execStream", () => { + it("parses SSE split across chunks and multibyte UTF-8 boundaries", async () => { + const enc = new TextEncoder(); + const bytes = enc.encode( + 'event: stdout\ndata: {"data":"héllo €"}\n\nevent: stderr\ndata: {"data":"warn"}\n\nevent: exit\ndata: {"exitCode":3}\n\n', + ); + // split at every byte: worst case for chunk and multibyte boundaries + const chunks = Array.from(bytes, (b) => new Uint8Array([b])); + const f = vi.fn().mockResolvedValue(sse(chunks)); + vi.stubGlobal("fetch", f); + const events = await collect(client.runtime.execStream("vm1", { command: "/bin/sh", argv: ["-c", "x"] })); + expect(events).toEqual([ + { type: "stdout", data: "héllo €" }, + { type: "stderr", data: "warn" }, + { type: "exit", exitCode: 3 }, + ]); + expect(f.mock.calls[0][1].headers.Accept).toBe("text/event-stream"); + }); + + it("throws on an error event", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue(sse([new TextEncoder().encode('event: error\ndata: {"error":"vm gone"}\n\n')]))); + await expect(collect(client.runtime.execStream("vm1", { command: "ls" }))).rejects.toBeInstanceOf(MessageStreamError); + }); + + it("throws LanternApiError on non-2xx", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue(new Response("conflict", { status: 409 }))); + await expect(collect(client.runtime.execStream("vm1", { command: "ls" }))).rejects.toBeInstanceOf(LanternApiError); + }); +}); + +describe("runtime.get/list normalization", () => { + it("unwraps {vm, events} and bare arrays", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValueOnce(json({ vm: { vmId: "v", state: "paused", snapshotId: "s" }, events: [] }))); + expect((await client.runtime.get("v")).snapshotId).toBe("s"); + vi.stubGlobal("fetch", vi.fn().mockResolvedValueOnce(json([{ vmId: "v", state: "running" }]))); + expect(await client.runtime.list()).toHaveLength(1); + }); +}); diff --git a/packages/sdk-ts/src/client.ts b/packages/sdk-ts/src/client.ts index 49206896..608a767d 100644 --- a/packages/sdk-ts/src/client.ts +++ b/packages/sdk-ts/src/client.ts @@ -11,6 +11,8 @@ import type { EvalSuite, Experiment, FeedbackEntry, + ExecResult, + ExecStreamEvent, FeedbackSummary, ForecastResult, MarketplaceAgent, @@ -24,6 +26,10 @@ import type { SessionStreamEvent, SignedReceipt, StreamEvent, + VmForkResult, + VmPauseResult, + VmResumeResult, + RuntimeVm, } from "./types.js"; import { isNetworkError, withRetry } from "./retry.js"; @@ -49,6 +55,10 @@ const messageCompletedSchema = z.object({ }), }); +const execExitSchema = z.object({ exitCode: z.number().int() }); +const execChunkSchema = z.object({ data: z.string() }); +const execErrorSchema = z.object({ error: z.string().optional(), message: z.string().optional() }); + const messageErrorSchema = z.object({ turnId: z.string().optional(), error: z.string(), @@ -127,6 +137,85 @@ export class LanternClient { this.fetch(`/v1/agents/${encodeURIComponent(name)}`, { method: "DELETE" }), }; + readonly runtime = { + /** Submit an AgentSpec; the spec shape is owned by runtime.proto. */ + schedule: (spec: Record): Promise<{ vmId: string } & Record> => + this.fetch("/v1/runtime/schedule", { method: "POST", body: JSON.stringify(spec) }), + + // The list endpoint returns bare rows; the detail endpoint wraps the row as + // `{vm, events}`. Both are normalized to plain RuntimeVm rows. + list: async (params?: { state?: string; limit?: number }): Promise => { + const data = await this.fetch(`/v1/runtime/vms${this.qs(params)}`); + return Array.isArray(data) ? data : data.vms ?? []; + }, + + get: async (id: string): Promise => { + const data = await this.fetch(`/v1/runtime/vms/${encodeURIComponent(id)}`); + return "vm" in data && data.vm ? (data.vm as RuntimeVm) : (data as RuntimeVm); + }, + + terminate: (id: string, params?: { grace?: string }): Promise => + this.fetch(`/v1/runtime/vms/${encodeURIComponent(id)}${this.qs(params)}`, { method: "DELETE" }), + + /** Buffered exec (execve-style: `command` is the executable, `argv` its args). */ + exec: (id: string, params: { command: string; argv?: string[] }): Promise => + this.fetch(`/v1/runtime/vms/${encodeURIComponent(id)}/exec`, { method: "POST", body: JSON.stringify(params) }), + + /** Streaming exec over SSE. Yields stdout/stderr chunks, then a final exit event. */ + execStream: (id: string, params: { command: string; argv?: string[] }): AsyncIterable => + this.execStream(id, params), + + /** Snapshot and pause a running VM. 409 if not running. */ + pause: (id: string): Promise => + this.fetch(`/v1/runtime/vms/${encodeURIComponent(id)}/pause`, { method: "POST" }), + + /** Resume a paused VM. 402 on quota exceeded, 409 if not paused. */ + resume: (id: string): Promise => + this.fetch(`/v1/runtime/vms/${encodeURIComponent(id)}/resume`, { method: "POST" }), + + /** Fork N (1..10, default 1) copies from a snapshot of the VM. Per-fork failures are in `forks[].error`. */ + fork: (id: string, params: { count?: number } = {}): Promise => + this.fetch(`/v1/runtime/vms/${encodeURIComponent(id)}/fork`, { + method: "POST", + body: JSON.stringify({ count: params.count ?? 1 }), + }), + }; + + private async *execStream( + id: string, + params: { command: string; argv?: string[] }, + ): AsyncIterable { + const res = await fetch(`${this.baseUrl}/v1/runtime/vms/${encodeURIComponent(id)}/exec`, { + method: "POST", + headers: { ...this.headers(), Accept: "text/event-stream" }, + body: JSON.stringify(params), + }); + if (!res.ok) { + throw new LanternApiError(res.status, await res.text().catch(() => "")); + } + for await (const { event, data } of this.rawSSEWithEvent(res)) { + let json: unknown; + try { + json = JSON.parse(data); + } catch { + continue; // malformed — skip + } + if (event === "stdout" || event === "stderr") { + const p = execChunkSchema.safeParse(json); + if (p.success) yield { type: event, data: p.data.data }; + } else if (event === "exit") { + const p = execExitSchema.safeParse(json); + if (p.success) { + yield { type: "exit", exitCode: p.data.exitCode }; + return; + } + } else if (event === "error") { + const p = execErrorSchema.safeParse(json); + throw new MessageStreamError(p.success ? (p.data.error ?? p.data.message ?? "exec stream error") : "exec stream error"); + } + } + } + readonly runs = { create: async (params: { agent: string; diff --git a/packages/sdk-ts/src/index.ts b/packages/sdk-ts/src/index.ts index 545f9bda..8076d3d1 100644 --- a/packages/sdk-ts/src/index.ts +++ b/packages/sdk-ts/src/index.ts @@ -109,4 +109,12 @@ export type { FeedbackSummary, RehearseResponse, ForecastResult, + VmState, + RuntimeVm, + VmPauseResult, + VmResumeResult, + VmForkResult, + VmForkEntry, + ExecStreamEvent, + ExecResult, } from "./types.js"; diff --git a/packages/sdk-ts/src/types.ts b/packages/sdk-ts/src/types.ts index e2edf41a..df1a1d7b 100644 --- a/packages/sdk-ts/src/types.ts +++ b/packages/sdk-ts/src/types.ts @@ -688,3 +688,51 @@ export interface ForecastResult { wouldExceedBudget: boolean; blockReason?: string; } + +// --------------------------------------------------------------------------- +// Runtime (microVM) types +// --------------------------------------------------------------------------- + +export type VmState = "pending" | "starting" | "running" | "paused" | "draining" | "terminated" | "failed" | (string & {}); + +export interface RuntimeVm { + vmId: string; + state: VmState; + /** Set once the VM has been paused/snapshotted. */ + snapshotId?: string; + /** Set on VMs created by fork(). */ + parentVmId?: string; + [key: string]: unknown; +} + +export interface VmPauseResult { + vmId: string; + snapshotId: string; + state: "paused"; +} + +export interface VmResumeResult { + vmId: string; + snapshotId: string; + state: "running"; +} + +export interface VmForkEntry { + vmId: string; + state: VmState; + /** Present when this particular fork failed. */ + error?: string; +} + +export interface VmForkResult { + sourceVmId: string; + snapshotId: string; + forks: VmForkEntry[]; +} + +export type ExecStreamEvent = + | { type: "stdout" | "stderr"; data: string } + | { type: "exit"; exitCode: number }; + +/** Buffered (non-streaming) exec response; shape is owned by the control-plane. */ +export type ExecResult = Record;