mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-11 05:11:49 +01:00
fix(heartbeat): gate dependent wakes on workspace-finalize barrier (PAPA-432)
When a predecessor issue is marked `done`, the dependent's `issue_blockers_resolved` wake used to fire purely on issue status. There was no barrier ensuring the predecessor's most recent run had synced its remote-side commits back into the local execution-workspace cwd before the dependent's `prepareWorkspaceForSshExecution` read it. The verified incident on PAPA-412 saw QA's run bundle the local worktree ~8m 39s before the predecessor's commit landed in it. This change adds a workspace-finalize barrier to the dependency- readiness check: - New workspaceOperations phase `workspace_finalize`. The heartbeat records `succeeded` after `adapter.execute` (and its restore finally block) returns cleanly, or `failed` if it throws. Failures now bubble up as run-level errors instead of being silent. - `listIssueDependencyReadinessMap` requires every done blocker's execution workspace to have a successful finalize as its latest op. Otherwise the dependent is marked not-ready with the blocker listed in the new `pendingFinalizeBlockerIssueIds` field. - `listWakeableBlockedDependents` now defers to the unified readiness check so the wake only fires when both status and finalize gates pass. - After a successful finalize for a `done` issue, the heartbeat re-fires `issue_blockers_resolved` wakes for any dependents that were previously gated, with `deferredFor: workspace_finalize` in the payload for traceability. Includes a new ssh-fixture contract test that asserts the round-trip propagates remote commits to the local worktree with no git remote configured at any point — documenting the architectural invariant that the local execution-workspace cwd is the only persistence boundary across runs. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
07266c07b7
commit
1acf2f1078
5 files changed
+463
-58
No files matched your search
@@ -451,6 +451,68 @@ describe("ssh env-lab fixture", () => {
|
||||
await expect(readFile(path.join(localRepo, "tracked.txt"), "utf8")).resolves.toBe("dirty remote\n");
|
||||
}, SSH_FIXTURE_TEST_TIMEOUT_MS);
|
||||
|
||||
it("propagates remote commits to the local worktree with no git remote configured (no-remote-git contract)", async () => {
|
||||
// Locks in the architectural contract documented in
|
||||
// packages/adapter-utils/README.md and packages/adapters/AUTHORING.md:
|
||||
// the local execution-workspace cwd is the only persistence boundary
|
||||
// across runs. No adapter may depend on a git remote for cross-run state.
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-ssh-fixture-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
const statePath = path.join(rootDir, "state.json");
|
||||
const localRepo = path.join(rootDir, "local-workspace");
|
||||
|
||||
await mkdir(localRepo, { recursive: true });
|
||||
await git(localRepo, ["init"]);
|
||||
await git(localRepo, ["checkout", "-b", "main"]);
|
||||
await git(localRepo, ["config", "user.name", "Paperclip Test"]);
|
||||
await git(localRepo, ["config", "user.email", "test@paperclip.dev"]);
|
||||
await writeFile(path.join(localRepo, "tracked.txt"), "base\n", "utf8");
|
||||
await git(localRepo, ["add", "tracked.txt"]);
|
||||
await git(localRepo, ["commit", "-m", "initial"]);
|
||||
|
||||
// Assert there is no git remote configured before we begin, and verify
|
||||
// that no point in the round-trip introduces one. `git remote` returns an
|
||||
// empty string when no remotes exist (and exit code 0).
|
||||
expect(await git(localRepo, ["remote"])).toBe("");
|
||||
|
||||
const started = await startSshEnvLabFixtureOrSkip(
|
||||
statePath,
|
||||
"no-remote-git contract test",
|
||||
);
|
||||
if (!started) return;
|
||||
const config = await buildSshEnvLabFixtureConfig(started);
|
||||
const spec = {
|
||||
...config,
|
||||
remoteCwd: started.workspaceDir,
|
||||
} as const;
|
||||
|
||||
const prepared = await prepareRemoteManagedRuntime({
|
||||
spec,
|
||||
runId: "run-no-remote",
|
||||
adapterKey: "test-adapter",
|
||||
workspaceLocalDir: localRepo,
|
||||
});
|
||||
|
||||
// Remote commit lands a deliverable that must show up locally via
|
||||
// sync-back alone — no `git push`, no fetch from any origin.
|
||||
await runSshCommand(
|
||||
config,
|
||||
`cd ${JSON.stringify(prepared.workspaceRemoteDir)} && git config user.name "Paperclip SSH" && git config user.email "ssh@paperclip.dev" && printf "deliverable\\n" > tracked.txt && git add tracked.txt && git commit -m "remote-only commit" >/dev/null`,
|
||||
{ timeoutMs: 30_000, maxBuffer: 256 * 1024 },
|
||||
);
|
||||
|
||||
await prepared.restoreWorkspace();
|
||||
|
||||
expect(await git(localRepo, ["log", "-1", "--pretty=%s"])).toBe(
|
||||
"remote-only commit",
|
||||
);
|
||||
expect(await readFile(path.join(localRepo, "tracked.txt"), "utf8")).toBe(
|
||||
"deliverable\n",
|
||||
);
|
||||
// Final assertion: still no git remote — restore did not silently add one.
|
||||
expect(await git(localRepo, ["remote"])).toBe("");
|
||||
}, SSH_FIXTURE_TEST_TIMEOUT_MS);
|
||||
|
||||
it("merges concurrent remote commits through the managed runtime restore path", async () => {
|
||||
const rootDir = await mkdtemp(path.join(os.tmpdir(), "paperclip-ssh-fixture-"));
|
||||
cleanupDirs.push(rootDir);
|
||||
|
||||
@@ -2,7 +2,8 @@ export type WorkspaceOperationPhase =
|
||||
| "worktree_prepare"
|
||||
| "workspace_provision"
|
||||
| "workspace_teardown"
|
||||
| "worktree_cleanup";
|
||||
| "worktree_cleanup"
|
||||
| "workspace_finalize";
|
||||
|
||||
export type WorkspaceOperationStatus = "running" | "succeeded" | "failed" | "skipped";
|
||||
|
||||
|
||||
@@ -18,6 +18,7 @@ import {
|
||||
issues,
|
||||
projectWorkspaces,
|
||||
projects,
|
||||
workspaceOperations,
|
||||
} from "@paperclipai/db";
|
||||
import {
|
||||
getEmbeddedPostgresTestSupport,
|
||||
@@ -2278,6 +2279,7 @@ describeEmbeddedPostgres("issueService blockers and dependency wake readiness",
|
||||
await db.delete(issueInboxArchives);
|
||||
await db.delete(activityLog);
|
||||
await db.delete(issues);
|
||||
await db.delete(workspaceOperations);
|
||||
await db.delete(executionWorkspaces);
|
||||
await db.delete(projectWorkspaces);
|
||||
await db.delete(projects);
|
||||
@@ -2447,6 +2449,179 @@ describeEmbeddedPostgres("issueService blockers and dependency wake readiness",
|
||||
]);
|
||||
});
|
||||
|
||||
it("gates dependents on the workspace-finalize barrier when a done blocker's execution workspace has not synced back", async () => {
|
||||
const companyId = randomUUID();
|
||||
const assigneeAgentId = randomUUID();
|
||||
const projectId = randomUUID();
|
||||
const projectWorkspaceId = randomUUID();
|
||||
const executionWorkspaceId = randomUUID();
|
||||
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
await db.insert(agents).values({
|
||||
id: assigneeAgentId,
|
||||
companyId,
|
||||
name: "QA",
|
||||
role: "qa",
|
||||
status: "active",
|
||||
adapterType: "claude_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
await db.insert(projects).values({
|
||||
id: projectId,
|
||||
companyId,
|
||||
name: "Shared workspace project",
|
||||
status: "in_progress",
|
||||
});
|
||||
await db.insert(projectWorkspaces).values({
|
||||
id: projectWorkspaceId,
|
||||
companyId,
|
||||
projectId,
|
||||
name: "Shared workspace",
|
||||
sourceType: "local_path",
|
||||
visibility: "default",
|
||||
isPrimary: true,
|
||||
});
|
||||
await db.insert(executionWorkspaces).values({
|
||||
id: executionWorkspaceId,
|
||||
companyId,
|
||||
projectId,
|
||||
projectWorkspaceId,
|
||||
mode: "isolated_workspace",
|
||||
strategyType: "git_worktree",
|
||||
name: "Shared exec workspace",
|
||||
status: "active",
|
||||
providerType: "git_worktree",
|
||||
});
|
||||
|
||||
const blockerId = randomUUID();
|
||||
const dependentId = randomUUID();
|
||||
await db.insert(issues).values([
|
||||
{
|
||||
id: blockerId,
|
||||
companyId,
|
||||
projectId,
|
||||
title: "Predecessor",
|
||||
status: "done",
|
||||
priority: "medium",
|
||||
executionWorkspaceId,
|
||||
},
|
||||
{
|
||||
id: dependentId,
|
||||
companyId,
|
||||
projectId,
|
||||
title: "Dependent",
|
||||
status: "blocked",
|
||||
priority: "medium",
|
||||
assigneeAgentId,
|
||||
},
|
||||
]);
|
||||
await svc.update(dependentId, { blockedByIssueIds: [blockerId] });
|
||||
|
||||
// A run touched the workspace (prepare phase) but has not yet recorded
|
||||
// workspace_finalize — the dependent must NOT wake.
|
||||
await db.insert(workspaceOperations).values({
|
||||
companyId,
|
||||
executionWorkspaceId,
|
||||
phase: "worktree_prepare",
|
||||
status: "succeeded",
|
||||
startedAt: new Date("2026-05-23T22:00:00.000Z"),
|
||||
});
|
||||
|
||||
expect(await svc.listWakeableBlockedDependents(blockerId)).toEqual([]);
|
||||
await expect(svc.getDependencyReadiness(dependentId)).resolves.toMatchObject({
|
||||
isDependencyReady: false,
|
||||
pendingFinalizeBlockerIssueIds: [blockerId],
|
||||
unresolvedBlockerIssueIds: [blockerId],
|
||||
});
|
||||
|
||||
// A failed finalize must keep the gate closed.
|
||||
await db.insert(workspaceOperations).values({
|
||||
companyId,
|
||||
executionWorkspaceId,
|
||||
phase: "workspace_finalize",
|
||||
status: "failed",
|
||||
startedAt: new Date("2026-05-23T22:05:00.000Z"),
|
||||
});
|
||||
expect(await svc.listWakeableBlockedDependents(blockerId)).toEqual([]);
|
||||
|
||||
// Once a workspace_finalize succeeded row lands AFTER the failed one,
|
||||
// the gate opens and the dependent is wakeable.
|
||||
await db.insert(workspaceOperations).values({
|
||||
companyId,
|
||||
executionWorkspaceId,
|
||||
phase: "workspace_finalize",
|
||||
status: "succeeded",
|
||||
startedAt: new Date("2026-05-23T22:10:00.000Z"),
|
||||
});
|
||||
|
||||
await expect(svc.listWakeableBlockedDependents(blockerId)).resolves.toEqual([
|
||||
expect.objectContaining({
|
||||
id: dependentId,
|
||||
assigneeAgentId,
|
||||
blockerIssueIds: [blockerId],
|
||||
}),
|
||||
]);
|
||||
await expect(svc.getDependencyReadiness(dependentId)).resolves.toMatchObject({
|
||||
isDependencyReady: true,
|
||||
pendingFinalizeBlockerIssueIds: [],
|
||||
});
|
||||
});
|
||||
|
||||
it("treats blockers with no executionWorkspaceId as not subject to the workspace-finalize barrier", async () => {
|
||||
const companyId = randomUUID();
|
||||
const assigneeAgentId = randomUUID();
|
||||
|
||||
await db.insert(companies).values({
|
||||
id: companyId,
|
||||
name: "Paperclip",
|
||||
issuePrefix: `T${companyId.replace(/-/g, "").slice(0, 6).toUpperCase()}`,
|
||||
requireBoardApprovalForNewAgents: false,
|
||||
});
|
||||
await db.insert(agents).values({
|
||||
id: assigneeAgentId,
|
||||
companyId,
|
||||
name: "QA",
|
||||
role: "qa",
|
||||
status: "active",
|
||||
adapterType: "claude_local",
|
||||
adapterConfig: {},
|
||||
runtimeConfig: {},
|
||||
permissions: {},
|
||||
});
|
||||
|
||||
const blockerId = randomUUID();
|
||||
const dependentId = randomUUID();
|
||||
await db.insert(issues).values([
|
||||
// Done blocker with no execution workspace ever attached (e.g. closed manually).
|
||||
{ id: blockerId, companyId, title: "Manual done blocker", status: "done", priority: "medium" },
|
||||
{
|
||||
id: dependentId,
|
||||
companyId,
|
||||
title: "Dependent",
|
||||
status: "blocked",
|
||||
priority: "medium",
|
||||
assigneeAgentId,
|
||||
},
|
||||
]);
|
||||
await svc.update(dependentId, { blockedByIssueIds: [blockerId] });
|
||||
|
||||
// No executionWorkspaceId → no barrier → dependent should be wakeable.
|
||||
await expect(svc.listWakeableBlockedDependents(blockerId)).resolves.toEqual([
|
||||
expect.objectContaining({
|
||||
id: dependentId,
|
||||
assigneeAgentId,
|
||||
blockerIssueIds: [blockerId],
|
||||
}),
|
||||
]);
|
||||
});
|
||||
|
||||
it("reports dependency readiness for blocked issue chains", async () => {
|
||||
const companyId = randomUUID();
|
||||
await db.insert(companies).values({
|
||||
|
||||
@@ -7808,31 +7808,72 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
"local agent jwt secret missing or invalid; running without injected PAPERCLIP_API_KEY",
|
||||
);
|
||||
}
|
||||
const adapterResult = await adapter.execute({
|
||||
runId: run.id,
|
||||
agent,
|
||||
runtime: runtimeForAdapter,
|
||||
config: runtimeConfig,
|
||||
context,
|
||||
runtimeCommandSpec: adapter.getRuntimeCommandSpec?.(runtimeConfig) ?? null,
|
||||
executionTarget,
|
||||
executionTransport: remoteExecution
|
||||
? { remoteExecution: remoteExecution as unknown as Record<string, unknown> }
|
||||
: undefined,
|
||||
onLog,
|
||||
onMeta: onAdapterMeta,
|
||||
onSpawn: async (meta) => {
|
||||
await persistRunProcessMetadata(run.id, {
|
||||
pid: meta.pid,
|
||||
processGroupId:
|
||||
"processGroupId" in meta && typeof meta.processGroupId === "number"
|
||||
? meta.processGroupId
|
||||
: null,
|
||||
startedAt: meta.startedAt,
|
||||
let adapterFinalizeOutcome: "succeeded" | "failed" | null = null;
|
||||
const recordWorkspaceFinalize = async (
|
||||
status: "succeeded" | "failed",
|
||||
metadata?: Record<string, unknown>,
|
||||
) => {
|
||||
if (adapterFinalizeOutcome) return;
|
||||
adapterFinalizeOutcome = status;
|
||||
try {
|
||||
await workspaceOperationRecorder.recordOperation({
|
||||
phase: "workspace_finalize",
|
||||
cwd: executionWorkspace.cwd,
|
||||
metadata: {
|
||||
adapterType: agent.adapterType,
|
||||
executionTargetKind: executionTarget?.kind ?? "local",
|
||||
...metadata,
|
||||
},
|
||||
run: async () => ({ status }),
|
||||
});
|
||||
},
|
||||
authToken: authToken ?? undefined,
|
||||
});
|
||||
} catch (recordErr) {
|
||||
logger.warn(
|
||||
{ err: recordErr, runId: run.id, executionWorkspaceId: persistedExecutionWorkspace?.id ?? null },
|
||||
"failed to record workspace_finalize operation",
|
||||
);
|
||||
}
|
||||
};
|
||||
|
||||
let adapterResult: Awaited<ReturnType<typeof adapter.execute>>;
|
||||
try {
|
||||
adapterResult = await adapter.execute({
|
||||
runId: run.id,
|
||||
agent,
|
||||
runtime: runtimeForAdapter,
|
||||
config: runtimeConfig,
|
||||
context,
|
||||
runtimeCommandSpec: adapter.getRuntimeCommandSpec?.(runtimeConfig) ?? null,
|
||||
executionTarget,
|
||||
executionTransport: remoteExecution
|
||||
? { remoteExecution: remoteExecution as unknown as Record<string, unknown> }
|
||||
: undefined,
|
||||
onLog,
|
||||
onMeta: onAdapterMeta,
|
||||
onSpawn: async (meta) => {
|
||||
await persistRunProcessMetadata(run.id, {
|
||||
pid: meta.pid,
|
||||
processGroupId:
|
||||
"processGroupId" in meta && typeof meta.processGroupId === "number"
|
||||
? meta.processGroupId
|
||||
: null,
|
||||
startedAt: meta.startedAt,
|
||||
});
|
||||
},
|
||||
authToken: authToken ?? undefined,
|
||||
});
|
||||
// Adapter returned cleanly, which means its workspace-restore finally
|
||||
// block also ran without throwing. Record the workspace_finalize
|
||||
// barrier so dependents that share this executionWorkspace can wake.
|
||||
await recordWorkspaceFinalize("succeeded");
|
||||
} catch (adapterErr) {
|
||||
// Adapter (or its restore finally) threw. The workspace may be in a
|
||||
// partial state — record finalize=failed so the dependent readiness
|
||||
// check keeps the gate closed instead of waking on stale local state.
|
||||
await recordWorkspaceFinalize("failed", {
|
||||
errorMessage: adapterErr instanceof Error ? adapterErr.message : String(adapterErr),
|
||||
});
|
||||
throw adapterErr;
|
||||
}
|
||||
const adapterManagedRuntimeServices = adapterResult.runtimeServices
|
||||
? await persistAdapterManagedRuntimeServices({
|
||||
db,
|
||||
@@ -8078,6 +8119,54 @@ export function heartbeatService(db: Db, options: HeartbeatServiceOptions = {})
|
||||
: livenessRun,
|
||||
agent,
|
||||
);
|
||||
|
||||
// Workspace-finalize wake re-fire: if this run's issue was marked done
|
||||
// mid-run (so the original `issue_blockers_resolved` wake was gated by
|
||||
// the readiness check waiting for workspace_finalize), the finalize
|
||||
// row we just recorded now lets dependents proceed. Fire wakes here.
|
||||
if (issueId && adapterFinalizeOutcome === "succeeded") {
|
||||
try {
|
||||
const blockerIssueStatus = await db
|
||||
.select({ status: issues.status })
|
||||
.from(issues)
|
||||
.where(eq(issues.id, issueId))
|
||||
.then((rows) => rows[0]?.status ?? null);
|
||||
if (blockerIssueStatus === "done") {
|
||||
const dependents = await issuesSvc.listWakeableBlockedDependents(issueId);
|
||||
for (const dependent of dependents) {
|
||||
await enqueueWakeup(dependent.assigneeAgentId, {
|
||||
source: "automation",
|
||||
triggerDetail: "system",
|
||||
reason: "issue_blockers_resolved",
|
||||
payload: {
|
||||
issueId: dependent.id,
|
||||
resolvedBlockerIssueId: issueId,
|
||||
blockerIssueIds: dependent.blockerIssueIds,
|
||||
deferredFor: "workspace_finalize",
|
||||
},
|
||||
contextSnapshot: {
|
||||
issueId: dependent.id,
|
||||
taskId: dependent.id,
|
||||
wakeReason: "issue_blockers_resolved",
|
||||
source: "workspace.finalize",
|
||||
resolvedBlockerIssueId: issueId,
|
||||
blockerIssueIds: dependent.blockerIssueIds,
|
||||
},
|
||||
}).catch((wakeErr) => {
|
||||
logger.warn(
|
||||
{ err: wakeErr, issueId, dependentIssueId: dependent.id, agentId: dependent.assigneeAgentId },
|
||||
"failed to fire deferred dependent wake after workspace_finalize",
|
||||
);
|
||||
});
|
||||
}
|
||||
}
|
||||
} catch (finalizeWakeErr) {
|
||||
logger.warn(
|
||||
{ err: finalizeWakeErr, runId: run.id, issueId },
|
||||
"failed to evaluate dependent wakes after workspace_finalize",
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (finalizedRun) {
|
||||
|
||||
+111
-33
@@ -27,6 +27,7 @@ import {
|
||||
labels,
|
||||
projectWorkspaces,
|
||||
projects,
|
||||
workspaceOperations,
|
||||
} from "@paperclipai/db";
|
||||
import type {
|
||||
IssueCommentAuthorType,
|
||||
@@ -312,6 +313,8 @@ export type IssueDependencyReadiness = {
|
||||
blockerIssueIds: string[];
|
||||
unresolvedBlockerIssueIds: string[];
|
||||
unresolvedBlockerCount: number;
|
||||
/** Blockers whose status is `done` but whose execution workspace has not yet finalized. */
|
||||
pendingFinalizeBlockerIssueIds: string[];
|
||||
allBlockersDone: boolean;
|
||||
isDependencyReady: boolean;
|
||||
};
|
||||
@@ -382,11 +385,70 @@ function createIssueDependencyReadiness(issueId: string): IssueDependencyReadine
|
||||
blockerIssueIds: [],
|
||||
unresolvedBlockerIssueIds: [],
|
||||
unresolvedBlockerCount: 0,
|
||||
pendingFinalizeBlockerIssueIds: [],
|
||||
allBlockersDone: true,
|
||||
isDependencyReady: true,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the set of execution-workspace ids whose most recent workspace operation
|
||||
* is NOT a successful `workspace_finalize`. These workspaces have either an in-flight
|
||||
* run, a failed finalize, or never reached the finalize barrier — dependents that
|
||||
* read this workspace must wait until finalize succeeds.
|
||||
*
|
||||
* Workspaces with no recorded operations are considered finalized (nothing has
|
||||
* touched them since they were realized).
|
||||
*/
|
||||
async function listUnfinalizedExecutionWorkspaceIds(
|
||||
dbOrTx: Pick<Db, "select">,
|
||||
companyId: string,
|
||||
executionWorkspaceIds: string[],
|
||||
): Promise<Set<string>> {
|
||||
const unfinalized = new Set<string>();
|
||||
if (executionWorkspaceIds.length === 0) return unfinalized;
|
||||
|
||||
// Pull every workspace op for the candidate workspaces and pick the latest per
|
||||
// workspace in memory. Per-workspace LATERAL queries would be tighter, but the
|
||||
// candidate set is tiny in practice (one workspace per blocker per readiness call).
|
||||
const rows = await dbOrTx
|
||||
.select({
|
||||
executionWorkspaceId: workspaceOperations.executionWorkspaceId,
|
||||
phase: workspaceOperations.phase,
|
||||
status: workspaceOperations.status,
|
||||
startedAt: workspaceOperations.startedAt,
|
||||
})
|
||||
.from(workspaceOperations)
|
||||
.where(
|
||||
and(
|
||||
eq(workspaceOperations.companyId, companyId),
|
||||
inArray(workspaceOperations.executionWorkspaceId, executionWorkspaceIds),
|
||||
),
|
||||
);
|
||||
|
||||
const latestByWorkspace = new Map<string, { phase: string; status: string; startedAt: Date }>();
|
||||
for (const row of rows) {
|
||||
if (!row.executionWorkspaceId) continue;
|
||||
const current = latestByWorkspace.get(row.executionWorkspaceId);
|
||||
if (!current || row.startedAt > current.startedAt) {
|
||||
latestByWorkspace.set(row.executionWorkspaceId, {
|
||||
phase: row.phase,
|
||||
status: row.status,
|
||||
startedAt: row.startedAt,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
for (const workspaceId of executionWorkspaceIds) {
|
||||
const latest = latestByWorkspace.get(workspaceId);
|
||||
if (!latest) continue; // no ops recorded → treat as finalized
|
||||
if (latest.phase === "workspace_finalize" && latest.status === "succeeded") continue;
|
||||
unfinalized.add(workspaceId);
|
||||
}
|
||||
|
||||
return unfinalized;
|
||||
}
|
||||
|
||||
async function listIssueDependencyReadinessMap(
|
||||
dbOrTx: Pick<Db, "select">,
|
||||
companyId: string,
|
||||
@@ -404,6 +466,7 @@ async function listIssueDependencyReadinessMap(
|
||||
issueId: issueRelations.relatedIssueId,
|
||||
blockerIssueId: issueRelations.issueId,
|
||||
blockerStatus: issues.status,
|
||||
blockerExecutionWorkspaceId: issues.executionWorkspaceId,
|
||||
})
|
||||
.from(issueRelations)
|
||||
.innerJoin(issues, eq(issueRelations.issueId, issues.id))
|
||||
@@ -415,6 +478,21 @@ async function listIssueDependencyReadinessMap(
|
||||
),
|
||||
);
|
||||
|
||||
// Collect executionWorkspaceIds of "done" blockers — these are the only ones
|
||||
// subject to the workspace-finalize barrier. Blockers that aren't done already
|
||||
// mark the dependent as not-ready and don't need a finalize check.
|
||||
const doneBlockerWorkspaceIds = new Set<string>();
|
||||
for (const row of blockerRows) {
|
||||
if (row.blockerStatus === "done" && row.blockerExecutionWorkspaceId) {
|
||||
doneBlockerWorkspaceIds.add(row.blockerExecutionWorkspaceId);
|
||||
}
|
||||
}
|
||||
const unfinalizedWorkspaceIds = await listUnfinalizedExecutionWorkspaceIds(
|
||||
dbOrTx,
|
||||
companyId,
|
||||
[...doneBlockerWorkspaceIds],
|
||||
);
|
||||
|
||||
for (const row of blockerRows) {
|
||||
const current = readinessMap.get(row.issueId) ?? createIssueDependencyReadiness(row.issueId);
|
||||
current.blockerIssueIds.push(row.blockerIssueId);
|
||||
@@ -425,6 +503,18 @@ async function listIssueDependencyReadinessMap(
|
||||
current.unresolvedBlockerCount += 1;
|
||||
current.allBlockersDone = false;
|
||||
current.isDependencyReady = false;
|
||||
} else if (
|
||||
row.blockerExecutionWorkspaceId &&
|
||||
unfinalizedWorkspaceIds.has(row.blockerExecutionWorkspaceId)
|
||||
) {
|
||||
// Workspace-finalize barrier: the blocker's most recent run on its
|
||||
// execution workspace hasn't recorded a successful workspace_finalize.
|
||||
// Treat the dependent as not-ready until sync-back lands (or the run
|
||||
// finalizes); a subsequent finalize wake will re-evaluate readiness.
|
||||
current.unresolvedBlockerIssueIds.push(row.blockerIssueId);
|
||||
current.unresolvedBlockerCount += 1;
|
||||
current.pendingFinalizeBlockerIssueIds.push(row.blockerIssueId);
|
||||
current.isDependencyReady = false;
|
||||
}
|
||||
readinessMap.set(row.issueId, current);
|
||||
}
|
||||
@@ -3891,45 +3981,33 @@ export function issueService(db: Db) {
|
||||
);
|
||||
if (candidates.length === 0) return [];
|
||||
|
||||
const candidateIds = candidates.map((candidate) => candidate.id);
|
||||
const blockerRows = await db
|
||||
.select({
|
||||
issueId: issueRelations.relatedIssueId,
|
||||
blockerIssueId: issueRelations.issueId,
|
||||
blockerStatus: issues.status,
|
||||
})
|
||||
.from(issueRelations)
|
||||
.innerJoin(issues, eq(issueRelations.issueId, issues.id))
|
||||
.where(
|
||||
and(
|
||||
eq(issueRelations.companyId, blockerIssue.companyId),
|
||||
eq(issueRelations.type, "blocks"),
|
||||
inArray(issueRelations.relatedIssueId, candidateIds),
|
||||
),
|
||||
);
|
||||
const wakeableCandidates = candidates.filter(
|
||||
(candidate) =>
|
||||
candidate.assigneeAgentId && !["backlog", "done", "cancelled"].includes(candidate.status),
|
||||
);
|
||||
if (wakeableCandidates.length === 0) return [];
|
||||
|
||||
const blockersByIssueId = new Map<string, Array<{ blockerIssueId: string; blockerStatus: string }>>();
|
||||
for (const row of blockerRows) {
|
||||
const list = blockersByIssueId.get(row.issueId) ?? [];
|
||||
list.push({ blockerIssueId: row.blockerIssueId, blockerStatus: row.blockerStatus });
|
||||
blockersByIssueId.set(row.issueId, list);
|
||||
}
|
||||
// Defer to the unified readiness check so that a dependent only fires when
|
||||
// (a) every blocker is done AND (b) every done blocker's workspace has
|
||||
// recorded a successful workspace_finalize. The finalize hook also calls
|
||||
// this function on completion, so a wake initially gated by an in-flight
|
||||
// sync-back will re-fire once the restore lands locally.
|
||||
const readinessMap = await listIssueDependencyReadinessMap(
|
||||
db,
|
||||
blockerIssue.companyId,
|
||||
wakeableCandidates.map((candidate) => candidate.id),
|
||||
);
|
||||
|
||||
return candidates
|
||||
.filter((candidate) => candidate.assigneeAgentId && !["backlog", "done", "cancelled"].includes(candidate.status))
|
||||
return wakeableCandidates
|
||||
.map((candidate) => {
|
||||
const blockers = blockersByIssueId.get(candidate.id) ?? [];
|
||||
return {
|
||||
...candidate,
|
||||
blockerIssueIds: blockers.map((blocker) => blocker.blockerIssueId),
|
||||
allBlockersDone: blockers.length > 0 && blockers.every((blocker) => blocker.blockerStatus === "done"),
|
||||
};
|
||||
const readiness = readinessMap.get(candidate.id) ?? createIssueDependencyReadiness(candidate.id);
|
||||
return { candidate, readiness };
|
||||
})
|
||||
.filter((candidate) => candidate.allBlockersDone)
|
||||
.map((candidate) => ({
|
||||
.filter(({ readiness }) => readiness.isDependencyReady && readiness.blockerIssueIds.length > 0)
|
||||
.map(({ candidate, readiness }) => ({
|
||||
id: candidate.id,
|
||||
assigneeAgentId: candidate.assigneeAgentId!,
|
||||
blockerIssueIds: candidate.blockerIssueIds,
|
||||
blockerIssueIds: readiness.blockerIssueIds,
|
||||
}));
|
||||
},
|
||||
|
||||
|
||||
Reference in new issue
Block a user