Skip to content

Commit e560c80

Browse files
committed
fix(models): surface refused credentials with recovery and handoff
1 parent ea054ea commit e560c80

29 files changed

Lines changed: 1084 additions & 65 deletions

‎apps/hub/src/app.ts‎

Lines changed: 11 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -85,7 +85,10 @@ import {
8585
correlationMiddleware,
8686
requestContext,
8787
} from "./middleware/correlation.ts";
88-
import { createAvailabilityRoutes } from "./availability.ts";
88+
import {
89+
createAvailabilityProbe,
90+
createAvailabilityRoutes,
91+
} from "./availability.ts";
8992
import { createGitTokenBoundary } from "./git-token-boundary.ts";
9093
import { createSidecarWebSocket } from "./sidecar-websocket.ts";
9194
import { createCoordinatorCredentialRefresh } from "./auth/coordinator-credential-refresh.ts";
@@ -318,6 +321,10 @@ export async function createHubRuntime(
318321
hubId: hexEncode(key.publicKey).slice(0, 24),
319322
});
320323
const allocation = workflows.services.workflowAllocationService;
324+
const availabilityProbe = createAvailabilityProbe(
325+
handle.db,
326+
credentialCipher,
327+
);
321328
const coordinatorStartup = createInstalledCoordinatorStartup(
322329
handle.db,
323330
assetService,
@@ -355,13 +362,14 @@ export async function createHubRuntime(
355362
},
356363
coordinatorStartTimeoutMs,
357364
workflows.services.workflowLifecycleService.requestCancellation,
365+
availabilityProbe,
358366
);
359367
const coordinatorUpgrades = createCoordinatorUpgrades({
360368
db: handle.db,
361369
runs: createWorkflowRunReader(agentRepoStore.repoStore),
362370
installed: (actor, projectId) =>
363371
installForTenant(
364-
{ db: handle.db, assets: assetService },
372+
{ db: handle.db, assets: assetService, probe: availabilityProbe },
365373
actor,
366374
projectId,
367375
),
@@ -525,6 +533,7 @@ export async function createHubRuntime(
525533
liveApps,
526534
workflows.services.workflowLifecycleService.requestCancellation,
527535
agentRepoStore.repoStore,
536+
availabilityProbe,
528537
),
529538
);
530539
app.route(

‎apps/hub/src/artifacts/activity.ts‎

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -108,12 +108,22 @@ export function createAgentActivity(events: Pick<SidecarEventEmitter, "on">) {
108108
delete run.said;
109109
return;
110110
}
111-
case "message.run.ended": {
111+
case "message.run.ended":
112+
case "reactor.done":
113+
case "reactor.error": {
112114
const run = entry(local);
113115
run.working = false;
114116
run.since = now;
115117
return;
116118
}
119+
case "inference.error": {
120+
if (event.data.error.category === "credential_failure") {
121+
const run = entry(local);
122+
run.working = false;
123+
run.since = now;
124+
}
125+
return;
126+
}
117127
case "tool.start": {
118128
const run = entry(local);
119129
// A turn whose start the hub missed, as after its own restart, dates from this call.

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

Lines changed: 15 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,9 @@
1-
import { and, eq } from "drizzle-orm";
2-
import type { DB } from "@intx/db";
3-
import { workflowRunExecutability } from "@intx/db";
1+
import { and, inArray } from "drizzle-orm";
2+
import {
3+
getDescendantTenants,
4+
workflowRunExecutability,
5+
type DB,
6+
} from "@intx/db";
47
import { workflowRun as runTable } from "@intx/db/schema";
58
import {
69
pushCredentialReconcile,
@@ -11,11 +14,10 @@ import type { CredentialCipher } from "@intx/types";
1114
import type { MiddlewareHandler } from "hono";
1215
import type { TenantEnv } from "@intx/hub-api";
1316
import { getLogger } from "@intx/log";
14-
import { projectCoordinator } from "../db/schema.ts";
1517

1618
const log = getLogger(["corbits-builder", "coordinator-credentials"]);
1719

18-
// After a credential's secret changes, pushes it to every active coordinator run that uses it.
20+
// After a credential's secret changes, pushes it to every active project run that uses it.
1921
export function createCoordinatorCredentialRefresh(options: {
2022
db: DB["db"];
2123
router: SidecarRouter;
@@ -29,19 +31,20 @@ export function createCoordinatorCredentialRefresh(options: {
2931
const body = await c.req.json();
3032
if (body.secret === undefined || body.status === "revoked") return;
3133
const credentialId = c.req.param("credentialId");
32-
if (!credentialId) return;
33-
const bindings = await options.db
34+
const tenantId = c.req.param("tenantId");
35+
if (!credentialId || !tenantId) return;
36+
const tenants = await getDescendantTenants(options.db, tenantId);
37+
const runs = await options.db
3438
.select({ run: runTable })
35-
.from(projectCoordinator)
36-
.innerJoin(runTable, eq(runTable.id, projectCoordinator.runId))
39+
.from(runTable)
3740
.where(
3841
and(
39-
eq(projectCoordinator.active, true),
40-
eq(runTable.tenantId, c.req.param("tenantId")!),
42+
inArray(runTable.tenantId, tenants.length ? tenants : [tenantId]),
43+
inArray(runTable.status, ["deployed", "running"]),
4144
),
4245
);
4346
await Promise.all(
44-
bindings
47+
runs
4548
.filter(
4649
({ run }) =>
4750
run.address &&

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

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -52,6 +52,11 @@ test("a project runs on the only connected provider's profile, a chosen one, or
5252
).toEqual({
5353
refused: "model_profile_required",
5454
});
55+
expect(
56+
resolveModelChoice(MODEL_PROVIDERS, { profile: "anthropic_default" }, []),
57+
).toEqual({
58+
refused: "model_provider_missing",
59+
});
5560
expect(
5661
frozenModelChoice(MODEL_PROVIDERS, { profile: "openai_default" }, [
5762
"anthropic",

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

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -467,9 +467,9 @@ export const MODEL_CHOICE_REFUSALS: Readonly<
467467
};
468468

469469
/**
470-
* The profile a project runs on: the person's choice when this server can run it, else the shipped
471-
* profile of the only connected provider. With none connected, or several and no runnable choice,
472-
* there is no profile.
470+
* The profile a project runs on: the person's choice when this server can run it on a connected
471+
* provider, else the shipped profile of the only connected provider. With none connected, or
472+
* several and no runnable choice, there is no profile.
473473
*/
474474
export function resolveModelChoice(
475475
directory: ModelDirectory,

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

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -12,6 +12,7 @@ import {
1212
import type { InstalledWorkflows } from "@corbits-builder/schemas/installed-workflows";
1313
import type { NativeSpecialistDeployment } from "@corbits-builder/schemas/native-specialist";
1414
import { createCoordinatorRegistry } from "../auth/coordinator-registry.ts";
15+
import type { AvailabilityState } from "../availability.ts";
1516
import { COORDINATOR_CAPACITY_FULL } from "../config/coordinator-start.ts";
1617
import {
1718
installForTenant,
@@ -265,11 +266,12 @@ export function createInstalledCoordinatorStartup(
265266
}) => Promise<{ runId: string }>,
266267
readyWaitMs: number,
267268
cancel: CancelRun,
269+
probe?: (tenantId: string, providerId: string) => Promise<AvailabilityState>,
268270
) {
269271
const registry = createCoordinatorRegistry(db);
270272
return createProjectCoordinatorStartup({
271273
installed(actor, projectId) {
272-
return installForTenant({ db, assets }, actor, projectId);
274+
return installForTenant({ db, assets, probe }, actor, projectId);
273275
},
274276
async boundRun(tenantId, projectId) {
275277
const [row] = await db

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

Lines changed: 15 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,6 +3,8 @@ import { type } from "arktype";
33
import type { DB } from "@intx/db";
44
import {
55
approval,
6+
modelOffering,
7+
modelProvider,
68
workflowRun,
79
workflowRunDispatch,
810
workflowRunLaunchSpec,
@@ -162,6 +164,7 @@ export function createCoordinatorUpgrades(deps: {
162164
status: workflowRun.status,
163165
cancellationRequestedAt: workflowRun.cancellationRequestedAt,
164166
cancellationReason: workflowRun.cancellationReason,
167+
credentialRefs: workflowRun.credentialRefs,
165168
principalId: workflowRunLaunchSpec.sourceAuthorityPrincipalId,
166169
bundle: workflowRunLaunchSpec.frozenApprovalBundle,
167170
})
@@ -196,8 +199,18 @@ export function createCoordinatorUpgrades(deps: {
196199
!["deployed", "running"].includes(latest.status)
197200
)
198201
return;
199-
const current = (await deps.installed(actor, projectId)).coordinator.pin;
200-
if (pinOf(latest.bundle) === current) return;
202+
const current = (await deps.installed(actor, projectId)).coordinator;
203+
const [offering] = await db
204+
.select({ credentialId: modelProvider.credentialId })
205+
.from(modelOffering)
206+
.innerJoin(modelProvider, eq(modelProvider.id, modelOffering.providerId))
207+
.where(eq(modelOffering.id, current.defaultSourceOfferingId));
208+
const sameCredential =
209+
!offering?.credentialId ||
210+
Boolean(
211+
latest.credentialRefs?.credentialIds.includes(offering.credentialId),
212+
);
213+
if (pinOf(latest.bundle) === current.pin && sameCredential) return;
201214
const parked = await atSafePoint(actor, {
202215
runId: latest.runId,
203216
address: latest.address,

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

Lines changed: 72 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -2,9 +2,11 @@ import { and, eq } from "drizzle-orm";
22
import { listVisibleProviders, type DB } from "@intx/db";
33
import { principal, tenant } from "@intx/db/schema";
44
import type { ModelChoice } from "@corbits-builder/schemas/models";
5+
import type { AvailabilityState } from "../availability.ts";
56
import {
67
MODEL_CHOICE_REFUSALS,
78
frozenModelChoice,
9+
pickedModelChoice,
810
providerOfRow,
911
resolveModelChoice,
1012
type ModelDirectory,
@@ -19,16 +21,28 @@ import { userPrincipalIn } from "./authorization.ts";
1921
import { ProjectError } from "./errors.ts";
2022
import type { Actor } from "../lib/actor.ts";
2123

22-
/** The provider entries a tenant can run on: each enabled connection with a credential. */
24+
/** The provider entries a tenant can run on: each enabled connection with a non-refused credential. */
2325
export async function connectedProviders(
2426
db: DB["db"],
2527
directory: ModelDirectory,
2628
tenantId: string,
29+
probe?: (tenantId: string, providerId: string) => Promise<AvailabilityState>,
2730
): Promise<string[]> {
28-
const ids = (await listVisibleProviders(db, tenantId))
29-
.filter(({ row }) => row.credentialId !== null && !row.disabled)
30-
.map(({ row }) => providerOfRow(directory, row)?.id)
31-
.filter((id) => id !== undefined);
31+
const visible = (await listVisibleProviders(db, tenantId)).filter(
32+
({ row }) => row.credentialId !== null && !row.disabled,
33+
);
34+
const ids: string[] = [];
35+
for (const { row } of visible) {
36+
const entry = providerOfRow(directory, row);
37+
if (!entry) continue;
38+
if (
39+
probe &&
40+
entry.kind !== "custom" &&
41+
(await probe(tenantId, row.id)) === "refused"
42+
)
43+
continue;
44+
ids.push(entry.id);
45+
}
3246
return [...new Set(ids)];
3347
}
3448

@@ -62,9 +76,15 @@ export async function projectModelChoice(
6276
actor: Actor,
6377
projectId: string,
6478
directory: ModelDirectory,
79+
probe?: (tenantId: string, providerId: string) => Promise<AvailabilityState>,
6580
): Promise<ModelChoice> {
6681
const { models } = await loadSettings(db, actor.tenantId, projectId);
67-
const connected = await connectedProviders(db, directory, actor.tenantId);
82+
const connected = await connectedProviders(
83+
db,
84+
directory,
85+
actor.tenantId,
86+
probe,
87+
);
6888
const kept = frozenModelChoice(directory, models.value, connected);
6989
if (kept) return kept;
7090
const resolved = resolveModelChoice(
@@ -94,3 +114,49 @@ export async function projectModelChoice(
94114
);
95115
return resolved.choice;
96116
}
117+
118+
/** Switches a project onto one of the caller's runnable profiles. */
119+
export async function switchProjectModelChoice(
120+
db: DB["db"],
121+
actor: Actor,
122+
projectId: string,
123+
profile: string,
124+
directory: ModelDirectory,
125+
probe?: (tenantId: string, providerId: string) => Promise<AvailabilityState>,
126+
): Promise<ModelChoice> {
127+
const { models } = await loadSettings(db, actor.tenantId, projectId);
128+
const personal = await personalChoice(db, actor);
129+
const picked =
130+
pickedModelChoice(personal, profile) ??
131+
pickedModelChoice(models.value, profile);
132+
if (!picked)
133+
throw new ProjectError(400, "model_profile_unknown", { profile });
134+
const connected = await connectedProviders(
135+
db,
136+
directory,
137+
actor.tenantId,
138+
probe,
139+
);
140+
const frozen = frozenModelChoice(directory, picked, connected);
141+
if (!frozen)
142+
throw new ProjectError(409, "model_provider_missing", {
143+
message: MODEL_CHOICE_REFUSALS.model_provider_missing,
144+
});
145+
const scopeKey = `project:${projectId}`;
146+
const fields = await loadSettingsAt(db, actor.tenantId, scopeKey);
147+
await db
148+
.update(settingsRecord)
149+
.set({
150+
fields: {
151+
...fields,
152+
models: { value: frozen, version: fields.models.version + 1 },
153+
},
154+
})
155+
.where(
156+
and(
157+
eq(settingsRecord.tenantId, actor.tenantId),
158+
eq(settingsRecord.scopeKey, scopeKey),
159+
),
160+
);
161+
return frozen;
162+
}

‎apps/hub/src/projects/routes.ts‎

Lines changed: 32 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -31,10 +31,18 @@ import { readProjectImport } from "@corbits-builder/schemas/project-import";
3131
import {
3232
createProjectService,
3333
ProjectError,
34+
projectResource,
3435
type CreateTenant,
3536
type ProjectCaller,
3637
} from "./service.ts";
37-
import { actorOf, resolveProjectTenant } from "./authorization.ts";
38+
import {
39+
actorOf,
40+
grantAuthorization,
41+
resolveProjectTenant,
42+
} from "./authorization.ts";
43+
import { switchProjectModelChoice } from "./model-choice.ts";
44+
import { modelDirectory } from "../settings/custom-providers.ts";
45+
import type { AvailabilityState } from "../availability.ts";
3846
import {
3947
createProjectNextStep,
4048
mountProjectNextStepRoutes,
@@ -188,6 +196,7 @@ export function createProjectRoutes(
188196
reason: string,
189197
) => Promise<unknown>,
190198
repoStore?: RepoStore,
199+
probe?: (tenantId: string, providerId: string) => Promise<AvailabilityState>,
191200
) {
192201
const app = new Hono<TenantEnv>();
193202
const service = createProjectService(db);
@@ -697,6 +706,28 @@ export function createProjectRoutes(
697706
const view = await service.read(actor, projectId);
698707
if (view.status !== "active")
699708
throw new ProjectError(409, "project_archived");
709+
const body = await readJson(c.req);
710+
const profile =
711+
body !== null &&
712+
typeof body === "object" &&
713+
"profile" in body &&
714+
typeof body.profile === "string"
715+
? body.profile.trim()
716+
: undefined;
717+
if (profile) {
718+
const allowed = await grantAuthorization(db, actor);
719+
if (!(await allowed(projectResource(projectId), "manage")))
720+
throw new ProjectError(403, "forbidden");
721+
const directory = await modelDirectory(db, actor.tenantId);
722+
await switchProjectModelChoice(
723+
db,
724+
actor,
725+
projectId,
726+
profile,
727+
directory,
728+
probe,
729+
);
730+
}
700731
await startCoordinator?.upgrade?.(actor, projectId);
701732
return c.json({ started: await resumeCoordinator(actor, projectId) });
702733
} catch (error) {

0 commit comments

Comments
 (0)