From ebd1aefe3081f134b8e5f8523dff986d732b5eb1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pascal=20Andr=C3=A9?= Date: Tue, 21 Jul 2026 00:04:58 +0200 Subject: [PATCH 01/17] feat(workflows): add host-managed workflow runs Add durable sequential workflow execution with bounded outputs, human review gates, cancellation, restart recovery, lineage-scoped locking, and SSE updates. Expose workflow creation and monitoring in the right panel and through the CodeNomad plugin bridge. Keep approval mutations on the authenticated user API so plugin credentials cannot bypass a human gate, and preserve workspace lineage across desktop restore. Cover runtime transitions, restart rebinding, abort containment, route scoping, client reconciliation, restore identity, and plugin messaging with focused tests. --- .github/workflows/pr-build.yml | 2 + packages/opencode-plugin/plugin/codenomad.ts | 5 +- .../plugin/lib/workflows.test.ts | 35 ++ .../opencode-plugin/plugin/lib/workflows.ts | 147 +++++ packages/opencode-plugin/tsconfig.json | 2 +- packages/server/src/api-types.ts | 54 ++ packages/server/src/index.ts | 10 + packages/server/src/server/http-server.ts | 4 + .../src/server/routes/workflows.test.ts | 60 ++ .../server/src/server/routes/workflows.ts | 199 ++++++ .../src/server/routes/workspaces.test.ts | 2 + .../server/src/server/routes/workspaces.ts | 2 + packages/server/src/shutdown.test.ts | 2 +- packages/server/src/shutdown.ts | 3 +- packages/server/src/workflows/manager.test.ts | 282 +++++++++ packages/server/src/workflows/manager.ts | 592 ++++++++++++++++++ .../__tests__/workspace-identity.test.ts | 13 + packages/server/src/workspaces/manager.ts | 20 +- packages/ui/src/components/agent-selector.tsx | 6 +- .../instance/shell/right-panel/RightPanel.tsx | 96 +++ .../shell/right-panel/tabs/WorkflowsTab.tsx | 358 +++++++++++ .../instance/shell/right-panel/types.ts | 2 +- .../src/components/instance/shell/storage.ts | 5 +- packages/ui/src/components/model-selector.tsx | 3 +- packages/ui/src/lib/api-client.ts | 38 ++ .../src/lib/hooks/use-app-session-capture.ts | 6 +- .../src/lib/hooks/use-app-session-restore.ts | 5 +- .../ui/src/lib/i18n/messages/de/instance.ts | 48 ++ .../ui/src/lib/i18n/messages/en/instance.ts | 48 ++ .../ui/src/lib/i18n/messages/es/instance.ts | 48 ++ .../ui/src/lib/i18n/messages/fr/instance.ts | 48 ++ .../ui/src/lib/i18n/messages/he/instance.ts | 48 ++ .../ui/src/lib/i18n/messages/ja/instance.ts | 48 ++ .../ui/src/lib/i18n/messages/ne/instance.ts | 48 ++ .../ui/src/lib/i18n/messages/ru/instance.ts | 48 ++ .../src/lib/i18n/messages/zh-Hans/instance.ts | 48 ++ packages/ui/src/lib/sse-manager.ts | 13 + .../stores/app-session-reconciliation.test.ts | 15 + .../src/stores/app-session-reconciliation.ts | 13 +- .../stores/app-session-snapshot-merge.test.ts | 19 + .../src/stores/app-session-snapshot-merge.ts | 29 +- .../ui/src/stores/client-state-codec.test.ts | 18 +- packages/ui/src/stores/client-state-codec.ts | 7 +- .../stores/instance-lifecycle-authority.ts | 1 + packages/ui/src/stores/instances.ts | 6 + .../ui/src/stores/workflow-reconciliation.ts | 32 + packages/ui/src/stores/workflows.test.ts | 48 ++ packages/ui/src/stores/workflows.ts | 142 +++++ packages/ui/src/styles/panels.css | 1 + packages/ui/src/styles/panels/workflows.css | 110 ++++ packages/ui/src/types/instance.ts | 1 + 51 files changed, 2811 insertions(+), 29 deletions(-) create mode 100644 packages/opencode-plugin/plugin/lib/workflows.test.ts create mode 100644 packages/opencode-plugin/plugin/lib/workflows.ts create mode 100644 packages/server/src/server/routes/workflows.test.ts create mode 100644 packages/server/src/server/routes/workflows.ts create mode 100644 packages/server/src/workflows/manager.test.ts create mode 100644 packages/server/src/workflows/manager.ts create mode 100644 packages/ui/src/components/instance/shell/right-panel/tabs/WorkflowsTab.tsx create mode 100644 packages/ui/src/stores/workflow-reconciliation.ts create mode 100644 packages/ui/src/stores/workflows.test.ts create mode 100644 packages/ui/src/stores/workflows.ts create mode 100644 packages/ui/src/styles/panels/workflows.css diff --git a/.github/workflows/pr-build.yml b/.github/workflows/pr-build.yml index aa599e288..e35170391 100644 --- a/.github/workflows/pr-build.yml +++ b/.github/workflows/pr-build.yml @@ -122,6 +122,8 @@ jobs: packages/ui/src/stores/session-metadata.test.ts packages/ui/src/stores/session-pagination.test.ts packages/ui/src/stores/workspace-list-reconciliation-fence.test.ts + packages/ui/src/stores/workflows.test.ts + packages/opencode-plugin/plugin/lib/workflows.test.ts - name: Test restore ownership integration run: >- diff --git a/packages/opencode-plugin/plugin/codenomad.ts b/packages/opencode-plugin/plugin/codenomad.ts index 61d1827f0..cee264840 100644 --- a/packages/opencode-plugin/plugin/codenomad.ts +++ b/packages/opencode-plugin/plugin/codenomad.ts @@ -1,17 +1,19 @@ import type { PluginInput } from "@opencode-ai/plugin" import { createCodeNomadClient, getCodeNomadConfig } from "./lib/client.js" import { createBackgroundProcessTools } from "./lib/background-process.js" +import { createWorkflowTools } from "./lib/workflows.js" let voiceModeEnabled = false export async function CodeNomadPlugin(input: PluginInput): Promise<{ - tool: ReturnType + tool: ReturnType & ReturnType "chat.message": CodeNomadChatMessageHook event: CodeNomadEventHook }> { const config = getCodeNomadConfig() const client = createCodeNomadClient(config) const backgroundProcessTools = createBackgroundProcessTools(config, { baseDir: input.directory }) + const workflowTools = createWorkflowTools(config) await client.startEvents((event) => { if (event.type === "codenomad.ping") { @@ -33,6 +35,7 @@ export async function CodeNomadPlugin(input: PluginInput): Promise<{ return { tool: { ...backgroundProcessTools, + ...workflowTools, }, async "chat.message"(_input: { sessionID: string }, output: { message: { system?: string } }) { if (!voiceModeEnabled) { diff --git a/packages/opencode-plugin/plugin/lib/workflows.test.ts b/packages/opencode-plugin/plugin/lib/workflows.test.ts new file mode 100644 index 000000000..ff8593323 --- /dev/null +++ b/packages/opencode-plugin/plugin/lib/workflows.test.ts @@ -0,0 +1,35 @@ +import assert from "node:assert/strict" +import test from "node:test" +import { describeWorkflowDetails, describeWorkflowStart } from "./workflows.js" + +const run = { + id: "run", + objective: "Ship it", + status: "running" as const, + steps: [{ id: "build", title: "Build", status: "pending" }], +} + +test("workflow start message reflects whether a human gate is configured", () => { + assert.match(describeWorkflowStart(run, true), /will pause/) + assert.match(describeWorkflowStart(run, false), /no human approval gate/) +}) + +test("workflow review messaging handles final and truncated gates", () => { + const waiting = { + ...run, + status: "waiting_for_review" as const, + pendingReviewStepId: "build", + steps: [{ + id: "build", + title: "Build", + status: "completed", + sessionId: "session-1", + output: "partial", + outputTruncated: true, + }], + } + const details = describeWorkflowDetails(waiting) + assert.match(details, /continue or complete/) + assert.match(details, /truncated/) + assert.match(details, /session-1/) +}) diff --git a/packages/opencode-plugin/plugin/lib/workflows.ts b/packages/opencode-plugin/plugin/lib/workflows.ts new file mode 100644 index 000000000..e8f3b7fc3 --- /dev/null +++ b/packages/opencode-plugin/plugin/lib/workflows.ts @@ -0,0 +1,147 @@ +import { tool } from "@opencode-ai/plugin/tool" +import { createCodeNomadRequester, type CodeNomadConfig } from "./request.js" + +type WorkflowStatus = "running" | "waiting_for_review" | "completed" | "failed" | "cancelled" | "interrupted" + +type WorkflowRun = { + id: string + objective: string + status: WorkflowStatus + error?: string + pendingReviewStepId?: string + steps: Array<{ + id: string + title: string + status: string + sessionId?: string + output?: unknown + outputTruncated?: boolean + }> +} + +function modelConfig(providerID?: string, modelID?: string) { + if (Boolean(providerID) !== Boolean(modelID)) { + throw new Error("Provider ID and model ID must be supplied together.") + } + return providerID && modelID ? { providerID, modelID } : undefined +} + +function summarize(run: WorkflowRun) { + const steps = run.steps + .map((step) => `${step.title}: ${step.status}${step.sessionId ? ` (${step.sessionId})` : ""}`) + .join("\n") + const review = run.status === "waiting_for_review" + ? "\nHuman review is required in CodeNomad before the workflow can continue or complete." + : "" + return `Workflow ${run.id}\nStatus: ${run.status}\n${steps}${review}${run.error ? `\nError: ${run.error}` : ""}` +} + +function details(run: WorkflowRun) { + const reviewed = run.steps.find((step) => step.id === run.pendingReviewStepId) + const output = reviewed?.output === undefined ? "" : `\nPending review:\n${JSON.stringify(reviewed.output, null, 2)}` + const truncated = reviewed?.outputTruncated + ? `\nThis output is truncated. Review the full generated session${reviewed.sessionId ? ` ${reviewed.sessionId}` : ""} before approval.` + : "" + return `${summarize(run)}${output}${truncated}` +} + +export const describeWorkflowDetails = details + +export function describeWorkflowStart(run: WorkflowRun, hasApprovalGate: boolean) { + const gateMessage = hasApprovalGate + ? "The workflow will pause at its configured human approval gates." + : "This workflow has no human approval gate." + return `${summarize(run)}\n${gateMessage}` +} + +export function createWorkflowTools(config: CodeNomadConfig) { + const requester = createCodeNomadRequester(config) + const request = (path: string, init?: RequestInit) => requester.requestJson(`/workflow-runs${path}`, init) + + return { + start_codenomad_workflow: tool({ + description: + "Start a host-managed sequential workflow. Stages may require explicit human review before the next stage runs.", + args: { + objective: tool.schema.string().describe("The objective for the workflow"), + stages: tool.schema.array(tool.schema.object({ + id: tool.schema.string().describe("Stable stage ID using letters, numbers, underscore, or dash"), + title: tool.schema.string().describe("Human-readable stage title"), + instructions: tool.schema.string().describe("Instructions for this stage"), + agent: tool.schema.string().optional().describe("Optional OpenCode agent"), + provider_id: tool.schema.string().optional().describe("Optional model provider ID"), + model_id: tool.schema.string().optional().describe("Optional model ID"), + requires_approval: tool.schema.boolean().optional().describe("Pause for human review after this stage"), + })).min(1).max(12).optional().describe("Ordered workflow stages; defaults to Planner then Implementer"), + }, + async execute(args, context) { + const stages = args.stages?.map((stage) => { + const model = modelConfig(stage.provider_id, stage.model_id) + return { + id: stage.id, + title: stage.title, + instructions: stage.instructions, + ...(stage.agent ? { agent: stage.agent } : {}), + ...(model ? { model } : {}), + requiresApproval: Boolean(stage.requires_approval), + } + }) ?? [ + { + id: "planner", + title: "Planner", + instructions: "Create a concise implementation plan with ordered, verifiable steps.", + requiresApproval: true, + }, + { + id: "implementer", + title: "Implementer", + instructions: "Implement the approved plan and run focused validation.", + requiresApproval: false, + }, + ] + const run = await request("", { + method: "POST", + body: JSON.stringify({ + objective: args.objective, + initiatorSessionId: context.sessionID, + stages, + }), + }) + return describeWorkflowStart(run, stages.some((stage) => stage.requiresApproval)) + }, + }), + list_codenomad_workflows: tool({ + description: "List workflow runs managed by CodeNomad for this workspace.", + args: {}, + async execute() { + const response = await request<{ runs: WorkflowRun[] }>("") + if (response.runs.length === 0) return "No CodeNomad workflow runs found." + return response.runs.map((run) => `${run.id} | ${run.status} | ${run.objective}`).join("\n") + }, + }), + get_codenomad_workflow: tool({ + description: "Inspect one CodeNomad workflow run and its role sessions.", + args: { run_id: tool.schema.string().describe("Workflow run ID") }, + async execute(args) { + return details(await request(`/${encodeURIComponent(args.run_id)}`)) + }, + }), + approve_codenomad_workflow: tool({ + description: + "Show the pending approval details. Only a user in the CodeNomad Workflows panel can approve and continue the workflow.", + args: { run_id: tool.schema.string().describe("Workflow run ID") }, + async execute(args) { + const run = await request(`/${encodeURIComponent(args.run_id)}`) + return `${details(run)}\nApproval was not applied. Ask the user to approve this run in the CodeNomad Workflows panel.` + }, + }), + cancel_codenomad_workflow: tool({ + description: "Cancel a running or review-pending CodeNomad workflow.", + args: { run_id: tool.schema.string().describe("Workflow run ID") }, + async execute(args) { + const run = await request(`/${encodeURIComponent(args.run_id)}/cancel`, { method: "POST" }) + return summarize(run) + }, + }), + } +} diff --git a/packages/opencode-plugin/tsconfig.json b/packages/opencode-plugin/tsconfig.json index 09a866276..a7da85a5f 100644 --- a/packages/opencode-plugin/tsconfig.json +++ b/packages/opencode-plugin/tsconfig.json @@ -13,5 +13,5 @@ "types": ["node"] }, "include": ["plugin/**/*.ts"], - "exclude": ["dist", "node_modules"] + "exclude": ["dist", "node_modules", "plugin/**/*.test.ts"] } diff --git a/packages/server/src/api-types.ts b/packages/server/src/api-types.ts index d61b12f0e..b39198e85 100644 --- a/packages/server/src/api-types.ts +++ b/packages/server/src/api-types.ts @@ -16,6 +16,8 @@ export type WorkspaceStatus = "starting" | "ready" | "stopped" | "error" export interface WorkspaceDescriptor { id: string + /** Stable across desktop restore; distinct for force-created instances of the same path. */ + lineageId?: string /** Correlates creation events with the client request that initiated them. */ requestId?: string /** Absolute path on the server host. */ @@ -39,6 +41,7 @@ export interface WorkspaceDescriptor { export interface WorkspaceCreateRequest { path: string + lineageId?: string name?: string binaryPath?: string requestId?: string @@ -293,6 +296,57 @@ export interface InstanceStreamEvent { [key: string]: unknown } +export type WorkflowRunStatus = "running" | "waiting_for_review" | "completed" | "failed" | "cancelled" | "interrupted" +export type WorkflowStepStatus = "pending" | "running" | "completed" | "failed" | "cancelled" + +export interface WorkflowModelSelection { + providerID: string + modelID: string +} + +export interface WorkflowStageConfig { + id: string + title: string + instructions: string + agent?: string + model?: WorkflowModelSelection + requiresApproval?: boolean +} + +export interface WorkflowRunStep extends WorkflowStageConfig { + status: WorkflowStepStatus + sessionId?: string + output?: unknown + outputTruncated?: boolean + error?: string + startedAt?: string + completedAt?: string +} + +export interface WorkflowRun { + id: string + workspaceId: string + workspaceLineageId: string + workspacePath: string + initiatorSessionId?: string + objective: string + status: WorkflowRunStatus + rootSessionId?: string + activeStepId?: string + pendingReviewStepId?: string + steps: WorkflowRunStep[] + error?: string + createdAt: string + updatedAt: string +} + +export interface WorkflowRunCreateRequest { + workspaceId: string + initiatorSessionId?: string + objective: string + stages: WorkflowStageConfig[] +} + export type SideCarKind = "port" export type SideCarPrefixMode = "strip" | "preserve" diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index 0f06a2cf0..933ff1039 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -36,6 +36,7 @@ import { createServerShutdownHandler, orchestrateServerShutdown, type ServerShut import { AutoAcceptManager } from "./permissions/auto-accept-manager" import { createOpencodePermissionReplier } from "./permissions/opencode-replier" import { createOpencodeYoloPersistence } from "./permissions/opencode-yolo-metadata" +import { WorkflowManager } from "./workflows/manager" const require = createRequire(import.meta.url) @@ -375,6 +376,12 @@ async function main() { getServerBaseUrl: () => serverMeta.localUrl, nodeExtraCaCertsPath, }) + const workflowManager = new WorkflowManager({ + workspaceManager, + eventBus, + storageDir: path.join(configDir, "workflow-runs"), + logger: logger.child({ component: "workflows" }), + }) const fileSystemBrowser = new FileSystemBrowser({ rootDir: options.rootDir, unrestricted: options.unrestrictedRoot, @@ -499,6 +506,7 @@ async function main() { remoteProxySessionManager, yoloManager, sessionMetadataPersistence, + workflowManager, uiStaticDir: uiResolution.uiStaticDir ?? DEFAULT_UI_STATIC_DIR, uiDevServerUrl: uiResolution.uiDevServerUrl, logger, @@ -528,6 +536,7 @@ async function main() { remoteProxySessionManager, yoloManager, sessionMetadataPersistence, + workflowManager, uiStaticDir: uiResolution.uiStaticDir ?? DEFAULT_UI_STATIC_DIR, uiDevServerUrl: undefined, logger, @@ -623,6 +632,7 @@ async function main() { orchestrateServerShutdown( { stopInstanceEventBridge: () => instanceEventBridge.shutdown(), + stopWorkflowRuns: () => workflowManager.shutdown(), stopSidecars: () => sidecarManager.shutdown(), stopClientConnections: () => clientConnectionManager.shutdown(), stopRemoteProxySessions: () => remoteProxySessionManager.shutdown(), diff --git a/packages/server/src/server/http-server.ts b/packages/server/src/server/http-server.ts index 8df2832be..65caec567 100644 --- a/packages/server/src/server/http-server.ts +++ b/packages/server/src/server/http-server.ts @@ -47,6 +47,8 @@ import type { SideCarManager } from "../sidecars/manager" import type { PreviewManager } from "../previews/manager" import type { RemoteProxySessionManager } from "./remote-proxy" import { createOpenCodeUpdateService } from "../opencode-update/service" +import type { WorkflowManager } from "../workflows/manager" +import { registerWorkflowRoutes } from "./routes/workflows" interface HttpServerDeps { bindHost: string @@ -71,6 +73,7 @@ interface HttpServerDeps { remoteProxySessionManager: RemoteProxySessionManager yoloManager: AutoAcceptManager sessionMetadataPersistence: OpencodeYoloPersistence + workflowManager: WorkflowManager uiStaticDir: string uiDevServerUrl?: string logger: Logger @@ -326,6 +329,7 @@ export function createHttpServer(deps: HttpServerDeps) { }) registerBackgroundProcessRoutes(app, { backgroundProcessManager }) registerYoloRoutes(app, { yoloManager: deps.yoloManager }) + registerWorkflowRoutes(app, { workflowManager: deps.workflowManager }) registerInstanceProxyRoutes(app, { workspaceManager: deps.workspaceManager, logger: proxyLogger }) diff --git a/packages/server/src/server/routes/workflows.test.ts b/packages/server/src/server/routes/workflows.test.ts new file mode 100644 index 000000000..fb41b20fc --- /dev/null +++ b/packages/server/src/server/routes/workflows.test.ts @@ -0,0 +1,60 @@ +import assert from "node:assert/strict" +import { describe, it } from "node:test" +import Fastify from "fastify" +import type { WorkflowManager } from "../../workflows/manager" +import { registerWorkflowRoutes } from "./workflows" + +describe("workflow routes", () => { + it("validates generic stages and scopes plugin requests to their workspace", async () => { + const calls: unknown[] = [] + const workflowManager = { + start: async (input: unknown) => { + calls.push(input) + return { id: "00000000-0000-4000-8000-000000000001", workspaceId: "workspace-a", status: "running" } + }, + list: async () => [], + get: async () => ({ id: "run", workspaceId: "workspace-b" }), + } as unknown as WorkflowManager + const app = Fastify({ logger: false }) + registerWorkflowRoutes(app, { workflowManager }) + + const duplicate = await app.inject({ + method: "POST", + url: "/workspaces/workspace-a/plugin/workflow-runs", + payload: { + objective: "Ship it", + stages: [ + { id: "same", title: "One", instructions: "First" }, + { id: "same", title: "Two", instructions: "Second" }, + ], + }, + }) + assert.equal(duplicate.statusCode, 400) + assert.equal(calls.length, 0) + + const created = await app.inject({ + method: "POST", + url: "/workspaces/workspace-a/plugin/workflow-runs", + payload: { objective: "Ship it", stages: [{ id: "build", title: "Build", instructions: "Implement" }] }, + }) + assert.equal(created.statusCode, 202) + assert.deepEqual(calls, [{ + workspaceId: "workspace-a", + objective: "Ship it", + stages: [{ id: "build", title: "Build", instructions: "Implement" }], + }]) + + const foreign = await app.inject({ + method: "GET", + url: "/workspaces/workspace-a/plugin/workflow-runs/00000000-0000-4000-8000-000000000001", + }) + assert.equal(foreign.statusCode, 404) + + const pluginApproval = await app.inject({ + method: "POST", + url: "/workspaces/workspace-a/plugin/workflow-runs/00000000-0000-4000-8000-000000000001/approve", + }) + assert.equal(pluginApproval.statusCode, 404) + await app.close() + }) +}) diff --git a/packages/server/src/server/routes/workflows.ts b/packages/server/src/server/routes/workflows.ts new file mode 100644 index 000000000..3e023bec2 --- /dev/null +++ b/packages/server/src/server/routes/workflows.ts @@ -0,0 +1,199 @@ +import type { FastifyInstance } from "fastify" +import { z } from "zod" +import type { WorkflowManager } from "../../workflows/manager" +import { WorkflowRunError } from "../../workflows/manager" + +interface RouteDeps { + workflowManager: WorkflowManager +} + +const ModelSchema = z.object({ + providerID: z.string().trim().min(1).max(200), + modelID: z.string().trim().min(1).max(200), +}) + +const StageSchema = z.object({ + id: z.string().trim().min(1).max(100).regex(/^[a-zA-Z0-9][a-zA-Z0-9_-]*$/), + title: z.string().trim().min(1).max(200), + instructions: z.string().trim().min(1).max(20_000), + agent: z.string().trim().min(1).max(200).optional(), + model: ModelSchema.optional(), + requiresApproval: z.boolean().optional(), +}) + +const CreateObjectSchema = z.object({ + workspaceId: z.string().trim().min(1).max(200), + initiatorSessionId: z.string().trim().min(1).max(200).optional(), + objective: z.string().trim().min(1).max(50_000), + stages: z.array(StageSchema).min(1).max(12), +}) + +const requireUniqueStageIds = (value: { stages: Array<{ id: string }> }, ctx: z.RefinementCtx) => { + const ids = new Set() + value.stages.forEach((stage, index) => { + if (ids.has(stage.id)) { + ctx.addIssue({ code: z.ZodIssueCode.custom, message: "Stage IDs must be unique", path: ["stages", index, "id"] }) + } + ids.add(stage.id) + }) +} + +const CreateSchema = CreateObjectSchema.superRefine(requireUniqueStageIds) + +const RunIdSchema = z.string().uuid() +const ListSchema = z.object({ workspaceId: z.string().trim().min(1).max(200).optional() }) +const PluginCreateSchema = CreateObjectSchema.omit({ workspaceId: true }).superRefine(requireUniqueStageIds) + +export function registerWorkflowRoutes(app: FastifyInstance, deps: RouteDeps) { + app.get<{ Params: { id: string } }>("/workspaces/:id/plugin/workflow-runs", async (request) => { + return { runs: await deps.workflowManager.list(request.params.id) } + }) + + app.post<{ Params: { id: string } }>("/workspaces/:id/plugin/workflow-runs", async (request, reply) => { + const parsed = PluginCreateSchema.safeParse(request.body) + if (!parsed.success) { + reply.code(400) + return { error: "Invalid workflow request", issues: parsed.error.flatten() } + } + try { + const run = await deps.workflowManager.start({ ...parsed.data, workspaceId: request.params.id }) + reply.code(202) + return run + } catch (error) { + if (error instanceof WorkflowRunError) { + reply.code(error.statusCode) + return { error: error.message } + } + throw error + } + }) + + app.get<{ Params: { id: string; runId: string } }>( + "/workspaces/:id/plugin/workflow-runs/:runId", + async (request, reply) => { + const parsed = RunIdSchema.safeParse(request.params.runId) + if (!parsed.success) { + reply.code(400) + return { error: "Invalid workflow run ID" } + } + const run = await deps.workflowManager.get(parsed.data, request.params.id) + if (!run || run.workspaceId !== request.params.id) { + reply.code(404) + return { error: "Workflow run not found" } + } + return run + }, + ) + + app.post<{ Params: { id: string; runId: string } }>( + "/workspaces/:id/plugin/workflow-runs/:runId/cancel", + async (request, reply) => { + const parsed = RunIdSchema.safeParse(request.params.runId) + if (!parsed.success) { + reply.code(400) + return { error: "Invalid workflow run ID" } + } + const existing = await deps.workflowManager.get(parsed.data, request.params.id) + if (!existing || existing.workspaceId !== request.params.id) { + reply.code(404) + return { error: "Workflow run not found" } + } + try { + return await deps.workflowManager.cancel(parsed.data) + } catch (error) { + if (error instanceof WorkflowRunError) { + reply.code(error.statusCode) + return { error: error.message } + } + throw error + } + }, + ) + + app.get("/api/workflow-runs", async (request, reply) => { + const parsed = ListSchema.safeParse(request.query) + if (!parsed.success) { + reply.code(400) + return { error: "Invalid workflow query" } + } + return { runs: await deps.workflowManager.list(parsed.data.workspaceId) } + }) + + app.post("/api/workflow-runs", async (request, reply) => { + const parsed = CreateSchema.safeParse(request.body) + if (!parsed.success) { + reply.code(400) + return { error: "Invalid workflow request", issues: parsed.error.flatten() } + } + + try { + const run = await deps.workflowManager.start(parsed.data) + reply.code(202) + return run + } catch (error) { + if (error instanceof WorkflowRunError) { + reply.code(error.statusCode) + return { error: error.message } + } + throw error + } + }) + + app.get<{ Params: { runId: string } }>("/api/workflow-runs/:runId", async (request, reply) => { + const parsed = RunIdSchema.safeParse(request.params.runId) + if (!parsed.success) { + reply.code(400) + return { error: "Invalid workflow run ID" } + } + const run = await deps.workflowManager.get(parsed.data) + if (!run) { + reply.code(404) + return { error: "Workflow run not found" } + } + return run + }) + + app.post<{ Params: { runId: string } }>("/api/workflow-runs/:runId/cancel", async (request, reply) => { + const parsed = RunIdSchema.safeParse(request.params.runId) + if (!parsed.success) { + reply.code(400) + return { error: "Invalid workflow run ID" } + } + try { + const run = await deps.workflowManager.cancel(parsed.data) + if (!run) { + reply.code(404) + return { error: "Workflow run not found" } + } + return run + } catch (error) { + if (error instanceof WorkflowRunError) { + reply.code(error.statusCode) + return { error: error.message } + } + throw error + } + }) + + app.post<{ Params: { runId: string } }>("/api/workflow-runs/:runId/approve", async (request, reply) => { + const parsed = RunIdSchema.safeParse(request.params.runId) + if (!parsed.success) { + reply.code(400) + return { error: "Invalid workflow run ID" } + } + try { + const run = await deps.workflowManager.approve(parsed.data) + if (!run) { + reply.code(404) + return { error: "Workflow run not found" } + } + return run + } catch (error) { + if (error instanceof WorkflowRunError) { + reply.code(error.statusCode) + return { error: error.message } + } + throw error + } + }) +} diff --git a/packages/server/src/server/routes/workspaces.test.ts b/packages/server/src/server/routes/workspaces.test.ts index e115c5b7a..9e0ac90e2 100644 --- a/packages/server/src/server/routes/workspaces.test.ts +++ b/packages/server/src/server/routes/workspaces.test.ts @@ -40,6 +40,7 @@ describe("workspace routes", () => { path: "C:/work", name: "Work", binaryPath: " C:/tools/opencode.exe ", + lineageId: "00000000-0000-4000-8000-000000000001", requestId: " restore-request ", forceNew: true, }, @@ -48,6 +49,7 @@ describe("workspace routes", () => { assert.equal(response.statusCode, 201) assert.deepEqual(calls, [["C:/work", "Work", { binaryPath: "C:/tools/opencode.exe", + lineageId: "00000000-0000-4000-8000-000000000001", requestId: "restore-request", forceNew: true, }]]) diff --git a/packages/server/src/server/routes/workspaces.ts b/packages/server/src/server/routes/workspaces.ts index e7f052136..8137823de 100644 --- a/packages/server/src/server/routes/workspaces.ts +++ b/packages/server/src/server/routes/workspaces.ts @@ -13,6 +13,7 @@ interface RouteDeps { const WorkspaceCreateSchema = z.object({ path: z.string(), + lineageId: z.string().uuid().optional(), name: z.string().optional(), binaryPath: z.string().trim().min(1).max(4096).optional(), requestId: z.string().trim().min(1).max(128).optional(), @@ -79,6 +80,7 @@ export function registerWorkspaceRoutes(app: FastifyInstance, deps: RouteDeps) { binaryPath: body.binaryPath, requestId: body.requestId, forceNew: body.forceNew, + ...(body.lineageId ? { lineageId: body.lineageId } : {}), }) reply.code(201) return result.created ? result.workspace : { ...result.workspace, reused: true as const } diff --git a/packages/server/src/shutdown.test.ts b/packages/server/src/shutdown.test.ts index 446e77918..33642d7cd 100644 --- a/packages/server/src/shutdown.test.ts +++ b/packages/server/src/shutdown.test.ts @@ -10,7 +10,7 @@ import { const logger = { info() {}, warn() {}, error() {} } const operations = (overrides: Partial = {}): ServerShutdownOperations => ({ - stopInstanceEventBridge() {}, stopSidecars() {}, stopClientConnections() {}, + stopInstanceEventBridge() {}, stopWorkflowRuns() {}, stopSidecars() {}, stopClientConnections() {}, stopRemoteProxySessions() {}, stopWorkspaces() {}, stopHttpServers() {}, stopReleaseMonitor() {}, ...overrides, }) diff --git a/packages/server/src/shutdown.ts b/packages/server/src/shutdown.ts index 3324aa255..9ee5a4d9e 100644 --- a/packages/server/src/shutdown.ts +++ b/packages/server/src/shutdown.ts @@ -6,7 +6,7 @@ export const SERVER_SHUTDOWN_COMPLETE = "CODENOMAD_SHUTDOWN_STATUS:complete" export const SERVER_SHUTDOWN_INCOMPLETE = "CODENOMAD_SHUTDOWN_STATUS:incomplete" export type ServerShutdownOperations = Record< - "stopInstanceEventBridge" | "stopSidecars" | "stopClientConnections" | "stopRemoteProxySessions" | "stopWorkspaces" | + "stopInstanceEventBridge" | "stopWorkflowRuns" | "stopSidecars" | "stopClientConnections" | "stopRemoteProxySessions" | "stopWorkspaces" | "stopHttpServers" | "stopReleaseMonitor", ShutdownOperation > @@ -92,6 +92,7 @@ export async function orchestrateServerShutdown( await Promise.all([ settle([ ["stopInstanceEventBridge", operations.stopInstanceEventBridge], ["stopSidecars", operations.stopSidecars], + ["stopWorkflowRuns", operations.stopWorkflowRuns], ["stopClientConnections", operations.stopClientConnections], ["stopRemoteProxySessions", operations.stopRemoteProxySessions], ]), workspaceShutdown, diff --git a/packages/server/src/workflows/manager.test.ts b/packages/server/src/workflows/manager.test.ts new file mode 100644 index 000000000..182cffec4 --- /dev/null +++ b/packages/server/src/workflows/manager.test.ts @@ -0,0 +1,282 @@ +import assert from "node:assert/strict" +import fs from "node:fs/promises" +import os from "node:os" +import path from "node:path" +import { describe, it } from "node:test" +import type { OpencodeClient } from "@opencode-ai/sdk/v2/client" +import type { EventBus } from "../events/bus" +import type { Logger } from "../logger" +import type { WorkspaceManager } from "../workspaces/manager" +import { WorkflowManager } from "./manager" + +describe("WorkflowManager", () => { + it("persists and hands a structured planner result to the implementer", async () => { + const storageDir = await fs.mkdtemp(path.join(os.tmpdir(), "codenomad-workflows-")) + const creates: Array | undefined> = [] + const prompts: Array> = [] + let session = 0 + const client = { + session: { + create: async (input?: Record) => { + creates.push(input) + return { data: { id: `session-${++session}` } } + }, + prompt: async (input: Record) => { + prompts.push(input) + if (prompts.length === 1) { + return { + data: { + info: { structured: { summary: "Plan", steps: ["Change code", "Run test"] } }, + parts: [], + }, + } + } + const text = prompts.length === 2 ? "Reviewed plan" : "Implemented" + return { data: { info: {}, parts: [{ type: "text", text }] } } + }, + abort: async () => ({ data: true }), + }, + } as unknown as OpencodeClient + const workspaceManager = { + get: () => ({ id: "workspace", lineageId: "lineage-a", path: "C:/workspace", status: "ready" }), + list: () => [{ id: "workspace", lineageId: "lineage-a", path: "C:/workspace", status: "ready" }], + } as unknown as WorkspaceManager + const events: unknown[] = [] + const eventBus = { publish: (event: unknown) => { events.push(event); return true } } as unknown as EventBus + const logger = { warn() {}, error() {} } as unknown as Logger + const manager = new WorkflowManager({ + workspaceManager, + eventBus, + logger, + storageDir, + createClient: () => client, + }) + let reloaded: WorkflowManager | undefined + + try { + const started = await manager.start({ + workspaceId: "workspace", + objective: "Add workflow support", + stages: [ + { id: "planner", title: "Planner", instructions: "Create a plan", requiresApproval: true }, + { id: "reviewer", title: "Reviewer", instructions: "Review the approved plan", requiresApproval: true }, + { id: "implementer", title: "Implementer", instructions: "Implement the reviewed plan", requiresApproval: true }, + ], + }) + let run = started + for (let attempt = 0; attempt < 50 && run.status === "running"; attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 5)) + run = (await manager.get(started.id))! + } + + assert.equal(run.status, "waiting_for_review") + assert.equal(run.rootSessionId, "session-1") + assert.deepEqual(run.steps.map((step) => [step.id, step.status, step.sessionId]), [ + ["planner", "completed", "session-2"], + ["reviewer", "pending", undefined], + ["implementer", "pending", undefined], + ]) + + run = (await manager.approve(started.id))! + for (let attempt = 0; attempt < 50 && run.status === "running"; attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 5)) + run = (await manager.get(started.id))! + } + + assert.equal(run.status, "waiting_for_review") + assert.deepEqual(run.steps.map((step) => [step.id, step.status, step.sessionId]), [ + ["planner", "completed", "session-2"], + ["reviewer", "completed", "session-3"], + ["implementer", "pending", undefined], + ]) + + run = (await manager.approve(started.id))! + for (let attempt = 0; attempt < 50 && run.status === "running"; attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 5)) + run = (await manager.get(started.id))! + } + + assert.equal(run.status, "waiting_for_review") + assert.deepEqual(run.steps.map((step) => [step.id, step.status, step.sessionId]), [ + ["planner", "completed", "session-2"], + ["reviewer", "completed", "session-3"], + ["implementer", "completed", "session-4"], + ]) + assert.equal(creates[1]?.parentID, "session-1") + assert.equal(creates[2]?.parentID, "session-1") + assert.match(JSON.stringify(prompts[1]), /Change code/) + assert.match(JSON.stringify(prompts[2]), /Reviewed plan/) + assert.ok(events.length >= 10) + + await manager.shutdown() + const restoredWorkspaceManager = { + get: (id: string) => id === "workspace-restored" + ? { id, lineageId: "lineage-a", path: "C:/workspace", status: "ready" } + : undefined, + list: () => [{ id: "workspace-restored", lineageId: "lineage-a", path: "C:/workspace", status: "ready" }], + } as unknown as WorkspaceManager + reloaded = new WorkflowManager({ + workspaceManager: restoredWorkspaceManager, + eventBus, + logger, + storageDir, + createClient: () => client, + }) + await assert.rejects( + reloaded.start({ + workspaceId: "workspace-restored", + objective: "Conflicting run", + stages: [{ id: "other", title: "Other", instructions: "Do other work" }], + }), + /workspace lineage/, + ) + run = (await reloaded.approve(started.id))! + for (let attempt = 0; attempt < 50 && run.status === "running"; attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 5)) + run = (await reloaded.get(started.id))! + } + assert.equal(run.status, "completed") + const [restored] = await reloaded.list("workspace-restored") + assert.equal(restored?.workspaceId, "workspace-restored") + } finally { + await manager.shutdown() + await reloaded?.shutdown() + await fs.rm(storageDir, { recursive: true, force: true }) + } + }) + + it("fails a stage when its OpenCode request exceeds the operation timeout", async () => { + const storageDir = await fs.mkdtemp(path.join(os.tmpdir(), "codenomad-workflow-timeout-")) + let session = 0 + let aborts = 0 + const client = { + session: { + create: async () => ({ data: { id: `session-${++session}` } }), + prompt: async (_input: unknown, options?: { signal?: AbortSignal }) => new Promise((_, reject) => { + const signal = options?.signal + if (!signal) return + signal.addEventListener("abort", () => reject(signal.reason), { once: true }) + }), + abort: async () => { aborts += 1; return { error: "abort failed" } }, + }, + } as unknown as OpencodeClient + const workspaceManager = { + get: () => ({ id: "workspace", lineageId: "lineage-timeout", path: "C:/timeout-workspace", status: "ready" }), + } as unknown as WorkspaceManager + const eventBus = { publish: () => true } as unknown as EventBus + const logger = { warn() {}, error() {} } as unknown as Logger + const manager = new WorkflowManager({ + workspaceManager, + eventBus, + logger, + storageDir, + createClient: () => client, + promptTimeoutMs: 10, + }) + + try { + const started = await manager.start({ + workspaceId: "workspace", + objective: "Never finish", + stages: [{ id: "blocked", title: "Blocked", instructions: "Wait forever" }], + }) + let run = started + for (let attempt = 0; attempt < 50 && run.status === "running"; attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 5)) + run = (await manager.get(started.id))! + } + assert.equal(run.status, "failed") + assert.equal(run.steps[0]?.status, "failed") + assert.equal(aborts, 1) + await assert.rejects(manager.start({ + workspaceId: "workspace", + objective: "Must remain blocked", + stages: [{ id: "next", title: "Next", instructions: "Do not start" }], + }), /already running/) + } finally { + await manager.shutdown() + await fs.rm(storageDir, { recursive: true, force: true }) + } + }) + + it("keeps force-created instances of the same path isolated by lineage", async () => { + const storageDir = await fs.mkdtemp(path.join(os.tmpdir(), "codenomad-workflow-lineage-")) + let session = 0 + const client = { + session: { + create: async () => ({ data: { id: `session-${++session}` } }), + prompt: async () => ({ data: { info: {}, parts: [{ type: "text", text: "Review me" }] } }), + abort: async () => ({ data: true }), + }, + } as unknown as OpencodeClient + const workspaceManager = { + get: (id: string) => ({ id, lineageId: id === "a" ? "lineage-a" : "lineage-b", path: "C:/same", status: "ready" }), + } as unknown as WorkspaceManager + const eventBus = { publish: () => true } as unknown as EventBus + const logger = { warn() {}, error() {} } as unknown as Logger + const manager = new WorkflowManager({ workspaceManager, eventBus, logger, storageDir, createClient: () => client }) + + try { + const started = await manager.start({ + workspaceId: "a", + objective: "Lineage A", + stages: [{ id: "only", title: "Only", instructions: "Run", requiresApproval: true }], + }) + let run = started + for (let attempt = 0; attempt < 50 && run.status === "running"; attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 5)) + run = (await manager.get(started.id))! + } + assert.equal(run.status, "waiting_for_review") + assert.deepEqual(await manager.list("b"), []) + assert.equal(await manager.get(started.id, "b"), undefined) + } finally { + await manager.shutdown() + await fs.rm(storageDir, { recursive: true, force: true }) + } + }) + + it("does not fail a completed run when history pruning fails", async () => { + const storageDir = await fs.mkdtemp(path.join(os.tmpdir(), "codenomad-workflow-prune-")) + let session = 0 + const client = { + session: { + create: async () => ({ data: { id: `session-${++session}` } }), + prompt: async () => ({ data: { info: {}, parts: [{ type: "text", text: "Complete" }] } }), + abort: async () => ({ data: true }), + }, + } as unknown as OpencodeClient + const workspaceManager = { + get: () => ({ id: "workspace", lineageId: "lineage", path: "C:/workspace", status: "ready" }), + } as unknown as WorkspaceManager + let warnings = 0 + const logger = { warn: () => { warnings += 1 }, error() {} } as unknown as Logger + const manager = new WorkflowManager({ + workspaceManager, + eventBus: { publish: () => true } as unknown as EventBus, + logger, + storageDir, + createClient: () => client, + }) + ;(manager as any).pruneHistory = async () => { throw new Error("prune failed") } + + try { + const started = await manager.start({ + workspaceId: "workspace", + objective: "Finish despite prune failure", + stages: [{ id: "only", title: "Only", instructions: "Complete" }], + }) + let run = started + for (let attempt = 0; attempt < 50 && run.status === "running"; attempt += 1) { + await new Promise((resolve) => setTimeout(resolve, 5)) + run = (await manager.get(started.id))! + } + + assert.equal(run.status, "completed") + assert.equal(warnings, 1) + } finally { + await manager.shutdown() + await fs.rm(storageDir, { recursive: true, force: true }) + } + }) +}) diff --git a/packages/server/src/workflows/manager.ts b/packages/server/src/workflows/manager.ts new file mode 100644 index 000000000..36ecccf1f --- /dev/null +++ b/packages/server/src/workflows/manager.ts @@ -0,0 +1,592 @@ +import { randomUUID } from "node:crypto" +import fs from "node:fs/promises" +import path from "node:path" +import type { OpencodeClient } from "@opencode-ai/sdk/v2/client" +import type { WorkflowRun, WorkflowRunCreateRequest, WorkflowRunStep } from "../api-types" +import type { EventBus } from "../events/bus" +import type { Logger } from "../logger" +import { createInstanceClient } from "../workspaces/instance-client" +import type { WorkspaceManager } from "../workspaces/manager" + +const PROMPT_TIMEOUT_MS = 30 * 60 * 1000 +const ABORT_TIMEOUT_MS = 5_000 +const SHUTDOWN_TIMEOUT_MS = 10_000 +const MAX_OUTPUT_CHARS = 16_000 +const WORKFLOW_HISTORY_LIMIT = 100 + +export class WorkflowRunError extends Error { + constructor(message: string, readonly statusCode: number) { + super(message) + } +} + +interface WorkflowManagerOptions { + workspaceManager: WorkspaceManager + eventBus: EventBus + logger: Logger + storageDir: string + createClient?: (workspaceId: string) => OpencodeClient | null + promptTimeoutMs?: number +} + +interface ActiveRun { + run: WorkflowRun + client: OpencodeClient + activeSessionId?: string + cancelRequested: boolean + completion?: Promise + abortController: AbortController + releaseBlocked: boolean +} + +class WorkflowCancelledError extends Error {} + +export class WorkflowManager { + private readonly activeRuns = new Map() + private readonly activeWorkspaces = new Map() + private readonly reservedLineages = new Map() + private readonly persistQueues = new Map>() + private readonly transitionQueues = new Map>() + private readonly createClient: (workspaceId: string) => OpencodeClient | null + private readonly promptTimeoutMs: number + private readonly initialized: Promise + private shuttingDown = false + private shutdownPromise?: Promise + + constructor(private readonly options: WorkflowManagerOptions) { + this.promptTimeoutMs = options.promptTimeoutMs ?? PROMPT_TIMEOUT_MS + this.createClient = options.createClient + ?? ((workspaceId) => createInstanceClient(options.workspaceManager, workspaceId, { timeoutMs: this.promptTimeoutMs })) + this.initialized = this.recoverInterruptedRuns() + } + + async start(input: WorkflowRunCreateRequest): Promise { + await this.initialized + const client = this.requireReadyClient(input.workspaceId) + const workspace = this.options.workspaceManager.get(input.workspaceId)! + + const now = new Date().toISOString() + const run: WorkflowRun = { + id: randomUUID(), + workspaceId: input.workspaceId, + workspaceLineageId: workspace.lineageId ?? workspace.id, + workspacePath: workspace.path, + ...(input.initiatorSessionId ? { initiatorSessionId: input.initiatorSessionId } : {}), + objective: input.objective, + status: "running", + steps: input.stages.map((stage) => ({ ...stage, status: "pending" })), + createdAt: now, + updatedAt: now, + } + const active: ActiveRun = { + run, client, cancelRequested: false, abortController: new AbortController(), releaseBlocked: false, + } + this.reserve(active) + try { + await this.persist(run) + } catch (error) { + this.release(active) + throw error + } + if (active.cancelRequested || this.shuttingDown) { + this.release(active) + return run + } + this.launch(active, (current) => this.executePendingStages(current)) + return run + } + + async get(runId: string, workspaceId?: string): Promise { + await this.initialized + const active = this.activeRuns.get(runId) + if (!active) { + const run = await this.read(runId) + return workspaceId && run ? this.bindWorkspace(run, workspaceId) : run + } + if (["waiting_for_review", "completed", "failed", "cancelled"].includes(active.run.status)) { + await active.completion + const run = await this.read(runId) ?? active.run + return workspaceId ? this.bindWorkspace(run, workspaceId) : run + } + return workspaceId ? this.bindWorkspace(active.run, workspaceId) : active.run + } + + async list(workspaceId?: string): Promise { + await this.initialized + let entries: string[] + try { + entries = await fs.readdir(this.options.storageDir) + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return [] + throw error + } + const runs = await Promise.all(entries + .filter((entry) => entry.endsWith(".json")) + .map((entry) => this.read(entry.slice(0, -5)))) + const matched: WorkflowRun[] = [] + for (const run of runs) { + if (!run) continue + const bound = workspaceId ? await this.bindWorkspace(run, workspaceId) : run + if (bound) matched.push(bound) + } + return matched.sort((left, right) => right.createdAt.localeCompare(left.createdAt)).slice(0, WORKFLOW_HISTORY_LIMIT) + } + + async approve(runId: string): Promise { + await this.initialized + return this.withRunTransition(runId, async () => { + const current = this.activeRuns.get(runId) + if (current?.run.status === "waiting_for_review") await current.completion + else if (current) { + throw new WorkflowRunError("Workflow stage is not ready for review", 409) + } + const run = await this.read(runId) + if (!run) return undefined + const reviewed = run.steps.find((step) => step.id === run.pendingReviewStepId) + if (run.status !== "waiting_for_review" || reviewed?.status !== "completed") { + throw new WorkflowRunError("Workflow run is not waiting for review", 409) + } + + const restoredWorkspace = this.options.workspaceManager.list().find((workspace) => + workspace.lineageId === run.workspaceLineageId && workspace.status === "ready") + if (restoredWorkspace && restoredWorkspace.id !== run.workspaceId) { + await this.bindWorkspace(run, restoredWorkspace.id) + } + const client = this.requireReadyClient(run.workspaceId, run.id) + run.status = "running" + delete run.pendingReviewStepId + delete run.error + const active: ActiveRun = { + run, client, cancelRequested: false, abortController: new AbortController(), releaseBlocked: false, + } + this.reserve(active) + try { + await this.persist(run) + } catch (error) { + run.status = "waiting_for_review" + this.release(active) + throw error + } + if (active.cancelRequested || this.shuttingDown) { + this.release(active) + return run + } + this.launch(active, (current) => this.executePendingStages(current)) + return run + }) + } + + async cancel(runId: string): Promise { + await this.initialized + if (this.shuttingDown) throw new WorkflowRunError("CodeNomad is shutting down", 503) + return this.cancelRun(runId) + } + + async shutdown(): Promise { + if (!this.shutdownPromise) { + this.shuttingDown = true + this.shutdownPromise = this.performShutdown() + } + await this.withTimeout(this.shutdownPromise, SHUTDOWN_TIMEOUT_MS, "Workflow shutdown timed out") + } + + private async cancelRun(runId: string): Promise { + return this.withRunTransition(runId, async () => { + const active = this.activeRuns.get(runId) + const run = active?.run ?? await this.read(runId) + if (!run) return undefined + if (run.status !== "running" && run.status !== "waiting_for_review") return run + + if (active) { + active.cancelRequested = true + active.abortController.abort() + if (active.activeSessionId) { + const sessionId = active.activeSessionId + active.activeSessionId = undefined + if (!await this.abortSession(active, sessionId)) active.releaseBlocked = true + } + } + this.markCancelled(run) + await this.persist(run) + if (!active && this.activeWorkspaces.get(run.workspaceId) === run.id) { + this.activeWorkspaces.delete(run.workspaceId) + } + if (this.reservedLineages.get(run.workspaceLineageId) === run.id) { + this.reservedLineages.delete(run.workspaceLineageId) + } + return run + }) + } + + private async performShutdown(): Promise { + await this.initialized + await Promise.all(Array.from(this.transitionQueues.values())) + const active = Array.from(this.activeRuns.values()) + await Promise.all(active.map(({ run }) => this.cancelRun(run.id))) + await Promise.all(active.map(({ completion }) => completion)) + await Promise.all(Array.from(this.persistQueues.values())) + } + + private async withTimeout(operation: Promise, timeoutMs: number, message: string): Promise { + let timeout: NodeJS.Timeout | undefined + try { + return await Promise.race([ + operation, + new Promise((_, reject) => { + timeout = setTimeout(() => reject(new Error(message)), timeoutMs) + }), + ]) + } finally { + if (timeout) clearTimeout(timeout) + } + } + + private requireReadyClient(workspaceId: string, runId?: string): OpencodeClient { + if (this.shuttingDown) throw new WorkflowRunError("CodeNomad is shutting down", 503) + const workspace = this.options.workspaceManager.get(workspaceId) + if (!workspace) throw new WorkflowRunError("Workspace not found", 404) + if (workspace.status !== "ready") throw new WorkflowRunError("Workspace instance is not ready", 409) + const activeRunId = this.activeWorkspaces.get(workspaceId) + if (activeRunId && activeRunId !== runId) { + throw new WorkflowRunError("A workflow is already running in this workspace", 409) + } + const lineageId = workspace.lineageId ?? workspace.id + const lineageRunId = this.reservedLineages.get(lineageId) + if (lineageRunId && lineageRunId !== runId) { + throw new WorkflowRunError("A workflow is already running for this workspace lineage", 409) + } + const client = this.createClient(workspaceId) + if (!client) throw new WorkflowRunError("Workspace instance is not ready", 409) + return client + } + + private reserve(active: ActiveRun) { + this.activeRuns.set(active.run.id, active) + // ponytail: one run per workspace; add worktree-aware concurrency only when parallel workflows are needed. + this.activeWorkspaces.set(active.run.workspaceId, active.run.id) + this.reservedLineages.set(active.run.workspaceLineageId, active.run.id) + } + + private release(active: ActiveRun) { + this.activeRuns.delete(active.run.id) + if (active.releaseBlocked) { + this.options.logger.error({ runId: active.run.id }, "Retaining workflow reservation after unconfirmed session abort") + return + } + if (active.run.status !== "waiting_for_review" && this.activeWorkspaces.get(active.run.workspaceId) === active.run.id) { + this.activeWorkspaces.delete(active.run.workspaceId) + } + if (active.run.status !== "waiting_for_review" && this.reservedLineages.get(active.run.workspaceLineageId) === active.run.id) { + this.reservedLineages.delete(active.run.workspaceLineageId) + } + } + + private async withRunTransition(runId: string, transition: () => Promise): Promise { + const previous = this.transitionQueues.get(runId) ?? Promise.resolve() + const queued = previous.catch(() => undefined).then(transition) + const marker = queued.then(() => undefined, () => undefined) + this.transitionQueues.set(runId, marker) + try { + return await queued + } finally { + if (this.transitionQueues.get(runId) === marker) this.transitionQueues.delete(runId) + } + } + + private launch(active: ActiveRun, execute: (active: ActiveRun) => Promise) { + active.completion = execute(active) + .catch((error) => this.handleExecutionError(active, error)) + .catch((error) => { + this.options.logger.error({ err: error, runId: active.run.id }, "Failed to persist workflow failure") + }) + .finally(() => this.release(active)) + } + + private async executePendingStages(active: ActiveRun): Promise { + const { run, client } = active + if (!run.rootSessionId) { + const root = await this.requireData(client.session.create({ + ...(run.initiatorSessionId ? { parentID: run.initiatorSessionId } : {}), + title: `Workflow: ${run.objective.slice(0, 80)}`, + metadata: this.sessionMetadata(run.id, "workflow"), + }, { signal: this.operationSignal(active) }), "create workflow session") + this.throwIfCancelled(active) + run.rootSessionId = root.id + await this.persist(run) + } + + while (true) { + const index = run.steps.findIndex((step) => step.status === "pending") + if (index < 0) break + const step = run.steps[index]! + const previous = index > 0 ? run.steps[index - 1]?.output : undefined + await this.runStep(active, step, this.buildStagePrompt(run, step, previous)) + this.throwIfCancelled(active) + if (step.requiresApproval) { + run.status = "waiting_for_review" + run.pendingReviewStepId = step.id + delete run.activeStepId + await this.persist(run) + return + } + } + + run.status = "completed" + delete run.activeStepId + delete run.pendingReviewStepId + await this.persist(run) + } + + private buildStagePrompt(run: WorkflowRun, step: WorkflowRunStep, previous: unknown): string { + return [ + `Workflow stage: ${step.title}`, + "", + `Objective:\n${run.objective}`, + "", + `Stage instructions:\n${step.instructions}`, + ...(previous === undefined ? [] : ["", `Previous stage handoff:\n${JSON.stringify(previous, null, 2)}`]), + ].join("\n") + } + + private async runStep( + active: ActiveRun, + step: WorkflowRunStep, + prompt: string, + ): Promise { + const { run, client } = active + this.throwIfCancelled(active) + step.status = "running" + step.startedAt = new Date().toISOString() + run.activeStepId = step.id + await this.persist(run) + this.throwIfCancelled(active) + + const session = await this.requireData(client.session.create({ + parentID: run.rootSessionId, + title: `${step.title}: ${run.objective.slice(0, 60)}`, + ...(step.agent ? { agent: step.agent } : {}), + metadata: this.sessionMetadata(run.id, step.id), + }, { signal: this.operationSignal(active) }), `create ${step.title} session`) + step.sessionId = session.id + active.activeSessionId = session.id + this.throwIfCancelled(active) + await this.persist(run) + this.throwIfCancelled(active) + + const response = await this.requireData(client.session.prompt({ + sessionID: session.id, + ...(step.agent ? { agent: step.agent } : {}), + ...(step.model ? { model: step.model } : {}), + parts: [{ type: "text", text: prompt }], + }, { signal: this.operationSignal(active) }), `run ${step.title} session`) + active.activeSessionId = undefined + this.throwIfCancelled(active) + if (response.info.error) throw new Error(this.errorMessage(response.info.error)) + + const output = response.info.structured ?? response.parts + .filter((part) => part.type === "text") + .map((part) => part.text) + .join("\n") + const bounded = this.boundOutput(output) + step.output = bounded.output + step.outputTruncated = bounded.truncated || undefined + step.status = "completed" + step.completedAt = new Date().toISOString() + active.activeSessionId = undefined + await this.persist(run) + return output + } + + private sessionMetadata(runId: string, role: string) { + return { codenomad: { version: 1, workflow: { runId, role } } } + } + + private operationSignal(active: ActiveRun): AbortSignal { + return AbortSignal.any([active.abortController.signal, AbortSignal.timeout(this.promptTimeoutMs)]) + } + + private boundOutput(output: unknown): { output: unknown; truncated: boolean } { + if (typeof output === "string") { + return output.length <= MAX_OUTPUT_CHARS + ? { output, truncated: false } + : { output: output.slice(0, MAX_OUTPUT_CHARS), truncated: true } + } + const serialized = JSON.stringify(output) + return serialized.length <= MAX_OUTPUT_CHARS + ? { output, truncated: false } + : { output: serialized.slice(0, MAX_OUTPUT_CHARS), truncated: true } + } + + private throwIfCancelled(active: ActiveRun) { + if (active.cancelRequested) throw new WorkflowCancelledError("Workflow run cancelled") + } + + private async abortSession(active: ActiveRun, sessionId: string): Promise { + try { + const response = await active.client.session.abort( + { sessionID: sessionId }, + { signal: AbortSignal.timeout(ABORT_TIMEOUT_MS) }, + ) + if (response.data === true && response.error === undefined) return true + this.options.logger.warn({ err: response.error, runId: active.run.id }, "Workflow session abort was not confirmed") + return false + } catch (error) { + this.options.logger.warn({ err: error, runId: active.run.id }, "Failed to abort workflow session") + return false + } + } + + private async handleExecutionError(active: ActiveRun, error: unknown): Promise { + const { run } = active + if (active.activeSessionId) { + const sessionId = active.activeSessionId + active.activeSessionId = undefined + if (!await this.abortSession(active, sessionId)) active.releaseBlocked = true + } + if (active.cancelRequested || error instanceof WorkflowCancelledError) { + this.markCancelled(run) + } else { + const message = this.errorMessage(error) + run.status = "failed" + run.error = message + const step = run.steps.find((candidate) => candidate.status === "running") + if (step) { + step.status = "failed" + step.error = message + step.completedAt = new Date().toISOString() + } + delete run.activeStepId + this.options.logger.error({ err: error, runId: run.id }, "Workflow run failed") + } + await this.persist(run) + } + + private markCancelled(run: WorkflowRun) { + run.status = "cancelled" + const step = run.steps.find((candidate) => candidate.status === "running") + if (step) { + step.status = "cancelled" + step.completedAt = new Date().toISOString() + } + delete run.activeStepId + delete run.pendingReviewStepId + } + + private async requireData(request: Promise<{ data?: T; error?: unknown }>, action: string): Promise { + const response = await request + if (response.data !== undefined) return response.data + throw new Error(`${action} failed: ${this.errorMessage(response.error)}`) + } + + private errorMessage(error: unknown): string { + if (error instanceof Error) return error.message + if (typeof error === "string") return error + try { + return JSON.stringify(error) || "Unknown error" + } catch { + return "Unknown error" + } + } + + private runPath(runId: string) { + return path.join(this.options.storageDir, `${runId}.json`) + } + + private async bindWorkspace(run: WorkflowRun, workspaceId: string): Promise { + if (run.workspaceId === workspaceId) return run + const workspace = this.options.workspaceManager.get(workspaceId) + if (!workspace || !workspace.lineageId || run.workspaceLineageId !== workspace.lineageId) return undefined + const previousId = run.workspaceId + run.workspaceId = workspaceId + if (this.activeWorkspaces.get(previousId) === run.id) this.activeWorkspaces.delete(previousId) + if (run.status === "waiting_for_review") this.activeWorkspaces.set(workspaceId, run.id) + await this.persist(run, false) + return run + } + + private async read(runId: string): Promise { + try { + return JSON.parse(await fs.readFile(this.runPath(runId), "utf8")) as WorkflowRun + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined + throw error + } + } + + private async recoverInterruptedRuns(): Promise { + let entries: string[] + try { + entries = await fs.readdir(this.options.storageDir) + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return + throw error + } + + for (const entry of entries.filter((candidate) => candidate.endsWith(".json"))) { + try { + const run = await this.read(entry.slice(0, -5)) + if (!run) continue + if (run.status === "waiting_for_review") { + this.reservedLineages.set(run.workspaceLineageId, run.id) + continue + } + if (run.status !== "running") continue + run.status = "interrupted" + run.error = "CodeNomad restarted before this workflow completed" + const step = run.steps.find((candidate) => candidate.status === "running") + if (step) { + step.status = "failed" + step.error = run.error + step.completedAt = new Date().toISOString() + } + delete run.activeStepId + await this.persist(run) + } catch (error) { + this.options.logger.error({ err: error, file: entry }, "Failed to recover workflow run") + } + } + } + + private async persist(run: WorkflowRun, touch = true): Promise { + if (touch) run.updatedAt = new Date().toISOString() + const snapshot = JSON.parse(JSON.stringify(run)) as WorkflowRun + const previous = this.persistQueues.get(run.id) ?? Promise.resolve() + const queued = previous.catch(() => undefined).then(async () => { + await fs.mkdir(this.options.storageDir, { recursive: true }) + const destination = this.runPath(run.id) + const temporary = `${destination}.${randomUUID()}.tmp` + await fs.writeFile(temporary, `${JSON.stringify(snapshot, null, 2)}\n`, "utf8") + await fs.rename(temporary, destination) + this.options.eventBus.publish({ + type: "instance.event", + instanceId: snapshot.workspaceId, + event: { type: "workflow.run.updated", properties: { run: snapshot } }, + }) + if (["completed", "failed", "cancelled", "interrupted"].includes(snapshot.status)) { + await this.pruneHistory(snapshot.workspaceLineageId).catch((error) => { + this.options.logger.warn({ err: error, runId: snapshot.id }, "Failed to prune workflow history") + }) + } + }) + this.persistQueues.set(run.id, queued) + try { + await queued + } finally { + if (this.persistQueues.get(run.id) === queued) this.persistQueues.delete(run.id) + } + } + + private async pruneHistory(workspaceLineageId: string): Promise { + const entries = await fs.readdir(this.options.storageDir) + const runs = (await Promise.all(entries + .filter((entry) => entry.endsWith(".json")) + .map((entry) => this.read(entry.slice(0, -5))))) + .filter((run): run is WorkflowRun => Boolean( + run + && run.workspaceLineageId === workspaceLineageId + && ["completed", "failed", "cancelled", "interrupted"].includes(run.status), + )) + .sort((left, right) => right.updatedAt.localeCompare(left.updatedAt)) + await Promise.all(runs.slice(WORKFLOW_HISTORY_LIMIT).map((run) => fs.rm(this.runPath(run.id), { force: true }))) + } +} diff --git a/packages/server/src/workspaces/__tests__/workspace-identity.test.ts b/packages/server/src/workspaces/__tests__/workspace-identity.test.ts index ccf8e6ca4..59e569b2f 100644 --- a/packages/server/src/workspaces/__tests__/workspace-identity.test.ts +++ b/packages/server/src/workspaces/__tests__/workspace-identity.test.ts @@ -183,4 +183,17 @@ describe("workspace identity", () => { assert.equal(reused.created, false) assert.equal(reused.workspace.id, normal.workspace.id) }) + + it("uses an explicit lineage before reusing a workspace by path", async () => { + const { root, target, link } = await createLinkedWorkspace() + const manager = createManager(root) + const normal = await manager.create(target) + const restored = await manager.create(link, undefined, { lineageId: "restored-lineage" }) + const repeated = await manager.create(target, undefined, { lineageId: "restored-lineage" }) + + assert.notEqual(restored.workspace.id, normal.workspace.id) + assert.equal(restored.workspace.lineageId, "restored-lineage") + assert.equal(repeated.workspace.id, restored.workspace.id) + assert.equal(repeated.created, false) + }) }) diff --git a/packages/server/src/workspaces/manager.ts b/packages/server/src/workspaces/manager.ts index 547bccd8f..8fa2cb1e5 100644 --- a/packages/server/src/workspaces/manager.ts +++ b/packages/server/src/workspaces/manager.ts @@ -121,6 +121,7 @@ export interface WorkspaceCreateOptions { binaryPath?: string requestId?: string forceNew?: boolean + lineageId?: string } type CreationRequestState = "active" | "cancelled" | "released" type WorkspaceCreationOwnership = Map @@ -258,7 +259,23 @@ export class WorkspaceManager { if (this.shuttingDown) { throw new Error("Workspace manager is shutting down") } - if (options.forceNew) { + if (options.lineageId) { + const lineageRecord = Array.from(this.workspaces.values()).find((record) => + record.lineageId === options.lineageId && !record[WORKSPACE_STATE].abortController.signal.aborted) + if (lineageRecord) { + if (lineageRecord.identityKey !== identityKey) { + throw new Error("Workspace lineage belongs to a different workspace") + } + const owner = options.requestId ?? ORDINARY_CREATION_OWNER + if (!lineageRecord.ownership.has(owner)) lineageRecord.ownership.set(owner, "active") + this.syncOwnership(lineageRecord) + const workspace = lineageRecord[WORKSPACE_STATE].creation + ? (await lineageRecord[WORKSPACE_STATE].creation).workspace + : lineageRecord + return this.finishCreation({ workspace, created: false }, options.requestId, lineageRecord.ownership) + } + } + if (options.forceNew || options.lineageId) { const ownership = this.createOwnership(options.requestId) const record = this.reserveWorkspace(workspacePath, identityKey, name, options, ownership, launchDeadlineAt) const result = await this.startCreation(record, options, launchDeadlineAt, launchTimeoutMs) @@ -319,6 +336,7 @@ export class WorkspaceManager { const record = { id, + lineageId: options.lineageId ?? randomUUID(), requestId: options.requestId, path: workspacePath, name, diff --git a/packages/ui/src/components/agent-selector.tsx b/packages/ui/src/components/agent-selector.tsx index be3616d35..5ede2af68 100644 --- a/packages/ui/src/components/agent-selector.tsx +++ b/packages/ui/src/components/agent-selector.tsx @@ -13,6 +13,8 @@ interface AgentSelectorProps { sessionId: string currentAgent: string onAgentChange: (agent: string) => Promise + allowChildAgents?: boolean + square?: boolean } export default function AgentSelector(props: AgentSelectorProps) { @@ -25,7 +27,7 @@ export default function AgentSelector(props: AgentSelectorProps) { }) const isChildSession = createMemo(() => { - return session()?.parentId !== null && session()?.parentId !== undefined + return props.allowChildAgents || (session()?.parentId !== null && session()?.parentId !== undefined) }) const availableAgents = createMemo(() => { @@ -105,7 +107,7 @@ export default function AgentSelector(props: AgentSelectorProps) { - + diff --git a/packages/ui/src/components/instance/shell/right-panel/RightPanel.tsx b/packages/ui/src/components/instance/shell/right-panel/RightPanel.tsx index 5f6865fb1..90ad6c359 100644 --- a/packages/ui/src/components/instance/shell/right-panel/RightPanel.tsx +++ b/packages/ui/src/components/instance/shell/right-panel/RightPanel.tsx @@ -4,6 +4,7 @@ import { createEffect, createMemo, createSignal, + createUniqueId, lazy, onCleanup, type Accessor, @@ -62,7 +63,9 @@ import { const LazyGitChangesTab = lazy(() => import("./tabs/GitChangesTab")) const LazyFilesTab = lazy(() => import("./tabs/FilesTab")) +const LazyWorkflowsTab = lazy(() => import("./tabs/WorkflowsTab")) const LazyStatusTab = lazy(() => import("./tabs/StatusTab")) +const RIGHT_PANEL_TABS: readonly RightPanelTab[] = ["workflows", "git-changes", "files", "status"] function RightPanelTabFallback() { return
@@ -98,6 +101,9 @@ interface RightPanelProps { const RightPanel: Component = (props) => { const [rightPanelTab, setRightPanelTab] = createSignal(readStoredRightPanelTab("git-changes")) + const [workflowsVisited, setWorkflowsVisited] = createSignal(rightPanelTab() === "workflows") + const tabGroupId = `right-panel-${createUniqueId()}` + const tabButtons: Partial> = {} const defaultStatusSectionIds = ["provider-usage", "yolo-mode", "plan", "background-processes", "mcp", "lsp", "plugins"] const [rightPanelExpandedItems, setRightPanelExpandedItems] = createSignal(defaultStatusSectionIds) @@ -213,6 +219,7 @@ const RightPanel: Component = (props) => { createEffect(() => { writeClientLayoutValue(RIGHT_PANEL_TAB_STORAGE_KEY, rightPanelTab()) + if (rightPanelTab() === "workflows") setWorkflowsVisited(true) }) createEffect(() => { @@ -651,6 +658,24 @@ const RightPanel: Component = (props) => { const tabClass = (tab: RightPanelTab) => `right-panel-tab ${rightPanelTab() === tab ? "right-panel-tab-active" : "right-panel-tab-inactive"}` + const tabId = (tab: RightPanelTab) => `${tabGroupId}-tab-${tab}` + const tabPanelId = (tab: RightPanelTab) => `${tabGroupId}-panel-${tab}` + const selectTabFromKeyboard = (tab: RightPanelTab) => { + setRightPanelTab(tab) + tabButtons[tab]?.focus() + } + const handleTabKeyDown = (event: KeyboardEvent, tab: RightPanelTab) => { + const index = RIGHT_PANEL_TABS.indexOf(tab) + let target: RightPanelTab | undefined + if (event.key === "ArrowLeft") target = RIGHT_PANEL_TABS[(index - 1 + RIGHT_PANEL_TABS.length) % RIGHT_PANEL_TABS.length] + if (event.key === "ArrowRight") target = RIGHT_PANEL_TABS[(index + 1) % RIGHT_PANEL_TABS.length] + if (event.key === "Home") target = RIGHT_PANEL_TABS[0] + if (event.key === "End") target = RIGHT_PANEL_TABS[RIGHT_PANEL_TABS.length - 1] + if (!target) return + event.preventDefault() + selectTabFromKeyboard(target) + } + return (
@@ -682,28 +707,57 @@ const RightPanel: Component = (props) => {
+
+ + + +
) diff --git a/packages/ui/src/components/instance/shell/right-panel/tabs/WorkflowsTab.tsx b/packages/ui/src/components/instance/shell/right-panel/tabs/WorkflowsTab.tsx new file mode 100644 index 000000000..089c9bd2a --- /dev/null +++ b/packages/ui/src/components/instance/shell/right-panel/tabs/WorkflowsTab.tsx @@ -0,0 +1,358 @@ +import { For, Show, createEffect, createMemo, createSignal, onMount, type Accessor, type Component } from "solid-js" +import { ArrowDown, ArrowUp, Plus, RefreshCw, Trash2 } from "lucide-solid" +import type { WorkflowRun, WorkflowRunStep } from "../../../../../../../server/src/api-types" +import AgentSelector from "../../../../agent-selector" +import ModelSelector from "../../../../model-selector" +import { Markdown } from "../../../../markdown" +import { showConfirmDialog } from "../../../../../stores/alerts" +import { hydrateRestoredSessionChain, setActiveSessionFromList } from "../../../../../stores/sessions" +import { + approveWorkflowRun, + cancelWorkflowRun, + createWorkflowRun, + getWorkflowRuns, + getWorkflowDraft, + loadWorkflowRuns, + setWorkflowDraft, + workflowErrors, + workflowLoading, + workflowStatusTransitions, + type WorkflowDraftStage, +} from "../../../../../stores/workflows" + +interface WorkflowsTabProps { + t: (key: string, vars?: Record) => string + instanceId: string + activeSessionId: Accessor + active: Accessor +} + +const MAX_STAGES = 12 +let nextStageId = 0 +const newStageId = () => `stage-${Date.now()}-${++nextStageId}` + +const WorkflowsTab: Component = (props) => { + const initialStages = (): WorkflowDraftStage[] => [ + { + id: newStageId(), + title: props.t("instanceShell.workflows.defaults.planner.title"), + instructions: props.t("instanceShell.workflows.defaults.planner.instructions"), + requiresApproval: true, + }, + { + id: newStageId(), + title: props.t("instanceShell.workflows.defaults.implementer.title"), + instructions: props.t("instanceShell.workflows.defaults.implementer.instructions"), + requiresApproval: false, + }, + ] + const savedDraft = getWorkflowDraft(props.instanceId) + const [objective, setObjective] = createSignal(savedDraft?.objective ?? "") + const [stages, setStages] = createSignal(savedDraft?.stages ?? initialStages()) + const [submitting, setSubmitting] = createSignal(false) + const [actionRunId, setActionRunId] = createSignal(null) + const [actionError, setActionError] = createSignal("") + const [statusAnnouncement, setStatusAnnouncement] = createSignal("") + const runs = createMemo(() => getWorkflowRuns(props.instanceId)) + const loading = createMemo(() => workflowLoading().get(props.instanceId) ?? false) + const loadError = createMemo(() => workflowErrors().get(props.instanceId) ?? "") + + createEffect(() => { + setWorkflowDraft(props.instanceId, { objective: objective(), stages: stages() }) + }) + + createEffect(() => { + if (!props.active()) return + const transition = workflowStatusTransitions().get(props.instanceId) + if (!transition) return + setStatusAnnouncement("") + queueMicrotask(() => { + if (props.active()) { + setStatusAnnouncement(props.t("instanceShell.workflows.live.statusChanged", { + status: props.t(`instanceShell.workflows.status.${transition.status}`), + })) + } + }) + }) + + onMount(() => void loadWorkflowRuns(props.instanceId)) + + const updateStage = (id: string, patch: Partial) => { + setStages((current) => current.map((stage) => (stage.id === id ? { ...stage, ...patch } : stage))) + } + + const moveStage = (index: number, offset: -1 | 1) => { + setStages((current) => { + const target = index + offset + if (target < 0 || target >= current.length) return current + const next = [...current] + ;[next[index], next[target]] = [next[target], next[index]] + return next + }) + } + + const addStage = () => { + if (stages().length >= MAX_STAGES) return + setStages((current) => [ + ...current, + { + id: newStageId(), + title: props.t("instanceShell.workflows.defaults.newStage.title", { number: current.length + 1 }), + instructions: "", + requiresApproval: false, + }, + ]) + } + + const submit = async (event: SubmitEvent) => { + event.preventDefault() + setSubmitting(true) + setActionError("") + try { + const sessionId = props.activeSessionId() + await createWorkflowRun(props.instanceId, { + objective: objective().trim(), + initiatorSessionId: sessionId && sessionId !== "info" ? sessionId : undefined, + stages: stages().map((stage) => ({ + ...stage, + title: stage.title.trim(), + instructions: stage.instructions.trim(), + agent: stage.agent?.trim() || undefined, + model: stage.model?.providerId && stage.model.modelId ? stage.model : undefined, + })), + }) + setObjective("") + } catch (error) { + setActionError(error instanceof Error ? error.message : props.t("instanceShell.workflows.errors.action")) + } finally { + setSubmitting(false) + } + } + + const approve = async (run: WorkflowRun) => { + const confirmed = await showConfirmDialog(props.t("instanceShell.workflows.confirm.approve.message"), { + confirmLabel: props.t("instanceShell.workflows.actions.approve"), + cancelLabel: props.t("instanceShell.workflows.actions.keepWaiting"), + dismissible: false, + }) + if (!confirmed) return + await runAction(run.id, () => approveWorkflowRun(props.instanceId, run.id)) + } + + const cancel = async (run: WorkflowRun) => { + const confirmed = await showConfirmDialog(props.t("instanceShell.workflows.confirm.cancel.message"), { + variant: "warning", + confirmLabel: props.t("instanceShell.workflows.actions.cancel"), + cancelLabel: props.t("instanceShell.workflows.actions.keepRunning"), + dismissible: false, + }) + if (!confirmed) return + await runAction(run.id, () => cancelWorkflowRun(props.instanceId, run.id)) + } + + const runAction = async (runId: string, action: () => Promise) => { + setActionRunId(runId) + setActionError("") + try { + await action() + } catch (error) { + setActionError(error instanceof Error ? error.message : props.t("instanceShell.workflows.errors.action")) + } finally { + setActionRunId(null) + } + } + + const openSession = async (sessionId: string) => { + setActionError("") + try { + await hydrateRestoredSessionChain(props.instanceId, [sessionId]) + setActiveSessionFromList(props.instanceId, sessionId) + } catch { + setActionError(props.t("instanceShell.workflows.errors.openSession")) + } + } + + const formatTime = (value: string) => new Intl.DateTimeFormat(undefined, { dateStyle: "medium", timeStyle: "short" }).format(new Date(value)) + const isCancellable = (run: WorkflowRun) => run.status === "running" || run.status === "waiting_for_review" + const statusLabel = (status: WorkflowRun["status"] | WorkflowRunStep["status"]) => + props.t(`instanceShell.workflows.status.${status}`) + + const renderOutput = (step: WorkflowRunStep) => ( + +
+ +
{props.t("instanceShell.workflows.output")}
+ {JSON.stringify(step.output, null, 2)}} + > + + + + +

{props.t("instanceShell.workflows.outputTruncated")}

+ +
+ + ) + + return ( +
+
{statusAnnouncement()}
+
+
+
+

{props.t("instanceShell.workflows.builder.title")}

+

{props.t("instanceShell.workflows.builder.description")}

+
+
+
+