[codex] prevent invalid agents from receiving assignments and runs (#7663)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The control plane owns agent lifecycle, issue assignment, routine dispatch, heartbeat wakeups, and recovery paths > - Terminated, paused, pending-approval, or otherwise invalid agents should not receive new work or new execution attempts > - The old behavior left eligibility checks spread across routes and services, so assignment and run paths could drift apart > - This pull request centralizes agent lifecycle eligibility and applies it consistently to assignment, invocation, routines, recovery, and UI affordances > - The benefit is safer autonomy: terminated agents stay paused, invalid org-chain agents are surfaced, and active agents keep receiving valid work ## Linked Issues or Issue Description Refs #5103 Related: #1864 Bug fix context: - What happened: agent assignment and heartbeat/run paths did not share one eligibility contract, so invalid lifecycle states could still be considered in some paths. - Expected behavior: terminated agents must never receive new assignments or heartbeat runs, and paused or otherwise invalid agents should be treated as non-invokable consistently. - Steps to reproduce: create or select an agent in an invalid lifecycle state, then attempt assignment, routine dispatch, or heartbeat/recovery wake paths. - Paperclip version/commit: fixed on top of `paperclipai/paperclip` `master` at the PR base. - Deployment mode: applies to the server control plane in local and authenticated deployments. ## What Changed - Added shared agent lifecycle eligibility helpers and exported the related shared types. - Centralized server-side assignability and invokability checks for issue assignment, agent routes, heartbeat dispatch, routines, recovery, and liveness logic. - Hardened issue assignment so invalid assignees are rejected instead of queued for work. - Hardened heartbeat/routine/recovery paths so terminated and otherwise invalid agents are not woken for new runs. - Updated board UI affordances to disable invalid agent actions and surface org-chain warnings where relevant. - Added targeted shared, server, and UI tests for the new eligibility behavior. ## Verification - `pnpm exec vitest run packages/shared/src/agent-eligibility.test.ts server/src/__tests__/agent-invokability.test.ts server/src/__tests__/heartbeat-archived-company-guard.test.ts server/src/__tests__/issue-liveness.test.ts server/src/__tests__/issues-service.test.ts server/src/__tests__/routines-service.test.ts ui/src/lib/company-members.test.ts ui/src/pages/Agents.test.tsx` — 8 files, 144 tests passed. - `pnpm --filter @paperclipai/shared typecheck && pnpm --filter @paperclipai/server typecheck && pnpm --filter @paperclipai/ui typecheck` — passed. - Checked the PR diff does not include `pnpm-lock.yaml` or `.github/workflows` changes. - Checked `ROADMAP.md`; this is a targeted control-plane safety fix and does not duplicate a planned core feature. - Searched GitHub for duplicate or related PRs/issues; closest related items are linked above. - CI and Greptile verification are pending on the opened PR and will be followed up before requesting merge. ## Risks Low to moderate risk. The intended behavioral shift is that invalid agents are refused earlier and more consistently, which could expose existing data with paused, pending, terminated, or broken org-chain assignees. The added tests cover the critical assignment, heartbeat, routine, recovery, shared helper, and UI paths. No database migrations are included. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used OpenAI GPT-5 Codex via the Paperclip `codex_local` adapter, with shell/git/GitHub CLI tool use. Reasoning mode and context window are managed by the adapter runtime and not exposed in this environment. ## 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 either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] If this change affects the UI, I have included before/after screenshots (not applicable: no design screenshots requested; UI behavior is covered by tests) - [x] I have updated relevant documentation to reflect my changes (not applicable: no user-facing command or schema docs changed) - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green (pending CI) - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups (pending Greptile) - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
@@ -1,7 +1,6 @@
|
||||
import { and, eq, inArray, ne, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import {
|
||||
agents,
|
||||
companyMemberships,
|
||||
instanceUserRoles,
|
||||
issues,
|
||||
@@ -9,6 +8,7 @@ import {
|
||||
} from "@paperclipai/db";
|
||||
import type { PermissionKey, PrincipalType } from "@paperclipai/shared";
|
||||
import { conflict } from "../errors.js";
|
||||
import { assertAssignableAgent } from "./agent-assignability.js";
|
||||
import { authorizationService, type AuthorizationActor, type AuthorizationResource } from "./authorization.js";
|
||||
import { ensureHumanRoleDefaultGrants } from "./principal-access-compatibility.js";
|
||||
|
||||
@@ -315,21 +315,7 @@ export function accessService(db: Db) {
|
||||
return;
|
||||
}
|
||||
|
||||
const agent = await tx
|
||||
.select({
|
||||
id: agents.id,
|
||||
companyId: agents.companyId,
|
||||
status: agents.status,
|
||||
})
|
||||
.from(agents)
|
||||
.where(eq(agents.id, input.assigneeAgentId!))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!agent || agent.companyId !== companyId) {
|
||||
throw conflict("Replacement agent must belong to the same company");
|
||||
}
|
||||
if (agent.status === "pending_approval" || agent.status === "terminated") {
|
||||
throw conflict("Replacement agent must be assignable");
|
||||
}
|
||||
await assertAssignableAgent(tx as Db, companyId, input.assigneeAgentId, { kind: "work" });
|
||||
}
|
||||
|
||||
async function archiveMember(companyId: string, memberId: string, input: MemberArchiveInput = {}) {
|
||||
|
||||
@@ -0,0 +1,171 @@
|
||||
import { eq } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { agents } from "@paperclipai/db";
|
||||
import {
|
||||
getAgentWorkEligibility,
|
||||
type AgentEligibilityAgent,
|
||||
type AgentOrgChainHealth,
|
||||
} from "@paperclipai/shared";
|
||||
import { conflict, notFound, unprocessable } from "../errors.js";
|
||||
|
||||
type AgentAssignmentKind = "work" | "routine";
|
||||
|
||||
type AssignabilityAgent = AgentEligibilityAgent;
|
||||
|
||||
type AgentAssignmentConflictReason =
|
||||
| "pending_approval"
|
||||
| "assignee_terminated"
|
||||
| "assignee_unknown_status"
|
||||
| "ancestor_terminated"
|
||||
| "ancestor_missing"
|
||||
| "ancestor_cross_company"
|
||||
| "ancestor_cycle"
|
||||
| "ancestor_depth_exceeded";
|
||||
|
||||
function assignmentMessage(kind: AgentAssignmentKind, reason: AgentAssignmentConflictReason) {
|
||||
if (reason === "pending_approval") {
|
||||
return kind === "routine"
|
||||
? "Cannot assign routines to pending approval agents"
|
||||
: "Cannot assign work to pending approval agents";
|
||||
}
|
||||
if (reason === "assignee_terminated") {
|
||||
return kind === "routine"
|
||||
? "Cannot assign routines to terminated agents"
|
||||
: "Cannot assign work to terminated agents";
|
||||
}
|
||||
if (reason === "assignee_unknown_status") {
|
||||
return kind === "routine"
|
||||
? "Cannot assign routines to agents with an unsupported lifecycle status"
|
||||
: "Cannot assign work to agents with an unsupported lifecycle status";
|
||||
}
|
||||
return kind === "routine"
|
||||
? "Cannot assign routines to agents with an invalid org chain"
|
||||
: "Cannot assign work to agents with an invalid org chain";
|
||||
}
|
||||
|
||||
function conflictDetails(input: {
|
||||
companyId: string;
|
||||
assigneeAgentId: string;
|
||||
reason: AgentAssignmentConflictReason;
|
||||
chain: AssignabilityAgent[];
|
||||
invalidAncestorAgentId?: string | null;
|
||||
missingAncestorAgentId?: string | null;
|
||||
}) {
|
||||
return {
|
||||
code: "agent_not_assignable",
|
||||
reason: input.reason,
|
||||
companyId: input.companyId,
|
||||
assigneeAgentId: input.assigneeAgentId,
|
||||
invalidAncestorAgentId: input.invalidAncestorAgentId ?? null,
|
||||
missingAncestorAgentId: input.missingAncestorAgentId ?? null,
|
||||
ancestorChain: input.chain.map((agent) => ({
|
||||
id: agent.id,
|
||||
companyId: agent.companyId,
|
||||
status: agent.status,
|
||||
reportsTo: agent.reportsTo,
|
||||
})),
|
||||
};
|
||||
}
|
||||
|
||||
async function getAgent(db: Db, agentId: string): Promise<AssignabilityAgent | null> {
|
||||
return db
|
||||
.select({
|
||||
id: agents.id,
|
||||
companyId: agents.companyId,
|
||||
name: agents.name,
|
||||
status: agents.status,
|
||||
reportsTo: agents.reportsTo,
|
||||
})
|
||||
.from(agents)
|
||||
.where(eq(agents.id, agentId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function listCompanyAgents(db: Db, companyId: string): Promise<AssignabilityAgent[]> {
|
||||
return db
|
||||
.select({
|
||||
id: agents.id,
|
||||
companyId: agents.companyId,
|
||||
name: agents.name,
|
||||
status: agents.status,
|
||||
reportsTo: agents.reportsTo,
|
||||
})
|
||||
.from(agents)
|
||||
.where(eq(agents.companyId, companyId));
|
||||
}
|
||||
|
||||
function assignmentReasonFromHealth(health: AgentOrgChainHealth): AgentAssignmentConflictReason {
|
||||
if (health.reason === "terminated_ancestor") return "ancestor_terminated";
|
||||
if (health.reason === "missing_manager") return "ancestor_missing";
|
||||
if (health.reason === "cycle") return "ancestor_cycle";
|
||||
return "ancestor_missing";
|
||||
}
|
||||
|
||||
export async function assertAssignableAgent(
|
||||
db: Db,
|
||||
companyId: string,
|
||||
agentId: string | null | undefined,
|
||||
options: { kind?: AgentAssignmentKind } = {},
|
||||
) {
|
||||
if (!agentId) return;
|
||||
const kind = options.kind ?? "work";
|
||||
const assignee = await getAgent(db, agentId);
|
||||
if (!assignee) throw notFound("Assignee agent not found");
|
||||
if (assignee.companyId !== companyId) {
|
||||
throw unprocessable("Assignee must belong to same company");
|
||||
}
|
||||
|
||||
const companyAgents = await listCompanyAgents(db, companyId);
|
||||
const eligibility = getAgentWorkEligibility({ agent: assignee, agents: companyAgents });
|
||||
const chain = eligibility.orgChainHealth.fullChain.map((entry) => ({
|
||||
id: entry.id,
|
||||
companyId: entry.companyId,
|
||||
name: entry.name,
|
||||
status: entry.status,
|
||||
reportsTo: entry.reportsTo,
|
||||
}));
|
||||
|
||||
if (eligibility.assignable) return;
|
||||
|
||||
if (eligibility.assignabilityReason === "pending_approval") {
|
||||
throw conflict(assignmentMessage(kind, "pending_approval"), conflictDetails({
|
||||
companyId,
|
||||
assigneeAgentId: agentId,
|
||||
reason: "pending_approval",
|
||||
chain,
|
||||
}));
|
||||
}
|
||||
if (eligibility.assignabilityReason === "terminated") {
|
||||
throw conflict(assignmentMessage(kind, "assignee_terminated"), conflictDetails({
|
||||
companyId,
|
||||
assigneeAgentId: agentId,
|
||||
reason: "assignee_terminated",
|
||||
chain,
|
||||
}));
|
||||
}
|
||||
if (eligibility.assignabilityReason === "unknown_status") {
|
||||
throw conflict(assignmentMessage(kind, "assignee_unknown_status"), conflictDetails({
|
||||
companyId,
|
||||
assigneeAgentId: agentId,
|
||||
reason: "assignee_unknown_status",
|
||||
chain,
|
||||
}));
|
||||
}
|
||||
|
||||
const reason = assignmentReasonFromHealth(eligibility.orgChainHealth);
|
||||
const firstInvalidAncestor = eligibility.orgChainHealth.firstInvalidAncestor;
|
||||
throw conflict(assignmentMessage(kind, reason), conflictDetails({
|
||||
companyId,
|
||||
assigneeAgentId: agentId,
|
||||
reason,
|
||||
chain,
|
||||
invalidAncestorAgentId:
|
||||
firstInvalidAncestor && firstInvalidAncestor.status !== "missing"
|
||||
? firstInvalidAncestor.id
|
||||
: null,
|
||||
missingAncestorAgentId:
|
||||
firstInvalidAncestor?.status === "missing"
|
||||
? firstInvalidAncestor.id
|
||||
: null,
|
||||
}));
|
||||
}
|
||||
@@ -0,0 +1,164 @@
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { agents } from "@paperclipai/db";
|
||||
import { getAgentWorkEligibility, type AgentEligibilityAgent, type AgentOrgChainHealth } from "@paperclipai/shared";
|
||||
import { eq } from "drizzle-orm";
|
||||
|
||||
type AgentStatus = (typeof agents.$inferSelect)["status"];
|
||||
|
||||
export type AgentOrgRow = Pick<
|
||||
typeof agents.$inferSelect,
|
||||
"id" | "companyId" | "name" | "reportsTo" | "status"
|
||||
>;
|
||||
|
||||
export type AgentInvokabilityBlockReason =
|
||||
| "missing"
|
||||
| "paused"
|
||||
| "terminated"
|
||||
| "pending_approval"
|
||||
| "unknown_status"
|
||||
| "manager_missing"
|
||||
| "manager_company_mismatch"
|
||||
| "manager_terminated"
|
||||
| "reporting_cycle"
|
||||
| "reporting_chain_too_deep";
|
||||
|
||||
export type AgentInvokability =
|
||||
| { invokable: true }
|
||||
| {
|
||||
invokable: false;
|
||||
reason: AgentInvokabilityBlockReason;
|
||||
message: string;
|
||||
details: Record<string, unknown>;
|
||||
invalidOrgChain: boolean;
|
||||
};
|
||||
|
||||
const DIRECT_NON_INVOKABLE_STATUSES = new Set<AgentStatus>([
|
||||
"paused",
|
||||
"terminated",
|
||||
"pending_approval",
|
||||
]);
|
||||
|
||||
function blocked(
|
||||
reason: AgentInvokabilityBlockReason,
|
||||
message: string,
|
||||
details: Record<string, unknown>,
|
||||
invalidOrgChain = false,
|
||||
): AgentInvokability {
|
||||
return { invokable: false, reason, message, details, invalidOrgChain };
|
||||
}
|
||||
|
||||
function statusBlockReason(status: AgentStatus): AgentInvokabilityBlockReason | null {
|
||||
if (status === "paused") return "paused";
|
||||
if (status === "terminated") return "terminated";
|
||||
if (status === "pending_approval") return "pending_approval";
|
||||
return null;
|
||||
}
|
||||
|
||||
function toEligibilityAgent(row: AgentOrgRow): AgentEligibilityAgent {
|
||||
return {
|
||||
id: row.id,
|
||||
companyId: row.companyId,
|
||||
name: row.name,
|
||||
status: row.status,
|
||||
reportsTo: row.reportsTo,
|
||||
};
|
||||
}
|
||||
|
||||
function invalidChainReason(health: AgentOrgChainHealth): AgentInvokabilityBlockReason {
|
||||
if (health.reason === "terminated_ancestor") return "manager_terminated";
|
||||
if (health.reason === "cycle") return "reporting_cycle";
|
||||
return "manager_missing";
|
||||
}
|
||||
|
||||
export function evaluateAgentInvokability(
|
||||
agent: AgentOrgRow | null | undefined,
|
||||
companyAgents: AgentOrgRow[],
|
||||
): AgentInvokability {
|
||||
if (!agent) {
|
||||
return blocked("missing", "Agent no longer exists", {}, false);
|
||||
}
|
||||
|
||||
const eligibility = getAgentWorkEligibility({
|
||||
agent: toEligibilityAgent(agent),
|
||||
agents: companyAgents.map(toEligibilityAgent),
|
||||
});
|
||||
|
||||
if (eligibility.invokable) return { invokable: true };
|
||||
|
||||
const directStatusReason = eligibility.invokabilityReason === "unknown_status"
|
||||
? "unknown_status"
|
||||
: statusBlockReason(agent.status);
|
||||
if (directStatusReason) {
|
||||
return blocked(
|
||||
directStatusReason,
|
||||
"Agent is not invokable in its current state",
|
||||
{ agentId: agent.id, agentStatus: agent.status },
|
||||
false,
|
||||
);
|
||||
}
|
||||
|
||||
const health = eligibility.orgChainHealth;
|
||||
const firstInvalidAncestor = health.firstInvalidAncestor;
|
||||
return blocked(
|
||||
invalidChainReason(health),
|
||||
"Agent is not invokable because its reporting chain is invalid",
|
||||
{
|
||||
agentId: agent.id,
|
||||
managerId: firstInvalidAncestor?.id ?? null,
|
||||
managerStatus: firstInvalidAncestor?.status ?? null,
|
||||
reportingChainAgentIds: health.fullChain
|
||||
.filter((entry) => entry.relation === "ancestor")
|
||||
.map((entry) => entry.id),
|
||||
orgChainHealth: health,
|
||||
},
|
||||
true,
|
||||
);
|
||||
}
|
||||
|
||||
export async function evaluateAgentInvokabilityFromDb(
|
||||
db: Db,
|
||||
agent: AgentOrgRow | null | undefined,
|
||||
): Promise<AgentInvokability> {
|
||||
if (!agent) return evaluateAgentInvokability(agent, []);
|
||||
const companyAgents = await db
|
||||
.select({
|
||||
id: agents.id,
|
||||
companyId: agents.companyId,
|
||||
name: agents.name,
|
||||
reportsTo: agents.reportsTo,
|
||||
status: agents.status,
|
||||
})
|
||||
.from(agents)
|
||||
.where(eq(agents.companyId, agent.companyId));
|
||||
return evaluateAgentInvokability(agent, companyAgents);
|
||||
}
|
||||
|
||||
export function listInvalidOrgChainDescendantIds(
|
||||
terminatedAgentId: string,
|
||||
companyAgents: AgentOrgRow[],
|
||||
): string[] {
|
||||
const byManager = new Map<string | null, AgentOrgRow[]>();
|
||||
for (const row of companyAgents) {
|
||||
const siblings = byManager.get(row.reportsTo ?? null) ?? [];
|
||||
siblings.push(row);
|
||||
byManager.set(row.reportsTo ?? null, siblings);
|
||||
}
|
||||
|
||||
const invalidDescendantIds: string[] = [];
|
||||
const stack = [...(byManager.get(terminatedAgentId) ?? [])];
|
||||
const seen = new Set<string>([terminatedAgentId]);
|
||||
while (stack.length > 0) {
|
||||
const current = stack.pop();
|
||||
if (!current || seen.has(current.id)) continue;
|
||||
seen.add(current.id);
|
||||
if (current.status !== "terminated") {
|
||||
invalidDescendantIds.push(current.id);
|
||||
}
|
||||
stack.push(...(byManager.get(current.id) ?? []));
|
||||
}
|
||||
return invalidDescendantIds;
|
||||
}
|
||||
|
||||
export function shouldCancelRunsForNonInvokableAgent(result: AgentInvokability) {
|
||||
return !result.invokable && (result.reason === "terminated" || result.invalidOrgChain);
|
||||
}
|
||||
@@ -16,7 +16,13 @@ import {
|
||||
issues,
|
||||
issueComments,
|
||||
} from "@paperclipai/db";
|
||||
import { AGENT_DEFAULT_MAX_CONCURRENT_RUNS, isUuidLike, normalizeAgentUrlKey } from "@paperclipai/shared";
|
||||
import {
|
||||
AGENT_DEFAULT_MAX_CONCURRENT_RUNS,
|
||||
getAgentWorkEligibility,
|
||||
isUuidLike,
|
||||
normalizeAgentUrlKey,
|
||||
type AgentEligibilityAgent,
|
||||
} from "@paperclipai/shared";
|
||||
import { conflict, notFound, unprocessable } from "../errors.js";
|
||||
import { normalizeAgentPermissions } from "./agent-permissions.js";
|
||||
import { REDACTED_EVENT_VALUE, sanitizeRecord } from "../redaction.js";
|
||||
@@ -228,13 +234,45 @@ export function agentService(db: Db) {
|
||||
};
|
||||
}
|
||||
|
||||
function normalizeAgentRow(row: typeof agents.$inferSelect) {
|
||||
function normalizeAgentBaseRow(row: typeof agents.$inferSelect) {
|
||||
return withUrlKey({
|
||||
...row,
|
||||
permissions: normalizeAgentPermissions(row.permissions, row.role),
|
||||
});
|
||||
}
|
||||
|
||||
function toEligibilityAgent(row: Pick<typeof agents.$inferSelect, "id" | "companyId" | "name" | "status" | "reportsTo">): AgentEligibilityAgent {
|
||||
return {
|
||||
id: row.id,
|
||||
companyId: row.companyId,
|
||||
name: row.name,
|
||||
status: row.status,
|
||||
reportsTo: row.reportsTo,
|
||||
};
|
||||
}
|
||||
|
||||
function normalizeAgentRows(rows: (typeof agents.$inferSelect)[], allCompanyRows = rows) {
|
||||
const eligibilityAgents = allCompanyRows.map(toEligibilityAgent);
|
||||
return rows.map((row) => {
|
||||
const base = normalizeAgentBaseRow(row);
|
||||
return {
|
||||
...base,
|
||||
orgChainHealth: getAgentWorkEligibility({
|
||||
agent: toEligibilityAgent(row),
|
||||
agents: eligibilityAgents,
|
||||
}).orgChainHealth,
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
function normalizeAgentRow(row: typeof agents.$inferSelect, allCompanyRows?: (typeof agents.$inferSelect)[]) {
|
||||
return normalizeAgentRows([row], allCompanyRows)[0]!;
|
||||
}
|
||||
|
||||
async function listCompanyAgentRows(companyId: string) {
|
||||
return db.select().from(agents).where(eq(agents.companyId, companyId));
|
||||
}
|
||||
|
||||
async function getMonthlySpendByAgentIds(companyId: string, agentIds: string[]) {
|
||||
if (agentIds.length === 0) return new Map<string, number>();
|
||||
const { start, end } = currentUtcMonthWindow();
|
||||
@@ -274,8 +312,17 @@ export function agentService(db: Db) {
|
||||
.where(eq(agents.id, id))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!row) return null;
|
||||
const [hydrated] = await hydrateAgentSpend([row]);
|
||||
return normalizeAgentRow(hydrated);
|
||||
const [companyRows, hydrated] = await Promise.all([
|
||||
listCompanyAgentRows(row.companyId),
|
||||
hydrateAgentSpend([row]).then((rows) => rows[0]!),
|
||||
]);
|
||||
return normalizeAgentRow(hydrated, companyRows);
|
||||
}
|
||||
|
||||
async function requireGetById(id: string) {
|
||||
const agent = await getById(id);
|
||||
if (!agent) throw notFound("Agent not found");
|
||||
return agent;
|
||||
}
|
||||
|
||||
async function ensureManager(companyId: string, managerId: string) {
|
||||
@@ -374,7 +421,7 @@ export function agentService(db: Db) {
|
||||
.where(eq(agents.id, id))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
const normalizedUpdated = updated ? normalizeAgentRow(updated) : null;
|
||||
const normalizedUpdated = updated ? await getById(updated.id) : null;
|
||||
|
||||
if (normalizedUpdated && shouldRecordRevision && beforeConfig) {
|
||||
const afterConfig = buildConfigSnapshot(normalizedUpdated);
|
||||
@@ -403,9 +450,12 @@ export function agentService(db: Db) {
|
||||
if (!options?.includeTerminated) {
|
||||
conditions.push(ne(agents.status, "terminated"));
|
||||
}
|
||||
const rows = await db.select().from(agents).where(and(...conditions));
|
||||
const [rows, allCompanyRows] = await Promise.all([
|
||||
db.select().from(agents).where(and(...conditions)),
|
||||
listCompanyAgentRows(companyId),
|
||||
]);
|
||||
const hydrated = await hydrateAgentSpend(rows);
|
||||
return hydrated.map(normalizeAgentRow);
|
||||
return normalizeAgentRows(hydrated, allCompanyRows);
|
||||
},
|
||||
|
||||
getById,
|
||||
@@ -430,7 +480,7 @@ export function agentService(db: Db) {
|
||||
.returning()
|
||||
.then((rows) => rows[0]);
|
||||
|
||||
return normalizeAgentRow(created);
|
||||
return requireGetById(created.id);
|
||||
},
|
||||
|
||||
update: updateAgent,
|
||||
@@ -451,7 +501,7 @@ export function agentService(db: Db) {
|
||||
.where(eq(agents.id, id))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return updated ? normalizeAgentRow(updated) : null;
|
||||
return updated ? getById(updated.id) : null;
|
||||
},
|
||||
|
||||
resume: async (id: string) => {
|
||||
@@ -473,7 +523,7 @@ export function agentService(db: Db) {
|
||||
.where(eq(agents.id, id))
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
return updated ? normalizeAgentRow(updated) : null;
|
||||
return updated ? getById(updated.id) : null;
|
||||
},
|
||||
|
||||
terminate: async (id: string) => {
|
||||
@@ -540,7 +590,7 @@ export function agentService(db: Db) {
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
if (updated) {
|
||||
return { agent: normalizeAgentRow(updated), activated: true };
|
||||
return { agent: await requireGetById(updated.id), activated: true };
|
||||
}
|
||||
|
||||
const existing = await getById(id);
|
||||
@@ -561,7 +611,7 @@ export function agentService(db: Db) {
|
||||
.returning()
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
return updated ? normalizeAgentRow(updated) : null;
|
||||
return updated ? getById(updated.id) : null;
|
||||
},
|
||||
|
||||
listConfigRevisions: async (id: string) =>
|
||||
@@ -670,14 +720,12 @@ export function agentService(db: Db) {
|
||||
},
|
||||
|
||||
orgForCompany: async (companyId: string) => {
|
||||
const rows = await db
|
||||
.select()
|
||||
.from(agents)
|
||||
.where(and(eq(agents.companyId, companyId), ne(agents.status, "terminated")));
|
||||
const normalizedRows = rows.map(normalizeAgentRow);
|
||||
const allCompanyRows = await listCompanyAgentRows(companyId);
|
||||
const rows = allCompanyRows.filter((row) => row.status !== "terminated");
|
||||
const normalizedRows = normalizeAgentRows(rows, allCompanyRows);
|
||||
const byManager = new Map<string | null, typeof normalizedRows>();
|
||||
for (const row of normalizedRows) {
|
||||
const key = row.reportsTo ?? null;
|
||||
const key = row.reportsTo && rows.some((candidate) => candidate.id === row.reportsTo) ? row.reportsTo : null;
|
||||
const group = byManager.get(key) ?? [];
|
||||
group.push(row);
|
||||
byManager.set(key, group);
|
||||
@@ -735,8 +783,7 @@ export function agentService(db: Db) {
|
||||
}
|
||||
|
||||
const rows = await db.select().from(agents).where(eq(agents.companyId, companyId));
|
||||
const matches = rows
|
||||
.map(normalizeAgentRow)
|
||||
const matches = normalizeAgentRows(rows, rows)
|
||||
.filter((agent) => agent.urlKey === urlKey && agent.status !== "terminated");
|
||||
if (matches.length === 1) {
|
||||
return { agent: matches[0] ?? null, ambiguous: false } as const;
|
||||
|
||||
@@ -155,6 +155,12 @@ import {
|
||||
import { recoveryService } from "./recovery/service.js";
|
||||
import { productivityReviewService } from "./productivity-review.js";
|
||||
import { withAgentStartLock } from "./agent-start-lock.js";
|
||||
import {
|
||||
evaluateAgentInvokability,
|
||||
evaluateAgentInvokabilityFromDb,
|
||||
shouldCancelRunsForNonInvokableAgent,
|
||||
type AgentOrgRow,
|
||||
} from "./agent-invokability.js";
|
||||
import {
|
||||
redactQuarantinedBodyForHigherTrust,
|
||||
sanitizeQuarantinedCommentForHigherTrust,
|
||||
@@ -3061,6 +3067,46 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
.then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function getAgentInvokability(agent: typeof agents.$inferSelect | null | undefined) {
|
||||
return evaluateAgentInvokabilityFromDb(db, agent);
|
||||
}
|
||||
|
||||
function toAgentOrgRow(agent: Pick<typeof agents.$inferSelect, "id" | "companyId" | "name" | "reportsTo" | "status">): AgentOrgRow {
|
||||
return {
|
||||
id: agent.id,
|
||||
companyId: agent.companyId,
|
||||
name: agent.name,
|
||||
reportsTo: agent.reportsTo,
|
||||
status: agent.status,
|
||||
};
|
||||
}
|
||||
|
||||
async function listCompanyAgentOrgRows(companyId: string): Promise<AgentOrgRow[]> {
|
||||
return db
|
||||
.select({
|
||||
id: agents.id,
|
||||
companyId: agents.companyId,
|
||||
name: agents.name,
|
||||
reportsTo: agents.reportsTo,
|
||||
status: agents.status,
|
||||
})
|
||||
.from(agents)
|
||||
.where(eq(agents.companyId, companyId));
|
||||
}
|
||||
|
||||
function groupAgentOrgRowsByCompany(agentRows: AgentOrgRow[]) {
|
||||
const byCompany = new Map<string, AgentOrgRow[]>();
|
||||
for (const agent of agentRows) {
|
||||
const companyAgents = byCompany.get(agent.companyId);
|
||||
if (companyAgents) {
|
||||
companyAgents.push(agent);
|
||||
} else {
|
||||
byCompany.set(agent.companyId, [agent]);
|
||||
}
|
||||
}
|
||||
return byCompany;
|
||||
}
|
||||
|
||||
async function getRun(runId: string, opts?: { unsafeFullResultJson?: boolean }) {
|
||||
const safeForLegacyEncoding = !opts?.unsafeFullResultJson && await hasUnsafeTextProjectionDatabase();
|
||||
return db
|
||||
@@ -5146,6 +5192,22 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
agent: typeof agents.$inferSelect,
|
||||
issueId: string,
|
||||
) {
|
||||
const invokability = await getAgentInvokability(agent);
|
||||
if (!invokability.invokable) {
|
||||
await appendRunEvent(run, await nextRunEventSeq(run.id), {
|
||||
eventType: "lifecycle",
|
||||
stream: "system",
|
||||
level: "warn",
|
||||
message: "Missing-comment retry suppressed because the agent is not invokable",
|
||||
payload: {
|
||||
reason: invokability.reason,
|
||||
invalidOrgChain: invokability.invalidOrgChain,
|
||||
...invokability.details,
|
||||
},
|
||||
});
|
||||
return null;
|
||||
}
|
||||
|
||||
const contextSnapshot = parseObject(run.contextSnapshot);
|
||||
const taskKey = deriveTaskKeyWithHeartbeatFallback(contextSnapshot, null);
|
||||
const sessionBefore = await resolveSessionBeforeForWakeup(agent, taskKey);
|
||||
@@ -5366,6 +5428,23 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
agent: typeof agents.$inferSelect,
|
||||
now: Date,
|
||||
) {
|
||||
const invokability = await getAgentInvokability(agent);
|
||||
if (!invokability.invokable) {
|
||||
await appendRunEvent(run, await nextRunEventSeq(run.id), {
|
||||
eventType: "lifecycle",
|
||||
stream: "system",
|
||||
level: "warn",
|
||||
message: "Process-loss retry suppressed because the agent is not invokable",
|
||||
payload: {
|
||||
reason: invokability.reason,
|
||||
invalidOrgChain: invokability.invalidOrgChain,
|
||||
...invokability.details,
|
||||
},
|
||||
});
|
||||
await releaseIssueExecutionAndPromote(run);
|
||||
return null;
|
||||
}
|
||||
|
||||
const contextSnapshot = parseObject(run.contextSnapshot);
|
||||
const issueId = readNonEmptyString(contextSnapshot.issueId);
|
||||
const taskKey = deriveTaskKeyWithHeartbeatFallback(contextSnapshot, null);
|
||||
@@ -5516,15 +5595,16 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
};
|
||||
}
|
||||
|
||||
if (agent.status === "paused" || agent.status === "terminated" || agent.status === "pending_approval") {
|
||||
const agentInvokability = await getAgentInvokability(agent);
|
||||
if (!agentInvokability.invokable) {
|
||||
return {
|
||||
allowed: false,
|
||||
reason: "Scheduled retry suppressed because the agent is not invokable",
|
||||
errorCode: "agent_not_invokable",
|
||||
issueId,
|
||||
details: {
|
||||
agentId: agent.id,
|
||||
agentStatus: agent.status,
|
||||
...agentInvokability.details,
|
||||
invalidOrgChain: agentInvokability.invalidOrgChain,
|
||||
},
|
||||
};
|
||||
}
|
||||
@@ -5896,6 +5976,35 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
maxAttempts,
|
||||
};
|
||||
}
|
||||
|
||||
if (retryReason !== MAX_TURN_CONTINUATION_RETRY_REASON) {
|
||||
const invokability = await getAgentInvokability(agent);
|
||||
if (!invokability.invokable) {
|
||||
const contextSnapshot = parseObject(run.contextSnapshot);
|
||||
const issueId = readNonEmptyString(contextSnapshot.issueId);
|
||||
await appendRunEvent(run, await nextRunEventSeq(run.id), {
|
||||
eventType: "lifecycle",
|
||||
stream: "system",
|
||||
level: "warn",
|
||||
message: "Scheduled retry suppressed because the agent is not invokable",
|
||||
payload: {
|
||||
retryReason,
|
||||
scheduledRetryAttempt: nextAttempt,
|
||||
maxAttempts,
|
||||
reason: invokability.reason,
|
||||
invalidOrgChain: invokability.invalidOrgChain,
|
||||
...invokability.details,
|
||||
},
|
||||
});
|
||||
return {
|
||||
outcome: "not_scheduled" as const,
|
||||
reason: "Scheduled retry suppressed because the agent is not invokable",
|
||||
errorCode: "agent_not_invokable" as const,
|
||||
issueId,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
const schedule =
|
||||
transientRetryNotBefore && transientRetryNotBefore.getTime() > baseSchedule.dueAt.getTime()
|
||||
? {
|
||||
@@ -6514,15 +6623,18 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
return Number(count ?? 0);
|
||||
}
|
||||
|
||||
async function claimQueuedRun(run: typeof heartbeatRuns.$inferSelect) {
|
||||
async function claimQueuedRun(run: typeof heartbeatRuns.$inferSelect, companyAgents?: AgentOrgRow[]) {
|
||||
if (run.status !== "queued") return run;
|
||||
const agent = await getAgent(run.agentId);
|
||||
if (!agent) {
|
||||
await cancelRunInternal(run.id, "Cancelled because the agent no longer exists");
|
||||
return null;
|
||||
}
|
||||
if (agent.status === "paused" || agent.status === "terminated" || agent.status === "pending_approval") {
|
||||
await cancelRunInternal(run.id, "Cancelled because the agent is not invokable");
|
||||
const invokability = companyAgents
|
||||
? evaluateAgentInvokability(toAgentOrgRow(agent), companyAgents)
|
||||
: await getAgentInvokability(agent);
|
||||
if (!invokability.invokable) {
|
||||
await cancelRunInternal(run.id, `Cancelled because the agent is not invokable: ${invokability.reason}`);
|
||||
return null;
|
||||
}
|
||||
|
||||
@@ -7433,7 +7545,11 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
return withAgentStartLock(agentId, async () => {
|
||||
const agent = await getAgent(agentId);
|
||||
if (!agent) return [];
|
||||
if (agent.status === "paused" || agent.status === "terminated" || agent.status === "pending_approval") {
|
||||
const invokability = await getAgentInvokability(agent);
|
||||
if (!invokability.invokable) {
|
||||
if (shouldCancelRunsForNonInvokableAgent(invokability)) {
|
||||
await cancelActiveForAgentInternal(agentId, `Cancelled because the agent is not invokable: ${invokability.reason}`);
|
||||
}
|
||||
return [];
|
||||
}
|
||||
const policy = parseHeartbeatPolicy(agent);
|
||||
@@ -7467,6 +7583,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
: sql`false`,
|
||||
);
|
||||
const issueById = new Map(issueRows.map((row) => [row.id, row]));
|
||||
const companyAgents = await listCompanyAgentOrgRows(agent.companyId);
|
||||
const prioritizedRuns = [...queuedRuns].sort((left, right) => {
|
||||
const leftIssueId = readNonEmptyString(parseObject(left.contextSnapshot).issueId);
|
||||
const rightIssueId = readNonEmptyString(parseObject(right.contextSnapshot).issueId);
|
||||
@@ -7488,7 +7605,7 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
const claimedRuns: Array<typeof heartbeatRuns.$inferSelect> = [];
|
||||
for (const queuedRun of prioritizedRuns) {
|
||||
if (claimedRuns.length >= availableSlots) break;
|
||||
const claimed = await claimQueuedRun(queuedRun);
|
||||
const claimed = await claimQueuedRun(queuedRun, companyAgents);
|
||||
if (claimed) claimedRuns.push(claimed);
|
||||
}
|
||||
if (claimedRuns.length === 0) return [];
|
||||
@@ -9278,13 +9395,24 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
.where(eq(agents.id, deferred.agentId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
if (
|
||||
!deferredAgent ||
|
||||
deferredAgent.companyId !== issue.companyId ||
|
||||
deferredAgent.status === "paused" ||
|
||||
deferredAgent.status === "terminated" ||
|
||||
deferredAgent.status === "pending_approval"
|
||||
) {
|
||||
const companyAgents = deferredAgent
|
||||
? await tx
|
||||
.select({
|
||||
id: agents.id,
|
||||
companyId: agents.companyId,
|
||||
name: agents.name,
|
||||
reportsTo: agents.reportsTo,
|
||||
status: agents.status,
|
||||
})
|
||||
.from(agents)
|
||||
.where(eq(agents.companyId, issue.companyId))
|
||||
: [];
|
||||
const deferredInvokability =
|
||||
deferredAgent?.companyId === issue.companyId
|
||||
? evaluateAgentInvokability(deferredAgent, companyAgents)
|
||||
: evaluateAgentInvokability(null, companyAgents);
|
||||
|
||||
if (!deferredAgent || deferredAgent.companyId !== issue.companyId || !deferredInvokability.invokable) {
|
||||
await tx
|
||||
.update(agentWakeupRequests)
|
||||
.set({
|
||||
@@ -9767,12 +9895,19 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
});
|
||||
}
|
||||
|
||||
if (
|
||||
agent.status === "paused" ||
|
||||
agent.status === "terminated" ||
|
||||
agent.status === "pending_approval"
|
||||
) {
|
||||
throw conflict("Agent is not invokable in its current state", { status: agent.status });
|
||||
const invokability = await getAgentInvokability(agent);
|
||||
if (!invokability.invokable) {
|
||||
if (opts.requestedByActorType !== "user") {
|
||||
await writeSkippedRequest("agent.not_invokable", {
|
||||
error: invokability.message,
|
||||
});
|
||||
}
|
||||
throw conflict(invokability.message, {
|
||||
status: agent.status,
|
||||
reason: invokability.reason,
|
||||
invalidOrgChain: invokability.invalidOrgChain,
|
||||
...invokability.details,
|
||||
});
|
||||
}
|
||||
|
||||
const policy = parseHeartbeatPolicy(agent);
|
||||
@@ -10576,6 +10711,52 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
return runs.length;
|
||||
}
|
||||
|
||||
async function cancelPendingWakeupsForAgentsInternal(agentIds: string[], reason: string) {
|
||||
const uniqueAgentIds = [...new Set(agentIds)].filter((agentId) => agentId.length > 0);
|
||||
if (uniqueAgentIds.length === 0) return 0;
|
||||
|
||||
const now = new Date();
|
||||
const wakeupIds = await db
|
||||
.select({ id: agentWakeupRequests.id })
|
||||
.from(agentWakeupRequests)
|
||||
.where(
|
||||
and(
|
||||
inArray(agentWakeupRequests.agentId, uniqueAgentIds),
|
||||
inArray(agentWakeupRequests.status, ["queued", "deferred_issue_execution"]),
|
||||
sql`${agentWakeupRequests.runId} is null`,
|
||||
),
|
||||
)
|
||||
.then((rows) => rows.map((row) => row.id));
|
||||
|
||||
if (wakeupIds.length === 0) return 0;
|
||||
|
||||
await db
|
||||
.update(agentWakeupRequests)
|
||||
.set({
|
||||
status: "cancelled",
|
||||
finishedAt: now,
|
||||
error: reason,
|
||||
updatedAt: now,
|
||||
})
|
||||
.where(inArray(agentWakeupRequests.id, wakeupIds));
|
||||
|
||||
return wakeupIds.length;
|
||||
}
|
||||
|
||||
async function cancelInvocationsForAgentsInternal(agentIds: string[], reason: string) {
|
||||
const uniqueAgentIds = [...new Set(agentIds)].filter((agentId) => agentId.length > 0);
|
||||
let runsCancelled = 0;
|
||||
for (const agentId of uniqueAgentIds) {
|
||||
runsCancelled += await cancelActiveForAgentInternal(agentId, reason);
|
||||
}
|
||||
const wakeupsCancelled = await cancelPendingWakeupsForAgentsInternal(uniqueAgentIds, reason);
|
||||
return {
|
||||
agentIds: uniqueAgentIds,
|
||||
runsCancelled,
|
||||
wakeupsCancelled,
|
||||
};
|
||||
}
|
||||
|
||||
async function cancelBudgetScopeWork(scope: BudgetEnforcementScope) {
|
||||
if (scope.scopeType === "agent") {
|
||||
await cancelActiveForAgentInternal(scope.scopeId, "Cancelled due to budget pause");
|
||||
@@ -10877,12 +11058,14 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
.from(agents)
|
||||
.innerJoin(companies, eq(companies.id, agents.companyId))
|
||||
.where(eq(companies.status, "active"));
|
||||
const agentsByCompany = groupAgentOrgRowsByCompany(allAgents.map(toAgentOrgRow));
|
||||
let checked = 0;
|
||||
let enqueued = 0;
|
||||
let skipped = 0;
|
||||
|
||||
for (const agent of allAgents) {
|
||||
if (agent.status === "paused" || agent.status === "terminated" || agent.status === "pending_approval") continue;
|
||||
const invokability = evaluateAgentInvokability(toAgentOrgRow(agent), agentsByCompany.get(agent.companyId) ?? []);
|
||||
if (!invokability.invokable) continue;
|
||||
const policy = parseHeartbeatPolicy(agent);
|
||||
if (!policy.enabled || policy.intervalSec <= 0) continue;
|
||||
|
||||
@@ -10920,6 +11103,9 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
|
||||
cancelActiveForAgent: (agentId: string, reason?: string) => cancelActiveForAgentInternal(agentId, reason),
|
||||
|
||||
cancelInvocationsForAgents: (agentIds: string[], reason: string) =>
|
||||
cancelInvocationsForAgentsInternal(agentIds, reason),
|
||||
|
||||
cancelBudgetScopeWork,
|
||||
|
||||
getRunIssueSummary: async (runId: string) => {
|
||||
|
||||
@@ -75,6 +75,7 @@ import { redactSensitiveText } from "../redaction.js";
|
||||
import { resolveIssueGoalId, resolveNextIssueGoalId } from "./issue-goal-fallback.js";
|
||||
import { getRunLogStore } from "./run-log-store.js";
|
||||
import { getDefaultCompanyGoal } from "./goals.js";
|
||||
import { assertAssignableAgent } from "./agent-assignability.js";
|
||||
import {
|
||||
isVerifiedIssueTreeControlInteractionWake,
|
||||
issueTreeControlService,
|
||||
@@ -3372,29 +3373,6 @@ export function issueService(db: Db) {
|
||||
});
|
||||
}
|
||||
|
||||
async function assertAssignableAgent(companyId: string, agentId: string) {
|
||||
const assignee = await db
|
||||
.select({
|
||||
id: agents.id,
|
||||
companyId: agents.companyId,
|
||||
status: agents.status,
|
||||
})
|
||||
.from(agents)
|
||||
.where(eq(agents.id, agentId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
|
||||
if (!assignee) throw notFound("Assignee agent not found");
|
||||
if (assignee.companyId !== companyId) {
|
||||
throw unprocessable("Assignee must belong to same company");
|
||||
}
|
||||
if (assignee.status === "pending_approval") {
|
||||
throw conflict("Cannot assign work to pending approval agents");
|
||||
}
|
||||
if (assignee.status === "terminated") {
|
||||
throw conflict("Cannot assign work to terminated agents");
|
||||
}
|
||||
}
|
||||
|
||||
async function isTreeHoldInteractionCheckoutAllowed(
|
||||
companyId: string,
|
||||
checkoutRunId: string | null,
|
||||
@@ -4733,7 +4711,7 @@ export function issueService(db: Db) {
|
||||
throw unprocessable("Issue can only have one assignee");
|
||||
}
|
||||
if (data.assigneeAgentId) {
|
||||
await assertAssignableAgent(companyId, data.assigneeAgentId);
|
||||
await assertAssignableAgent(db, companyId, data.assigneeAgentId, { kind: "work" });
|
||||
}
|
||||
if (data.assigneeUserId) {
|
||||
await assertAssignableUser(companyId, data.assigneeUserId);
|
||||
@@ -5021,8 +4999,11 @@ export function issueService(db: Db) {
|
||||
throw unprocessable("Issue is blocked by unresolved blockers", { unresolvedBlockerIssueIds });
|
||||
}
|
||||
}
|
||||
if (issueData.assigneeAgentId) {
|
||||
await assertAssignableAgent(existing.companyId, issueData.assigneeAgentId);
|
||||
const shouldValidateNextAssignee =
|
||||
Boolean(nextAssigneeAgentId) &&
|
||||
(issueData.assigneeAgentId !== undefined || patch.status === "in_progress");
|
||||
if (shouldValidateNextAssignee) {
|
||||
await assertAssignableAgent(dbOrTx as Db, existing.companyId, nextAssigneeAgentId, { kind: "work" });
|
||||
}
|
||||
if (issueData.assigneeUserId) {
|
||||
await assertAssignableUser(existing.companyId, issueData.assigneeUserId);
|
||||
@@ -5347,7 +5328,7 @@ export function issueService(db: Db) {
|
||||
.where(eq(issues.id, id))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!issueCompany) throw notFound("Issue not found");
|
||||
await assertAssignableAgent(issueCompany.companyId, agentId);
|
||||
await assertAssignableAgent(db, issueCompany.companyId, agentId, { kind: "work" });
|
||||
|
||||
const now = new Date();
|
||||
const activePauseHold = await treeControlSvc.getActivePauseHoldGate(issueCompany.companyId, id);
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { getAgentWorkEligibility, isAgentInvokable } from "@paperclipai/shared";
|
||||
import { buildIssueGraphLivenessIncidentKey } from "./origins.js";
|
||||
|
||||
export type IssueLivenessSeverity = "warning" | "critical";
|
||||
@@ -106,9 +107,6 @@ export interface IssueGraphLivenessInput {
|
||||
now?: Date | string;
|
||||
}
|
||||
|
||||
const INVOKABLE_AGENT_STATUSES = new Set(["active", "idle", "running", "error"]);
|
||||
const BLOCKING_AGENT_STATUSES = new Set(["paused", "terminated", "pending_approval"]);
|
||||
|
||||
function issueLabel(issue: IssueLivenessIssueInput) {
|
||||
return issue.identifier ?? issue.id;
|
||||
}
|
||||
@@ -122,8 +120,11 @@ function pathEntry(issue: IssueLivenessIssueInput): IssueLivenessDependencyPathE
|
||||
};
|
||||
}
|
||||
|
||||
function isInvokableAgent(agent: IssueLivenessAgentInput | null | undefined) {
|
||||
return Boolean(agent && INVOKABLE_AGENT_STATUSES.has(agent.status));
|
||||
function isInvokableAgent(
|
||||
agent: IssueLivenessAgentInput | null | undefined,
|
||||
agentsById: Map<string, IssueLivenessAgentInput>,
|
||||
) {
|
||||
return Boolean(agent && isAgentInvokable({ agent, agents: [...agentsById.values()] }));
|
||||
}
|
||||
|
||||
function hasActiveExecutionPath(
|
||||
@@ -209,7 +210,7 @@ function addOwnerCandidate(
|
||||
) {
|
||||
if (!agentId || seen.has(agentId)) return;
|
||||
const agent = agentsById.get(agentId);
|
||||
if (!agent || agent.companyId !== companyId || !isInvokableAgent(agent)) return;
|
||||
if (!agent || agent.companyId !== companyId || !isInvokableAgent(agent, agentsById)) return;
|
||||
seen.add(agentId);
|
||||
candidates.push({ agentId, reason, sourceIssueId });
|
||||
}
|
||||
@@ -236,9 +237,13 @@ function addAgentChainCandidates(
|
||||
}
|
||||
}
|
||||
|
||||
function orderedInvokableAgents(agents: IssueLivenessAgentInput[], companyId: string) {
|
||||
function orderedInvokableAgents(
|
||||
agents: IssueLivenessAgentInput[],
|
||||
agentsById: Map<string, IssueLivenessAgentInput>,
|
||||
companyId: string,
|
||||
) {
|
||||
return agents
|
||||
.filter((agent) => agent.companyId === companyId && isInvokableAgent(agent))
|
||||
.filter((agent) => agent.companyId === companyId && isInvokableAgent(agent, agentsById))
|
||||
.sort((left, right) => left.id.localeCompare(right.id));
|
||||
}
|
||||
|
||||
@@ -284,7 +289,7 @@ function ownerCandidatesForRecoveryIssue(
|
||||
issue.id,
|
||||
);
|
||||
|
||||
const invokableAgents = orderedInvokableAgents(agents, issue.companyId);
|
||||
const invokableAgents = orderedInvokableAgents(agents, agentsById, issue.companyId);
|
||||
for (const agent of invokableAgents) {
|
||||
if (!agent.reportsTo) {
|
||||
addOwnerCandidate(candidates, seen, agentsById, issue.companyId, agent.id, "root_agent", issue.id);
|
||||
@@ -419,7 +424,7 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu
|
||||
const participantAgentId = readPrincipalAgentId(participant);
|
||||
if (participantAgentId) {
|
||||
const participantAgent = agentsById.get(participantAgentId);
|
||||
if (isInvokableAgent(participantAgent) && participantAgent?.companyId === reviewIssue.companyId) return null;
|
||||
if (isInvokableAgent(participantAgent, agentsById) && participantAgent?.companyId === reviewIssue.companyId) return null;
|
||||
|
||||
return finding({
|
||||
issue: source,
|
||||
@@ -532,12 +537,15 @@ export function classifyIssueGraphLiveness(input: IssueGraphLivenessInput): Issu
|
||||
if (!blocker.assigneeAgentId) return null;
|
||||
|
||||
const blockerAgent = agentsById.get(blocker.assigneeAgentId);
|
||||
if (!blockerAgent || blockerAgent.companyId !== source.companyId || BLOCKING_AGENT_STATUSES.has(blockerAgent.status)) {
|
||||
const blockerEligibility = blockerAgent
|
||||
? getAgentWorkEligibility({ agent: blockerAgent, agents: input.agents })
|
||||
: null;
|
||||
if (!blockerAgent || blockerAgent.companyId !== source.companyId || !blockerEligibility?.invokable) {
|
||||
return finding({
|
||||
issue: source,
|
||||
state: "blocked_by_uninvokable_assignee",
|
||||
reason: blockerAgent
|
||||
? `${issueLabel(source)} is blocked by ${issueLabel(blocker)}, but its assignee is ${blockerAgent.status}.`
|
||||
? `${issueLabel(source)} is blocked by ${issueLabel(blocker)}, but its assignee is ${blockerEligibility?.invokabilityReason === "invalid_org_chain" ? "in an invalid org chain" : blockerAgent.status}.`
|
||||
: `${issueLabel(source)} is blocked by ${issueLabel(blocker)}, but its assignee no longer exists.`,
|
||||
dependencyPath,
|
||||
recoveryIssue: blocker,
|
||||
|
||||
@@ -36,6 +36,7 @@ import { instanceSettingsService } from "../instance-settings.js";
|
||||
import { issueRecoveryActionService } from "../issue-recovery-actions.js";
|
||||
import { issueTreeControlService } from "../issue-tree-control.js";
|
||||
import { issueService } from "../issues.js";
|
||||
import { evaluateAgentInvokabilityFromDb } from "../agent-invokability.js";
|
||||
import { getRunLogStore } from "../run-log-store.js";
|
||||
import {
|
||||
DEFAULT_MAX_SUCCESSFUL_RUN_HANDOFF_ATTEMPTS,
|
||||
@@ -327,10 +328,6 @@ function unwrapDatabaseConflictError(error: unknown) {
|
||||
};
|
||||
}
|
||||
|
||||
function isAgentInvokable(agent: typeof agents.$inferSelect | null | undefined) {
|
||||
return Boolean(agent && !["paused", "terminated", "pending_approval"].includes(agent.status));
|
||||
}
|
||||
|
||||
function isStrandedIssueRecoveryIssue(issue: Pick<typeof issues.$inferSelect, "originKind">) {
|
||||
return isStrandedIssueRecoveryOriginKind(issue.originKind);
|
||||
}
|
||||
@@ -466,6 +463,10 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
return db.select().from(agents).where(eq(agents.id, agentId)).then((rows) => rows[0] ?? null);
|
||||
}
|
||||
|
||||
async function isAgentInvokable(agent: typeof agents.$inferSelect | null | undefined) {
|
||||
return (await evaluateAgentInvokabilityFromDb(db, agent)).invokable;
|
||||
}
|
||||
|
||||
async function getLatestIssueRun(companyId: string, issueId: string): Promise<LatestIssueRun> {
|
||||
return db
|
||||
.select({
|
||||
@@ -697,7 +698,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
continue;
|
||||
}
|
||||
const creatorAgent = await getAgent(creatorAgentId);
|
||||
if (!creatorAgent || creatorAgent.companyId !== candidate.companyId || !isAgentInvokable(creatorAgent)) {
|
||||
if (!creatorAgent || creatorAgent.companyId !== candidate.companyId || !(await isAgentInvokable(creatorAgent))) {
|
||||
skipped += 1;
|
||||
continue;
|
||||
}
|
||||
@@ -1249,7 +1250,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
issueId: input.sourceIssue?.id ?? null,
|
||||
projectId: input.sourceIssue?.projectId ?? null,
|
||||
});
|
||||
if (isAgentInvokable(candidate) && !budgetBlock) return candidate.id;
|
||||
if ((await isAgentInvokable(candidate)) && !budgetBlock) return candidate.id;
|
||||
}
|
||||
|
||||
return null;
|
||||
@@ -1872,7 +1873,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
issueId: issue.id,
|
||||
projectId: issue.projectId,
|
||||
});
|
||||
if (isAgentInvokable(candidate) && !budgetBlock) return candidate.id;
|
||||
if ((await isAgentInvokable(candidate)) && !budgetBlock) return candidate.id;
|
||||
}
|
||||
|
||||
return null;
|
||||
@@ -2486,7 +2487,7 @@ export function recoveryService(db: Db, deps: { enqueueWakeup: RecoveryWakeup })
|
||||
}
|
||||
|
||||
const agent = await getAgent(agentId);
|
||||
if (!agent || agent.companyId !== issue.companyId || !isAgentInvokable(agent)) {
|
||||
if (!agent || agent.companyId !== issue.companyId || !(await isAgentInvokable(agent))) {
|
||||
result.skipped += 1;
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -52,6 +52,7 @@ import { logger } from "../middleware/logger.js";
|
||||
import { getTelemetryClient } from "../telemetry.js";
|
||||
import { getConfiguredSecretProvider } from "../secrets/configured-provider.js";
|
||||
import { issueService } from "./issues.js";
|
||||
import { assertAssignableAgent } from "./agent-assignability.js";
|
||||
import { secretService } from "./secrets.js";
|
||||
import { getSecretProvider } from "../secrets/provider-registry.js";
|
||||
import { parseCron, validateCron } from "./cron.js";
|
||||
@@ -610,25 +611,12 @@ export function routineService(
|
||||
return routine;
|
||||
}
|
||||
|
||||
async function assertAssignableAgent(companyId: string, agentId: string | null | undefined) {
|
||||
if (!agentId) return;
|
||||
const agent = await db
|
||||
.select({ id: agents.id, companyId: agents.companyId, status: agents.status })
|
||||
.from(agents)
|
||||
.where(eq(agents.id, agentId))
|
||||
.then((rows) => rows[0] ?? null);
|
||||
if (!agent) throw notFound("Assignee agent not found");
|
||||
if (agent.companyId !== companyId) throw unprocessable("Assignee must belong to same company");
|
||||
if (agent.status === "pending_approval") throw conflict("Cannot assign routines to pending approval agents");
|
||||
if (agent.status === "terminated") throw conflict("Cannot assign routines to terminated agents");
|
||||
}
|
||||
|
||||
async function assertRestorableAssignee(
|
||||
companyId: string,
|
||||
assigneeAgentId: string | null | undefined,
|
||||
actor: Actor,
|
||||
) {
|
||||
await assertAssignableAgent(companyId, assigneeAgentId);
|
||||
await assertAssignableAgent(db, companyId, assigneeAgentId, { kind: "routine" });
|
||||
if (actor.agentId && assigneeAgentId !== actor.agentId) {
|
||||
throw forbidden("Agents can only restore routine revisions assigned to themselves");
|
||||
}
|
||||
@@ -1170,6 +1158,7 @@ export function routineService(
|
||||
if (!assigneeAgentId) {
|
||||
throw unprocessable("Default agent required");
|
||||
}
|
||||
await assertAssignableAgent(db, input.routine.companyId, assigneeAgentId, { kind: "routine" });
|
||||
const automaticVariables: Record<string, string | number | boolean> = {};
|
||||
if (input.executionWorkspaceId && routineUsesWorkspaceBranch(input.routine)) {
|
||||
const workspace = await db
|
||||
@@ -1581,7 +1570,7 @@ export function routineService(
|
||||
|
||||
create: async (companyId: string, input: CreateRoutine, actor: Actor): Promise<Routine> => {
|
||||
await assertProject(companyId, input.projectId ?? null);
|
||||
await assertAssignableAgent(companyId, input.assigneeAgentId ?? null);
|
||||
await assertAssignableAgent(db, companyId, input.assigneeAgentId ?? null, { kind: "routine" });
|
||||
if (input.goalId) await assertGoal(companyId, input.goalId);
|
||||
if (input.parentIssueId) await assertParentIssue(companyId, input.parentIssueId);
|
||||
const env = input.env === undefined || input.env === null
|
||||
@@ -1663,7 +1652,9 @@ export function routineService(
|
||||
patch.variables === undefined ? existing.variables : sanitizeRoutineVariableInputs(patch.variables),
|
||||
);
|
||||
if (patch.projectId !== undefined) await assertProject(existing.companyId, nextProjectId);
|
||||
if (patch.assigneeAgentId !== undefined) await assertAssignableAgent(existing.companyId, nextAssigneeAgentId);
|
||||
if (patch.assigneeAgentId !== undefined || patch.status === "active") {
|
||||
await assertAssignableAgent(db, existing.companyId, nextAssigneeAgentId, { kind: "routine" });
|
||||
}
|
||||
if (patch.goalId) await assertGoal(existing.companyId, patch.goalId);
|
||||
if (patch.parentIssueId) await assertParentIssue(existing.companyId, patch.parentIssueId);
|
||||
assertRoutineVariableDefinitions(nextVariables);
|
||||
@@ -2187,7 +2178,8 @@ export function routineService(
|
||||
if (!routine) throw notFound("Routine not found");
|
||||
if (routine.status === "archived") throw conflict("Routine is archived");
|
||||
await assertProject(routine.companyId, input.projectId ?? null);
|
||||
await assertAssignableAgent(routine.companyId, input.assigneeAgentId ?? null);
|
||||
const assigneeAgentId = input.assigneeAgentId ?? routine.assigneeAgentId ?? null;
|
||||
await assertAssignableAgent(db, routine.companyId, assigneeAgentId, { kind: "routine" });
|
||||
const trigger = input.triggerId ? await getTriggerById(input.triggerId) : null;
|
||||
if (trigger && trigger.routineId !== routine.id) throw forbidden("Trigger does not belong to routine");
|
||||
if (trigger && !trigger.enabled) throw conflict("Routine trigger is not active");
|
||||
|
||||
Reference in New Issue
Block a user