mirror of
https://github.com/alkimake/paperclip.git
synced 2026-06-17 03:10:38 +09:00
[codex] Make heartbeat scheduling blocker-aware (#4157)
## Thinking Path > - Paperclip orchestrates AI agents through issue-driven heartbeats, checkouts, and wake scheduling. > - This change sits in the server heartbeat and issue services that decide which queued runs are allowed to start. > - Before this branch, queued heartbeats could be selected even when their issue still had unresolved blocker relationships. > - That let blocked descendant work compete with actually-ready work and risked auto-checking out issues that were not dependency-ready. > - This pull request teaches the scheduler and checkout path to consult issue dependency readiness before claiming queued runs. > - It also exposes dependency readiness in the agent inbox so agents can see which assigned issues are still blocked. > - The result is that heartbeat execution follows the DAG of blocked dependencies instead of waking work out of order. ## What Changed - Added `IssueDependencyReadiness` helpers to `issueService`, including unresolved blocker lookup for single issues and bulk issue lists. - Prevented issue checkout and `in_progress` transitions when unresolved blockers still exist. - Made heartbeat queued-run claiming and prioritization dependency-aware so ready work starts before blocked descendants. - Included dependency readiness fields in `/api/agents/me/inbox-lite` for agent heartbeat selection. - Added regression coverage for dependency-aware heartbeat promotion and issue-service participation filtering. ## Verification - `pnpm run preflight:workspace-links` - `pnpm exec vitest run server/src/__tests__/heartbeat-dependency-scheduling.test.ts server/src/__tests__/issues-service.test.ts` - On this host, the Vitest command passed, but the embedded-Postgres portions of those files were skipped because `@embedded-postgres/darwin-x64` is not installed. ## Risks - Scheduler ordering now prefers dependency-ready runs, so any hidden assumptions about strict FIFO ordering could surface in edge cases. - The new guardrails reject checkout or `in_progress` transitions for blocked issues; callers depending on the old permissive behavior would now get `422` errors. - Local verification did not execute the embedded-Postgres integration paths on this macOS host because the platform binary package was missing. > I checked `ROADMAP.md`; this is a targeted execution/scheduling fix and does not duplicate planned roadmap feature work. ## Model Used - OpenAI Codex via the Paperclip `codex_local` adapter in this workspace. Exact backend model ID is not surfaced in the runtime here; tool-enabled coding agent with terminal execution and repository editing capabilities. ## 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 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 - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] I will address all Greptile and reviewer comments before requesting merge
This commit is contained in:
Dotta
GitHub
parent
1bf2424377
commit
1266954a4e
5 changed files with 581 additions and 4 deletions
|
|
@ -1211,9 +1211,11 @@ function shouldAutoCheckoutIssueForWake(input: {
|
|||
contextSnapshot: Record<string, unknown> | null | undefined;
|
||||
issueStatus: string | null;
|
||||
issueAssigneeAgentId: string | null;
|
||||
isDependencyReady: boolean;
|
||||
agentId: string;
|
||||
}) {
|
||||
if (input.issueAssigneeAgentId !== input.agentId) return false;
|
||||
if (!input.isDependencyReady) return false;
|
||||
|
||||
const issueStatus = readNonEmptyString(input.issueStatus);
|
||||
if (
|
||||
|
|
@ -3062,6 +3064,36 @@ export function heartbeatService(db: Db) {
|
|||
};
|
||||
}
|
||||
|
||||
function issueRunPriorityRank(priority: string | null | undefined) {
|
||||
switch (priority) {
|
||||
case "critical":
|
||||
return 0;
|
||||
case "high":
|
||||
return 1;
|
||||
case "medium":
|
||||
return 2;
|
||||
case "low":
|
||||
return 3;
|
||||
default:
|
||||
return 4;
|
||||
}
|
||||
}
|
||||
|
||||
async function listQueuedRunDependencyReadiness(
|
||||
companyId: string,
|
||||
queuedRuns: Array<typeof heartbeatRuns.$inferSelect>,
|
||||
) {
|
||||
const issueIds = [...new Set(
|
||||
queuedRuns
|
||||
.map((run) => readNonEmptyString(parseObject(run.contextSnapshot).issueId))
|
||||
.filter((issueId): issueId is string => Boolean(issueId)),
|
||||
)];
|
||||
if (issueIds.length === 0) {
|
||||
return new Map<string, Awaited<ReturnType<typeof issuesSvc.getDependencyReadiness>>>();
|
||||
}
|
||||
return issuesSvc.listDependencyReadiness(companyId, issueIds);
|
||||
}
|
||||
|
||||
async function countRunningRunsForAgent(agentId: string) {
|
||||
const [{ count }] = await db
|
||||
.select({ count: sql<number>`count(*)` })
|
||||
|
|
@ -3092,6 +3124,16 @@ export function heartbeatService(db: Db) {
|
|||
return null;
|
||||
}
|
||||
|
||||
const issueId = readNonEmptyString(context.issueId);
|
||||
if (issueId) {
|
||||
const dependencyReadiness = await issuesSvc.listDependencyReadiness(run.companyId, [issueId]);
|
||||
const unresolvedBlockerCount = dependencyReadiness.get(issueId)?.unresolvedBlockerCount ?? 0;
|
||||
if (unresolvedBlockerCount > 0) {
|
||||
logger.debug({ runId: run.id, issueId, unresolvedBlockerCount }, "claimQueuedRun: skipping blocked run");
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
const claimedAt = new Date();
|
||||
const claimed = await db
|
||||
.update(heartbeatRuns)
|
||||
|
|
@ -3859,12 +3901,49 @@ export function heartbeatService(db: Db) {
|
|||
.select()
|
||||
.from(heartbeatRuns)
|
||||
.where(and(eq(heartbeatRuns.agentId, agentId), eq(heartbeatRuns.status, "queued")))
|
||||
.orderBy(asc(heartbeatRuns.createdAt))
|
||||
.limit(availableSlots);
|
||||
.orderBy(asc(heartbeatRuns.createdAt));
|
||||
if (queuedRuns.length === 0) return [];
|
||||
|
||||
const dependencyReadiness = await listQueuedRunDependencyReadiness(agent.companyId, queuedRuns);
|
||||
const queuedIssueIds = [...new Set(
|
||||
queuedRuns
|
||||
.map((run) => readNonEmptyString(parseObject(run.contextSnapshot).issueId))
|
||||
.filter((issueId): issueId is string => Boolean(issueId)),
|
||||
)];
|
||||
const issueRows = await db
|
||||
.select({
|
||||
id: issues.id,
|
||||
status: issues.status,
|
||||
priority: issues.priority,
|
||||
})
|
||||
.from(issues)
|
||||
.where(
|
||||
queuedIssueIds.length > 0
|
||||
? and(eq(issues.companyId, agent.companyId), inArray(issues.id, queuedIssueIds))
|
||||
: sql`false`,
|
||||
);
|
||||
const issueById = new Map(issueRows.map((row) => [row.id, row]));
|
||||
const prioritizedRuns = [...queuedRuns].sort((left, right) => {
|
||||
const leftIssueId = readNonEmptyString(parseObject(left.contextSnapshot).issueId);
|
||||
const rightIssueId = readNonEmptyString(parseObject(right.contextSnapshot).issueId);
|
||||
const leftReadiness = leftIssueId ? dependencyReadiness.get(leftIssueId) : null;
|
||||
const rightReadiness = rightIssueId ? dependencyReadiness.get(rightIssueId) : null;
|
||||
const leftReady = leftIssueId ? (leftReadiness?.isDependencyReady ?? true) : true;
|
||||
const rightReady = rightIssueId ? (rightReadiness?.isDependencyReady ?? true) : true;
|
||||
const leftIssue = leftIssueId ? issueById.get(leftIssueId) : null;
|
||||
const rightIssue = rightIssueId ? issueById.get(rightIssueId) : null;
|
||||
const leftRank = leftIssueId ? (leftReady ? (leftIssue?.status === "in_progress" ? 0 : 1) : 3) : 2;
|
||||
const rightRank = rightIssueId ? (rightReady ? (rightIssue?.status === "in_progress" ? 0 : 1) : 3) : 2;
|
||||
if (leftRank !== rightRank) return leftRank - rightRank;
|
||||
const leftPriorityRank = issueRunPriorityRank(leftIssue?.priority);
|
||||
const rightPriorityRank = issueRunPriorityRank(rightIssue?.priority);
|
||||
if (leftPriorityRank !== rightPriorityRank) return leftPriorityRank - rightPriorityRank;
|
||||
return left.createdAt.getTime() - right.createdAt.getTime();
|
||||
});
|
||||
|
||||
const claimedRuns: Array<typeof heartbeatRuns.$inferSelect> = [];
|
||||
for (const queuedRun of queuedRuns) {
|
||||
for (const queuedRun of prioritizedRuns) {
|
||||
if (claimedRuns.length >= availableSlots) break;
|
||||
const claimed = await claimQueuedRun(queuedRun);
|
||||
if (claimed) claimedRuns.push(claimed);
|
||||
}
|
||||
|
|
@ -3887,7 +3966,7 @@ export function heartbeatService(db: Db) {
|
|||
if (run.status === "queued") {
|
||||
const claimed = await claimQueuedRun(run);
|
||||
if (!claimed) {
|
||||
// Another worker has already claimed or finalized this run.
|
||||
// claimQueuedRun can also leave the run queued when dependencies are unresolved.
|
||||
return;
|
||||
}
|
||||
run = claimed;
|
||||
|
|
@ -3918,6 +3997,9 @@ export function heartbeatService(db: Db) {
|
|||
const sessionCodec = getAdapterSessionCodec(agent.adapterType);
|
||||
const issueId = readNonEmptyString(context.issueId);
|
||||
let issueContext = issueId ? await getIssueExecutionContext(agent.companyId, issueId) : null;
|
||||
const issueDependencyReadiness = issueId
|
||||
? await issuesSvc.listDependencyReadiness(agent.companyId, [issueId]).then((rows) => rows.get(issueId) ?? null)
|
||||
: null;
|
||||
if (
|
||||
issueId &&
|
||||
issueContext &&
|
||||
|
|
@ -3925,6 +4007,7 @@ export function heartbeatService(db: Db) {
|
|||
contextSnapshot: context,
|
||||
issueStatus: issueContext.status,
|
||||
issueAssigneeAgentId: issueContext.assigneeAgentId,
|
||||
isDependencyReady: issueDependencyReadiness?.isDependencyReady ?? true,
|
||||
agentId: agent.id,
|
||||
})
|
||||
) {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue