feat: offer Update in the project menu to move its runs onto the current tool code

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);

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",
+  ]);
+});

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<T extends { role: WorkflowRole; bundle: unknown }>(
+  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,23 +123,26 @@ 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);
     }
     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<ProjectUpdateReceipt> {
+    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<string, { opened: boolean; work: Promise<void> }>();
+  const inflight = new Map<
+    string,
+    { opened: boolean; work: Promise<unknown> }
+  >();
   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<void> {
       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<ProjectUpdateReceipt> {
+      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,
   };
 }

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<boolean>;
   attached: (actor: Actor, projectId: string) => Promise<boolean>;
   upgrade?: (actor: Actor, projectId: string) => Promise<void>;
+  /** Moves the project's runs onto the current tool code when the person asks. */
+  update?: (actor: Actor, projectId: string) => Promise<ProjectUpdateReceipt>;
 };
 
 // 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",

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,