From e9157de454bfa9700586fbf951ef769483fac34b Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 4 Oct 2026 18:23:13 -0700 Subject: [PATCH 1/4] Add tests for dead worker liveness and capped auto-redeploy --- apps/web/src/chat/agent-redeploy.test.ts | 53 ++++++++ apps/web/src/chat/deployment-liveness.test.ts | 100 ++++++++++++++++ apps/web/src/chat/list-chat-agents.test.ts | 113 ++++++++++++++++++ 3 files changed, 266 insertions(+) create mode 100644 apps/web/src/chat/agent-redeploy.test.ts create mode 100644 apps/web/src/chat/deployment-liveness.test.ts create mode 100644 apps/web/src/chat/list-chat-agents.test.ts diff --git a/apps/web/src/chat/agent-redeploy.test.ts b/apps/web/src/chat/agent-redeploy.test.ts new file mode 100644 index 000000000..2826b74fe --- /dev/null +++ b/apps/web/src/chat/agent-redeploy.test.ts @@ -0,0 +1,53 @@ +import { describe, expect, test } from "bun:test"; + +import { agentsToAutoDeploy, stoppedAgents } from "./agent-redeploy"; +import { redeployMode, workerState } from "./deployment-liveness"; +import type { WorkbenchParticipant } from "./threads-api"; +import { workerStatus } from "../pages/workers-page"; + +const agent = (over: Partial): WorkbenchParticipant => ({ + id: "a1", + kind: "agent", + name: "Worker", + address: "", + assetName: "worker", + ...over, +}); + +describe("agent redeploy selection", () => { + test("an agent marked auto is redeployed on its own", () => { + const released = agent({ redeploy: "auto" }); + expect(agentsToAutoDeploy([released]).map((a) => a.id)).toEqual([released.id]); + expect(stoppedAgents([released])).toEqual([]); + }); + + test("an agent marked manual waits for Restart", () => { + const capped = agent({ redeploy: "manual" }); + expect(agentsToAutoDeploy([capped])).toEqual([]); + expect(stoppedAgents([capped]).map((a) => a.id)).toEqual([capped.id]); + }); + + test("a coming-up or live agent is neither", () => { + for (const other of [agent({}), agent({ address: "worker@bench.example" })]) { + expect(agentsToAutoDeploy([other])).toEqual([]); + expect(stoppedAgents([other])).toEqual([]); + } + }); +}); + +describe("workers list and chat agree", () => { + const shapes = [ + { liveAddress: "w@bench.example", latest: "live", capped: false }, + { liveAddress: null, latest: "starting", capped: false }, + { liveAddress: null, latest: "stopped", capped: false }, + { liveAddress: null, latest: "stopped", capped: true }, + ] as const; + + for (const shape of shapes) { + test(`${shape.latest} live=${shape.liveAddress !== null} capped=${shape.capped}`, () => { + const stopped = redeployMode(shape) === "manual"; + expect(workerStatus(shape).tone === "ready").toBe(stopped); + expect(workerState(shape) === "stopped").toBe(stopped); + }); + } +}); diff --git a/apps/web/src/chat/deployment-liveness.test.ts b/apps/web/src/chat/deployment-liveness.test.ts new file mode 100644 index 000000000..12cb41aab --- /dev/null +++ b/apps/web/src/chat/deployment-liveness.test.ts @@ -0,0 +1,100 @@ +import { describe, expect, test } from "bun:test"; + +import { deploymentLiveness, diedQuickly, redeployMode, workerState } from "./deployment-liveness"; + +describe("deploymentLiveness", () => { + test("a deployed projection over an errored run is stopped", () => { + expect(deploymentLiveness("deployed", "error")).toBe("stopped"); + }); + + test("a deployed projection over a stopped run is stopped", () => { + expect(deploymentLiveness("deployed", "stopped")).toBe("stopped"); + }); + + test("a deployed projection over a serving run is live", () => { + expect(deploymentLiveness("deployed", "running")).toBe("live"); + expect(deploymentLiveness("deployed", "deployed")).toBe("live"); + expect(deploymentLiveness("deployed", "updating")).toBe("live"); + }); + + test("pending and recovering projections are starting", () => { + expect(deploymentLiveness("pending", "running")).toBe("starting"); + expect(deploymentLiveness("recovering", "updating")).toBe("starting"); + }); + + test("a deployment with no run is judged by its projection alone", () => { + expect(deploymentLiveness("deployed", undefined)).toBe("live"); + }); + + test("terminal projections are stopped", () => { + for (const status of ["failed", "released", "releasing", "destroy_failed"] as const) { + expect(deploymentLiveness(status, "running")).toBe("stopped"); + } + }); +}); + +describe("diedQuickly", () => { + const run = { createdAt: "2026-10-04T10:00:00.000Z" }; + + test("a run that ended seconds after it started died quickly", () => { + expect( + diedQuickly({ + ...run, + updatedAt: "2026-10-04T10:00:20.000Z", + endedAt: "2026-10-04T10:00:20.000Z", + }), + ).toBe(true); + }); + + test("a run that served for an hour before dying did not", () => { + expect( + diedQuickly({ + ...run, + updatedAt: "2026-10-04T11:00:00.000Z", + endedAt: "2026-10-04T11:00:00.000Z", + }), + ).toBe(false); + }); + + test("falls back to updatedAt when endedAt is absent", () => { + expect(diedQuickly({ ...run, updatedAt: "2026-10-04T10:00:05.000Z" })).toBe(true); + expect(diedQuickly({ ...run, updatedAt: "2026-10-04T10:30:00.000Z", endedAt: null })).toBe( + false, + ); + }); +}); + +describe("2-minute boundary", () => { + const createdAt = "2026-10-04T10:00:00.000Z"; + + test("dying just inside two minutes is quick, at two minutes it is not", () => { + expect(diedQuickly({ createdAt, updatedAt: "2026-10-04T10:01:59.999Z" })).toBe(true); + expect(diedQuickly({ createdAt, updatedAt: "2026-10-04T10:02:00.000Z" })).toBe(false); + }); +}); + +describe("redeployMode and workerState", () => { + const dead = { liveAddress: null, latest: "stopped", capped: false } as const; + + test("a live agent is left alone", () => { + const live = { ...dead, liveAddress: "w@bench.example" }; + expect(redeployMode(live)).toBeUndefined(); + expect(workerState(live)).toBe("live"); + }); + + test("a coming-up agent is left alone and reads starting", () => { + const coming = { liveAddress: null, latest: "starting", capped: false } as const; + expect(redeployMode(coming)).toBeUndefined(); + expect(workerState(coming)).toBe("starting"); + }); + + test("a dead agent redeploys on its own and reads starting", () => { + expect(redeployMode(dead)).toBe("auto"); + expect(workerState(dead)).toBe("starting"); + }); + + test("a capped dead agent waits for Restart and reads stopped", () => { + expect(redeployMode({ ...dead, capped: true })).toBe("manual"); + expect(workerState({ ...dead, capped: true })).toBe("stopped"); + }); +}); diff --git a/apps/web/src/chat/list-chat-agents.test.ts b/apps/web/src/chat/list-chat-agents.test.ts new file mode 100644 index 000000000..c28f95432 --- /dev/null +++ b/apps/web/src/chat/list-chat-agents.test.ts @@ -0,0 +1,113 @@ +import { afterEach, describe, expect, spyOn, test } from "bun:test"; + +import { redeployMode } from "./deployment-liveness"; +import { markRedeployFailed, clearRedeployFailure } from "./redeploy-failures"; +import { listChatAgents } from "./threads-api"; + +const fetchSpy = spyOn(globalThis, "fetch"); +afterEach(() => { + fetchSpy.mockReset(); + clearRedeployFailure("asset-1"); +}); + +type Deployment = { id: string; status: string; createdAt: string }; +type Run = { id: string; status: string; createdAt: string; updatedAt: string; endedAt?: string }; + +const T0 = "2026-10-04T10:00:00.000Z"; +const T1H = "2026-10-04T11:00:00.000Z"; +const T10S = "2026-10-04T11:00:10.000Z"; + +function hub(deployments: Deployment[], runs: Run[]) { + fetchSpy.mockImplementation(((input: string | URL | Request) => { + const url = String(input); + const body = url.includes("/workflows/deployments") + ? deployments.map((d) => ({ + ...d, + tenantId: "t1", + definitionAssetId: "asset-1", + })) + : url.includes("/workflows/runs") + ? { + data: runs.map((r) => ({ + ...r, + definitionId: "def", + definitionName: "worker", + tenantId: "t1", + address: `${r.id}@bench.example`, + })), + nextCursor: null, + } + : [{ id: "asset-1", name: "worker" }]; + return Promise.resolve(new Response(JSON.stringify(body))); + }) as typeof fetch); +} + +async function agent() { + const [only] = await listChatAgents("t1"); + if (only === undefined) throw new Error("no agent listed"); + return only; +} + +describe("listChatAgents self-heal join", () => { + test("a hub restart (released deployment, long-lived errored run) redeploys on its own", async () => { + hub( + [{ id: "r1", status: "released", createdAt: T0 }], + [{ id: "r1", status: "error", createdAt: T0, updatedAt: T1H, endedAt: T1H }], + ); + const listed = await agent(); + expect(listed.liveAddress).toBeNull(); + expect(redeployMode(listed)).toBe("auto"); + }); + + test("a deployed projection over a dead run is not live", async () => { + hub( + [{ id: "r1", status: "deployed", createdAt: T0 }], + [{ id: "r1", status: "stopped", createdAt: T0, updatedAt: T1H, endedAt: T1H }], + ); + expect((await agent()).liveAddress).toBeNull(); + }); + + test("a redeploy that died within moments waits for Restart", async () => { + hub( + [ + { id: "r2", status: "deployed", createdAt: T1H }, + { id: "r1", status: "released", createdAt: T0 }, + ], + [ + { id: "r2", status: "error", createdAt: T1H, updatedAt: T10S, endedAt: T10S }, + { id: "r1", status: "error", createdAt: T0, updatedAt: T1H, endedAt: T1H }, + ], + ); + expect(redeployMode(await agent())).toBe("manual"); + }); + + test("a fresh deployment whose run is not addressed yet is starting, not redeployed again", async () => { + hub( + [ + { id: "r2", status: "pending", createdAt: T1H }, + { id: "r1", status: "released", createdAt: T0 }, + ], + [{ id: "r1", status: "error", createdAt: T0, updatedAt: T1H, endedAt: T1H }], + ); + expect(redeployMode(await agent())).toBeUndefined(); + }); + + test("a failed redeploy caps the agent", async () => { + hub( + [{ id: "r1", status: "released", createdAt: T0 }], + [{ id: "r1", status: "error", createdAt: T0, updatedAt: T1H, endedAt: T1H }], + ); + markRedeployFailed("asset-1"); + expect(redeployMode(await agent())).toBe("manual"); + }); + + test("a live run is left alone", async () => { + hub( + [{ id: "r1", status: "deployed", createdAt: T0 }], + [{ id: "r1", status: "running", createdAt: T0, updatedAt: T1H }], + ); + const listed = await agent(); + expect(listed.liveAddress).toBe("r1@bench.example"); + expect(redeployMode(listed)).toBeUndefined(); + }); +}); From 2357eb646fe2967330f643b6b031a49df2207d88 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 4 Oct 2026 18:24:39 -0700 Subject: [PATCH 2/4] A dead worker redeploys once, then waits for Restart (CL-9913) --- apps/web/src/chat/agent-redeploy.ts | 16 +++++ apps/web/src/chat/deployment-liveness.ts | 52 ++++++++++++++++ apps/web/src/chat/redeploy-failures.ts | 7 +++ apps/web/src/chat/thread.css | 8 +++ apps/web/src/chat/threads-api.ts | 79 +++++++++++++++--------- apps/web/src/pages/workbench-page.tsx | 42 +++++++------ apps/web/src/pages/workers-page.tsx | 46 ++++++-------- apps/web/src/routines-api.ts | 11 +++- apps/web/src/use-restart-agent.ts | 27 ++++++++ apps/web/src/workbench-create.ts | 21 +++++++ apps/web/src/worker-status.ts | 12 +++- 11 files changed, 241 insertions(+), 80 deletions(-) create mode 100644 apps/web/src/chat/agent-redeploy.ts create mode 100644 apps/web/src/chat/deployment-liveness.ts create mode 100644 apps/web/src/chat/redeploy-failures.ts create mode 100644 apps/web/src/use-restart-agent.ts diff --git a/apps/web/src/chat/agent-redeploy.ts b/apps/web/src/chat/agent-redeploy.ts new file mode 100644 index 000000000..03e183133 --- /dev/null +++ b/apps/web/src/chat/agent-redeploy.ts @@ -0,0 +1,16 @@ +import type { WorkbenchParticipant } from "./threads-api"; + +type RedeployableAgent = WorkbenchParticipant & { readonly assetName: string }; + +function withMode(mode: "auto" | "manual") { + return (agent: WorkbenchParticipant): agent is RedeployableAgent => + agent.kind === "agent" && agent.assetName !== undefined && agent.redeploy === mode; +} + +export function agentsToAutoDeploy(agents: readonly WorkbenchParticipant[]): RedeployableAgent[] { + return agents.filter(withMode("auto")); +} + +export function stoppedAgents(agents: readonly WorkbenchParticipant[]): RedeployableAgent[] { + return agents.filter(withMode("manual")); +} diff --git a/apps/web/src/chat/deployment-liveness.ts b/apps/web/src/chat/deployment-liveness.ts new file mode 100644 index 000000000..829663594 --- /dev/null +++ b/apps/web/src/chat/deployment-liveness.ts @@ -0,0 +1,52 @@ +import type { WorkflowDeploymentResponse, WorkflowRunStatus } from "@intx/types"; + +export type DeploymentLiveness = "live" | "starting" | "stopped"; + +const DEAD_RUN_STATUSES = new Set(["error", "stopped"]); + +/** The deployments projection never consults run status, so a dead run still + * reads "deployed"; the run's own status settles it. */ +export function deploymentLiveness( + deploymentStatus: WorkflowDeploymentResponse["status"], + runStatus: WorkflowRunStatus | undefined, +): DeploymentLiveness { + if (runStatus !== undefined && DEAD_RUN_STATUSES.has(runStatus)) return "stopped"; + if (deploymentStatus === "deployed") return "live"; + if (deploymentStatus === "pending" || deploymentStatus === "recovering") return "starting"; + return "stopped"; +} + +const QUICK_DEATH_MS = 2 * 60 * 1000; + +/** A run that ended within moments of starting: redeploying it again would + * only loop, so the person restarts it instead. */ +export function diedQuickly(run: { + readonly createdAt: string; + readonly updatedAt: string; + readonly endedAt?: string | null; +}): boolean { + const end = Date.parse(run.endedAt ?? run.updatedAt); + return end - Date.parse(run.createdAt) < QUICK_DEATH_MS; +} + +export type RedeployMode = "auto" | "manual"; +export type WorkerState = "live" | "starting" | "stopped"; + +type AgentLiveness = { + readonly liveAddress: string | null; + readonly latest: DeploymentLiveness; + readonly capped: boolean; +}; + +/** What happens to a worker with nothing live: a dead one redeploys on its + * own once, then, if that died again quickly, waits for the person. */ +export function redeployMode(agent: AgentLiveness): RedeployMode | undefined { + if (agent.liveAddress !== null || agent.latest !== "stopped") return undefined; + return agent.capped ? "manual" : "auto"; +} + +/** The one worker state every surface reads. */ +export function workerState(agent: AgentLiveness): WorkerState { + if (agent.liveAddress !== null) return "live"; + return redeployMode(agent) === "manual" ? "stopped" : "starting"; +} diff --git a/apps/web/src/chat/redeploy-failures.ts b/apps/web/src/chat/redeploy-failures.ts new file mode 100644 index 000000000..cb9a556ed --- /dev/null +++ b/apps/web/src/chat/redeploy-failures.ts @@ -0,0 +1,7 @@ +// Assets whose redeploy threw this session. Their old dead run is still the +// latest one, so without this every remount would redeploy again. +const failed = new Set(); + +export const markRedeployFailed = (assetId: string): void => void failed.add(assetId); +export const clearRedeployFailure = (assetId: string): void => void failed.delete(assetId); +export const hasRedeployFailed = (assetId: string): boolean => failed.has(assetId); diff --git a/apps/web/src/chat/thread.css b/apps/web/src/chat/thread.css index b9deb5ee9..f65c3202e 100644 --- a/apps/web/src/chat/thread.css +++ b/apps/web/src/chat/thread.css @@ -179,3 +179,11 @@ display: none; } } + +.wb-restart { + position: absolute; + top: 54px; + left: 50%; + transform: translateX(-50%); + z-index: 20; +} diff --git a/apps/web/src/chat/threads-api.ts b/apps/web/src/chat/threads-api.ts index 31d56c5f0..e9c67908d 100644 --- a/apps/web/src/chat/threads-api.ts +++ b/apps/web/src/chat/threads-api.ts @@ -10,6 +10,14 @@ import { agentSlugFromSourceAssetName } from "../agent-deploy"; import { listTopLevelRuns } from "../agents-api"; import { WORKER_SOURCE_CONFIG } from "../worker-source"; import { personMailAddress } from "../mail-address"; +import { + deploymentLiveness, + diedQuickly, + redeployMode, + type DeploymentLiveness, + type RedeployMode, +} from "./deployment-liveness"; +import { hasRedeployFailed } from "./redeploy-failures"; import { mentionedAgents } from "./mentions"; import { appendRoster } from "./workbench-roster"; @@ -36,15 +44,13 @@ export type ChatAgent = { /** The address of the agent's currently live run, or null when none is * live (mid-redeploy). */ readonly liveAddress: string | null; - // A terminal status means nothing is running and nothing is coming - // unless someone restarts it. - readonly latestStatus: string | undefined; + /** The newest deployment's liveness; stopped means nothing is running. */ + readonly latest: DeploymentLiveness; + /** Redeploying automatically would loop: the newest run never lived, or + * the last redeploy threw. */ + readonly capped: boolean; }; -const LIVE_DEPLOYMENT_STATUSES = new Set(["deployed", "pending", "recovering"]); -// On the way up but not live yet, distinct from a terminal status. -export const STARTING_DEPLOYMENT_STATUSES = new Set(["pending", "recovering"]); - const DeploymentsSchema = WorkflowDeploymentResponse.array(); const WorkflowAssetSchema = type({ id: "string", name: "string" }).array(); @@ -96,26 +102,38 @@ export async function listChatAgents(tenantId: string): Promise [run.id, run.address])); + const runById = new Map(runs.map((run) => [run.id, run])); const nameByAssetId = new Map(assets.map((asset) => [asset.id, asset.name])); const byAsset = new Map< string, - { addresses: Set; liveAddress: string | null; latestStatus: string | undefined } + { + addresses: Set; + liveAddress: string | null; + latest: DeploymentLiveness | undefined; + capped: boolean; + } >(); for (const deployment of deployments) { - const address = addressByRunId.get(deployment.id); - if (address === undefined || address.length === 0) continue; + const run = runById.get(deployment.id); + const address = run?.address ?? ""; const entry = byAsset.get(deployment.definitionAssetId) ?? { addresses: new Set(), liveAddress: null, - latestStatus: undefined, + latest: undefined, + capped: hasRedeployFailed(deployment.definitionAssetId), }; - entry.addresses.add(address); + if (address.length > 0) entry.addresses.add(address); // Deployments come back newest-first, so the first one seen per asset - // is the latest, and the first live one seen is the current one. - if (entry.latestStatus === undefined) entry.latestStatus = deployment.status; - if (entry.liveAddress === null && LIVE_DEPLOYMENT_STATUSES.has(deployment.status)) { + // is the latest, and the first live one seen is the current one. A run + // the list does not carry yet is still coming up, unless the deployment + // itself is terminal, in which case it never lived. + const liveness = deploymentLiveness(deployment.status, run?.status); + if (entry.latest === undefined) { + entry.latest = liveness; + if (run === undefined ? liveness === "stopped" : diedQuickly(run)) entry.capped = true; + } + if (entry.liveAddress === null && liveness !== "stopped" && address.length > 0) { entry.liveAddress = address; } byAsset.set(deployment.definitionAssetId, entry); @@ -129,17 +147,12 @@ export async function listChatAgents(tenantId: string): Promise): boolean { - if (agent.liveAddress !== null) return false; - return agent.latestStatus === undefined || !STARTING_DEPLOYMENT_STATUSES.has(agent.latestStatus); -} - /** The agent an `@name` first message picks, matched case-insensitively * against the agent's display name with spaces removed. */ export function agentFromMention( @@ -281,6 +294,8 @@ export type WorkbenchParticipant = { /** The workflow asset's raw name — only present for a `kind: "agent"` * row; what a released agent's redeploy re-reads/re-pushes source by. */ readonly assetName?: string; + /** What becomes of an agent with nothing live: redeployed on its own, or left for the person. */ + readonly redeploy?: RedeployMode; }; const PrincipalPage = type({ @@ -313,13 +328,17 @@ export async function listWorkbenchParticipants( name: principal.displayName, address: personMailAddress(principal.refId, tenantDomain), })); - const agents = chatAgents.map((agent): WorkbenchParticipant => ({ - id: agent.id, - kind: "agent", - name: agent.name, - address: agent.liveAddress ?? "", - assetName: agent.assetName, - })); + const agents = chatAgents.map((agent): WorkbenchParticipant => { + const redeploy = redeployMode(agent); + return { + id: agent.id, + kind: "agent", + name: agent.name, + address: agent.liveAddress ?? "", + assetName: agent.assetName, + ...(redeploy === undefined ? {} : { redeploy }), + }; + }); return [...people, ...agents]; } diff --git a/apps/web/src/pages/workbench-page.tsx b/apps/web/src/pages/workbench-page.tsx index b3d2e07c3..d5a0edea9 100644 --- a/apps/web/src/pages/workbench-page.tsx +++ b/apps/web/src/pages/workbench-page.tsx @@ -2,7 +2,6 @@ // the thread left; the drawer's Information tab carries the bench overview. import { Button, EmptyState, PageShell, Skeleton } from "@corbits/react-ui"; -import { toast } from "@corbits/react-ui/ui/toast"; import { PanelLeft, WarningCircle } from "@/lib/icons"; import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { Fragment, useCallback, useContext, useEffect, useRef, useState } from "react"; @@ -15,6 +14,7 @@ import "@/chat/thread.css"; import { Markdown } from "@/chat/markdown"; import { MessageAttachments } from "@/chat/message-attachments"; import { resolveMessagePackage } from "@/chat/deployable-package"; +import { agentsToAutoDeploy, stoppedAgents } from "@/chat/agent-redeploy"; import { stripRoster } from "@/chat/workbench-roster"; import { ancestorChain, @@ -54,7 +54,7 @@ import { recordLastWorkbenchId } from "../last-workbench"; import { PROVIDER_SETTINGS_PATH, ProviderSkipBanner } from "../provider-skip-banner"; import { useNavigate } from "../navigation"; import { isClassifiedInferenceFailureText } from "@/chat/inference-failure"; -import { redeployWorkbenchAgent } from "../workbench-create"; +import { useRestartAgent } from "../use-restart-agent"; import { workbenchIdFromPath } from "../workbench-path"; function messagePlaceholder(name: string | undefined): string { @@ -202,20 +202,12 @@ function AgentRedeployer({ readonly assetName: string; }; }) { - const queryClient = useQueryClient(); const started = useRef(false); - const redeploy = useMutation({ - mutationFn: () => redeployWorkbenchAgent(workbenchTenantId, agent), - onSuccess: () => - queryClient.invalidateQueries({ - queryKey: workbenchKeys.scope(workbenchTenantId), - }), - onError: (cause) => toast(errorText(cause)), - }); + const redeploy = useRestartAgent(); useEffect(() => { if (started.current || redeploy.isPending || redeploy.isSuccess) return; started.current = true; - redeploy.mutate(); + redeploy.mutate({ tenantId: workbenchTenantId, agent }); }); return null; } @@ -242,7 +234,11 @@ function Workbench({ workbenchTenantId }: { readonly workbenchTenantId: string } // Poll while any agent has no live run yet, so the workbench notices its own // redeploy finishing without a manual refresh. refetchInterval: (query) => - (query.state.data ?? []).some((p) => p.kind === "agent" && p.address === "") ? 3000 : false, + (query.state.data ?? []).some( + (p) => p.kind === "agent" && p.address === "" && p.redeploy !== "manual", + ) + ? 3000 + : false, }); const chatAgentsQuery = useQuery({ queryKey: tenantKeys.agents(workbenchTenantId), @@ -292,13 +288,11 @@ function Workbench({ workbenchTenantId }: { readonly workbenchTenantId: string } }, [statedName, saveName]); const agents = roster.filter((participant) => participant.kind === "agent"); - // Released by a hub restart: the asset is still here but nothing is live. - const releasedAgents = agents.filter( - (agent): agent is WorkbenchParticipant & { assetName: string } => - agent.address === "" && agent.assetName !== undefined, - ); + const releasedAgents = agentsToAutoDeploy(agents); + const stoppedAgent = stoppedAgents(agents)[0]; const startingAgent = releasedAgents[0]; const workerStatus = useWorkerStatus(workbenchTenantId, agents[0]?.id); + const restart = useRestartAgent(); // Only a live agent can be addressed, so only one can be mentioned. const mentionables = agents .filter((agent) => agent.address.includes("@")) @@ -379,6 +373,18 @@ function Workbench({ workbenchTenantId }: { readonly workbenchTenantId: string } open={drawerOpen} onToggle={() => setDrawerOpen((open) => !open)} /> + {stoppedAgent === undefined ? null : ( +
+ +
+ )}
diff --git a/apps/web/src/pages/workers-page.tsx b/apps/web/src/pages/workers-page.tsx index 1fd4e61a2..df5d95ab9 100644 --- a/apps/web/src/pages/workers-page.tsx +++ b/apps/web/src/pages/workers-page.tsx @@ -4,15 +4,13 @@ import { useState } from "react"; import { Button, RichEmptyState, Skeleton } from "@corbits/react-ui"; import { Menu, MenuContent, MenuItem, MenuTrigger } from "@corbits/react-ui/ui/menu"; -import { toast } from "@corbits/react-ui/ui/toast"; import { Hash, Plus, Robot } from "@/lib/icons"; -import { useMutation, useQueryClient } from "@tanstack/react-query"; -import { reportError } from "@corbits/error-sink"; +import { useQueryClient } from "@tanstack/react-query"; -import { tenantKeys } from "../query-client"; -import { isAgentNotRunning, type ChatAgent } from "@/chat/threads-api"; +import { workerState } from "@/chat/deployment-liveness"; +import type { ChatAgent } from "@/chat/threads-api"; import { WorkbenchAvatar } from "@/chat/avatar"; -import { describeRestartFailure, redeployWorkbenchAgent } from "../workbench-create"; +import { useRestartAgent } from "../use-restart-agent"; import { useBench } from "../bench-context"; import type { HubTenant } from "../needs-converge"; import { useSidebarSections } from "../shell/sidebar-sections"; @@ -28,17 +26,21 @@ import "./workers-page.css"; export type WorkerTone = "working" | "ready" | "idle"; -/** What a worker is doing right now, from its deployment alone: coming up is - * working, stopped is waiting on the person to restart it, live and waiting - * for mail is idle. */ -export function workerStatus(agent: Pick): { +/** What a worker is doing right now, from its deployment alone: coming up + * (including a dead one about to be redeployed) is working, stopped is + * waiting on the person to restart it, live and waiting for mail is idle. */ +export function workerStatus(agent: Pick): { readonly tone: WorkerTone; readonly text: string; } { - if (agent.liveAddress !== null) return { tone: "idle", text: "Waiting for mail" }; - return isAgentNotRunning(agent) - ? { tone: "ready", text: "Stopped. Restart it to put it back to work." } - : { tone: "working", text: "Starting up" }; + switch (workerState(agent)) { + case "live": + return { tone: "idle", text: "Waiting for mail" }; + case "stopped": + return { tone: "ready", text: "Stopped. Restart it to put it back to work." }; + case "starting": + return { tone: "working", text: "Starting up" }; + } } export function workerPath(agentId: string): string { @@ -82,19 +84,7 @@ const FILTERS: readonly (readonly [WorkerTone | "all", string])[] = [ export function WorkersRosterList({ workers }: { readonly workers: readonly BenchWorker[] }) { const [filter, setFilter] = useState("all"); const [query, setQuery] = useState(""); - const queryClient = useQueryClient(); - const restart = useMutation({ - mutationFn: ({ agent, bench }: BenchWorker) => redeployWorkbenchAgent(bench.id, agent), - onSuccess: (_result, { bench }) => { - void queryClient.invalidateQueries({ - queryKey: tenantKeys.agents(bench.id), - }); - }, - onError: (cause, { bench }) => { - reportError(cause, { operation: "agent_restart", tenantId: bench.id }); - toast(describeRestartFailure(cause)); - }, - }); + const restart = useRestartAgent(); if (workers.length === 0) { return ( restart.mutate({ agent, bench })} + onClick={() => restart.mutate({ tenantId: bench.id, agent })} > {restart.isPending && restart.variables?.agent.id === agent.id ? "Starting…" diff --git a/apps/web/src/routines-api.ts b/apps/web/src/routines-api.ts index 7597c15b5..331808cce 100644 --- a/apps/web/src/routines-api.ts +++ b/apps/web/src/routines-api.ts @@ -7,6 +7,8 @@ import { useQuery } from "@tanstack/react-query"; import { WorkflowDeploymentResponse } from "@intx/types"; import type { APIQuery } from "@/lib/api-query"; import { ApiQueryError, UnauthenticatedError, toAPIQuery } from "@/lib/api-query"; +import { listTopLevelRuns } from "@/agents-api"; +import { deploymentLiveness } from "@/chat/deployment-liveness"; import { isAgentDeploySourceAssetName } from "@/agent-deploy"; import { WORKER_SOURCE_CONFIG } from "@/worker-source"; @@ -182,11 +184,13 @@ function isAgentAssetName(name: string): boolean { export async function listScheduledWorkflows( tenantId: string, ): Promise { - const [deployments, assets, schedules] = await Promise.all([ + const [deployments, assets, schedules, runs] = await Promise.all([ fetchJSON(deploymentsPath(tenantId), DeploymentsSchema), fetchJSON(workflowAssetsPath(tenantId), WorkflowAssetsSchema), listCronSchedules(tenantId), + listTopLevelRuns(tenantId), ]); + const runStatusById = new Map(runs.map((run) => [run.id, run.status])); const nameByAssetId = new Map(assets.map((asset) => [asset.id, asset.name])); // A schedule names its target by definition name, which is the source // asset's name — so a redeploy keeps the row joined to the same row here. @@ -206,7 +210,10 @@ export async function listScheduledWorkflows( assetId: deployment.definitionAssetId, name: name ?? "Untitled workflow", tenantId: deployment.tenantId, - status: deployment.status === "deployed" ? "deployed" : "stopped", + status: + deploymentLiveness(deployment.status, runStatusById.get(deployment.id)) === "stopped" + ? "stopped" + : "deployed", createdAt: deployment.createdAt, updatedAt: deployment.createdAt, schedule: row?.expression ?? null, diff --git a/apps/web/src/use-restart-agent.ts b/apps/web/src/use-restart-agent.ts new file mode 100644 index 000000000..40e645534 --- /dev/null +++ b/apps/web/src/use-restart-agent.ts @@ -0,0 +1,27 @@ +import { toast } from "@corbits/react-ui/ui/toast"; +import { useMutation, useQueryClient } from "@tanstack/react-query"; + +import { workbenchKeys } from "./chat-path"; +import { tenantKeys } from "./query-client"; +import { redeployWorkbenchAgent, reportRestartFailure } from "./workbench-create"; + +/** The one way an agent is put back to work: a person's Restart and the + * automatic redeploy of a dead worker both go through it. */ +export function useRestartAgent() { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: ({ + tenantId, + agent, + }: { + readonly tenantId: string; + readonly agent: { readonly id: string; readonly name: string; readonly assetName: string }; + }) => redeployWorkbenchAgent(tenantId, agent), + onSettled: (_result, _error, { tenantId }) => + Promise.all([ + queryClient.invalidateQueries({ queryKey: workbenchKeys.scope(tenantId) }), + queryClient.invalidateQueries({ queryKey: tenantKeys.agents(tenantId) }), + ]), + onError: (cause, { tenantId }) => toast(reportRestartFailure(cause, tenantId)), + }); +} diff --git a/apps/web/src/workbench-create.ts b/apps/web/src/workbench-create.ts index baf4b727c..484d3c6c3 100644 --- a/apps/web/src/workbench-create.ts +++ b/apps/web/src/workbench-create.ts @@ -1,7 +1,9 @@ // An agent joins a workbench by being deployed into the child tenant // itself — its deploy route rejects the parent's inherited asset. +import { clearRedeployFailure, markRedeployFailed } from "@/chat/redeploy-failures"; import { isDefaultWorker } from "@/chat/threads-api"; +import { reportError } from "@corbits/error-sink"; import { agentSlugFromSourceAssetName, deployAgentSource } from "./agent-deploy"; import { readAgentSource } from "./agent-source-read"; import { deployWorkerSource } from "./worker-deploy"; @@ -133,6 +135,25 @@ export function describeRestartFailure(cause: unknown): string { export async function redeployWorkbenchAgent( workbenchTenantId: string, agent: { readonly id: string; readonly name: string; readonly assetName: string }, +): Promise { + try { + await deployExistingAgent(workbenchTenantId, agent); + clearRedeployFailure(agent.id); + } catch (cause) { + markRedeployFailed(agent.id); + throw cause; + } +} + +/** Reports a failed restart and returns the text to show the person. */ +export function reportRestartFailure(cause: unknown, tenantId: string): string { + reportError(cause, { operation: "agent_restart", tenantId }); + return describeRestartFailure(cause); +} + +async function deployExistingAgent( + workbenchTenantId: string, + agent: { readonly id: string; readonly name: string; readonly assetName: string }, ): Promise { if (isDefaultWorker(agent)) { const hub = createFetchStockHub(); diff --git a/apps/web/src/worker-status.ts b/apps/web/src/worker-status.ts index a64d60e9b..b9cf9ec76 100644 --- a/apps/web/src/worker-status.ts +++ b/apps/web/src/worker-status.ts @@ -41,7 +41,13 @@ function turnAnswered(timeline: readonly WorkbenchMessage[] | undefined, sentAt: return (timeline ?? []).some((m) => m.author === "other" && Date.parse(m.at) > sentAt); } -const RANK: Record = { "Needs you": 3, Working: 2, Live: 1, Starting: 0 }; +const RANK: Record = { + "Needs you": 3, + Working: 2, + Live: 1, + Starting: 0, + Stopped: 0, +}; /** The status of the worker `isWorker` matches, across the given benches: * needs-you beats working beats live. */ @@ -89,7 +95,9 @@ export function useBenchWorkerStatus( benches.forEach((bench, index) => { const agent = participants[index]?.data?.find((p) => p.kind === "agent" && isWorker(p)); if (agent === undefined) return; - if (agent.address === "") return offer({ tone: "idle", text: "Starting" }); + if (agent.address === "") { + return offer({ tone: "idle", text: agent.redeploy === "manual" ? "Stopped" : "Starting" }); + } const needsYou = (approvals[index]?.data?.data ?? []).some( (row) => row.status === "pending" && sameAddress(row.agentAddress, agent.address), ); From 986614edbde58d33eacd4f5600c144e1e6d356d9 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 4 Oct 2026 18:25:22 -0700 Subject: [PATCH 3/4] Add tests for auto-redeploy dedupe --- apps/web/src/chat/redeploy-failures.test.ts | 15 +++++++++++++++ 1 file changed, 15 insertions(+) create mode 100644 apps/web/src/chat/redeploy-failures.test.ts diff --git a/apps/web/src/chat/redeploy-failures.test.ts b/apps/web/src/chat/redeploy-failures.test.ts new file mode 100644 index 000000000..f3978b51a --- /dev/null +++ b/apps/web/src/chat/redeploy-failures.test.ts @@ -0,0 +1,15 @@ +import { describe, expect, test } from "bun:test"; + +import { claimAutoRedeploy, releaseAutoRedeploy } from "./redeploy-failures"; + +describe("auto-redeploy claim", () => { + test("only the first surface claims an asset until it settles", () => { + expect(claimAutoRedeploy("asset-x")).toBe(true); + expect(claimAutoRedeploy("asset-x")).toBe(false); + expect(claimAutoRedeploy("asset-y")).toBe(true); + releaseAutoRedeploy("asset-x"); + releaseAutoRedeploy("asset-y"); + expect(claimAutoRedeploy("asset-x")).toBe(true); + releaseAutoRedeploy("asset-x"); + }); +}); From 758318a531aea149724ef18826140ded971f99b1 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 4 Oct 2026 18:25:53 -0700 Subject: [PATCH 4/4] Workers list redeploys dead workers too, once per agent (CL-9913) --- apps/web/src/agent-redeployer.tsx | 31 ++++++++++++++++++++++++++ apps/web/src/chat/redeploy-failures.ts | 12 ++++++++++ apps/web/src/pages/workbench-page.tsx | 28 ++--------------------- apps/web/src/pages/workers-page.tsx | 7 +++++- apps/web/src/use-restart-agent.ts | 9 +++++--- 5 files changed, 57 insertions(+), 30 deletions(-) create mode 100644 apps/web/src/agent-redeployer.tsx diff --git a/apps/web/src/agent-redeployer.tsx b/apps/web/src/agent-redeployer.tsx new file mode 100644 index 000000000..79e723775 --- /dev/null +++ b/apps/web/src/agent-redeployer.tsx @@ -0,0 +1,31 @@ +import { useEffect, useRef } from "react"; + +import { claimAutoRedeploy } from "@/chat/redeploy-failures"; +import { useRestartAgent } from "./use-restart-agent"; + +// Renders nothing; a mount's own ref plus the mutation's `isPending`/ +// `isSuccess` keep StrictMode's double effect from firing it twice. Fires +// from an effect, never during render: mutating mid-render updates the +// mutation's own state on a fiber that hasn't mounted yet (React warns). +// The claim keeps the chat page and the workers list from both redeploying +// the same agent. +export function AgentRedeployer({ + workbenchTenantId, + agent, +}: { + readonly workbenchTenantId: string; + readonly agent: { + readonly id: string; + readonly name: string; + readonly assetName: string; + }; +}) { + const started = useRef(false); + const redeploy = useRestartAgent(); + useEffect(() => { + if (started.current || redeploy.isPending || redeploy.isSuccess) return; + started.current = true; + if (claimAutoRedeploy(agent.id)) redeploy.mutate({ tenantId: workbenchTenantId, agent }); + }); + return null; +} diff --git a/apps/web/src/chat/redeploy-failures.ts b/apps/web/src/chat/redeploy-failures.ts index cb9a556ed..458043215 100644 --- a/apps/web/src/chat/redeploy-failures.ts +++ b/apps/web/src/chat/redeploy-failures.ts @@ -5,3 +5,15 @@ const failed = new Set(); export const markRedeployFailed = (assetId: string): void => void failed.add(assetId); export const clearRedeployFailure = (assetId: string): void => void failed.delete(assetId); export const hasRedeployFailed = (assetId: string): boolean => failed.has(assetId); + +const inFlight = new Set(); + +/** One surface at a time redeploys an asset on its own: false when another + * already holds it. */ +export function claimAutoRedeploy(assetId: string): boolean { + if (inFlight.has(assetId)) return false; + inFlight.add(assetId); + return true; +} + +export const releaseAutoRedeploy = (assetId: string): void => void inFlight.delete(assetId); diff --git a/apps/web/src/pages/workbench-page.tsx b/apps/web/src/pages/workbench-page.tsx index d5a0edea9..74dd933fc 100644 --- a/apps/web/src/pages/workbench-page.tsx +++ b/apps/web/src/pages/workbench-page.tsx @@ -4,7 +4,7 @@ import { Button, EmptyState, PageShell, Skeleton } from "@corbits/react-ui"; import { PanelLeft, WarningCircle } from "@/lib/icons"; import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; -import { Fragment, useCallback, useContext, useEffect, useRef, useState } from "react"; +import { Fragment, useCallback, useContext, useEffect, useState } from "react"; import { IdentityAvatar } from "@/chat/avatar"; import { Composer } from "@/chat/composer"; @@ -54,6 +54,7 @@ import { recordLastWorkbenchId } from "../last-workbench"; import { PROVIDER_SETTINGS_PATH, ProviderSkipBanner } from "../provider-skip-banner"; import { useNavigate } from "../navigation"; import { isClassifiedInferenceFailureText } from "@/chat/inference-failure"; +import { AgentRedeployer } from "../agent-redeployer"; import { useRestartAgent } from "../use-restart-agent"; import { workbenchIdFromPath } from "../workbench-path"; @@ -187,31 +188,6 @@ function WorkbenchMessageRow({ ); } -// Renders nothing; a mount's own ref plus the mutation's `isPending`/ -// `isSuccess` keep StrictMode's double effect from firing it twice. Fires -// from an effect, never during render: mutating mid-render updates the -// mutation's own state on a fiber that hasn't mounted yet (React warns). -function AgentRedeployer({ - workbenchTenantId, - agent, -}: { - readonly workbenchTenantId: string; - readonly agent: { - readonly id: string; - readonly name: string; - readonly assetName: string; - }; -}) { - const started = useRef(false); - const redeploy = useRestartAgent(); - useEffect(() => { - if (started.current || redeploy.isPending || redeploy.isSuccess) return; - started.current = true; - redeploy.mutate({ tenantId: workbenchTenantId, agent }); - }); - return null; -} - function Workbench({ workbenchTenantId }: { readonly workbenchTenantId: string }) { const queryClient = useQueryClient(); const [openThread, setOpenThread] = useState(null); diff --git a/apps/web/src/pages/workers-page.tsx b/apps/web/src/pages/workers-page.tsx index df5d95ab9..184251f2b 100644 --- a/apps/web/src/pages/workers-page.tsx +++ b/apps/web/src/pages/workers-page.tsx @@ -7,7 +7,8 @@ import { Menu, MenuContent, MenuItem, MenuTrigger } from "@corbits/react-ui/ui/m import { Hash, Plus, Robot } from "@/lib/icons"; import { useQueryClient } from "@tanstack/react-query"; -import { workerState } from "@/chat/deployment-liveness"; +import { AgentRedeployer } from "../agent-redeployer"; +import { redeployMode, workerState } from "@/chat/deployment-liveness"; import type { ChatAgent } from "@/chat/threads-api"; import { WorkbenchAvatar } from "@/chat/avatar"; import { useRestartAgent } from "../use-restart-agent"; @@ -85,6 +86,7 @@ export function WorkersRosterList({ workers }: { readonly workers: readonly Benc const [filter, setFilter] = useState("all"); const [query, setQuery] = useState(""); const restart = useRestartAgent(); + const autoRedeploy = workers.filter(({ agent }) => redeployMode(agent) === "auto"); if (workers.length === 0) { return ( + {autoRedeploy.map(({ agent, bench }) => ( + + ))}
{FILTERS.map(([key, label]) => ( diff --git a/apps/web/src/use-restart-agent.ts b/apps/web/src/use-restart-agent.ts index 40e645534..ea155bb3a 100644 --- a/apps/web/src/use-restart-agent.ts +++ b/apps/web/src/use-restart-agent.ts @@ -1,6 +1,7 @@ import { toast } from "@corbits/react-ui/ui/toast"; import { useMutation, useQueryClient } from "@tanstack/react-query"; +import { releaseAutoRedeploy } from "@/chat/redeploy-failures"; import { workbenchKeys } from "./chat-path"; import { tenantKeys } from "./query-client"; import { redeployWorkbenchAgent, reportRestartFailure } from "./workbench-create"; @@ -17,11 +18,13 @@ export function useRestartAgent() { readonly tenantId: string; readonly agent: { readonly id: string; readonly name: string; readonly assetName: string }; }) => redeployWorkbenchAgent(tenantId, agent), - onSettled: (_result, _error, { tenantId }) => - Promise.all([ + onSettled: (_result, _error, { tenantId, agent }) => { + releaseAutoRedeploy(agent.id); + return Promise.all([ queryClient.invalidateQueries({ queryKey: workbenchKeys.scope(tenantId) }), queryClient.invalidateQueries({ queryKey: tenantKeys.agents(tenantId) }), - ]), + ]); + }, onError: (cause, { tenantId }) => toast(reportRestartFailure(cause, tenantId)), }); }