diff --git a/apps/hub/src/app.ts b/apps/hub/src/app.ts index 849700c9..27de5b1a 100644 --- a/apps/hub/src/app.ts +++ b/apps/hub/src/app.ts @@ -504,6 +504,7 @@ export async function createHubRuntime( ...(coordinatorUpgrade.enabled ? { upgrade: coordinatorUpgrades.open } : {}), + update: coordinatorUpgrades.update, }, async (actor, projectId) => { const bound = await coordinatorStartup.release(actor, projectId); diff --git a/apps/hub/src/projects/coordinator-upgrade.test.ts b/apps/hub/src/projects/coordinator-upgrade.test.ts new file mode 100644 index 00000000..41ed89ec --- /dev/null +++ b/apps/hub/src/projects/coordinator-upgrade.test.ts @@ -0,0 +1,64 @@ +import { expect, test } from "bun:test"; +import type { InstalledWorkflows } from "@corbits-builder/schemas/installed-workflows"; +import { staleRuns } from "./coordinator-upgrade.ts"; + +const deployment = (pin: string) => ({ + source: { + kind: "asset" as const, + assetId: "asset", + package: { format: "tarball" as const }, + }, + entry: "dist/workflow.mjs", + pin, + sourceOfferingIds: ["offering"], + defaultSourceOfferingId: "offering", +}); +const installed: InstalledWorkflows = { + schemaVersion: 1, + coordinator: deployment("@corbits-builder/coordinator-native@1.0.0-sha-new"), + writer: deployment("@corbits-builder/writer-native@1.0.0-sha-new"), + reviewer: deployment("@corbits-builder/reviewer-native@1.0.0-sha-new"), + coder: deployment("@corbits-builder/coder-native@1.0.0-sha-new"), + coderAnyPackage: deployment("@corbits-builder/coder-native@1.0.0-sha-any"), + designer: deployment("@corbits-builder/designer-native@1.0.0-sha-new"), + presenter: deployment("@corbits-builder/presenter-native@1.0.0-sha-new"), +}; +const frozen = (name: string, version: string) => ({ + closure: { topLevel: [{ name, version }] }, +}); + +test("a run is stale when its frozen pin is not one installed for its role", () => { + const runs = [ + { + runId: "old-coder", + role: "coder" as const, + bundle: frozen("@corbits-builder/coder-native", "1.0.0-sha-old"), + }, + { + runId: "coder", + role: "coder" as const, + bundle: frozen("@corbits-builder/coder-native", "1.0.0-sha-new"), + }, + { + runId: "any-coder", + role: "coder" as const, + bundle: frozen("@corbits-builder/coder-native", "1.0.0-sha-any"), + }, + { + runId: "designer", + role: "designer" as const, + bundle: frozen("@corbits-builder/designer-native", "1.0.0-sha-new"), + }, + { + runId: "old-designer", + role: "designer" as const, + bundle: frozen("@corbits-builder/designer-native", "1.0.0-sha-old"), + }, + { runId: "unreadable", role: "writer" as const, bundle: {} }, + ]; + expect(staleRuns(runs, installed).map((run) => run.runId)).toEqual([ + "old-coder", + "old-designer", + "unreadable", + ]); +}); diff --git a/apps/hub/src/projects/coordinator-upgrade.ts b/apps/hub/src/projects/coordinator-upgrade.ts index d631c195..f2dc0314 100644 --- a/apps/hub/src/projects/coordinator-upgrade.ts +++ b/apps/hub/src/projects/coordinator-upgrade.ts @@ -13,7 +13,11 @@ import { type WorkflowRunReader, } from "@intx/hub-sessions"; import { getLogger } from "@intx/log"; -import type { InstalledWorkflows } from "@corbits-builder/schemas/installed-workflows"; +import type { + InstalledWorkflows, + WorkflowRole, +} from "@corbits-builder/schemas/installed-workflows"; +import type { ProjectUpdateReceipt } from "@corbits-builder/schemas/project"; import { spawnedWorkflowRole } from "../workflows/spawn-role.ts"; import { MODEL_CHOICE_REFUSALS } from "../config/models.ts"; import { project, projectCoordinator } from "../db/schema.ts"; @@ -24,6 +28,11 @@ import { ProjectError } from "./errors.ts"; const log = getLogger(["corbits-builder", "hub", "coordinator-upgrade"]); export const COORDINATOR_UPGRADED = "coordinator_upgraded"; +// The person asked for the project's runs to move onto the current tool code. +export const PROJECT_UPDATED = "project_updated"; +// A specialist replaced on its own, read by the coordinator that keeps running. +export const SPECIALIST_REPLACED = + "project_updated: Builder was updated and this run was replaced; start it again to continue its work on the current version"; // The packaged definition a run was frozen with; its version carries the bundle's content hash. const FrozenDefinition = type({ @@ -41,6 +50,23 @@ const pinOf = (bundle: unknown) => { return top && top.name + "@" + top.version; }; +// The pins a role may run on: a coder is either coder package, by the project's package choice. +const pinsFor = (installed: InstalledWorkflows, role: WorkflowRole) => + role === "coder" + ? [installed.coder.pin, installed.coderAnyPackage.pin] + : [installed[role].pin]; + +/** The runs whose frozen package is not the one installed for their role, or has no readable pin. */ +export function staleRuns( + runs: readonly T[], + installed: InstalledWorkflows, +): T[] { + return runs.filter((run) => { + const pin = pinOf(run.bundle); + return pin === undefined || !pinsFor(installed, run.role).includes(pin); + }); +} + /** * Moves each project's coordinator onto the current coordinator definition at a safe point: parked on * the person's input with no unsettled inbound delivery, no pending approval and no specialist at work. @@ -78,7 +104,8 @@ export function createCoordinatorUpgrades(deps: { // The specialists to close with the coordinator, or undefined while one is still at work. A designer // or writer parked on input has filed its work and only waits for follow-up mail; a coder's // workspace holds unfiled progress, so any coder holds the upgrade back. - async function settledSpecialists(tenantId: string) { + // The tenant's live specialist runs with the role each was spawned as. + async function liveSpecialists(tenantId: string) { const live = await db .select({ runId: workflowRun.id, @@ -96,11 +123,16 @@ export function createCoordinatorUpgrades(deps: { inArray(workflowRun.status, ["deployed", "running"]), ), ); - const parked: string[] = []; - for (const run of live) { + return live.flatMap((run) => { const role = spawnedWorkflowRole(run.bundle); - if (!role) continue; - if (role === "coder" || !run.address) return undefined; + return role ? [{ ...run, role }] : []; + }); + } + + async function settledSpecialists(tenantId: string) { + const parked: string[] = []; + for (const run of await liveSpecialists(tenantId)) { + if (run.role === "coder" || !run.address) return undefined; if (!(await parkedOnInput({ runId: run.runId, address: run.address }))) return undefined; parked.push(run.runId); @@ -108,11 +140,9 @@ export function createCoordinatorUpgrades(deps: { return parked; } - async function atSafePoint( - actor: Actor, - run: { runId: string; address: string }, - ) { - if (!(await parkedOnInput(run))) return undefined; + // Parked on the person's input with every delivery to it settled. + async function coordinatorIdle(run: { runId: string; address: string }) { + if (!(await parkedOnInput(run))) return false; const [delivery] = await db .select({ id: workflowRunDispatch.id }) .from(workflowRunDispatch) @@ -123,18 +153,26 @@ export function createCoordinatorUpgrades(deps: { ), ) .limit(1); - if (delivery) return undefined; + return delivery === undefined; + } + + async function approvalPending(tenantId: string) { const [asked] = await db .select({ id: approval.id }) .from(approval) .where( - and( - eq(approval.tenantId, actor.tenantId), - eq(approval.status, "pending"), - ), + and(eq(approval.tenantId, tenantId), eq(approval.status, "pending")), ) .limit(1); - if (asked) return undefined; + return asked !== undefined; + } + + async function atSafePoint( + actor: Actor, + run: { runId: string; address: string }, + ) { + if (!(await coordinatorIdle(run))) return undefined; + if (await approvalPending(actor.tenantId)) return undefined; return settledSpecialists(actor.tenantId); } @@ -153,7 +191,8 @@ export function createCoordinatorUpgrades(deps: { log.info("Coordinator upgraded", { projectId, runId: bound.runId }); } - async function upgrade(projectId: string, tenantId: string, opener?: Actor) { + // The project's active coordinator, or the latest one when none is active. + async function latestCoordinator(projectId: string, tenantId: string) { const [latest] = await db .select({ runId: workflowRun.id, @@ -182,12 +221,20 @@ export function createCoordinatorUpgrades(deps: { desc(projectCoordinator.createdAt), ) .limit(1); + return latest; + } + + // A replacement that cancelled the old run but did not get its successor bound finishes with the + // next pass or open. + const replaced = (reason: string | null) => + reason === COORDINATOR_UPGRADED || reason === PROJECT_UPDATED; + + async function upgrade(projectId: string, tenantId: string, opener?: Actor) { + const latest = await latestCoordinator(projectId, tenantId); if (!latest) return; const actor = opener ?? { tenantId, principalId: latest.principalId }; - // An upgrade that cancelled the old run but did not get its successor bound finishes here. if (!latest.active) { - if (latest.cancellationReason === COORDINATOR_UPGRADED) - await start(actor, projectId); + if (replaced(latest.cancellationReason)) await start(actor, projectId); return; } if ( @@ -210,12 +257,76 @@ export function createCoordinatorUpgrades(deps: { await start(actor, projectId); } + /** + * Moves the project's runs onto the current tool code when the person asks: the specialists whose + * frozen package is behind the installed one are cancelled, and the coordinator with every parked + * specialist when it is the one behind, so a fresh coordinator starts as an upgrade does. Unlike + * the pass, a parked coder is replaced too; its unfiled work is what the person gives up. A run + * mid-turn is left alone and named, so nothing is cut off. + */ + async function update( + actor: Actor, + projectId: string, + ): Promise { + const latest = await latestCoordinator(projectId, actor.tenantId); + if (!latest) return { updated: false, reason: "no_coordinator" }; + if (!latest.active) { + if (!replaced(latest.cancellationReason)) + return { updated: false, reason: "no_coordinator" }; + await start(actor, projectId); + return { updated: true }; + } + if ( + !latest.address || + latest.cancellationRequestedAt !== null || + !["deployed", "running"].includes(latest.status) + ) + return { updated: false, reason: "no_coordinator" }; + const installed = await deps.installed(actor, projectId); + const specialists = await liveSpecialists(actor.tenantId); + const coordinatorStale = + staleRuns([{ role: "coordinator", bundle: latest.bundle }], installed) + .length > 0; + const stale = staleRuns(specialists, installed); + if (!coordinatorStale && stale.length === 0) + return { updated: false, reason: "current" }; + const coordinator = { runId: latest.runId, address: latest.address }; + if (!(await coordinatorIdle(coordinator))) + return { updated: false, reason: "coordinator_busy" }; + if (await approvalPending(actor.tenantId)) + return { updated: false, reason: "approval_pending" }; + for (const run of specialists) + if ( + !run.address || + !(await parkedOnInput({ runId: run.runId, address: run.address })) + ) + return { updated: false, reason: "specialist_working" }; + if (!coordinatorStale) { + for (const run of stale) + await deps.cancel(actor.tenantId, run.runId, SPECIALIST_REPLACED); + log.info("Project updated: specialists replaced", { + projectId, + runIds: stale.map((run) => run.runId), + }); + return { updated: true }; + } + await deps.startup.release(actor, projectId); + for (const run of specialists) + await deps.cancel(actor.tenantId, run.runId, PROJECT_UPDATED); + await deps.cancel(actor.tenantId, latest.runId, PROJECT_UPDATED); + await start(actor, projectId); + return { updated: true }; + } + // One upgrade per project at a time, so a pass and an open never both cancel and start. An open // runs as its opener, so it waits out a pass rather than sharing the pass's starter. - const inflight = new Map }>(); + const inflight = new Map< + string, + { opened: boolean; work: Promise } + >(); function upgradeOnce(projectId: string, tenantId: string, opener?: Actor) { const pending = inflight.get(projectId); - if (pending && (pending.opened || !opener)) return pending.work; + if (pending && (pending.opened || !opener)) return settle(pending.work); async function upgradeAfterPending(): Promise { try { await settle(pending?.work); @@ -256,9 +367,26 @@ export function createCoordinatorUpgrades(deps: { return false; } + // An update waits out a pass or an open on the project, and holds the next one back. + function updateOnce(actor: Actor, projectId: string) { + const pending = inflight.get(projectId); + async function updateAfterPending(): Promise { + try { + await settle(pending?.work); + return await update(actor, projectId); + } finally { + if (inflight.get(projectId)?.work === work) inflight.delete(projectId); + } + } + const work = updateAfterPending(); + inflight.set(projectId, { opened: true, work }); + return work; + } + return { reconcile, open: (actor: Actor, projectId: string) => upgradeOnce(projectId, actor.tenantId, actor), + update: updateOnce, }; } diff --git a/apps/hub/src/projects/routes.ts b/apps/hub/src/projects/routes.ts index 4c2793e8..6144af92 100644 --- a/apps/hub/src/projects/routes.ts +++ b/apps/hub/src/projects/routes.ts @@ -24,6 +24,7 @@ import { ProjectMutation, ProjectMutationReceipt, ProjectShipReceipt, + ProjectUpdateReceipt, ProjectView, ProjectSummaryList, } from "@corbits-builder/schemas/project"; @@ -53,6 +54,8 @@ type CoordinatorStart = { ensure: (actor: Actor, projectId: string) => Promise; attached: (actor: Actor, projectId: string) => Promise; upgrade?: (actor: Actor, projectId: string) => Promise; + /** Moves the project's runs onto the current tool code when the person asks. */ + update?: (actor: Actor, projectId: string) => Promise; }; // The coordinator takes a turn only when mail arrives, so a new project's first mail is the @@ -635,6 +638,24 @@ export function createProjectRoutes( return receipt; }, ); + projectAction( + "/:projectId/update", + "updateProject", + { + description: + "Whether the project's runs behind the installed tool code were replaced, and when not, why: every run is current, no coordinator is live, the coordinator or a specialist is mid-turn, or an approval is pending", + schema: ProjectUpdateReceipt, + }, + "Project is archived", + async (actor, id) => { + const row = await service.manage(actor, id); + if (row.status !== "active") + throw new ProjectError(409, "project_archived"); + if (!startCoordinator?.update) + throw new ProjectError(503, "project_update_unavailable"); + return startCoordinator.update(actor, id); + }, + ); const ship = { description: "The person's shipped mark, set or cleared; repeating it is idempotent. The project stays active, so its documents, app and coordinator stay usable", diff --git a/apps/hub/src/projects/service.ts b/apps/hub/src/projects/service.ts index 6833ef2c..56540c9d 100644 --- a/apps/hub/src/projects/service.ts +++ b/apps/hub/src/projects/service.ts @@ -332,6 +332,15 @@ export function createProjectService(db: DB["db"]) { } return ProjectSummaryList.assert({ projects }); }, + /** The project, for an action on its runs that needs manage authority; read without a lock. */ + async manage(actor: Actor, rawId: unknown) { + const id = ProjectId(rawId); + if (id instanceof type.errors) + throw new ProjectError(400, "invalid_project_id"); + const allowed = await authorizationFor(actor); + await requireActor(db, actor, id, "manage", allowed); + return load(db, actor, id); + }, async lifecycle( actor: Actor, rawId: unknown, diff --git a/apps/web/src/api/projects.ts b/apps/web/src/api/projects.ts index 4ab054ad..b2033900 100644 --- a/apps/web/src/api/projects.ts +++ b/apps/web/src/api/projects.ts @@ -17,6 +17,7 @@ import { exportSection, ProjectExport, type ProjectSummary as ProjectSummaryRow, + type ProjectUpdateReceipt, } from "@corbits-builder/schemas/project"; import { readProjectImport, @@ -149,6 +150,14 @@ export async function restoreProject(projectId: string): Promise { await productClient().restoreProject(tenantId, projectId); } +/** Moves the project's runs onto the current tool code, or says why that waits. */ +export async function updateProject( + projectId: string, +): Promise { + const tenantId = await workspaceTenantId(); + return productClient().updateProject(tenantId, projectId); +} + /** Sets or clears the person's shipped mark; the project stays active either way. */ export async function shipProject( projectId: string, diff --git a/apps/web/src/config.ts b/apps/web/src/config.ts index f55bb9d8..a6fc5eeb 100644 --- a/apps/web/src/config.ts +++ b/apps/web/src/config.ts @@ -246,6 +246,27 @@ export const wording = { }, action: "Export", }, + updateProject: { + menu: "Update…", + title: "Update this project", + lead: `Moves the project onto the current ${productName}: the assistants running on older tool code are replaced with fresh ones. The documents, the design and the filed app stay. Work the app builder has not filed yet is lost, so wait for it to finish first.`, + after: + "Afterwards, ask the coordinator to carry on; it picks up where the project left off.", + action: "Update", + done: "The project is updated. Ask the coordinator to carry on.", + reason: { + current: `Every assistant on this project already runs the current ${productName}.`, + no_coordinator: + "This project has no coordinator running to update. Open the project and send it a message first.", + coordinator_busy: + "The coordinator is mid-turn. Wait for its reply, then try again.", + approval_pending: + "A decision is waiting on you. Answer it, then try again.", + specialist_working: + "An assistant is mid-task. Wait for it to report, then try again.", + } as Record, + unexplained: "The project was not updated.", + }, importProject: { action: "Import project", source: `A project ${productName} or the original Solution Builder exported`, diff --git a/apps/web/src/pages/project-menu.test.ts b/apps/web/src/pages/project-menu.test.ts new file mode 100644 index 00000000..4624182d --- /dev/null +++ b/apps/web/src/pages/project-menu.test.ts @@ -0,0 +1,18 @@ +import { expect, test } from "bun:test"; +import { updateOutcome } from "./project-menu.tsx"; + +test("an update that replaced runs reads as done, and one that waits says why", () => { + expect(updateOutcome({ updated: true })).toEqual([ + "The project is updated. Ask the coordinator to carry on.", + true, + ]); + const [why, complete] = updateOutcome({ + updated: false, + reason: "specialist_working", + }); + expect(why).toMatch(/mid-task/); + expect(complete).toBe(false); + expect(updateOutcome({ updated: false })[0]).toBe( + "The project was not updated.", + ); +}); diff --git a/apps/web/src/pages/project-menu.tsx b/apps/web/src/pages/project-menu.tsx index 8b8a14ff..c3d09667 100644 --- a/apps/web/src/pages/project-menu.tsx +++ b/apps/web/src/pages/project-menu.tsx @@ -1,8 +1,8 @@ /** * A project's options menu, the same wherever a project is named: the * rail row and the title in the top bar. Info (and rename), - * export, the shipped mark, archive, and a two-step delete, matching the - * project operations the hub offers. The dialogs it opens live here too, + * export, the shipped mark, update, archive, and a two-step delete, matching + * the project operations the hub offers. The dialogs it opens live here too, * so the menu is complete on its own. */ import { wording } from "../config.ts"; @@ -10,6 +10,7 @@ import { Checkbox, Input } from "@corbits/react-ui"; import { EXPORT_SECTIONS, type ExportSection, + type ProjectUpdateReceipt, } from "@corbits-builder/schemas/project"; import { Dialog, @@ -34,6 +35,7 @@ import { renameProject, restoreProject, shipProject, + updateProject, } from "../api/projects.ts"; import { Button, type NoticeAction } from "../components.jsx"; import { Dictated } from "../dictation.jsx"; @@ -99,6 +101,8 @@ export function ProjectMenu({ const [confirming, setConfirming] = useState(false); // Export asks which parts to put in the zip. const [exporting, setExporting] = useState(false); + // Update says what it gives up before it replaces anything. + const [updating, setUpdating] = useState(false); const act = async (work: () => Promise) => { try { @@ -109,6 +113,18 @@ export function ProjectMenu({ } }; + // The dialog closes either way; what the hub answered, or why it failed, is the notice. + const runUpdate = async () => { + try { + await act(async () => { + const receipt = await updateProject(project.id); + onNotice?.(...updateOutcome(receipt)); + }); + } finally { + setUpdating(false); + } + }; + const openInfo = (withName = false) => { setFocusName(withName); setInfoOpen(true); @@ -138,6 +154,11 @@ export function ProjectMenu({ : wording.action.markShipped} )} + {project.archivedAt ? null : ( + setUpdating(true)}> + {wording.updateProject.menu} + + )} @@ -197,6 +218,20 @@ export function ProjectMenu({ /> ) : null} + {updating ? ( +
event.stopPropagation()} + onPointerDown={(event) => event.stopPropagation()} + onContextMenu={(event) => event.stopPropagation()} + onKeyDown={(event) => event.stopPropagation()} + > + setUpdating(false)} + onUpdate={() => void runUpdate()} + /> +
+ ) : null} {exporting ? (
void; + onUpdate: () => void; +}) { + const text = wording.updateProject; + return ( + (open ? undefined : onClose())}> + + + {text.title} + + +

{text.lead}

+

{text.after}

+
+ + + + +
+
+ ); +} + /** Offers the sections the project has; a failed check offers the planning documents and says why. */ function ExportDialog({ projectId, diff --git a/packages/client/src/client/projects.ts b/packages/client/src/client/projects.ts index bbd822cd..64a38182 100644 --- a/packages/client/src/client/projects.ts +++ b/packages/client/src/client/projects.ts @@ -31,6 +31,7 @@ export function buildProjectsMethods( | "archiveProject" | "restoreProject" | "shipProject" + | "updateProject" | "startProjectCoordinator" | "projectNextStep" | "projectRunStates" @@ -64,6 +65,9 @@ export function buildProjectsMethods( restoreProject(tenantId, projectId, signal) { return request(`${projectPath(tenantId, projectId)}/restore`, signal, {}); }, + updateProject(tenantId, projectId, signal) { + return request(`${projectPath(tenantId, projectId)}/update`, signal, {}); + }, shipProject(tenantId, projectId, shipped, signal) { return request( `${projectPath(tenantId, projectId)}/${shipped ? "ship" : "unship"}`, diff --git a/packages/client/src/client/shared.ts b/packages/client/src/client/shared.ts index 747b8b3d..8179414c 100644 --- a/packages/client/src/client/shared.ts +++ b/packages/client/src/client/shared.ts @@ -34,6 +34,7 @@ import type { ProjectSummaryList, ProjectLifecycleReceipt, ProjectShipReceipt, + ProjectUpdateReceipt, ProjectView, } from "@corbits-builder/schemas/project"; import type { InstalledWorkflows } from "@corbits-builder/schemas/installed-workflows"; @@ -346,6 +347,12 @@ export type ProductApiClient = Readonly<{ projectId: string, signal?: AbortSignal, ): Promise; + /** Moves the project's runs onto the current tool code, or says why not. */ + updateProject( + tenantId: string, + projectId: string, + signal?: AbortSignal, + ): Promise; /** Marks the project shipped, or clears the mark; either way it stays active and usable. */ shipProject( tenantId: string, diff --git a/packages/schemas/src/project.ts b/packages/schemas/src/project.ts index 30bbda20..a1844337 100644 --- a/packages/schemas/src/project.ts +++ b/packages/schemas/src/project.ts @@ -87,6 +87,18 @@ export const ProjectLifecycleReceipt = type({ status: ProjectLifecycleStatus, }); export type ProjectLifecycleReceipt = typeof ProjectLifecycleReceipt.infer; +/** + * What Update did to a project's runs: replaced the ones behind the installed tool code, or nothing, + * and why not: every run is current, there is no live coordinator, the coordinator is mid-turn or + * has an unsettled delivery, an approval is pending, or a specialist is mid-turn. + */ +export const ProjectUpdateReceipt = type({ + "+": "reject", + updated: "boolean", + "reason?": + "'current' | 'no_coordinator' | 'coordinator_busy' | 'approval_pending' | 'specialist_working'", +}); +export type ProjectUpdateReceipt = typeof ProjectUpdateReceipt.infer; export const ProjectShipReceipt = type({ "+": "reject", ok: "true",