mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 01:57:31 +01:00
feat(connections): add self-serve intent runtime (#12345)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Agents need a governed way to request app connections during issue work. > - The catalog now describes the available providers and setup methods. > - A request must become a durable, company-scoped intent before an operator acts on it. > - This pull request adds that intent runtime across server, agent, CLI, and shared contracts. > - The benefit is a safe bridge from agent need to operator-approved setup. ## Linked Issues or Issue Description Refs #11965 This is stack 7 of 11. It depends on stack 6 and replaces another reviewable part of #11965. ## What Changed - Add connection intent types, validation, service logic, and routes. - Add agent runtime tools and CLI support for connection requests. - Add issue-thread interaction support for connection intents. - Add runtime, route, adapter, and contract tests. - Hold the final resolved-continuation row lock through asynchronous adapter preparation until an actual process spawn, so parking or reassignment cannot cross that boundary. - Report Hermes Gateway's first remote run request through the shared dispatch hook so the resolved-intent lock is released at the true dispatch boundary. - Revalidate the addressed user's live non-viewer membership and connection-management authority for every intent mutation, including OAuth completion. ## Verification - `pnpm --filter @paperclipai/server typecheck` - `pnpm --filter @paperclipai/server exec vitest run src/__tests__/tool-access-service.test.ts` - Result: 176 tests passed. - `pnpm build` - `pnpm --filter @paperclipai/server exec vitest run src/__tests__/heartbeat-stale-queue-invalidation.test.ts` (32 passed; includes non-process dispatch lock-release coverage) - `pnpm exec vitest run --project @paperclipai/server server/src/__tests__/connection-intents-service.test.ts -t "addressed-user mutation"` (1 passed) - `pnpm exec vitest run --project @paperclipai/server server/src/__tests__/tool-access-service.test.ts -t "binds OAuth callback completion to the initiating board session"` (1 passed) - `pnpm --filter @paperclipai/hermes-paperclip-adapter test -- src/gateway/server/execute.test.ts` (23 passed; includes dispatch-hook ordering and exactly-once coverage) - `pnpm --filter @paperclipai/hermes-paperclip-adapter typecheck` ## Risks - A malformed intent could create an unusable operator request. - Validators and company checks reject invalid or cross-company requests. - The final continuation gate holds the issue row lock through adapter preparation until process or remote dispatch; later operator changes use the normal active-run interruption path. - The change does not add a database migration. > I checked `ROADMAP.md`. This stack continues the existing app connection work from #11965 and does not duplicate another planned item. ## Model Used OpenAI Codex, GPT-5. The runtime model ID and context window were not exposed. The model used reasoning, tool use, and code execution. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have linked the public source pull request with `Refs #` - [x] I have not referenced internal or instance-local Paperclip issues or links - [x] My branch name describes the change and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip authored and GitHub committed
1 parent
fcb2e99e8f
commit
b3343dbd64
96 files changed
+8695
-971
No files matched your search
@@ -351,7 +351,7 @@ function agentIsInSubtree(
|
||||
return false;
|
||||
}
|
||||
|
||||
async function loadCompanyAgentHierarchy(db: Db, companyId: string) {
|
||||
async function loadCompanyAgentHierarchy(db: Db | DbTransaction, companyId: string) {
|
||||
const rows = await db
|
||||
.select({ id: agents.id, reportsTo: agents.reportsTo })
|
||||
.from(agents)
|
||||
@@ -359,7 +359,12 @@ async function loadCompanyAgentHierarchy(db: Db, companyId: string) {
|
||||
return new Map(rows.map((agent) => [agent.id, agent]));
|
||||
}
|
||||
|
||||
async function isAgentInSubtree(db: Db, companyId: string, rootAgentId: string, targetAgentId: string) {
|
||||
async function isAgentInSubtree(
|
||||
db: Db | DbTransaction,
|
||||
companyId: string,
|
||||
rootAgentId: string,
|
||||
targetAgentId: string,
|
||||
) {
|
||||
return agentIsInSubtree(
|
||||
await loadCompanyAgentHierarchy(db, companyId),
|
||||
rootAgentId,
|
||||
@@ -368,7 +373,7 @@ async function isAgentInSubtree(db: Db, companyId: string, rootAgentId: string,
|
||||
}
|
||||
|
||||
async function scopeAllows(
|
||||
db: Db,
|
||||
db: Db | DbTransaction,
|
||||
companyId: string,
|
||||
grantScope: Record<string, unknown> | null,
|
||||
requestedScope: Record<string, unknown> | null | undefined,
|
||||
@@ -526,7 +531,9 @@ export function authorizationDeniedDetails(decision: AuthorizationDecision) {
|
||||
};
|
||||
}
|
||||
|
||||
export function authorizationService(db: Db) {
|
||||
type DbTransaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
|
||||
|
||||
export function authorizationService(db: Db | DbTransaction) {
|
||||
async function isInstanceAdmin(userId: string | null | undefined): Promise<boolean> {
|
||||
if (!userId) return false;
|
||||
if (
|
||||
|
||||
@@ -0,0 +1,530 @@
|
||||
import { and, eq } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
agents,
|
||||
companyMemberships,
|
||||
heartbeatRuns,
|
||||
issueThreadInteractions,
|
||||
issues,
|
||||
} from "@paperclipai/db";
|
||||
import {
|
||||
CONNECTABLE_APP_DEFINITIONS,
|
||||
connectionIntentPayloadSchema,
|
||||
getAvailableConnectionMethods,
|
||||
getConnectableAppDefinition,
|
||||
type ConnectionIntentInteraction,
|
||||
type ConnectionIntentSetupOptions,
|
||||
type ConnectionRequestResult,
|
||||
type ConnectionsSearchResult,
|
||||
type ToolApplication,
|
||||
type ToolConnection,
|
||||
} from "@paperclipai/shared";
|
||||
import { conflict, forbidden, notFound, unprocessable } from "../errors.js";
|
||||
import type { RuntimeToolsTokenClaims } from "../runtime-tools-token.js";
|
||||
import { issueThreadInteractionService } from "./issue-thread-interactions.js";
|
||||
import { toolAccessService } from "./tool-access.js";
|
||||
|
||||
type DbTransaction = Parameters<Parameters<Db["transaction"]>[0]>[0];
|
||||
|
||||
function record(value: unknown): Record<string, unknown> | null {
|
||||
return value && typeof value === "object" && !Array.isArray(value)
|
||||
? value as Record<string, unknown>
|
||||
: null;
|
||||
}
|
||||
|
||||
function text(value: unknown) {
|
||||
return typeof value === "string" && value.trim() ? value.trim() : null;
|
||||
}
|
||||
|
||||
function sourceSlugForApplication(application: ToolApplication | undefined) {
|
||||
return text(application?.metadata?.sourceTemplateKey) ?? text(application?.metadata?.galleryKey);
|
||||
}
|
||||
|
||||
function sourceSlugForConnection(
|
||||
connection: ToolConnection,
|
||||
applications: ReadonlyMap<string, ToolApplication>,
|
||||
) {
|
||||
return text(connection.config?.sourceTemplateKey)
|
||||
?? text(connection.transportConfig?.sourceTemplateKey)
|
||||
?? sourceSlugForApplication(applications.get(connection.applicationId));
|
||||
}
|
||||
|
||||
function displayDescription(app: (typeof CONNECTABLE_APP_DEFINITIONS)[number]) {
|
||||
return text(app.description) ?? null;
|
||||
}
|
||||
|
||||
export function connectionIntentService(db: Db) {
|
||||
const interactions = issueThreadInteractionService(db);
|
||||
const access = toolAccessService(db);
|
||||
|
||||
async function assertCurrentUserWriteAccess(
|
||||
companyId: string,
|
||||
userId: string,
|
||||
bypassCurrentMembershipCheck = false,
|
||||
) {
|
||||
if (bypassCurrentMembershipCheck) return;
|
||||
const membership = await db
|
||||
.select({
|
||||
status: companyMemberships.status,
|
||||
membershipRole: companyMemberships.membershipRole,
|
||||
})
|
||||
.from(companyMemberships)
|
||||
.where(and(
|
||||
eq(companyMemberships.companyId, companyId),
|
||||
eq(companyMemberships.principalType, "user"),
|
||||
eq(companyMemberships.principalId, userId),
|
||||
))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!membership
|
||||
|| membership.status !== "active"
|
||||
|| !membership.membershipRole
|
||||
|| membership.membershipRole === "viewer"
|
||||
) {
|
||||
throw forbidden("Addressed user is no longer authorized for company write access");
|
||||
}
|
||||
}
|
||||
|
||||
async function lockCurrentUserWriteAccess(
|
||||
tx: DbTransaction,
|
||||
companyId: string,
|
||||
userId: string,
|
||||
bypassCurrentMembershipCheck = false,
|
||||
) {
|
||||
if (bypassCurrentMembershipCheck) return;
|
||||
const membership = await tx
|
||||
.select({
|
||||
status: companyMemberships.status,
|
||||
membershipRole: companyMemberships.membershipRole,
|
||||
})
|
||||
.from(companyMemberships)
|
||||
.where(and(
|
||||
eq(companyMemberships.companyId, companyId),
|
||||
eq(companyMemberships.principalType, "user"),
|
||||
eq(companyMemberships.principalId, userId),
|
||||
))
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!membership
|
||||
|| membership.status !== "active"
|
||||
|| !membership.membershipRole
|
||||
|| membership.membershipRole === "viewer"
|
||||
) {
|
||||
throw forbidden("Addressed user is no longer authorized for company write access");
|
||||
}
|
||||
}
|
||||
|
||||
async function loadRunContext(claims: RuntimeToolsTokenClaims) {
|
||||
const run = await db
|
||||
.select({
|
||||
id: heartbeatRuns.id,
|
||||
companyId: heartbeatRuns.companyId,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
status: heartbeatRuns.status,
|
||||
responsibleUserId: heartbeatRuns.responsibleUserId,
|
||||
contextSnapshot: heartbeatRuns.contextSnapshot,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, claims.run_id))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (
|
||||
!run
|
||||
|| run.companyId !== claims.company_id
|
||||
|| run.agentId !== claims.sub
|
||||
|| run.responsibleUserId !== claims.responsible_user_id
|
||||
) throw forbidden("Runtime tool token does not match its heartbeat run");
|
||||
if (run.status !== "running") throw forbidden("Runtime tool token is no longer active");
|
||||
const snapshot = record(run.contextSnapshot);
|
||||
const issueId = text(snapshot?.issueId) ?? text(snapshot?.taskId);
|
||||
if (!issueId) throw unprocessable("Connection requests require a task-bound heartbeat run");
|
||||
const [issue, agent, responsibleMembership] = await Promise.all([
|
||||
db.select({
|
||||
id: issues.id,
|
||||
companyId: issues.companyId,
|
||||
status: issues.status,
|
||||
assigneeAgentId: issues.assigneeAgentId,
|
||||
}).from(issues).where(and(eq(issues.id, issueId), eq(issues.companyId, run.companyId))).then((rows) => rows[0] ?? null),
|
||||
db.select({ id: agents.id, companyId: agents.companyId, name: agents.name })
|
||||
.from(agents)
|
||||
.where(and(eq(agents.id, run.agentId), eq(agents.companyId, run.companyId)))
|
||||
.then((rows) => rows[0] ?? null),
|
||||
db.select({
|
||||
status: companyMemberships.status,
|
||||
membershipRole: companyMemberships.membershipRole,
|
||||
}).from(companyMemberships).where(and(
|
||||
eq(companyMemberships.companyId, run.companyId),
|
||||
eq(companyMemberships.principalType, "user"),
|
||||
eq(companyMemberships.principalId, run.responsibleUserId!),
|
||||
)).then((rows) => rows[0] ?? null),
|
||||
]);
|
||||
if (!issue || !agent) throw notFound("Runtime task or agent was not found");
|
||||
if (
|
||||
!responsibleMembership
|
||||
|| responsibleMembership.status !== "active"
|
||||
|| !responsibleMembership.membershipRole
|
||||
|| responsibleMembership.membershipRole === "viewer"
|
||||
) {
|
||||
throw forbidden("Responsible user is no longer authorized for company write access");
|
||||
}
|
||||
if (issue.status === "done" || issue.status === "cancelled") {
|
||||
throw conflict("Connection requests cannot be created on a closed task");
|
||||
}
|
||||
return { run, issue, agent };
|
||||
}
|
||||
|
||||
async function connectionInventory(companyId: string) {
|
||||
const [applications, connections] = await Promise.all([
|
||||
access.listApplications(companyId),
|
||||
access.listConnections(companyId),
|
||||
]);
|
||||
return {
|
||||
applications,
|
||||
connections,
|
||||
applicationsById: new Map(applications.map((application) => [application.id, application] as const)),
|
||||
};
|
||||
}
|
||||
|
||||
async function usableConnectionForAgent(input: {
|
||||
companyId: string;
|
||||
agentId: string;
|
||||
responsibleUserId: string;
|
||||
serviceSlug: string;
|
||||
inventory?: Awaited<ReturnType<typeof connectionInventory>>;
|
||||
}) {
|
||||
const inventory = input.inventory ?? await connectionInventory(input.companyId);
|
||||
const matching = inventory.connections.filter((connection) =>
|
||||
sourceSlugForConnection(connection, inventory.applicationsById) === input.serviceSlug
|
||||
&& connection.status === "active"
|
||||
&& connection.enabled
|
||||
);
|
||||
if (matching.length === 0) return null;
|
||||
const effective = await access.getEffectiveProfilesForAgent(input.companyId, input.agentId);
|
||||
const installedIds = new Set(effective.installedConnections.map((connection) => connection.id));
|
||||
for (const connection of matching) {
|
||||
if (!installedIds.has(connection.id)) continue;
|
||||
const { grants } = await access.listConnectionGrants(connection.id, input.companyId);
|
||||
const usable = grants.some((grant) => {
|
||||
if (grant.status !== "active") return false;
|
||||
if (grant.kind === "organization") return true;
|
||||
return grant.subjectUserId === input.responsibleUserId
|
||||
&& grant.delegations.some((delegation) => delegation.agentId === input.agentId);
|
||||
});
|
||||
if (usable) return connection;
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
async function search(claims: RuntimeToolsTokenClaims, query: string): Promise<ConnectionsSearchResult> {
|
||||
const { run, agent } = await loadRunContext(claims);
|
||||
const normalized = query.trim().toLocaleLowerCase();
|
||||
const inventory = await connectionInventory(run.companyId);
|
||||
const results = await Promise.all(CONNECTABLE_APP_DEFINITIONS
|
||||
.filter((app) => !normalized
|
||||
|| app.slug.toLocaleLowerCase().includes(normalized)
|
||||
|| app.name.toLocaleLowerCase().includes(normalized)
|
||||
|| displayDescription(app)?.toLocaleLowerCase().includes(normalized))
|
||||
.map(async (app) => {
|
||||
const methods = getAvailableConnectionMethods(app);
|
||||
const matching = inventory.connections.filter((connection) =>
|
||||
sourceSlugForConnection(connection, inventory.applicationsById) === app.slug
|
||||
&& connection.status !== "archived"
|
||||
);
|
||||
const ready = await usableConnectionForAgent({
|
||||
companyId: run.companyId,
|
||||
agentId: agent.id,
|
||||
responsibleUserId: run.responsibleUserId!,
|
||||
serviceSlug: app.slug,
|
||||
inventory,
|
||||
});
|
||||
return {
|
||||
service: app.slug,
|
||||
name: app.name,
|
||||
description: displayDescription(app),
|
||||
logoUrl: app.branding.logoUrl ?? null,
|
||||
methods: methods.map((method) => ({
|
||||
key: method.key,
|
||||
label: method.label ?? method.key,
|
||||
auth: method.auth,
|
||||
})),
|
||||
state: app.availability?.available === false || methods.length === 0
|
||||
? "unavailable" as const
|
||||
: ready
|
||||
? "ready" as const
|
||||
: matching.length > 0
|
||||
? "needs_user_action" as const
|
||||
: "available" as const,
|
||||
connectionId: ready?.id ?? null,
|
||||
};
|
||||
}));
|
||||
return { version: 1, query, results };
|
||||
}
|
||||
|
||||
async function request(
|
||||
claims: RuntimeToolsTokenClaims,
|
||||
serviceSlug: string,
|
||||
): Promise<ConnectionRequestResult> {
|
||||
const context = await loadRunContext(claims);
|
||||
const app = getConnectableAppDefinition(serviceSlug);
|
||||
if (!app || app.availability?.available === false || getAvailableConnectionMethods(app).length === 0) {
|
||||
throw unprocessable(`Connection service ${serviceSlug} is not available`);
|
||||
}
|
||||
const ready = await usableConnectionForAgent({
|
||||
companyId: context.run.companyId,
|
||||
agentId: context.agent.id,
|
||||
responsibleUserId: context.run.responsibleUserId!,
|
||||
serviceSlug: app.slug,
|
||||
});
|
||||
if (ready) {
|
||||
return {
|
||||
version: 1,
|
||||
service: app.slug,
|
||||
state: "ready",
|
||||
connectionId: ready.id,
|
||||
interactionId: null,
|
||||
instruction: `${app.name} is connected and available in this run.`,
|
||||
};
|
||||
}
|
||||
const interaction = await interactions.createConnectionIntent(
|
||||
context.issue,
|
||||
{
|
||||
payload: {
|
||||
version: 1,
|
||||
serviceSlug: app.slug,
|
||||
serviceName: app.name,
|
||||
serviceLogoUrl: app.branding.logoUrl ?? null,
|
||||
serviceDarkLogoUrl: app.branding.darkLogoUrl ?? null,
|
||||
requestingAgentId: context.agent.id,
|
||||
requestingAgentName: context.agent.name,
|
||||
phase: "requested",
|
||||
},
|
||||
sourceRunId: context.run.id,
|
||||
addresseeUserId: context.run.responsibleUserId!,
|
||||
idempotencyKey: `connection-intent:${context.run.id}:${app.slug}`,
|
||||
},
|
||||
);
|
||||
return {
|
||||
version: 1,
|
||||
service: app.slug,
|
||||
state: "needs_user_action",
|
||||
connectionId: null,
|
||||
interactionId: interaction.id,
|
||||
instruction: `A connection card was sent to the responsible user. End this run and wait for continuation.`,
|
||||
};
|
||||
}
|
||||
|
||||
async function loadIntent(interactionId: string) {
|
||||
const row = await db
|
||||
.select({ interaction: issueThreadInteractions, issue: issues })
|
||||
.from(issueThreadInteractions)
|
||||
.innerJoin(issues, eq(issueThreadInteractions.issueId, issues.id))
|
||||
.where(eq(issueThreadInteractions.id, interactionId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!row || row.interaction.kind !== "connection_intent") throw notFound("Connection intent not found");
|
||||
const interaction = await interactions.getForIssue(row.issue, interactionId) as ConnectionIntentInteraction;
|
||||
return { ...row, interaction };
|
||||
}
|
||||
|
||||
async function setupOptions(interactionId: string): Promise<ConnectionIntentSetupOptions> {
|
||||
const loaded = await loadIntent(interactionId);
|
||||
const payload = connectionIntentPayloadSchema.parse(loaded.interaction.payload);
|
||||
const app = getConnectableAppDefinition(payload.serviceSlug);
|
||||
if (!app) throw notFound("Connection service is no longer available");
|
||||
const inventory = await connectionInventory(loaded.issue.companyId);
|
||||
const matchingConnections = inventory.connections.filter((connection) =>
|
||||
sourceSlugForConnection(connection, inventory.applicationsById) === app.slug
|
||||
&& connection.status === "active"
|
||||
&& connection.enabled
|
||||
);
|
||||
const existingConnections = (await Promise.all(matchingConnections.map(async (connection) => {
|
||||
const { grants } = await access.listConnectionGrants(connection.id, loaded.issue.companyId);
|
||||
const eligible = grants.some((grant) =>
|
||||
grant.status === "active"
|
||||
&& (grant.kind === "organization" || grant.subjectUserId === loaded.interaction.addresseeUserId)
|
||||
);
|
||||
return eligible ? connection : null;
|
||||
}))).filter((connection): connection is ToolConnection => connection !== null);
|
||||
return {
|
||||
version: 1,
|
||||
interaction: loaded.interaction,
|
||||
service: {
|
||||
service: app.slug,
|
||||
name: app.name,
|
||||
description: displayDescription(app),
|
||||
logoUrl: app.branding.logoUrl ?? null,
|
||||
methods: getAvailableConnectionMethods(app).map((method) => ({
|
||||
key: method.key,
|
||||
label: method.label ?? method.key,
|
||||
auth: method.auth,
|
||||
})),
|
||||
state: existingConnections.length > 0 ? "needs_user_action" : "available",
|
||||
connectionId: null,
|
||||
},
|
||||
existingConnections,
|
||||
requestedAgentId: payload.requestingAgentId,
|
||||
};
|
||||
}
|
||||
|
||||
async function complete(
|
||||
interactionId: string,
|
||||
connectionId: string,
|
||||
userId: string,
|
||||
options: {
|
||||
canManageOrganizationGrant?: boolean;
|
||||
bypassCurrentMembershipCheck?: boolean;
|
||||
} = {},
|
||||
) {
|
||||
const loaded = await loadIntent(interactionId);
|
||||
if (loaded.interaction.status !== "pending") throw conflict("Connection intent is already resolved");
|
||||
if (loaded.interaction.addresseeUserId !== userId) throw forbidden("Only the addressed user can connect this service");
|
||||
await assertCurrentUserWriteAccess(
|
||||
loaded.issue.companyId,
|
||||
userId,
|
||||
options.bypassCurrentMembershipCheck,
|
||||
);
|
||||
const payload = connectionIntentPayloadSchema.parse(loaded.interaction.payload);
|
||||
return db.transaction(async (tx) => {
|
||||
// Membership downgrade/removal takes the same row lock. Whichever side
|
||||
// commits first is authoritative: a completed revocation makes this
|
||||
// revalidation fail, while completion holds authority through OAuth
|
||||
// finalization, every install/delegation, and intent resolution.
|
||||
await lockCurrentUserWriteAccess(
|
||||
tx,
|
||||
loaded.issue.companyId,
|
||||
userId,
|
||||
options.bypassCurrentMembershipCheck,
|
||||
);
|
||||
const txDb = tx as unknown as Db;
|
||||
const txAccess = toolAccessService(txDb);
|
||||
const txInteractions = issueThreadInteractionService(txDb);
|
||||
let selectedConnection = await txAccess.getConnection(connectionId, loaded.issue.companyId);
|
||||
const selectedApplication = await txAccess.getApplication(
|
||||
selectedConnection.applicationId,
|
||||
loaded.issue.companyId,
|
||||
);
|
||||
if (sourceSlugForConnection(
|
||||
selectedConnection,
|
||||
new Map([[selectedApplication.id, selectedApplication]]),
|
||||
) !== payload.serviceSlug) {
|
||||
throw notFound("Connection does not match this intent");
|
||||
}
|
||||
if (selectedConnection.status !== "active" || !selectedConnection.enabled) {
|
||||
throw conflict("Finish and test this connection before using it for the task");
|
||||
}
|
||||
|
||||
let { grants } = await txAccess.listConnectionGrants(
|
||||
selectedConnection.id,
|
||||
loaded.issue.companyId,
|
||||
);
|
||||
const pendingPersonalGrant = grants.find((grant) =>
|
||||
grant.kind === "user" && grant.status === "active" && grant.subjectUserId === userId
|
||||
);
|
||||
if (selectedConnection.authKind === "oauth" && pendingPersonalGrant) {
|
||||
// txAccess is bound to the outer transaction. Its internal transactions
|
||||
// become savepoints, so activation, credential bindings, the all-agents
|
||||
// profile, and the company install roll back with any later failure.
|
||||
await txAccess.finalizeOAuthAccess(
|
||||
loaded.issue.companyId,
|
||||
selectedConnection.id,
|
||||
{ grantKind: "user" },
|
||||
{ actorType: "user", actorId: userId },
|
||||
);
|
||||
selectedConnection = await txAccess.getConnection(
|
||||
selectedConnection.id,
|
||||
loaded.issue.companyId,
|
||||
);
|
||||
({ grants } = await txAccess.listConnectionGrants(
|
||||
selectedConnection.id,
|
||||
loaded.issue.companyId,
|
||||
));
|
||||
}
|
||||
const personalGrant = grants.find((grant) =>
|
||||
grant.kind === "user" && grant.status === "active" && grant.subjectUserId === userId
|
||||
);
|
||||
const organizationGrant = grants.find((grant) =>
|
||||
grant.kind === "organization" && grant.status === "active"
|
||||
);
|
||||
if (!personalGrant && !organizationGrant) {
|
||||
throw conflict("This connection has no usable identity grant");
|
||||
}
|
||||
if (!personalGrant && !options.canManageOrganizationGrant) {
|
||||
throw forbidden("Sharing a company connection requires connection-management authority");
|
||||
}
|
||||
|
||||
if (personalGrant) {
|
||||
await txAccess.createConnectionGrantDelegation(
|
||||
selectedConnection.id,
|
||||
personalGrant.id,
|
||||
payload.requestingAgentId,
|
||||
userId,
|
||||
);
|
||||
}
|
||||
|
||||
const installs = await txAccess.listConnectionInstalls(
|
||||
selectedConnection.id,
|
||||
loaded.issue.companyId,
|
||||
);
|
||||
const requestedInstall = { targetType: "agent" as const, targetId: payload.requestingAgentId };
|
||||
const additiveInstalls = installs.some((install) =>
|
||||
install.targetType === requestedInstall.targetType && install.targetId === requestedInstall.targetId
|
||||
) ? installs : [...installs, requestedInstall];
|
||||
await txAccess.putConnectionInstalls(selectedConnection.id, { installs: additiveInstalls }, {
|
||||
actorType: "user",
|
||||
actorId: userId,
|
||||
});
|
||||
|
||||
return txInteractions.resolveConnectionIntent(
|
||||
loaded.issue,
|
||||
interactionId,
|
||||
{ version: 1, outcome: "connected", connectionId: selectedConnection.id },
|
||||
{ userId },
|
||||
);
|
||||
});
|
||||
}
|
||||
|
||||
async function decline(
|
||||
interactionId: string,
|
||||
userId: string,
|
||||
reason?: string,
|
||||
options: { bypassCurrentMembershipCheck?: boolean } = {},
|
||||
) {
|
||||
const loaded = await loadIntent(interactionId);
|
||||
if (loaded.interaction.addresseeUserId !== userId) throw forbidden("Only the addressed user can decline this request");
|
||||
await assertCurrentUserWriteAccess(
|
||||
loaded.issue.companyId,
|
||||
userId,
|
||||
options.bypassCurrentMembershipCheck,
|
||||
);
|
||||
return interactions.resolveConnectionIntent(
|
||||
loaded.issue,
|
||||
interactionId,
|
||||
{ version: 1, outcome: "declined", reason: reason?.trim() || null },
|
||||
{ userId },
|
||||
);
|
||||
}
|
||||
|
||||
return {
|
||||
validate: loadRunContext,
|
||||
search,
|
||||
request,
|
||||
loadIntent,
|
||||
setupOptions,
|
||||
complete,
|
||||
decline,
|
||||
updatePhase: async (
|
||||
interactionId: string,
|
||||
phase: "requested" | "authorizing" | "needs_retry",
|
||||
userId: string,
|
||||
options: { bypassCurrentMembershipCheck?: boolean } = {},
|
||||
) => {
|
||||
const loaded = await loadIntent(interactionId);
|
||||
if (loaded.interaction.addresseeUserId !== userId) throw forbidden("Only the addressed user can update this request");
|
||||
await assertCurrentUserWriteAccess(
|
||||
loaded.issue.companyId,
|
||||
userId,
|
||||
options.bypassCurrentMembershipCheck,
|
||||
);
|
||||
return interactions.updateConnectionIntentPhase(loaded.issue, interactionId, phase, { userId });
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -7,6 +7,8 @@ import { and, asc, desc, eq, getTableColumns, gt, gte, inArray, isNull, lt, lte,
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
AGENT_DEFAULT_MAX_CONCURRENT_RUNS,
|
||||
CONNECTION_INTENT_AGENT_GUIDANCE,
|
||||
CONNECTION_RUNTIME_TOOL_NAMES,
|
||||
ISSUE_CONTINUATION_SUMMARY_DOCUMENT_KEY,
|
||||
ISSUE_DISPOSITION_REPAIR_RETRY_REASON,
|
||||
MAX_TASK_DRAIN_TTL_MS,
|
||||
@@ -99,10 +101,12 @@ import type {
|
||||
AdapterRuntimeEvent,
|
||||
AdapterRuntimeMcpAccess,
|
||||
AdapterRuntimeMcpServer,
|
||||
AdapterRuntimeToolAccess,
|
||||
AdapterSessionCodec,
|
||||
UsageSummary,
|
||||
} from "../adapters/index.js";
|
||||
import { createLocalAgentJwt } from "../agent-auth-jwt.js";
|
||||
import { createRuntimeToolsToken } from "../runtime-tools-token.js";
|
||||
import { parseObject, asBoolean, asNumber, appendWithByteCap, MAX_EXCERPT_BYTES } from "../adapters/utils.js";
|
||||
import { costService } from "./costs.js";
|
||||
import { trackAgentFirstHeartbeat } from "@paperclipai/shared/telemetry";
|
||||
@@ -2749,6 +2753,10 @@ interface WakeupOptions {
|
||||
requestedByActorType?: "user" | "agent" | "system";
|
||||
requestedByActorId?: string | null;
|
||||
contextSnapshot?: Record<string, unknown>;
|
||||
issueStateGuard?: {
|
||||
statuses: string[];
|
||||
assigneeAgentId: string;
|
||||
};
|
||||
}
|
||||
|
||||
type UsageTotals = {
|
||||
@@ -3467,12 +3475,19 @@ type ManagedMcpGatewayRunConfig = {
|
||||
}>;
|
||||
};
|
||||
|
||||
function paperclipApiBaseUrl(): string {
|
||||
function configuredPaperclipApiBaseUrl(): string | null {
|
||||
const configured = readNonEmptyString(process.env.PAPERCLIP_API_URL);
|
||||
return configured
|
||||
? configured.replace(/\/+$/, "").replace(/\/api$/, "")
|
||||
: null;
|
||||
}
|
||||
|
||||
function paperclipApiBaseUrl(): string {
|
||||
const configured = configuredPaperclipApiBaseUrl();
|
||||
if (!configured) {
|
||||
throw new Error("PAPERCLIP_API_URL is required to deliver managed runtime MCP servers");
|
||||
}
|
||||
return configured.replace(/\/+$/, "").replace(/\/api$/, "");
|
||||
return configured;
|
||||
}
|
||||
|
||||
export async function revokeHeartbeatRunGatewayTokens(input: {
|
||||
@@ -3607,7 +3622,11 @@ export async function buildPaperclipRuntimeMcpServers(input: {
|
||||
});
|
||||
servers.push({
|
||||
name: connection.name,
|
||||
url: `${paperclipApiBaseUrl()}/api/tool-gateway/gateways/${gateway.id}/mcp`,
|
||||
// Runtime MCP clients authenticate with a short-lived gateway bearer, not
|
||||
// a Paperclip agent JWT. Route them through the public gateway protocol
|
||||
// endpoint mounted ahead of the API auth middleware; the gateway service
|
||||
// still validates the bearer and its run binding on every request.
|
||||
url: `${paperclipApiBaseUrl()}/mcp/gateways/${gateway.gatewayPublicId}`,
|
||||
token: token.token,
|
||||
connectionId: connection.id,
|
||||
});
|
||||
@@ -3633,6 +3652,40 @@ function createAdapterRuntimeMcpAccess(
|
||||
});
|
||||
}
|
||||
|
||||
function createAdapterRuntimeToolAccess(input: {
|
||||
agentId: string;
|
||||
companyId: string;
|
||||
runId: string;
|
||||
responsibleUserId: string | null;
|
||||
}): AdapterRuntimeToolAccess | undefined {
|
||||
if (!input.responsibleUserId) return undefined;
|
||||
const minted = createRuntimeToolsToken({
|
||||
agentId: input.agentId,
|
||||
companyId: input.companyId,
|
||||
runId: input.runId,
|
||||
responsibleUserId: input.responsibleUserId,
|
||||
});
|
||||
if (!minted) return undefined;
|
||||
// The normal server bootstrap always exports PAPERCLIP_API_URL. Some service
|
||||
// tests invoke heartbeat execution without booting an HTTP server, however;
|
||||
// in that context there is no reachable endpoint to advertise and runtime
|
||||
// tools should simply remain unavailable instead of failing the run.
|
||||
const baseUrl = configuredPaperclipApiBaseUrl();
|
||||
if (!baseUrl) return undefined;
|
||||
return Object.freeze({
|
||||
version: 1,
|
||||
guidance: CONNECTION_INTENT_AGENT_GUIDANCE,
|
||||
mcpEndpoint: `${baseUrl}/mcp/runtime-tools`,
|
||||
rest: {
|
||||
connectionsSearch: `${baseUrl}/runtime-tools/connections/search`,
|
||||
connectionRequest: `${baseUrl}/runtime-tools/connections/request`,
|
||||
},
|
||||
bearerToken: minted.token,
|
||||
expiresAt: minted.expiresAt,
|
||||
tools: CONNECTION_RUNTIME_TOOL_NAMES,
|
||||
});
|
||||
}
|
||||
|
||||
const MANAGED_MCP_LOCAL_ADAPTERS = new Set(["codex_local"]);
|
||||
|
||||
function adapterSupportsManagedMcpConfig(adapterType: string): boolean {
|
||||
@@ -3787,7 +3840,9 @@ export async function createManagedMcpRunConfig(input: {
|
||||
managedGateways.push({
|
||||
id: gateway.id,
|
||||
name: gateway.name,
|
||||
endpointPath: `/api/tool-gateway/gateways/${gateway.id}/mcp`,
|
||||
// This path must bypass the normal /api agent-JWT middleware. The MCP
|
||||
// gateway performs its own bearer validation for the run-scoped token.
|
||||
endpointPath: `/mcp/gateways/${gateway.gatewayPublicId}`,
|
||||
bearerToken: token.token,
|
||||
tokenPrefix: token.tokenPrefix,
|
||||
});
|
||||
@@ -6909,6 +6964,16 @@ export interface HeartbeatServiceOptions {
|
||||
pluginWorkerManager?: PluginWorkerManager;
|
||||
environmentRuntime?: HeartbeatEnvironmentRuntime;
|
||||
runtimeEnv?: Record<string, string | undefined>;
|
||||
/** Test seam for changing a continuation issue at the final pre-dispatch boundary. */
|
||||
beforeResolvedInteractionContinuationDispatchCheck?: (input: {
|
||||
runId: string;
|
||||
issueId: string;
|
||||
}) => Promise<void>;
|
||||
/** Test seam for racing an issue mutation after validation while its row lock is held. */
|
||||
afterResolvedInteractionContinuationDispatchCheck?: (input: {
|
||||
runId: string;
|
||||
issueId: string;
|
||||
}) => Promise<void>;
|
||||
}
|
||||
|
||||
type WorkspaceReadyCommentWriter = {
|
||||
@@ -12920,7 +12985,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
|
||||
const staleness = await evaluateQueuedRunStaleness(run, issueId, context);
|
||||
if (staleness.stale) {
|
||||
await cancelQueuedRunForStaleIssue(run, issueId, staleness);
|
||||
await cancelRunForStaleIssue(run, issueId, staleness);
|
||||
logger.info(
|
||||
{ runId: run.id, issueId, errorCode: staleness.errorCode },
|
||||
"claimQueuedRun: cancelled stale queued run",
|
||||
@@ -13212,8 +13277,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
run: typeof heartbeatRuns.$inferSelect,
|
||||
issueId: string,
|
||||
context: Record<string, unknown>,
|
||||
dbOrTx: Db = db,
|
||||
): Promise<QueuedRunStaleness> {
|
||||
const issue = await db
|
||||
const issue = await dbOrTx
|
||||
.select({
|
||||
id: issues.id,
|
||||
status: issues.status,
|
||||
@@ -13241,6 +13307,30 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
const retryReason = readNonEmptyString(context.retryReason) ?? run.scheduledRetryReason ?? null;
|
||||
const interactionResolvedAt = readNonEmptyString(context.interactionResolvedAt);
|
||||
const hasResolvedInteractionEvidence = interactionResolvedAt !== null && !Number.isNaN(Date.parse(interactionResolvedAt));
|
||||
const isResolvedInteractionContinuation = isResolvedInteractionContinuationWakeContext(context);
|
||||
|
||||
if (isResolvedInteractionContinuation && issue.status !== "in_progress") {
|
||||
return {
|
||||
stale: true,
|
||||
errorCode: "issue_not_in_progress",
|
||||
reason: `Cancelled because resolved-interaction continuation issue is no longer in_progress (current status: ${issue.status}) before the queued run could start`,
|
||||
details: { issueId, currentStatus: issue.status, requiredStatus: "in_progress" },
|
||||
};
|
||||
}
|
||||
|
||||
if (isResolvedInteractionContinuation && issue.assigneeAgentId !== run.agentId) {
|
||||
return {
|
||||
stale: true,
|
||||
errorCode: "issue_assignee_changed",
|
||||
reason:
|
||||
"Cancelled because resolved-interaction continuation issue changed assignee before the queued run could start",
|
||||
details: {
|
||||
issueId,
|
||||
previousAssigneeAgentId: run.agentId,
|
||||
currentAssigneeAgentId: issue.assigneeAgentId,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
if (
|
||||
issue.status === "in_progress" &&
|
||||
@@ -13254,7 +13344,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
readNonEmptyString(parseObject(queuedWake.continuationSummary).body);
|
||||
const currentContinuationSummary = queuedContinuationSummary
|
||||
? null
|
||||
: await getIssueContinuationSummaryDocument(db, issueId);
|
||||
: await getIssueContinuationSummaryDocument(dbOrTx, issueId);
|
||||
const continuationSummaryBody = queuedContinuationSummary ?? currentContinuationSummary?.body ?? null;
|
||||
if (continuationSummaryParksExecutor(continuationSummaryBody)) {
|
||||
return {
|
||||
@@ -13281,7 +13371,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
|
||||
const recoveryActionId = readNonEmptyString(context.recoveryActionId);
|
||||
const authorizedSourceScopedRecovery = wakeReason === "source_scoped_recovery_action" && recoveryActionId
|
||||
? await db
|
||||
? await dbOrTx
|
||||
.select({ id: issueRecoveryActions.id })
|
||||
.from(issueRecoveryActions)
|
||||
.where(and(
|
||||
@@ -13373,7 +13463,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
return { stale: false };
|
||||
}
|
||||
|
||||
async function cancelQueuedRunForStaleIssue(
|
||||
async function cancelRunForStaleIssue(
|
||||
run: typeof heartbeatRuns.$inferSelect,
|
||||
issueId: string,
|
||||
staleness: Extract<QueuedRunStaleness, { stale: true }>,
|
||||
@@ -14560,9 +14650,29 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
const issueDependencyReadiness = issueId
|
||||
? await issuesSvc.listDependencyReadiness(agent.companyId, [issueId]).then((rows) => rows.get(issueId) ?? null)
|
||||
: null;
|
||||
if (issueId && issueContext && isResolvedInteractionContinuationWakeContext(context)) {
|
||||
try {
|
||||
// Claim the issue under the same in_progress predicate used by the
|
||||
// queued-run staleness gate. This is the final atomic guard before
|
||||
// dispatch: an operator parking the issue after claim but before this
|
||||
// checkout must not be overwritten by the continuation.
|
||||
await issuesSvc.checkout(issueId, agent.id, ["in_progress"], run.id);
|
||||
context[PAPERCLIP_HARNESS_CHECKOUT_KEY] = true;
|
||||
} catch (error) {
|
||||
if (!isCheckoutConflictError(error)) throw error;
|
||||
const staleness = await evaluateQueuedRunStaleness(run, issueId, context);
|
||||
if (staleness.stale) {
|
||||
await cancelRunForStaleIssue(run, issueId, staleness);
|
||||
return;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
issueContext = await getIssueExecutionContext(agent.companyId, issueId);
|
||||
}
|
||||
if (
|
||||
issueId &&
|
||||
issueContext &&
|
||||
!isResolvedInteractionContinuationWakeContext(context) &&
|
||||
shouldAutoCheckoutIssueForWake({
|
||||
contextSnapshot: context,
|
||||
issueStatus: issueContext.status,
|
||||
@@ -16375,6 +16485,76 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
executionTarget,
|
||||
});
|
||||
const adapter = getServerAdapter(agent.adapterType);
|
||||
const dispatchResolvedInteractionContinuationWithAtomicGate = async <T>(
|
||||
dispatch: (markDispatchStarted: () => void) => Promise<T>,
|
||||
): Promise<
|
||||
| { dispatched: true; resultPromise: Promise<T> }
|
||||
| { dispatched: false }
|
||||
> => {
|
||||
if (!issueId || !isResolvedInteractionContinuationWakeContext(context)) {
|
||||
return { dispatched: true, resultPromise: dispatch(() => {}) };
|
||||
}
|
||||
await options.beforeResolvedInteractionContinuationDispatchCheck?.({ runId: run.id, issueId });
|
||||
|
||||
const gate = await db.transaction(async (tx) => {
|
||||
const lockedIssue = await tx
|
||||
.select({ executionRunId: issues.executionRunId })
|
||||
.from(issues)
|
||||
.where(and(eq(issues.id, issueId), eq(issues.companyId, run.companyId)))
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const staleness = await evaluateQueuedRunStaleness(
|
||||
run,
|
||||
issueId,
|
||||
context,
|
||||
tx as unknown as Db,
|
||||
);
|
||||
if (staleness.stale) {
|
||||
return { dispatched: false as const, staleness };
|
||||
}
|
||||
if (lockedIssue?.executionRunId !== run.id) {
|
||||
return {
|
||||
dispatched: false as const,
|
||||
staleness: {
|
||||
stale: true as const,
|
||||
errorCode: "issue_execution_lock_changed" as const,
|
||||
reason:
|
||||
"Cancelled because resolved-interaction continuation no longer owns the issue execution lock before adapter dispatch",
|
||||
details: {
|
||||
issueId,
|
||||
expectedExecutionRunId: run.id,
|
||||
currentExecutionRunId: lockedIssue?.executionRunId ?? null,
|
||||
},
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
await options.afterResolvedInteractionContinuationDispatchCheck?.({ runId: run.id, issueId });
|
||||
let dispatchStarted = false;
|
||||
let resolveDispatchStarted!: () => void;
|
||||
const dispatchStartedPromise = new Promise<void>((resolve) => {
|
||||
resolveDispatchStarted = resolve;
|
||||
});
|
||||
const markDispatchStarted = () => {
|
||||
if (dispatchStarted) return;
|
||||
dispatchStarted = true;
|
||||
resolveDispatchStarted();
|
||||
};
|
||||
|
||||
// Keep the issue row locked through the adapter's asynchronous
|
||||
// preparation and release it only once the adapter reports an
|
||||
// actual process spawn. If preparation fails or returns without a
|
||||
// spawn, settling the adapter promise also releases the gate.
|
||||
const resultPromise = dispatch(markDispatchStarted);
|
||||
void resultPromise.then(markDispatchStarted, markDispatchStarted);
|
||||
await dispatchStartedPromise;
|
||||
return { dispatched: true as const, resultPromise };
|
||||
});
|
||||
|
||||
if (gate.dispatched) return gate;
|
||||
await cancelRunForStaleIssue(run, issueId, gate.staleness);
|
||||
return { dispatched: false };
|
||||
};
|
||||
const localAgentJwtScope =
|
||||
issueRef?.workMode === "skill_test"
|
||||
? { kind: "skill_test" as const, issueId: issueRef.id }
|
||||
@@ -16617,7 +16797,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
promptMetrics: { promptChars: prompt.length },
|
||||
context: { provider: "codex", protocolVersion: 1 },
|
||||
});
|
||||
adapterResult = await executeNativeCodexRunner({
|
||||
const guardedDispatch = await dispatchResolvedInteractionContinuationWithAtomicGate((markDispatchStarted) =>
|
||||
executeNativeCodexRunner({
|
||||
db,
|
||||
companyId: agent.companyId,
|
||||
issueId: issueRef.id,
|
||||
@@ -16636,16 +16817,50 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
timeoutMs,
|
||||
environment,
|
||||
onLog,
|
||||
onSpawn,
|
||||
});
|
||||
onSpawn: async (meta) => {
|
||||
markDispatchStarted();
|
||||
await onSpawn(meta);
|
||||
},
|
||||
}),
|
||||
);
|
||||
if (!guardedDispatch.dispatched) return;
|
||||
adapterResult = await guardedDispatch.resultPromise;
|
||||
} else {
|
||||
const adapterContext = { ...context };
|
||||
const runtimeTools = createAdapterRuntimeToolAccess({
|
||||
agentId: agent.id,
|
||||
companyId: agent.companyId,
|
||||
runId: run.id,
|
||||
responsibleUserId: run.responsibleUserId,
|
||||
});
|
||||
if (!runtimeTools) {
|
||||
logger.warn(
|
||||
{
|
||||
companyId: agent.companyId,
|
||||
agentId: agent.id,
|
||||
runId: run.id,
|
||||
},
|
||||
"runtime connection tools could not be delivered",
|
||||
);
|
||||
}
|
||||
const runtimeMcpServers = await buildPaperclipRuntimeMcpServers({
|
||||
db,
|
||||
agent,
|
||||
runId: run.id,
|
||||
});
|
||||
const runtimeToolDelivery = adapter.runtimeToolDelivery ?? "invocation_context";
|
||||
if (runtimeTools && runtimeToolDelivery === "native_mcp") {
|
||||
runtimeMcpServers.unshift({
|
||||
name: "Paperclip connections",
|
||||
url: runtimeTools.mcpEndpoint,
|
||||
token: runtimeTools.bearerToken,
|
||||
connectionId: "paperclip-runtime-tools",
|
||||
});
|
||||
}
|
||||
const runtimeMcp = createAdapterRuntimeMcpAccess(runtimeMcpServers);
|
||||
if (runtimeTools && runtimeToolDelivery === "invocation_context") {
|
||||
adapterContext.paperclipRuntimeTools = runtimeTools;
|
||||
}
|
||||
const managedMcpConfig = await createManagedMcpRunConfig({
|
||||
db,
|
||||
agent,
|
||||
@@ -16657,7 +16872,8 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
if (managedMcpConfig) {
|
||||
adapterContext.paperclipManagedMcp = managedMcpConfig;
|
||||
}
|
||||
adapterResult = await adapter.execute({
|
||||
const guardedDispatch = await dispatchResolvedInteractionContinuationWithAtomicGate((markDispatchStarted) =>
|
||||
adapter.execute({
|
||||
runId: run.id,
|
||||
agent,
|
||||
runtime: runtimeForAdapter,
|
||||
@@ -16669,6 +16885,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
? { remoteExecution: remoteExecution as unknown as Record<string, unknown> }
|
||||
: undefined,
|
||||
runtimeMcp,
|
||||
runtimeTools,
|
||||
onLog,
|
||||
onMeta: onAdapterMeta,
|
||||
onEvent: onAdapterEvent,
|
||||
@@ -16680,9 +16897,16 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
onRuntimeProgress: async (progress) => {
|
||||
await recordCurrentHeartbeatRunRuntimeProgress(run, progress, issueId);
|
||||
},
|
||||
onSpawn,
|
||||
onDispatch: markDispatchStarted,
|
||||
onSpawn: async (meta) => {
|
||||
markDispatchStarted();
|
||||
await onSpawn(meta);
|
||||
},
|
||||
authToken: authToken ?? undefined,
|
||||
});
|
||||
}),
|
||||
);
|
||||
if (!guardedDispatch.dispatched) return;
|
||||
adapterResult = await guardedDispatch.resultPromise;
|
||||
}
|
||||
// Adapter returned cleanly, which means its workspace-restore finally
|
||||
// block also ran without throwing. Record the workspace_finalize
|
||||
@@ -18603,6 +18827,40 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
return { kind: "skipped" as const };
|
||||
}
|
||||
|
||||
const issueStateGuard = opts.issueStateGuard;
|
||||
if (
|
||||
issueStateGuard
|
||||
&& (
|
||||
!issueStateGuard.statuses.includes(issue.status)
|
||||
|| issue.assigneeAgentId !== issueStateGuard.assigneeAgentId
|
||||
)
|
||||
) {
|
||||
await tx.insert(agentWakeupRequests).values({
|
||||
companyId: agent.companyId,
|
||||
agentId,
|
||||
source,
|
||||
triggerDetail,
|
||||
reason: "issue_state_guard_mismatch",
|
||||
payload: {
|
||||
...(payload ?? {}),
|
||||
heartbeatSkip: {
|
||||
reason: "Issue status or assignee changed before the wake could be queued.",
|
||||
issueId: issue.id,
|
||||
expectedStatuses: issueStateGuard.statuses,
|
||||
actualStatus: issue.status,
|
||||
expectedAssigneeAgentId: issueStateGuard.assigneeAgentId,
|
||||
actualAssigneeAgentId: issue.assigneeAgentId,
|
||||
},
|
||||
},
|
||||
status: "skipped",
|
||||
requestedByActorType: opts.requestedByActorType ?? null,
|
||||
requestedByActorId: opts.requestedByActorId ?? null,
|
||||
idempotencyKey: opts.idempotencyKey ?? null,
|
||||
finishedAt: new Date(),
|
||||
});
|
||||
return { kind: "skipped" as const };
|
||||
}
|
||||
|
||||
if (worktreeExecutionCutoff && issue.createdAt < worktreeExecutionCutoff) {
|
||||
await tx.insert(agentWakeupRequests).values({
|
||||
companyId: agent.companyId,
|
||||
|
||||
@@ -13,6 +13,7 @@ import {
|
||||
issueThreadInteractions,
|
||||
issues,
|
||||
toolActionRequests,
|
||||
toolOauthStates,
|
||||
} from "@paperclipai/db";
|
||||
import { trackInteractionCreated, trackInteractionResolved } from "@paperclipai/shared/telemetry";
|
||||
import type {
|
||||
@@ -20,6 +21,7 @@ import type {
|
||||
AskUserQuestionsAnswer,
|
||||
AskUserQuestionsInteraction,
|
||||
CancelIssueThreadInteraction,
|
||||
ConnectionIntentInteraction,
|
||||
CreateIssueThreadInteraction,
|
||||
InteractionResolverGovernance,
|
||||
IssueReviewPolicy,
|
||||
@@ -47,6 +49,8 @@ import {
|
||||
askUserQuestionsPayloadSchema,
|
||||
askUserQuestionsResultSchema,
|
||||
cancelIssueThreadInteractionSchema,
|
||||
connectionIntentPayloadSchema,
|
||||
connectionIntentResultSchema,
|
||||
createIssueThreadInteractionSchema,
|
||||
legacyIssueThreadInteractionResolverPolicyAlias,
|
||||
normalizeIssueThreadInteractionResolverPolicy,
|
||||
@@ -254,6 +258,7 @@ export const DEFAULT_RESOLVER_POLICY_BY_KIND: Record<
|
||||
request_confirmation: "anyone",
|
||||
request_checkbox_confirmation: "anyone",
|
||||
request_item_verdicts: "anyone",
|
||||
connection_intent: "human_only",
|
||||
};
|
||||
|
||||
const RESOLVER_POLICY_RESTRICTION_RANK: Record<IssueThreadInteractionCanonicalResolverPolicy, number> = {
|
||||
@@ -387,11 +392,13 @@ type TargetBoundInteraction =
|
||||
const USER_COMMENT_SUPERSEDABLE_INTERACTION_KINDS = [
|
||||
...TARGET_BOUND_INTERACTION_KINDS,
|
||||
"ask_user_questions",
|
||||
"connection_intent",
|
||||
] as const;
|
||||
type UserCommentSupersedableKind = (typeof USER_COMMENT_SUPERSEDABLE_INTERACTION_KINDS)[number];
|
||||
type UserCommentSupersedableInteraction =
|
||||
| TargetBoundInteraction
|
||||
| AskUserQuestionsInteraction;
|
||||
| AskUserQuestionsInteraction
|
||||
| ConnectionIntentInteraction;
|
||||
|
||||
function isRequestConfirmationLikeKind(kind: string): kind is RequestConfirmationLikeKind {
|
||||
return (REQUEST_CONFIRMATION_INTERACTION_KINDS as readonly string[]).includes(kind);
|
||||
@@ -529,6 +536,13 @@ function hydrateInteraction(
|
||||
payload: requestItemVerdictsPayloadSchema.parse(row.payload),
|
||||
result: parseStoredInteractionResult(requestItemVerdictsResultSchema, row.result, row),
|
||||
} satisfies RequestItemVerdictsInteraction;
|
||||
case "connection_intent":
|
||||
return {
|
||||
...base,
|
||||
kind: "connection_intent",
|
||||
payload: connectionIntentPayloadSchema.parse(row.payload),
|
||||
result: parseStoredInteractionResult(connectionIntentResultSchema, row.result, row),
|
||||
} satisfies ConnectionIntentInteraction;
|
||||
default:
|
||||
throw unprocessable(`Unknown interaction kind: ${row.kind}`);
|
||||
}
|
||||
@@ -608,6 +622,7 @@ function shouldReturnAcceptedConfirmationToCreatorAgent(args: {
|
||||
}
|
||||
|
||||
function shouldSupersedeInteractionOnUserComment(interaction: UserCommentSupersedableInteraction) {
|
||||
if (interaction.kind === "connection_intent") return true;
|
||||
return interaction.payload.supersedeOnUserComment === true;
|
||||
}
|
||||
|
||||
@@ -651,6 +666,13 @@ function normalizeCreateInteractionInput(input: CreateIssueThreadInteraction): C
|
||||
}
|
||||
|
||||
function buildSupersededByCommentResult(row: IssueThreadInteractionRow, commentId: string) {
|
||||
if (row.kind === "connection_intent") {
|
||||
return {
|
||||
version: 1,
|
||||
outcome: "expired",
|
||||
reason: "Superseded by a newer user comment",
|
||||
} as const;
|
||||
}
|
||||
if (row.kind === "ask_user_questions") {
|
||||
return {
|
||||
version: 1,
|
||||
@@ -729,6 +751,9 @@ function buildAdministrativeOutcomeResult(
|
||||
outcome: "withdrawn" | "issue_closed" | "addressee_deleted",
|
||||
reason: string | null = null,
|
||||
) {
|
||||
if (row.kind === "connection_intent") {
|
||||
return { version: 1, outcome: "expired", reason } as const;
|
||||
}
|
||||
if (row.kind === "ask_user_questions") {
|
||||
return { version: 1, outcome, reason, answers: [], summaryMarkdown: null } as const;
|
||||
}
|
||||
@@ -909,6 +934,9 @@ function deriveResolutionReason(interaction: IssueThreadInteraction) {
|
||||
case "cancelled":
|
||||
return "cancelled";
|
||||
case "expired": {
|
||||
if (interaction.kind === "connection_intent") {
|
||||
return interaction.result?.outcome ?? "expired";
|
||||
}
|
||||
if (interaction.kind === "ask_user_questions") {
|
||||
return interaction.result?.expirationReason ?? "expired";
|
||||
}
|
||||
@@ -1007,7 +1035,7 @@ async function emitInteractionResolvedTelemetry(
|
||||
: undefined;
|
||||
|
||||
trackInteractionResolved(telemetryClient, {
|
||||
interactionKind: interaction.kind,
|
||||
interactionKind: interaction.kind === "connection_intent" ? "other" : interaction.kind,
|
||||
status: interaction.status,
|
||||
resolvedByKind: resolveActorKind(interaction),
|
||||
resolutionReason: deriveResolutionReason(interaction),
|
||||
@@ -1034,7 +1062,10 @@ function emitInteractionCreatedTelemetry(args: {
|
||||
if (!telemetryClient) return;
|
||||
|
||||
try {
|
||||
trackInteractionCreated(telemetryClient, args);
|
||||
trackInteractionCreated(telemetryClient, {
|
||||
...args,
|
||||
interactionKind: args.interactionKind === "connection_intent" ? "other" : args.interactionKind,
|
||||
});
|
||||
} catch (error) {
|
||||
console.error("[paperclip] Failed to emit interaction.created telemetry", error);
|
||||
}
|
||||
@@ -1853,6 +1884,181 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti
|
||||
|
||||
return {
|
||||
getForIssue,
|
||||
createConnectionIntent: async (
|
||||
issue: { id: string; companyId: string },
|
||||
input: {
|
||||
payload: ConnectionIntentInteraction["payload"];
|
||||
sourceRunId: string;
|
||||
addresseeUserId: string;
|
||||
idempotencyKey: string;
|
||||
},
|
||||
) => {
|
||||
const payload = connectionIntentPayloadSchema.parse(input.payload);
|
||||
const existing = await getIdempotentInteraction({
|
||||
issueId: issue.id,
|
||||
companyId: issue.companyId,
|
||||
idempotencyKey: input.idempotencyKey,
|
||||
});
|
||||
if (existing) {
|
||||
if (
|
||||
existing.kind !== "connection_intent"
|
||||
|| existing.sourceRunId !== input.sourceRunId
|
||||
|| existing.addresseeUserId !== input.addresseeUserId
|
||||
|| !isDeepStrictEqual(existing.payload, payload)
|
||||
) {
|
||||
throw conflict("Interaction idempotency key already exists for a different request", {
|
||||
idempotencyKey: input.idempotencyKey,
|
||||
});
|
||||
}
|
||||
return hydrateInteraction(existing) as ConnectionIntentInteraction;
|
||||
}
|
||||
|
||||
const created = await db.transaction(async (tx) => {
|
||||
const issueRow = await tx
|
||||
.select({ status: issues.status })
|
||||
.from(issues)
|
||||
.where(and(eq(issues.id, issue.id), eq(issues.companyId, issue.companyId)))
|
||||
.for("update")
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!issueRow || isTerminalIssueStatus(issueRow.status)) {
|
||||
throw conflict("Cannot create an interaction on a closed issue");
|
||||
}
|
||||
|
||||
const [row] = await tx
|
||||
.insert(issueThreadInteractions)
|
||||
.values({
|
||||
companyId: issue.companyId,
|
||||
issueId: issue.id,
|
||||
kind: "connection_intent",
|
||||
status: "pending",
|
||||
continuationPolicy: "wake_assignee",
|
||||
requestedResolverPolicy: "human_only",
|
||||
effectiveResolverPolicy: "human_only",
|
||||
resolverPolicyProvenance: "explicit",
|
||||
effectiveResolverPolicySource: "governed_action",
|
||||
idempotencyKey: input.idempotencyKey,
|
||||
sourceRunId: input.sourceRunId,
|
||||
title: `Connect ${payload.serviceName}`,
|
||||
summary: `${payload.requestingAgentName} needs this connection to continue.`,
|
||||
createdByAgentId: payload.requestingAgentId,
|
||||
addresseeUserId: input.addresseeUserId,
|
||||
payload,
|
||||
})
|
||||
.returning();
|
||||
|
||||
const olderPending = await tx
|
||||
.select()
|
||||
.from(issueThreadInteractions)
|
||||
.where(and(
|
||||
eq(issueThreadInteractions.companyId, issue.companyId),
|
||||
eq(issueThreadInteractions.issueId, issue.id),
|
||||
eq(issueThreadInteractions.kind, "connection_intent"),
|
||||
eq(issueThreadInteractions.createdByAgentId, payload.requestingAgentId),
|
||||
eq(issueThreadInteractions.status, "pending"),
|
||||
ne(issueThreadInteractions.id, row.id),
|
||||
));
|
||||
const supersededIds = olderPending
|
||||
.filter((candidate) => {
|
||||
const candidatePayload = connectionIntentPayloadSchema.safeParse(candidate.payload);
|
||||
return candidatePayload.success && candidatePayload.data.serviceSlug === payload.serviceSlug;
|
||||
})
|
||||
.map((candidate) => candidate.id);
|
||||
if (supersededIds.length > 0) {
|
||||
await tx
|
||||
.delete(toolOauthStates)
|
||||
.where(inArray(toolOauthStates.interactionId, supersededIds));
|
||||
await tx
|
||||
.update(issueThreadInteractions)
|
||||
.set({
|
||||
status: "expired",
|
||||
result: {
|
||||
version: 1,
|
||||
outcome: "superseded",
|
||||
supersededByInteractionId: row.id,
|
||||
},
|
||||
resolvedAt: now(),
|
||||
updatedAt: now(),
|
||||
})
|
||||
.where(inArray(issueThreadInteractions.id, supersededIds));
|
||||
}
|
||||
await touchIssue(tx, issue.id);
|
||||
return row;
|
||||
});
|
||||
|
||||
const interaction = hydrateInteraction(created) as ConnectionIntentInteraction;
|
||||
emitInteractionCreatedTelemetry({
|
||||
interactionKind: "connection_intent",
|
||||
usedDeprecatedResolverPolicyAlias: false,
|
||||
});
|
||||
return interaction;
|
||||
},
|
||||
updateConnectionIntentPhase: async (
|
||||
issue: { id: string; companyId: string },
|
||||
interactionId: string,
|
||||
phase: ConnectionIntentInteraction["payload"]["phase"],
|
||||
actor: InteractionActor,
|
||||
) => {
|
||||
const current = await getPendingInteractionForResolution({ issue, interactionId });
|
||||
if (current.kind !== "connection_intent") {
|
||||
throw unprocessable("Only connection_intent interactions have a connection phase");
|
||||
}
|
||||
assertInteractionResolutionAllowed(current, actor);
|
||||
const payload = connectionIntentPayloadSchema.parse(current.payload);
|
||||
const [updated] = await db
|
||||
.update(issueThreadInteractions)
|
||||
.set({ payload: { ...payload, phase }, updatedAt: now() })
|
||||
.where(and(
|
||||
eq(issueThreadInteractions.id, interactionId),
|
||||
eq(issueThreadInteractions.companyId, issue.companyId),
|
||||
eq(issueThreadInteractions.issueId, issue.id),
|
||||
eq(issueThreadInteractions.status, "pending"),
|
||||
))
|
||||
.returning();
|
||||
if (!updated) throw interactionAlreadyResolvedError();
|
||||
await touchIssue(db, issue.id);
|
||||
return hydrateInteraction(updated) as ConnectionIntentInteraction;
|
||||
},
|
||||
resolveConnectionIntent: async (
|
||||
issue: { id: string; companyId: string },
|
||||
interactionId: string,
|
||||
resultInput: ConnectionIntentInteraction["result"] extends infer T ? NonNullable<T> : never,
|
||||
actor: InteractionActor,
|
||||
) => {
|
||||
const result = connectionIntentResultSchema.parse(resultInput);
|
||||
const current = await getPendingInteractionForResolution({ issue, interactionId });
|
||||
if (current.kind !== "connection_intent") {
|
||||
throw unprocessable("Only connection_intent interactions can be resolved by this operation");
|
||||
}
|
||||
if (!actor.userId) throw forbidden("Connection intents require a human resolver");
|
||||
assertInteractionResolutionAllowed(current, actor);
|
||||
const status = result.outcome === "connected"
|
||||
? "accepted"
|
||||
: result.outcome === "declined"
|
||||
? "rejected"
|
||||
: "expired";
|
||||
const resolvedAt = now();
|
||||
const [updated] = await db
|
||||
.update(issueThreadInteractions)
|
||||
.set({
|
||||
status,
|
||||
result,
|
||||
resolvedByUserId: actor.userId,
|
||||
resolvedAt,
|
||||
updatedAt: resolvedAt,
|
||||
})
|
||||
.where(and(
|
||||
eq(issueThreadInteractions.id, interactionId),
|
||||
eq(issueThreadInteractions.companyId, issue.companyId),
|
||||
eq(issueThreadInteractions.issueId, issue.id),
|
||||
eq(issueThreadInteractions.status, "pending"),
|
||||
))
|
||||
.returning();
|
||||
if (!updated) throw interactionAlreadyResolvedError();
|
||||
await touchIssue(db, issue.id);
|
||||
const interaction = hydrateInteraction(updated) as ConnectionIntentInteraction;
|
||||
await emitInteractionResolvedTelemetry(db, interaction);
|
||||
return interaction;
|
||||
},
|
||||
sweepMergedPullRequestConfirmations: async () => {
|
||||
const rows = await db
|
||||
.select({
|
||||
@@ -2928,21 +3134,26 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti
|
||||
const now = new Date();
|
||||
const expired: IssueThreadInteraction[] = [];
|
||||
for (const row of superseded) {
|
||||
const [updated] = await db
|
||||
.update(issueThreadInteractions)
|
||||
.set({
|
||||
status: "expired",
|
||||
result: buildSupersededByCommentResult(row, comment.id),
|
||||
resolvedByAgentId: actor.agentId ?? null,
|
||||
resolvedByUserId: actor.userId ?? null,
|
||||
resolvedAt: now,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(and(
|
||||
eq(issueThreadInteractions.id, row.id),
|
||||
eq(issueThreadInteractions.status, "pending"),
|
||||
))
|
||||
.returning();
|
||||
const [updated] = await db.transaction(async (tx) => {
|
||||
if (row.kind === "connection_intent") {
|
||||
await tx.delete(toolOauthStates).where(eq(toolOauthStates.interactionId, row.id));
|
||||
}
|
||||
return tx
|
||||
.update(issueThreadInteractions)
|
||||
.set({
|
||||
status: "expired",
|
||||
result: buildSupersededByCommentResult(row, comment.id),
|
||||
resolvedByAgentId: actor.agentId ?? null,
|
||||
resolvedByUserId: actor.userId ?? null,
|
||||
resolvedAt: now,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(and(
|
||||
eq(issueThreadInteractions.id, row.id),
|
||||
eq(issueThreadInteractions.status, "pending"),
|
||||
))
|
||||
.returning();
|
||||
});
|
||||
if (updated) expired.push(hydrateInteraction(updated));
|
||||
}
|
||||
|
||||
@@ -3025,6 +3236,8 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti
|
||||
.map((row) => row.id);
|
||||
const itemVerdictRows = commentRows
|
||||
.filter((row) => row.kind === "request_item_verdicts");
|
||||
const connectionIntentRows = commentRows
|
||||
.filter((row) => row.kind === "connection_intent");
|
||||
|
||||
if (questionRowIds.length > 0) {
|
||||
const sampleQuestionRow = commentRows.find((row) => row.kind === "ask_user_questions");
|
||||
@@ -3086,6 +3299,28 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti
|
||||
.returning();
|
||||
if (updated) expired.push(hydrateInteraction(updated));
|
||||
}
|
||||
|
||||
for (const row of connectionIntentRows) {
|
||||
const [updated] = await db.transaction(async (tx) => {
|
||||
await tx.delete(toolOauthStates).where(eq(toolOauthStates.interactionId, row.id));
|
||||
return tx
|
||||
.update(issueThreadInteractions)
|
||||
.set({
|
||||
status: "expired",
|
||||
result: buildSupersededByCommentResult(row, comment.id),
|
||||
resolvedByAgentId: null,
|
||||
resolvedByUserId: comment.authorUserId,
|
||||
resolvedAt: now,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(and(
|
||||
eq(issueThreadInteractions.id, row.id),
|
||||
eq(issueThreadInteractions.status, "pending"),
|
||||
))
|
||||
.returning();
|
||||
});
|
||||
if (updated) expired.push(hydrateInteraction(updated));
|
||||
}
|
||||
}
|
||||
|
||||
if (expired.length > 0) {
|
||||
@@ -3194,6 +3429,11 @@ export function issueThreadInteractionService(db: Db, opts: IssueThreadInteracti
|
||||
// expires and the execution result lands on it via the gateway's
|
||||
// lifecycle reflection.
|
||||
const updated = await db.transaction(async (tx) => {
|
||||
if (row.kind === "connection_intent") {
|
||||
await tx
|
||||
.delete(toolOauthStates)
|
||||
.where(eq(toolOauthStates.interactionId, row.id));
|
||||
}
|
||||
await resolveLinkedToolActionRequests(tx, row, {
|
||||
status: "expired",
|
||||
fromStatuses: ["pending", "approved"],
|
||||
|
||||
@@ -918,7 +918,7 @@ function assertSelectableProviderConfig(config: {
|
||||
}
|
||||
}
|
||||
|
||||
export function secretService(db: Db) {
|
||||
export function secretService(db: Db | DbTransaction) {
|
||||
const authorization = authorizationService(db);
|
||||
|
||||
type NormalizeEnvOptions = {
|
||||
@@ -1629,7 +1629,7 @@ export function secretService(db: Db) {
|
||||
accessContext,
|
||||
});
|
||||
await context.registerForRedaction(resolution.value);
|
||||
await logActivity(db, {
|
||||
await logActivity(db as Db, {
|
||||
companyId,
|
||||
actorType: "agent",
|
||||
actorId: context.agentId,
|
||||
@@ -1651,7 +1651,7 @@ export function secretService(db: Db) {
|
||||
};
|
||||
} catch (error) {
|
||||
const errorCode = secretResolutionErrorCode(error);
|
||||
await logActivity(db, {
|
||||
await logActivity(db as Db, {
|
||||
companyId,
|
||||
actorType: "agent",
|
||||
actorId: context.agentId,
|
||||
|
||||
+1285
-358
File diff suppressed because it is too large.
Load diff
@@ -1,12 +1,13 @@
|
||||
import { spawn } from "node:child_process";
|
||||
import { createHash, randomBytes, randomUUID } from "node:crypto";
|
||||
import { and, desc, eq, inArray, isNull, lte, ne, sql } from "drizzle-orm";
|
||||
import { and, desc, eq, inArray, isNull, lte, ne, or, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
agents,
|
||||
approvals,
|
||||
companies,
|
||||
companyMemberships,
|
||||
companySecrets,
|
||||
connectionGrantMembers,
|
||||
connectionGrantDelegations,
|
||||
connectionGrants,
|
||||
@@ -208,6 +209,8 @@ export interface ToolGatewaySession {
|
||||
gatewayTokenAllowedActions?: ToolMcpGatewayTokenAction[];
|
||||
actorType?: "agent" | "user" | "system" | "plugin";
|
||||
actorId?: string | null;
|
||||
/** Human whose personal connection grant applies to this execution. */
|
||||
responsibleUserId?: string | null;
|
||||
createdAt: Date;
|
||||
expiresAt: Date;
|
||||
}
|
||||
@@ -922,7 +925,15 @@ export function createToolGatewayService(
|
||||
inArray(toolConnections.transport, ["mcp_remote", "local_stdio"]),
|
||||
eq(toolConnections.status, "active"),
|
||||
eq(toolConnections.enabled, true),
|
||||
inArray(toolConnections.healthStatus, ["ok", "healthy"]),
|
||||
// A personal connection has no company-level credential to probe. A
|
||||
// credential-less health sweep can therefore mark it as errored even
|
||||
// while the responsible user's grant is valid. Keep its cached active
|
||||
// catalog discoverable; execution resolves and validates that user's
|
||||
// grant, and a successful call restores the shared health indicator.
|
||||
or(
|
||||
inArray(toolConnections.healthStatus, ["ok", "healthy"]),
|
||||
eq(toolConnections.credentialPolicy, "per_user"),
|
||||
),
|
||||
eq(toolApplications.companyId, companyId),
|
||||
inArray(toolApplications.type, ["mcp_http", "mcp_stdio"]),
|
||||
eq(toolApplications.status, "active"),
|
||||
@@ -2446,6 +2457,73 @@ export function createToolGatewayService(
|
||||
);
|
||||
}
|
||||
|
||||
async function resolveGrantSecretValue(
|
||||
session: ToolGatewaySession,
|
||||
connection: typeof toolConnections.$inferSelect,
|
||||
grant: typeof connectionGrants.$inferSelect,
|
||||
ref: ToolCredentialSecretRef,
|
||||
configPath = ref.configPath,
|
||||
): Promise<string> {
|
||||
const accessContext = {
|
||||
consumerType: "tool_connection" as const,
|
||||
consumerId: connection.id,
|
||||
configPath,
|
||||
actorType: "system" as const,
|
||||
actorId: session.agentId,
|
||||
responsibleUserId: grant.subjectUserId,
|
||||
issueId: session.issueId,
|
||||
heartbeatRunId: session.runId,
|
||||
};
|
||||
if (grant.kind !== "user") {
|
||||
return secrets.resolveSecretValue(
|
||||
connection.companyId,
|
||||
ref.secretId,
|
||||
ref.versionSelector ?? "latest",
|
||||
{ accessContext },
|
||||
);
|
||||
}
|
||||
if (!grant.subjectUserId) {
|
||||
throw new ToolGatewayHttpError(422, "Personal authorization has no owner", "grant_owner_missing", {
|
||||
connectionId: connection.id,
|
||||
grantId: grant.id,
|
||||
});
|
||||
}
|
||||
const [secret] = await db.select({
|
||||
scope: companySecrets.scope,
|
||||
ownerUserId: companySecrets.ownerUserId,
|
||||
userSecretDefinitionId: companySecrets.userSecretDefinitionId,
|
||||
}).from(companySecrets).where(and(
|
||||
eq(companySecrets.id, ref.secretId),
|
||||
eq(companySecrets.companyId, connection.companyId),
|
||||
)).limit(1);
|
||||
if (
|
||||
!secret
|
||||
|| secret.scope !== "user"
|
||||
|| secret.ownerUserId !== grant.subjectUserId
|
||||
|| !secret.userSecretDefinitionId
|
||||
) {
|
||||
throw new ToolGatewayHttpError(422, "Personal authorization has an invalid credential", "grant_credential_invalid", {
|
||||
connectionId: connection.id,
|
||||
grantId: grant.id,
|
||||
credential: configPath,
|
||||
});
|
||||
}
|
||||
const resolved = await secrets.resolveUserSecretValue(connection.companyId, {
|
||||
definitionId: secret.userSecretDefinitionId,
|
||||
responsibleUserId: grant.subjectUserId,
|
||||
version: ref.versionSelector ?? "latest",
|
||||
required: ref.required ?? true,
|
||||
}, accessContext);
|
||||
if (!resolved) {
|
||||
throw new ToolGatewayHttpError(422, "Personal credential is not configured", "user_secret_missing", {
|
||||
connectionId: connection.id,
|
||||
grantId: grant.id,
|
||||
credential: configPath,
|
||||
});
|
||||
}
|
||||
return resolved.value;
|
||||
}
|
||||
|
||||
async function maybeRefreshPaperclipIdGmailGrant(
|
||||
session: ToolGatewaySession,
|
||||
connection: typeof toolConnections.$inferSelect,
|
||||
@@ -2480,22 +2558,7 @@ export function createToolGatewayService(
|
||||
grantId: grant.id,
|
||||
});
|
||||
}
|
||||
const refreshToken = await secrets.resolveSecretValue(
|
||||
connection.companyId,
|
||||
refreshRef.secretId,
|
||||
refreshRef.versionSelector ?? "latest",
|
||||
{
|
||||
accessContext: {
|
||||
consumerType: "tool_connection",
|
||||
consumerId: connection.id,
|
||||
configPath: refreshRef.configPath,
|
||||
actorType: "system",
|
||||
actorId: session.agentId,
|
||||
issueId: session.issueId,
|
||||
heartbeatRunId: session.runId,
|
||||
},
|
||||
},
|
||||
);
|
||||
const refreshToken = await resolveGrantSecretValue(session, connection, grant, refreshRef);
|
||||
try {
|
||||
const credentials = await gmailConnector.refresh({
|
||||
subject: grant.subjectUserId,
|
||||
@@ -2562,17 +2625,13 @@ export function createToolGatewayService(
|
||||
const grantRef = grantRefForHeader(grant, ref);
|
||||
if (!grantRef) continue;
|
||||
try {
|
||||
const value = await secrets.resolveSecretValue(connection.companyId, grantRef.secretId, grantRef.versionSelector ?? "latest", {
|
||||
accessContext: {
|
||||
consumerType: "tool_connection",
|
||||
consumerId: connection.id,
|
||||
configPath: `credentials.${ref.name}`,
|
||||
actorType: "system",
|
||||
actorId: session.agentId,
|
||||
issueId: session.issueId,
|
||||
heartbeatRunId: session.runId,
|
||||
},
|
||||
});
|
||||
const value = await resolveGrantSecretValue(
|
||||
session,
|
||||
connection,
|
||||
grant,
|
||||
grantRef,
|
||||
`credentials.${ref.name}`,
|
||||
);
|
||||
headers[ref.key] = `${ref.prefix ?? ""}${value}`;
|
||||
} catch {
|
||||
await markRemoteConnectionHealth(connection, "missing_secret", "A configured credential secret could not be resolved.");
|
||||
@@ -2587,22 +2646,7 @@ export function createToolGatewayService(
|
||||
const oauthAccessRef = grant.credentialSecretRefs.find((ref) => ref.configPath === "oauth.access_token");
|
||||
if (oauthAccessRef && headers.Authorization === undefined) {
|
||||
try {
|
||||
const value = await secrets.resolveSecretValue(
|
||||
connection.companyId,
|
||||
oauthAccessRef.secretId,
|
||||
oauthAccessRef.versionSelector ?? "latest",
|
||||
{
|
||||
accessContext: {
|
||||
consumerType: "tool_connection",
|
||||
consumerId: connection.id,
|
||||
configPath: oauthAccessRef.configPath,
|
||||
actorType: "system",
|
||||
actorId: session.agentId,
|
||||
issueId: session.issueId,
|
||||
heartbeatRunId: session.runId,
|
||||
},
|
||||
},
|
||||
);
|
||||
const value = await resolveGrantSecretValue(session, connection, grant, oauthAccessRef);
|
||||
headers.Authorization = `Bearer ${value}`;
|
||||
} catch {
|
||||
await markRemoteConnectionHealth(connection, "missing_secret", "A configured credential secret could not be resolved.");
|
||||
@@ -2852,7 +2896,7 @@ export function createToolGatewayService(
|
||||
eq(heartbeatRuns.companyId, session.companyId),
|
||||
)).limit(1)
|
||||
: [];
|
||||
const actingUserId = run?.responsibleUserId ?? null;
|
||||
const actingUserId = run?.responsibleUserId ?? session.responsibleUserId ?? null;
|
||||
const autonomous = run?.invocationSource === "automation" || run?.invocationSource === "timer";
|
||||
const findUserGrant = async () => {
|
||||
if (!actingUserId) return undefined;
|
||||
@@ -3077,22 +3121,7 @@ export function createToolGatewayService(
|
||||
const grantRef = grant.credentialSecretRefs.find((ref) => ref.configPath === `env.${key}`);
|
||||
if (!grantRef) continue;
|
||||
try {
|
||||
env[key] = await secrets.resolveSecretValue(
|
||||
connection.companyId,
|
||||
grantRef.secretId,
|
||||
grantRef.versionSelector ?? "latest",
|
||||
{
|
||||
accessContext: {
|
||||
consumerType: "tool_connection",
|
||||
consumerId: connection.id,
|
||||
configPath: grantRef.configPath,
|
||||
actorType: "system",
|
||||
actorId: session.agentId,
|
||||
issueId: session.issueId,
|
||||
heartbeatRunId: session.runId,
|
||||
},
|
||||
},
|
||||
);
|
||||
env[key] = await resolveGrantSecretValue(session, connection, grant, grantRef);
|
||||
} catch {
|
||||
await markRemoteConnectionHealth(connection, "missing_secret", "A configured local stdio credential could not be resolved.");
|
||||
throw new ToolGatewayHttpError(
|
||||
@@ -4171,6 +4200,7 @@ export function createToolGatewayService(
|
||||
}
|
||||
let agentId = row.gateway.agentId;
|
||||
let runId: string | null = null;
|
||||
let responsibleUserId: string | null = null;
|
||||
let issueId = row.gateway.issueId;
|
||||
let projectId = row.gateway.projectId;
|
||||
if (row.token.subjectType === "heartbeat_run") {
|
||||
@@ -4189,6 +4219,7 @@ export function createToolGatewayService(
|
||||
companyId: heartbeatRuns.companyId,
|
||||
agentId: heartbeatRuns.agentId,
|
||||
status: heartbeatRuns.status,
|
||||
responsibleUserId: heartbeatRuns.responsibleUserId,
|
||||
})
|
||||
.from(heartbeatRuns)
|
||||
.where(eq(heartbeatRuns.id, tokenRunId))
|
||||
@@ -4230,6 +4261,7 @@ export function createToolGatewayService(
|
||||
});
|
||||
agentId = run.agentId;
|
||||
runId = tokenRunId;
|
||||
responsibleUserId = run.responsibleUserId;
|
||||
issueId = runContext.issueId;
|
||||
projectId = runContext.projectId;
|
||||
} catch {
|
||||
@@ -4262,6 +4294,7 @@ export function createToolGatewayService(
|
||||
gatewayTokenAllowedActions: normalizeGatewayTokenActions(row.token.allowedActions),
|
||||
actorType: runId ? "agent" : "system",
|
||||
actorId: runId ? agentId : row.token.id,
|
||||
responsibleUserId,
|
||||
createdAt: row.token.createdAt,
|
||||
expiresAt: row.token.expiresAt ?? new Date(Date.now() + 3650 * 24 * 60 * 60 * 1000),
|
||||
};
|
||||
@@ -4504,6 +4537,7 @@ export function createToolGatewayService(
|
||||
projectId: null,
|
||||
actorType: "user",
|
||||
actorId: userId,
|
||||
responsibleUserId: userId,
|
||||
createdAt: new Date(),
|
||||
expiresAt: new Date(Date.now() + DEFAULT_SESSION_TTL_MS),
|
||||
};
|
||||
@@ -5536,6 +5570,7 @@ export function createToolGatewayService(
|
||||
projectId: null,
|
||||
actorType: "user",
|
||||
actorId: input.userId,
|
||||
responsibleUserId: input.userId,
|
||||
createdAt: new Date(),
|
||||
expiresAt: new Date(Date.now() + DEFAULT_SESSION_TTL_MS),
|
||||
};
|
||||
|
||||
Reference in new issue
Block a user