Skip to content

Commit b39681e

Browse files
committed
refactor(hub): tighten comments in hub composition and process provisioner
1 parent d5932fd commit b39681e

2 files changed

Lines changed: 58 additions & 135 deletions

File tree

‎apps/hub/src/corbits-mount.ts‎

Lines changed: 41 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,3 @@
1-
// Corbits mount block — the single composition point where Workbench adds
2-
// Corbits libraries (@corbits/artifacts, @corbits/agent-token, @corbits/mailbox,
3-
// @corbits/cron, @corbits/memory, @corbits/oauth-core, @corbits/mcp,
4-
// @corbits/webhooks) and live workflow-run SSE streaming on top of stock
5-
// @intx/hub-api routes. See AGENTS.md: any other hub mount is cutover debt,
6-
// never a pattern to extend.
71
import { createAgentTokenVerifier, mountAgentTokens } from "@corbits/agent-token";
82
import {
93
InlineContentStore,
@@ -66,11 +60,7 @@ import {
6660
createRequireGrant,
6761
type TenantEnv,
6862
} from "@intx/hub-api";
69-
import {
70-
pushSourceUpdates,
71-
type SidecarLookups,
72-
type SidecarRouter,
73-
} from "@intx/hub-sessions";
63+
import { pushSourceUpdates, type SidecarLookups, type SidecarRouter } from "@intx/hub-sessions";
7464
import { getLogger } from "@intx/log";
7565
import type { ConditionRegistry } from "@intx/types/authz";
7666
import { and, eq } from "drizzle-orm";
@@ -80,10 +70,6 @@ import { captureMailboxRequest, createMailboxDeliver } from "./mailbox-send";
8070
import { createWorkflowRunStreamRoutes } from "./run-stream";
8171
import { createWorkflowRunAuthenticator } from "./workflow-run-authenticator";
8272

83-
// The same condition registry `mountHubRoutes` builds by default when no
84-
// registry is supplied -- kept as one local constant so every Corbits
85-
// route factory below shares the identical registry rather than each
86-
// re-deriving stock's own default.
8773
const grantConditionRegistry: ConditionRegistry = {
8874
time_window: timeWindowEvaluator,
8975
};
@@ -130,6 +116,8 @@ export async function mountCorbits({
130116
authorizeSender: createHubMailboxAuthorizeSender(db),
131117
bus: mailboxBus,
132118
});
119+
// `createSidecarRouter` captured `lookups` before this wrapper existed;
120+
// mutating it in place ensures agent outbound mail also lands in principal inboxes.
133121
lookups.persistMail = persistMail;
134122

135123
{
@@ -152,16 +140,13 @@ export async function mountCorbits({
152140
);
153141

154142
{
155-
// Minting an agent token is minting a credential, so it is gated by the
156-
// same stock grant the credential routes are, not by tenant membership.
157143
const agentTokensApp = new Hono<TenantEnv>();
158144
mountAgentTokens(agentTokensApp, {
159145
db,
160146
requireGrant: requireGrant("credential:*", "create"),
161147
resolveTenantId: (ctx) => (ctx as { get(key: "tenant"): { id: string } }).get("tenant").id,
162-
// A token is scoped to the agent's `workflow` source asset: the
163-
// tenant-owned thing that already exists when the workbench mints the
164-
// token, before the deploy that would create a run.
148+
// Scoped to the `workflow` source asset so the workbench can mint the
149+
// token before the deploy that creates the run.
165150
resolveDefinition: async (tenantId, definitionId) => {
166151
const row = await db.query.asset.findFirst({
167152
where: and(
@@ -176,17 +161,18 @@ export async function mountCorbits({
176161
app.route(TENANT_PREFIX, agentTokensApp);
177162
}
178163

179-
// A deployed agent has no sidecar token; it presents the bearer this hub
180-
// minted for its definition and is scoped to the run it names. One
181-
// verifier and one run lookup serve every run-scoped Corbits mount.
182164
const verifyAgentToken = createAgentTokenVerifier({ db });
183165
const resolveAgentRun = async (runAddress: string) => {
184166
if (runAddress === "") return null;
185167
const run = await db.query.workflowRun.findFirst({
186168
where: eq(workflowRun.address, runAddress),
187169
});
188170
if (run === undefined || run.principalId === null) return null;
189-
return { tenantId: run.tenantId, principalId: run.principalId, runId: run.id };
171+
return {
172+
tenantId: run.tenantId,
173+
principalId: run.principalId,
174+
runId: run.id,
175+
};
190176
};
191177

192178
{
@@ -196,24 +182,27 @@ export async function mountCorbits({
196182
db,
197183
contentStore: InlineContentStore,
198184
resolveRunScope: (token, runAddress) => workflowRunAuthenticator.resolve(token, runAddress),
199-
agentToken: { verify: (ctx) => verifyAgentToken(ctx), resolveRun: resolveAgentRun },
185+
agentToken: {
186+
verify: (ctx) => verifyAgentToken(ctx),
187+
resolveRun: resolveAgentRun,
188+
},
200189
});
201190
app.route("/api/workflow-artifacts", workflowArtifactsApi);
202191
}
203192

204193
{
205194
const mailboxApp = new Hono<TenantEnv>();
206-
// Registered before the routes: Hono runs handlers in registration order.
207195
mailboxApp.use("/me/inbox/send", captureMailboxRequest());
208196
mountMailbox(mailboxApp, {
209197
db: mailboxDb,
210198
bus: mailboxBus,
211199
resolvePrincipal: (ctx) => {
212200
const c = ctx as { get(key: "tenant" | "principal"): { id: string } };
213-
return { tenantId: c.get("tenant").id, principalId: c.get("principal").id };
201+
return {
202+
tenantId: c.get("tenant").id,
203+
principalId: c.get("principal").id,
204+
};
214205
},
215-
// The address stock itself stamps on a person's outbound mail, so a
216-
// Sent copy and the agent's reply share one identity.
217206
senderAddressFor: async (principal) => {
218207
const [tenantRow] = await db
219208
.select({ domain: tenantTable.domain })
@@ -231,19 +220,14 @@ export async function mountCorbits({
231220
if (principalRow === undefined) {
232221
throw new Error(`no principal "${principal.principalId}" to address a mailbox sender as`);
233222
}
234-
// A person's refId is mixed case; a small model lowercases an address
235-
// before replying, so the canonical stamped form is lowercase.
223+
// Small models lowercase addresses before replying, so stamp lowercase canonically.
236224
return `${principalRow.refId}@${tenantRow.domain}`.toLowerCase();
237225
},
238226
deliver: createMailboxDeliver({ app, persistMail }),
239227
});
240228
app.route(`${TENANT_PREFIX}/mailbox`, mailboxApp);
241229
}
242230

243-
// The router every system-originated trigger (webhook, cron) goes
244-
// through. `HookMailRouter` types its payloads as `unknown` at the
245-
// package boundary; this just narrows them back to `sidecarRouter`'s own
246-
// types on the way through, with no behavior change.
247231
const systemTriggerMailRouter: HookMailRouter = {
248232
routeMail: (address, rawMessage, authenticatedSender, messageId) => {
249233
if (!sidecarRouter.getRoutableAddresses().includes(address)) {
@@ -276,11 +260,8 @@ export async function mountCorbits({
276260
}),
277261
);
278262

279-
// A due schedule fires with nobody signed in, so it cannot ride the
280-
// mailbox persist path, which authorizes its sender against a live
281-
// routable endpoint. It is a system trigger like an inbound webhook,
282-
// so it takes the same route: the run is the authenticated sender of
283-
// its own signed trigger mail, with its grants materialized first.
263+
// Due schedules fire without a signed-in user, so they use the system-trigger
264+
// path rather than mailbox persist (which requires a live routable sender).
284265
const cronTicker = createCronTicker({
285266
db,
286267
intervalMs: 60_000,
@@ -304,8 +285,6 @@ export async function mountCorbits({
304285
return tenantRow.domain;
305286
},
306287
senderLocalPart: "cron",
307-
// A durable per-tenant key the recipient can verify against;
308-
// the sidecar rejects trigger mail from an unknown sender.
309288
systemSender: createTenantSystemSender({ db, principalKeyStore }),
310289
}),
311290
),
@@ -315,17 +294,13 @@ export async function mountCorbits({
315294
extra: { scheduleId: schedule.id, tenantId: schedule.tenantId },
316295
});
317296
},
318-
// Waiting is the ordinary gap between an agent's runs, not a fault —
319-
// it resolves itself the moment the agent comes back.
320297
onScheduleWaiting: (schedule) => {
321298
log.info("Cron schedule waiting for agent {definitionName}", {
322299
scheduleId: schedule.id,
323300
tenantId: schedule.tenantId,
324301
definitionName: schedule.definitionName,
325302
});
326303
},
327-
// A stopped schedule never fires again and no redeploy resumes it, so
328-
// the one report it gets has to be durable.
329304
onScheduleStopped: (schedule) => {
330305
reportError(
331306
new Error(`cron schedule stopped: ${schedule.reason} (${schedule.definitionName})`),
@@ -353,36 +328,32 @@ export async function mountCorbits({
353328
});
354329
app.route("/", memoryApp);
355330

356-
// The same plane, reached by a deployed agent's bearer rather than a
357-
// browser session, and scoped to the run the bearer names.
358331
const workflowMemoryApi = new Hono<WorkflowMemoryEnv>();
359332
mountWorkflowMemory(workflowMemoryApi, {
360333
memory,
361-
agentToken: { verify: (ctx) => verifyAgentToken(ctx), resolveRun: resolveAgentRun },
334+
agentToken: {
335+
verify: (ctx) => verifyAgentToken(ctx),
336+
resolveRun: resolveAgentRun,
337+
},
362338
});
363339
app.route("/api/workflow-memory", workflowMemoryApi);
364340
}
365341

366342
const hubPublicUrl = process.env["HUB_PUBLIC_URL"] ?? `http://127.0.0.1:${String(port)}`;
367-
// A loopback hub is local development, where MCP servers may be plain-http
368-
// loopback too; a public hub keeps discovery https-only.
369343
const hubHost = new URL(hubPublicUrl).hostname;
370344
const localDevHub = hubHost === "localhost" || hubHost.startsWith("127.") || hubHost === "[::1]";
371345

372-
// "Continue with Codex"/"Continue with xAI": the whole loopback PKCE
373-
// flow runs here, so the verifier and the callback listener never
374-
// leave this process and the browser only learns a credential id.
375346
const oauthLoginApi = new Hono<TenantEnv>();
376347
const oauthProviders: OAuthLoginProviders = {
377348
[CODEX_PROVIDER]: {
378349
oauthConfig: codexOAuthConfig,
379350
exchange: (code: string, verifier: string, now: number) =>
380351
exchangeCodexCode(code, verifier, now),
381-
// The account id the refresher carries forward lives on the
382-
// credential, so the prior tokens need only supply the secret.
383352
refresh: (refreshSecret: string, now: number) =>
384-
refreshCodexTokens(refreshSecret, now, { access: "", refresh: refreshSecret }),
385-
// The Codex backend rejects inference without this header value.
353+
refreshCodexTokens(refreshSecret, now, {
354+
access: "",
355+
refresh: refreshSecret,
356+
}),
386357
metadata: (tokens) =>
387358
"accountId" in tokens && typeof tokens.accountId === "string"
388359
? { accountId: tokens.accountId }
@@ -395,12 +366,12 @@ export async function mountCorbits({
395366
refresh: (refreshSecret: string, now: number) => refreshXaiTokens(refreshSecret, now),
396367
},
397368
};
398-
// Resource-URL logins (MCP servers) redirect back to a hub-root route
399-
// outside any tenant; it shares the login store with the tenant routes.
400369
const oauthLoginStore = createOAuthLoginStore();
401-
// The mount takes a plain Hono, so it rides its own router on the root.
402370
const oauthCallbackApi = new Hono();
403-
mountOAuthCallback(oauthCallbackApi, { store: oauthLoginStore, path: "/api/oauth/callback" });
371+
mountOAuthCallback(oauthCallbackApi, {
372+
store: oauthLoginStore,
373+
path: "/api/oauth/callback",
374+
});
404375
app.route("/", oauthCallbackApi);
405376
mountOAuthLogin(oauthLoginApi, {
406377
store: oauthLoginStore,
@@ -411,22 +382,20 @@ export async function mountCorbits({
411382
requireGrant: requireGrant("credential:*", "create"),
412383
providers: oauthProviders,
413384
onError: (error, { provider }) => {
414-
reportError(error, { operation: "hub.oauth-login", extra: { provider } });
385+
reportError(error, {
386+
operation: "hub.oauth-login",
387+
extra: { provider },
388+
});
415389
},
416390
});
417391
app.route(TENANT_PREFIX, oauthLoginApi);
418392

419-
// Stock Interchange has no serving-time refresh hook, so a subscription
420-
// token that lapses between inference calls would simply fail the next
421-
// one; this renews it a little before expiry instead.
422393
const oauthTokenRefresher = createOAuthTokenRefresher({
423394
db,
424395
cipher: credentialCipher,
425396
providers: oauthProviders,
426397
intervalMs: 60_000,
427398
onRefreshed: ({ tenantId, credentialId }) => {
428-
// The same push the stock credentials route fires after a secret
429-
// rotation, so running sidecars get the new material.
430399
void pushSourceUpdates(db, sidecarRouter, tenantId, credentialCipher).catch(
431400
(error: unknown) => {
432401
reportError(error, {
@@ -439,15 +408,16 @@ export async function mountCorbits({
439408
onError: (error, { provider, credentialId }) => {
440409
reportError(error, {
441410
operation: "hub.oauth-refresh",
442-
extra: { provider: provider ?? "", credentialId: credentialId ?? "" },
411+
extra: {
412+
provider: provider ?? "",
413+
credentialId: credentialId ?? "",
414+
},
443415
});
444416
},
445417
});
446418
oauthTokenRefresher.start();
447419

448420
{
449-
// The browser never holds an MCP server's token, so the one place a
450-
// catalog can be read is here, with the secret decrypted in-process.
451421
const mcpApi = new Hono<TenantEnv>();
452422
mountMcpDiscovery(mcpApi, {
453423
allowLoopback: localDevHub,

0 commit comments

Comments
 (0)