fix(models): surface refused credentials with recovery and handoff

apps/hub/src/app.ts

@@ -85,7 +85,10 @@ import {
   correlationMiddleware,
   requestContext,
 } from "./middleware/correlation.ts";
-import { createAvailabilityRoutes } from "./availability.ts";
+import {
+  createAvailabilityProbe,
+  createAvailabilityRoutes,
+} from "./availability.ts";
 import { createGitTokenBoundary } from "./git-token-boundary.ts";
 import { createSidecarWebSocket } from "./sidecar-websocket.ts";
 import { createCoordinatorCredentialRefresh } from "./auth/coordinator-credential-refresh.ts";
@@ -318,6 +321,10 @@ export async function createHubRuntime(
       hubId: hexEncode(key.publicKey).slice(0, 24),
     });
     const allocation = workflows.services.workflowAllocationService;
+    const availabilityProbe = createAvailabilityProbe(
+      handle.db,
+      credentialCipher,
+    );
     const coordinatorStartup = createInstalledCoordinatorStartup(
       handle.db,
       assetService,
@@ -355,13 +362,14 @@ export async function createHubRuntime(
       },
       coordinatorStartTimeoutMs,
       workflows.services.workflowLifecycleService.requestCancellation,
+      availabilityProbe,
     );
     const coordinatorUpgrades = createCoordinatorUpgrades({
       db: handle.db,
       runs: createWorkflowRunReader(agentRepoStore.repoStore),
       installed: (actor, projectId) =>
         installForTenant(
-          { db: handle.db, assets: assetService },
+          { db: handle.db, assets: assetService, probe: availabilityProbe },
           actor,
           projectId,
         ),
@@ -525,6 +533,7 @@ export async function createHubRuntime(
         liveApps,
         workflows.services.workflowLifecycleService.requestCancellation,
         agentRepoStore.repoStore,
+        availabilityProbe,
       ),
     );
     app.route(

apps/hub/src/artifacts/activity.ts

@@ -108,12 +108,22 @@ export function createAgentActivity(events: Pick<SidecarEventEmitter, "on">) {
         delete run.said;
         return;
       }
-      case "message.run.ended": {
+      case "message.run.ended":
+      case "reactor.done":
+      case "reactor.error": {
         const run = entry(local);
         run.working = false;
         run.since = now;
         return;
       }
+      case "inference.error": {
+        if (event.data.error.category === "credential_failure") {
+          const run = entry(local);
+          run.working = false;
+          run.since = now;
+        }
+        return;
+      }
       case "tool.start": {
         const run = entry(local);
         // A turn whose start the hub missed, as after its own restart, dates from this call.

apps/hub/src/auth/coordinator-credential-refresh.ts

@@ -1,6 +1,9 @@
-import { and, eq } from "drizzle-orm";
-import type { DB } from "@intx/db";
-import { workflowRunExecutability } from "@intx/db";
+import { and, inArray } from "drizzle-orm";
+import {
+  getDescendantTenants,
+  workflowRunExecutability,
+  type DB,
+} from "@intx/db";
 import { workflowRun as runTable } from "@intx/db/schema";
 import {
   pushCredentialReconcile,
@@ -11,11 +14,10 @@ import type { CredentialCipher } from "@intx/types";
 import type { MiddlewareHandler } from "hono";
 import type { TenantEnv } from "@intx/hub-api";
 import { getLogger } from "@intx/log";
-import { projectCoordinator } from "../db/schema.ts";
 
 const log = getLogger(["corbits-builder", "coordinator-credentials"]);
 
-// After a credential's secret changes, pushes it to every active coordinator run that uses it.
+// After a credential's secret changes, pushes it to every active project run that uses it.
 export function createCoordinatorCredentialRefresh(options: {
   db: DB["db"];
   router: SidecarRouter;
@@ -29,19 +31,20 @@ export function createCoordinatorCredentialRefresh(options: {
       const body = await c.req.json();
       if (body.secret === undefined || body.status === "revoked") return;
       const credentialId = c.req.param("credentialId");
-      if (!credentialId) return;
-      const bindings = await options.db
+      const tenantId = c.req.param("tenantId");
+      if (!credentialId || !tenantId) return;
+      const tenants = await getDescendantTenants(options.db, tenantId);
+      const runs = await options.db
         .select({ run: runTable })
-        .from(projectCoordinator)
-        .innerJoin(runTable, eq(runTable.id, projectCoordinator.runId))
+        .from(runTable)
         .where(
           and(
-            eq(projectCoordinator.active, true),
-            eq(runTable.tenantId, c.req.param("tenantId")!),
+            inArray(runTable.tenantId, tenants.length ? tenants : [tenantId]),
+            inArray(runTable.status, ["deployed", "running"]),
           ),
         );
       await Promise.all(
-        bindings
+        runs
           .filter(
             ({ run }) =>
               run.address &&

apps/hub/src/config/models.test.ts

@@ -52,6 +52,11 @@ test("a project runs on the only connected provider's profile, a chosen one, or
   ).toEqual({
     refused: "model_profile_required",
   });
+  expect(
+    resolveModelChoice(MODEL_PROVIDERS, { profile: "anthropic_default" }, []),
+  ).toEqual({
+    refused: "model_provider_missing",
+  });
   expect(
     frozenModelChoice(MODEL_PROVIDERS, { profile: "openai_default" }, [
       "anthropic",

apps/hub/src/config/models.ts

@@ -467,9 +467,9 @@ export const MODEL_CHOICE_REFUSALS: Readonly<
 };
 
 /**
- * The profile a project runs on: the person's choice when this server can run it, else the shipped
- * profile of the only connected provider. With none connected, or several and no runnable choice,
- * there is no profile.
+ * The profile a project runs on: the person's choice when this server can run it on a connected
+ * provider, else the shipped profile of the only connected provider. With none connected, or
+ * several and no runnable choice, there is no profile.
  */
 export function resolveModelChoice(
   directory: ModelDirectory,

apps/hub/src/projects/coordinator-startup.ts

@@ -12,6 +12,7 @@ import {
 import type { InstalledWorkflows } from "@corbits-builder/schemas/installed-workflows";
 import type { NativeSpecialistDeployment } from "@corbits-builder/schemas/native-specialist";
 import { createCoordinatorRegistry } from "../auth/coordinator-registry.ts";
+import type { AvailabilityState } from "../availability.ts";
 import { COORDINATOR_CAPACITY_FULL } from "../config/coordinator-start.ts";
 import {
   installForTenant,
@@ -265,11 +266,12 @@ export function createInstalledCoordinatorStartup(
   }) => Promise<{ runId: string }>,
   readyWaitMs: number,
   cancel: CancelRun,
+  probe?: (tenantId: string, providerId: string) => Promise<AvailabilityState>,
 ) {
   const registry = createCoordinatorRegistry(db);
   return createProjectCoordinatorStartup({
     installed(actor, projectId) {
-      return installForTenant({ db, assets }, actor, projectId);
+      return installForTenant({ db, assets, probe }, actor, projectId);
     },
     async boundRun(tenantId, projectId) {
       const [row] = await db

apps/hub/src/projects/coordinator-upgrade.ts

@@ -3,6 +3,8 @@ import { type } from "arktype";
 import type { DB } from "@intx/db";
 import {
   approval,
+  modelOffering,
+  modelProvider,
   workflowRun,
   workflowRunDispatch,
   workflowRunLaunchSpec,
@@ -162,6 +164,7 @@ export function createCoordinatorUpgrades(deps: {
         status: workflowRun.status,
         cancellationRequestedAt: workflowRun.cancellationRequestedAt,
         cancellationReason: workflowRun.cancellationReason,
+        credentialRefs: workflowRun.credentialRefs,
         principalId: workflowRunLaunchSpec.sourceAuthorityPrincipalId,
         bundle: workflowRunLaunchSpec.frozenApprovalBundle,
       })
@@ -196,8 +199,18 @@ export function createCoordinatorUpgrades(deps: {
       !["deployed", "running"].includes(latest.status)
     )
       return;
-    const current = (await deps.installed(actor, projectId)).coordinator.pin;
-    if (pinOf(latest.bundle) === current) return;
+    const current = (await deps.installed(actor, projectId)).coordinator;
+    const [offering] = await db
+      .select({ credentialId: modelProvider.credentialId })
+      .from(modelOffering)
+      .innerJoin(modelProvider, eq(modelProvider.id, modelOffering.providerId))
+      .where(eq(modelOffering.id, current.defaultSourceOfferingId));
+    const sameCredential =
+      !offering?.credentialId ||
+      Boolean(
+        latest.credentialRefs?.credentialIds.includes(offering.credentialId),
+      );
+    if (pinOf(latest.bundle) === current.pin && sameCredential) return;
     const parked = await atSafePoint(actor, {
       runId: latest.runId,
       address: latest.address,

apps/hub/src/projects/model-choice.ts

@@ -2,9 +2,11 @@ import { and, eq } from "drizzle-orm";
 import { listVisibleProviders, type DB } from "@intx/db";
 import { principal, tenant } from "@intx/db/schema";
 import type { ModelChoice } from "@corbits-builder/schemas/models";
+import type { AvailabilityState } from "../availability.ts";
 import {
   MODEL_CHOICE_REFUSALS,
   frozenModelChoice,
+  pickedModelChoice,
   providerOfRow,
   resolveModelChoice,
   type ModelDirectory,
@@ -19,16 +21,28 @@ import { userPrincipalIn } from "./authorization.ts";
 import { ProjectError } from "./errors.ts";
 import type { Actor } from "../lib/actor.ts";
 
-/** The provider entries a tenant can run on: each enabled connection with a credential. */
+/** The provider entries a tenant can run on: each enabled connection with a non-refused credential. */
 export async function connectedProviders(
   db: DB["db"],
   directory: ModelDirectory,
   tenantId: string,
+  probe?: (tenantId: string, providerId: string) => Promise<AvailabilityState>,
 ): Promise<string[]> {
-  const ids = (await listVisibleProviders(db, tenantId))
-    .filter(({ row }) => row.credentialId !== null && !row.disabled)
-    .map(({ row }) => providerOfRow(directory, row)?.id)
-    .filter((id) => id !== undefined);
+  const visible = (await listVisibleProviders(db, tenantId)).filter(
+    ({ row }) => row.credentialId !== null && !row.disabled,
+  );
+  const ids: string[] = [];
+  for (const { row } of visible) {
+    const entry = providerOfRow(directory, row);
+    if (!entry) continue;
+    if (
+      probe &&
+      entry.kind !== "custom" &&
+      (await probe(tenantId, row.id)) === "refused"
+    )
+      continue;
+    ids.push(entry.id);
+  }
   return [...new Set(ids)];
 }
 
@@ -62,9 +76,15 @@ export async function projectModelChoice(
   actor: Actor,
   projectId: string,
   directory: ModelDirectory,
+  probe?: (tenantId: string, providerId: string) => Promise<AvailabilityState>,
 ): Promise<ModelChoice> {
   const { models } = await loadSettings(db, actor.tenantId, projectId);
-  const connected = await connectedProviders(db, directory, actor.tenantId);
+  const connected = await connectedProviders(
+    db,
+    directory,
+    actor.tenantId,
+    probe,
+  );
   const kept = frozenModelChoice(directory, models.value, connected);
   if (kept) return kept;
   const resolved = resolveModelChoice(
@@ -94,3 +114,49 @@ export async function projectModelChoice(
     );
   return resolved.choice;
 }
+
+/** Switches a project onto one of the caller's runnable profiles. */
+export async function switchProjectModelChoice(
+  db: DB["db"],
+  actor: Actor,
+  projectId: string,
+  profile: string,
+  directory: ModelDirectory,
+  probe?: (tenantId: string, providerId: string) => Promise<AvailabilityState>,
+): Promise<ModelChoice> {
+  const { models } = await loadSettings(db, actor.tenantId, projectId);
+  const personal = await personalChoice(db, actor);
+  const picked =
+    pickedModelChoice(personal, profile) ??
+    pickedModelChoice(models.value, profile);
+  if (!picked)
+    throw new ProjectError(400, "model_profile_unknown", { profile });
+  const connected = await connectedProviders(
+    db,
+    directory,
+    actor.tenantId,
+    probe,
+  );
+  const frozen = frozenModelChoice(directory, picked, connected);
+  if (!frozen)
+    throw new ProjectError(409, "model_provider_missing", {
+      message: MODEL_CHOICE_REFUSALS.model_provider_missing,
+    });
+  const scopeKey = `project:${projectId}`;
+  const fields = await loadSettingsAt(db, actor.tenantId, scopeKey);
+  await db
+    .update(settingsRecord)
+    .set({
+      fields: {
+        ...fields,
+        models: { value: frozen, version: fields.models.version + 1 },
+      },
+    })
+    .where(
+      and(
+        eq(settingsRecord.tenantId, actor.tenantId),
+        eq(settingsRecord.scopeKey, scopeKey),
+      ),
+    );
+  return frozen;
+}

apps/hub/src/projects/routes.ts

@@ -31,10 +31,18 @@ import { readProjectImport } from "@corbits-builder/schemas/project-import";
 import {
   createProjectService,
   ProjectError,
+  projectResource,
   type CreateTenant,
   type ProjectCaller,
 } from "./service.ts";
-import { actorOf, resolveProjectTenant } from "./authorization.ts";
+import {
+  actorOf,
+  grantAuthorization,
+  resolveProjectTenant,
+} from "./authorization.ts";
+import { switchProjectModelChoice } from "./model-choice.ts";
+import { modelDirectory } from "../settings/custom-providers.ts";
+import type { AvailabilityState } from "../availability.ts";
 import {
   createProjectNextStep,
   mountProjectNextStepRoutes,
@@ -188,6 +196,7 @@ export function createProjectRoutes(
     reason: string,
   ) => Promise<unknown>,
   repoStore?: RepoStore,
+  probe?: (tenantId: string, providerId: string) => Promise<AvailabilityState>,
 ) {
   const app = new Hono<TenantEnv>();
   const service = createProjectService(db);
@@ -697,6 +706,28 @@ export function createProjectRoutes(
         const view = await service.read(actor, projectId);
         if (view.status !== "active")
           throw new ProjectError(409, "project_archived");
+        const body = await readJson(c.req);
+        const profile =
+          body !== null &&
+          typeof body === "object" &&
+          "profile" in body &&
+          typeof body.profile === "string"
+            ? body.profile.trim()
+            : undefined;
+        if (profile) {
+          const allowed = await grantAuthorization(db, actor);
+          if (!(await allowed(projectResource(projectId), "manage")))
+            throw new ProjectError(403, "forbidden");
+          const directory = await modelDirectory(db, actor.tenantId);
+          await switchProjectModelChoice(
+            db,
+            actor,
+            projectId,
+            profile,
+            directory,
+            probe,
+          );
+        }
         await startCoordinator?.upgrade?.(actor, projectId);
         return c.json({ started: await resumeCoordinator(actor, projectId) });
       } catch (error) {