mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-30 13:20:54 +01:00
refactor(heartbeat): extract runtime service coordination (#15980)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Heartbeat runs start workspace services and receive service reports from adapters. > - Service context updates and readiness notices are spread through the executor. > - These steps share a clear responsibility, while process and lease ownership already live elsewhere. > - This pull request puts service coordination in one module. > - The executor retains the same dispatch and cleanup order. ## Linked Issues or Issue Description **What existing behavior does this improve?** Maintenance of heartbeat execution. This is the first of two final focused extractions after #15973. Related #15653 changes runtime callback origins; this PR preserves that policy. **Current behavior** `heartbeat.ts` has 4,372 lines and 178,459 bytes. It validates service intents, starts services, persists adapter reports and primary URLs, and posts readiness notices inline. **Proposed behavior** `heartbeat/runtime-services.ts` owns those steps in 212 lines. `heartbeat.ts` becomes 4,215 lines and 173,657 bytes. Process registries and leases stay in `workspace-runtime.ts`. The executor still releases services in its original cleanup path. **Reason and benefit** Service coordination has named preparation and result phases. The execution path is easier to read without changing service lifetime or moving cleanup policy. **Breaking changes** None intended. All 140 named public heartbeat exports remain available, including `postWorkspaceReadyComment`. ## What Changed - Move service-intent validation, startup, adapter-result persistence, and readiness notices into one module. - Preserve mutable run context, primary URL selection, notice error handling, and release ordering. - Add four public-path tests using disposable PostgreSQL, real HTTP service processes, and a fixture adapter. - Extend browser recovery acceptance to fetch the actual service URL, check the readiness notice, and verify listener cleanup. - Update the module map and acceptance instructions. ## Verification - Before extraction: all five tests pass, including the four new characterization cases and existing notice formatting test. - After extraction: all 17 tests pass across service coordination, notice formatting, output recording, and runtime authorization. - Structural comparison confirms that all moved bodies and the reconstructed executor match the baseline. Module boundary checks pass. - `pnpm -r typecheck` and `pnpm build` pass. - Extended isolated browser acceptance: **2/2 pass** (3.7 minutes), covering actual service readiness, recovery, bounded failure, and cleanup. At `8e33e2032de681082f128470eb8ab0711b382afe`, all 55 current-head checks pass (two configured skips), including hosted full-suite tests. Greptile is 5/5 with no findings or review threads. ## Risks - Service context must be persisted before adapter dispatch and before run completion. The tests inspect both phases and preserve the first service URL when adapter reports arrive. - Notice failures must remain nonfatal. The public-path test checks both startup and result notices. - Shared services must survive run cleanup; ephemeral services must stop after success or failure. Tests use actual processes and HTTP requests. - The fixture supplies UUID service report IDs, as required by the database. The existing generated-ID fallback for reports without IDs is outside this extraction. - Test adapters and services are local fixtures; these tests make no live provider calls. ## Model Used OpenAI Codex, based on GPT-6, assisted with reasoning, code edits, tool execution, and verification. The exact deployment model ID and context-window size are not exposed in this session. ## 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 not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
914b70572c
commit
94e4218310
9 files changed
+405
-168
No files matched your search
@@ -1278,6 +1278,12 @@ When a workspace service runs Paperclip for browser OAuth QA, configure its `exp
|
||||
|
||||
Keep relevant heartbeat extractions and their focused tests in `server/src/services/heartbeat/`.
|
||||
|
||||
Runtime-service coordination is in `server/src/services/heartbeat/runtime-services.ts`.
|
||||
It validates service intents, starts configured services, records adapter-managed
|
||||
services and primary URLs, and posts workspace-ready notices. Process registries
|
||||
and leases stay in `workspace-runtime.ts`; the executor keeps release ordering in
|
||||
its existing cleanup path. Public-path tests use real local service processes.
|
||||
|
||||
Task assignment Markdown is rendered by `server/src/services/heartbeat/task-markdown.ts`.
|
||||
`heartbeat.ts` re-exports it for existing callers. Keep prompt
|
||||
formatting changes in the renderer and its tests, separate from run orchestration.
|
||||
|
||||
@@ -0,0 +1,152 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { mkdtemp, rm, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { eq } from "drizzle-orm";
|
||||
import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
|
||||
import { agents, authUsers, companies, companyMemberships, createDb, heartbeatRuns, issueComments, issues, workspaceRuntimeServices } from "@paperclipai/db";
|
||||
import { registerServerAdapter, unregisterServerAdapter, type AdapterExecutionContext } from "../adapters/index.js";
|
||||
import { heartbeatService } from "../services/heartbeat.js";
|
||||
import { stopRuntimeServicesForExecutionWorkspace, type RuntimeServiceRef } from "../services/workspace-runtime.js";
|
||||
import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js";
|
||||
|
||||
const failedNotices = vi.hoisted(() => new Set<string>());
|
||||
vi.mock("../services/issues.js", async importOriginal => {
|
||||
const actual = await importOriginal<typeof import("../services/issues.js")>();
|
||||
return { ...actual, issueService: (...args: Parameters<typeof actual.issueService>) => {
|
||||
const service = actual.issueService(...args);
|
||||
return { ...service, addComment: (...input: Parameters<typeof service.addComment>) => {
|
||||
if (failedNotices.has(input[0]) && input[1].startsWith("## Workspace Ready")) {
|
||||
throw new Error("Fixture notice unavailable");
|
||||
}
|
||||
return service.addComment(...input);
|
||||
} };
|
||||
} };
|
||||
});
|
||||
|
||||
const support = await getEmbeddedPostgresTestSupport();
|
||||
const describePostgres = support.supported ? describe : describe.skip;
|
||||
if (!support.supported) console.warn(`Skipping heartbeat runtime services tests: ${support.reason}`);
|
||||
|
||||
describePostgres("heartbeat runtime services through adapter dispatch", () => {
|
||||
const adapterType = "runtime_services_test";
|
||||
let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
|
||||
let db: ReturnType<typeof createDb>;
|
||||
let heartbeat: ReturnType<typeof heartbeatService>;
|
||||
const roots: string[] = [];
|
||||
const captured = new Map<string, { services: RuntimeServiceRef[]; primaryUrl: unknown; intents: unknown }>();
|
||||
const failures = new Set<string>();
|
||||
|
||||
beforeAll(async () => {
|
||||
temporary = await startEmbeddedPostgresTestDatabase("heartbeat-runtime-services-");
|
||||
db = createDb(temporary.connectionString);
|
||||
heartbeat = heartbeatService(db);
|
||||
registerServerAdapter({
|
||||
type: adapterType, supportsLocalAgentJwt: false,
|
||||
execute: async (input: AdapterExecutionContext) => {
|
||||
const services = input.context.paperclipRuntimeServices as RuntimeServiceRef[];
|
||||
captured.set(input.runId, { services: structuredClone(services), primaryUrl: input.context.paperclipRuntimePrimaryUrl, intents: input.context.paperclipRuntimeServiceIntents });
|
||||
expect(await fetch(services[0]!.url!).then(response => response.text())).toBe("heartbeat runtime fixture");
|
||||
// Stop unfinished-task recovery from racing subsequent fixture runs.
|
||||
await db.update(issues).set({ status: "in_review" }).where(eq(issues.id, String(input.context.issueId)));
|
||||
if (failures.has(input.agent.id)) throw new Error("Invalid fixture adapter configuration");
|
||||
return { exitCode: 0, signal: null, timedOut: false, resultJson: {}, runtimeServices: [{ id: randomUUID(), serviceName: "adapter-preview", url: "https://preview.example.test/fixture", providerRef: "fixture-preview" }] };
|
||||
},
|
||||
testEnvironment: async () => ({ adapterType, status: "pass", checks: [], testedAt: new Date().toISOString() }),
|
||||
});
|
||||
}, 30_000);
|
||||
|
||||
afterAll(async () => {
|
||||
await heartbeat?.drainActiveRunExecutions();
|
||||
for (const root of roots) {
|
||||
await stopRuntimeServicesForExecutionWorkspace({ db, executionWorkspaceId: randomUUID(), workspaceCwd: root });
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
unregisterServerAdapter(adapterType);
|
||||
await db?.$client.end({ timeout: 0 });
|
||||
await temporary?.cleanup();
|
||||
});
|
||||
|
||||
async function seed(shared = false) {
|
||||
const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID();
|
||||
const root = await mkdtemp(path.join(os.tmpdir(), "heartbeat-runtime-service-"));
|
||||
roots.push(root);
|
||||
await writeFile(path.join(root, "serve.cjs"), "require('node:http').createServer((req,res)=>res.end('heartbeat runtime fixture')).listen(Number(process.env.PORT),'127.0.0.1');\n");
|
||||
await db.insert(companies).values({ id: companyId, name: "Runtime service fixture", issuePrefix: `R${companyId.slice(0, 6)}` });
|
||||
const userId = randomUUID();
|
||||
await db.insert(authUsers).values({ id: userId, name: "Runtime user", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() });
|
||||
await db.insert(companyMemberships).values([{ companyId, principalType: "user", principalId: userId, membershipRole: "operator" }, { companyId, principalType: "agent", principalId: agentId, membershipRole: "member" }]);
|
||||
await db.insert(agents).values({ id: agentId, companyId, name: "Runtime service reader", adapterType, status: "idle", adapterConfig: {
|
||||
cwd: root, workspaceRuntime: { services: [{ name: "web", command: `${JSON.stringify(process.execPath)} ${JSON.stringify(path.join(root, "serve.cjs"))}`,
|
||||
port: { type: "auto" }, readiness: { type: "http", urlTemplate: "http://127.0.0.1:{{port}}", timeoutSec: 10, intervalMs: 100 },
|
||||
lifecycle: shared ? "shared" : "ephemeral", reuseScope: "agent", stopPolicy: { type: shared ? "manual" : "on_run_finish" },
|
||||
}] },
|
||||
} });
|
||||
await db.insert(issues).values({ id: issueId, companyId, title: "Runtime services fixture", status: "todo", assigneeAgentId: agentId, responsibleUserId: userId, issueNumber: 1 });
|
||||
return { companyId, agentId, issueId, root };
|
||||
}
|
||||
|
||||
async function execute(f: Awaited<ReturnType<typeof seed>>) {
|
||||
const run = await heartbeat.wakeup(f.agentId, { source: "on_demand", triggerDetail: "manual", reason: "manual", payload: { issueId: f.issueId }, contextSnapshot: { issueId: f.issueId } });
|
||||
expect(run).not.toBeNull();
|
||||
await heartbeat.drainActiveRunExecutions();
|
||||
const [finished] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, run!.id));
|
||||
const services = await db.select().from(workspaceRuntimeServices).where(eq(workspaceRuntimeServices.companyId, f.companyId));
|
||||
const comments = await db.select().from(issueComments).where(eq(issueComments.issueId, f.issueId));
|
||||
const input = captured.get(run!.id)!;
|
||||
return { finished: finished!, services: services.filter(service => service.id === input.services[0]!.id || service.startedByRunId === run!.id),
|
||||
notices: comments.filter(comment => comment.createdByRunId === run!.id && comment.body.startsWith("## Workspace Ready")), input };
|
||||
}
|
||||
|
||||
it("starts a real service before dispatch, retains its primary URL, and persists adapter-reported services", async () => {
|
||||
const f = await seed();
|
||||
const { finished, input, services, notices } = await execute(f);
|
||||
expect(finished.status, finished.error ?? undefined).toBe("succeeded");
|
||||
expect(input.intents).toEqual([expect.objectContaining({ name: "web" })]);
|
||||
expect(input.services).toEqual([expect.objectContaining({ companyId: f.companyId, serviceName: "web", status: "running", reused: false, startedByRunId: finished.id })]);
|
||||
expect(finished.contextSnapshot?.paperclipRuntimePrimaryUrl).toBe(input.primaryUrl);
|
||||
expect(finished.contextSnapshot?.paperclipRuntimeServices).toEqual([
|
||||
expect.objectContaining({ id: input.services[0]!.id }),
|
||||
expect.objectContaining({ serviceName: "adapter-preview", provider: "adapter_managed", url: "https://preview.example.test/fixture" }),
|
||||
]);
|
||||
expect(services, finished.stderrExcerpt ?? undefined).toHaveLength(2);
|
||||
expect(services.find(service => service.provider === "local_process")).toMatchObject({ id: input.services[0]!.id, status: "stopped" });
|
||||
expect(services.find(service => service.provider === "adapter_managed")).toMatchObject({ companyId: f.companyId, issueId: f.issueId, status: "running", startedByRunId: finished.id });
|
||||
expect(notices).toHaveLength(2);
|
||||
expect(notices.every(notice => notice.authorAgentId === f.agentId && notice.createdByRunId === finished.id)).toBe(true);
|
||||
expect(notices.some(notice => notice.body.includes(String(input.primaryUrl)))).toBe(true);
|
||||
expect(notices.some(notice => notice.body.includes("https://preview.example.test/fixture"))).toBe(true);
|
||||
});
|
||||
|
||||
it("reuses a shared service without posting another startup notice", async () => {
|
||||
const f = await seed(true);
|
||||
const first = await execute(f);
|
||||
const second = await execute(f);
|
||||
expect(second.finished.status, second.finished.error ?? undefined).toBe("succeeded");
|
||||
expect(second.input.services[0]).toMatchObject({ id: first.input.services[0]!.id, reused: true });
|
||||
expect(second.services.find(service => service.provider === "local_process")?.status).toBe("running");
|
||||
expect(first.notices).toHaveLength(2);
|
||||
expect(second.notices, second.finished.stderrExcerpt ?? undefined).toHaveLength(1);
|
||||
expect(second.notices[0]!.body).toContain("https://preview.example.test/fixture");
|
||||
});
|
||||
|
||||
it("releases a started ephemeral service when adapter execution fails", async () => {
|
||||
const f = await seed();
|
||||
failures.add(f.agentId);
|
||||
const { finished, services } = await execute(f);
|
||||
expect(finished.status).toBe("failed");
|
||||
expect(finished.error).toContain("Invalid fixture adapter configuration");
|
||||
expect(services).toEqual([expect.objectContaining({ provider: "local_process", status: "stopped" })]);
|
||||
});
|
||||
|
||||
it("keeps both service records and a successful result when readiness notices fail", async () => {
|
||||
const f = await seed();
|
||||
failedNotices.add(f.issueId);
|
||||
const { finished, services, notices } = await execute(f);
|
||||
expect(finished.status, finished.error ?? undefined).toBe("succeeded");
|
||||
expect(services, finished.stderrExcerpt ?? undefined).toHaveLength(2);
|
||||
expect(notices).toEqual([]);
|
||||
expect(finished.stderrExcerpt).toContain("Failed to post workspace-ready comment: Fixture notice unavailable");
|
||||
expect(finished.stderrExcerpt).toContain("Failed to post adapter-managed runtime comment: Fixture notice unavailable");
|
||||
});
|
||||
});
|
||||
@@ -1,3 +1,4 @@
|
||||
import { prepareHeartbeatRuntimeServiceIntents, startHeartbeatRuntimeServices, recordHeartbeatAdapterRuntimeServices } from "./heartbeat/runtime-services.js";
|
||||
import { prepareHeartbeatExecutionConfiguration } from "./heartbeat/execution-configuration.js";
|
||||
import { createHeartbeatRunOutput } from "./heartbeat/run-output.js";
|
||||
import { prepareHeartbeatSession, resolveHeartbeatAdapterSession } from "./heartbeat/session-preparation.js";
|
||||
@@ -400,14 +401,7 @@ import {
|
||||
type LogActivityInput,
|
||||
} from "./activity-log.js";
|
||||
import {
|
||||
buildWorkspaceReadyComment,
|
||||
buildWorkspaceReadyMetadata,
|
||||
buildWorkspaceReadyPresentation,
|
||||
ensureRuntimeServicesForRun,
|
||||
persistAdapterManagedRuntimeServices,
|
||||
releaseRuntimeServicesForRun,
|
||||
type RealizedExecutionWorkspace,
|
||||
type RuntimeServiceRef,
|
||||
} from "./workspace-runtime.js";
|
||||
|
||||
import { issueService } from "./issues.js";
|
||||
@@ -509,10 +503,6 @@ import {
|
||||
sweepExpiredHeartbeatRunRuntimeStatuses,
|
||||
} from "./heartbeat-run-runtime-status.js";
|
||||
|
||||
import {
|
||||
assertLowTrustRuntimeServicesAllowed,
|
||||
} from "./low-trust-runtime-containment.js";
|
||||
|
||||
import { resolveAndRetainRunTrustPreset } from "./run-trust-preset.js";
|
||||
|
||||
import type { PluginWorkerManager } from "./plugin-worker-manager.js";
|
||||
@@ -796,40 +786,7 @@ export interface HeartbeatServiceOptions {
|
||||
}) => Promise<void>;
|
||||
}
|
||||
|
||||
type WorkspaceReadyCommentWriter = {
|
||||
addComment: (
|
||||
issueId: string,
|
||||
body: string,
|
||||
actor: { agentId?: string; userId?: string; runId?: string | null },
|
||||
options?: {
|
||||
presentation?: ReturnType<typeof buildWorkspaceReadyPresentation>;
|
||||
metadata?: ReturnType<typeof buildWorkspaceReadyMetadata>;
|
||||
},
|
||||
) => Promise<unknown>;
|
||||
};
|
||||
|
||||
export function postWorkspaceReadyComment(input: {
|
||||
issuesSvc: WorkspaceReadyCommentWriter;
|
||||
issueId: string;
|
||||
agentId: string;
|
||||
runId: string;
|
||||
workspace: RealizedExecutionWorkspace;
|
||||
runtimeServices: RuntimeServiceRef[];
|
||||
}) {
|
||||
const workspaceReadyInput = {
|
||||
workspace: input.workspace,
|
||||
runtimeServices: input.runtimeServices,
|
||||
};
|
||||
return input.issuesSvc.addComment(
|
||||
input.issueId,
|
||||
buildWorkspaceReadyComment(workspaceReadyInput),
|
||||
{ agentId: input.agentId, runId: input.runId },
|
||||
{
|
||||
presentation: buildWorkspaceReadyPresentation(workspaceReadyInput),
|
||||
metadata: buildWorkspaceReadyMetadata(workspaceReadyInput),
|
||||
},
|
||||
);
|
||||
}
|
||||
export { postWorkspaceReadyComment } from "./heartbeat/runtime-services.js";
|
||||
|
||||
export { postNativeModelFallbackWarning } from "./heartbeat/runtime-selection.js";
|
||||
|
||||
@@ -3137,26 +3094,7 @@ export function heartbeatService(
|
||||
executionWorkspace: { branchName: executionWorkspace.branchName },
|
||||
};
|
||||
}
|
||||
const runtimeServiceIntents = (() => {
|
||||
const runtimeConfig = parseObject(
|
||||
hostExecutionWorkspaceConfig.workspaceRuntime,
|
||||
);
|
||||
return Array.isArray(runtimeConfig.services)
|
||||
? runtimeConfig.services.filter(
|
||||
(value): value is Record<string, unknown> =>
|
||||
typeof value === "object" && value !== null,
|
||||
)
|
||||
: [];
|
||||
})();
|
||||
assertLowTrustRuntimeServicesAllowed({
|
||||
resolution: trustPreset,
|
||||
runtimeServiceCount: runtimeServiceIntents.length,
|
||||
});
|
||||
if (runtimeServiceIntents.length > 0) {
|
||||
context.paperclipRuntimeServiceIntents = runtimeServiceIntents;
|
||||
} else {
|
||||
delete context.paperclipRuntimeServiceIntents;
|
||||
}
|
||||
prepareHeartbeatRuntimeServiceIntents({ hostExecutionWorkspaceConfig, context, trustPreset });
|
||||
if (
|
||||
executionWorkspace.projectId &&
|
||||
!readNonEmptyString(context.projectId)
|
||||
@@ -3300,59 +3238,10 @@ export function heartbeatService(
|
||||
typeof entry[0] === "string" && typeof entry[1] === "string",
|
||||
),
|
||||
);
|
||||
const runtimeServices = await ensureRuntimeServicesForRun({
|
||||
db,
|
||||
runId: run.id,
|
||||
agent: {
|
||||
id: agent.id,
|
||||
name: agent.name,
|
||||
companyId: agent.companyId,
|
||||
},
|
||||
issue: issueRef,
|
||||
workspace: executionWorkspace,
|
||||
executionWorkspaceId:
|
||||
persistedExecutionWorkspace?.id ??
|
||||
issueRef?.executionWorkspaceId ??
|
||||
null,
|
||||
config: hostExecutionWorkspaceConfig,
|
||||
adapterEnv,
|
||||
onLog,
|
||||
recorder: workspaceOperationRecorder,
|
||||
const runtimeServices = await startHeartbeatRuntimeServices(db, {
|
||||
run, agent, issueId, issueRef, executionWorkspace, context, issuesSvc, onLog,
|
||||
persistedExecutionWorkspace, hostExecutionWorkspaceConfig, adapterEnv, workspaceOperationRecorder,
|
||||
});
|
||||
if (runtimeServices.length > 0) {
|
||||
context.paperclipRuntimeServices = runtimeServices;
|
||||
context.paperclipRuntimePrimaryUrl =
|
||||
runtimeServices.find((service) => readNonEmptyString(service.url))
|
||||
?.url ?? null;
|
||||
await db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
contextSnapshot: context,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, run.id));
|
||||
}
|
||||
if (
|
||||
issueId &&
|
||||
(executionWorkspace.created ||
|
||||
runtimeServices.some((service) => !service.reused))
|
||||
) {
|
||||
try {
|
||||
await postWorkspaceReadyComment({
|
||||
issuesSvc,
|
||||
issueId,
|
||||
agentId: agent.id,
|
||||
runId: run.id,
|
||||
workspace: executionWorkspace,
|
||||
runtimeServices,
|
||||
});
|
||||
} catch (err) {
|
||||
await onLog(
|
||||
"stderr",
|
||||
`[paperclip] Failed to post workspace-ready comment: ${err instanceof Error ? err.message : String(err)}\n`,
|
||||
);
|
||||
}
|
||||
}
|
||||
const adapter = getServerAdapter(agent.adapterType);
|
||||
const durableGoalControlRun =
|
||||
readNonEmptyString(context.goalControlRequestId) !== null ||
|
||||
@@ -3689,56 +3578,10 @@ export function heartbeatService(
|
||||
"run referenced-project remote staging",
|
||||
);
|
||||
}
|
||||
const adapterManagedRuntimeServices = adapterResult.runtimeServices
|
||||
? await persistAdapterManagedRuntimeServices({
|
||||
db,
|
||||
adapterType: agent.adapterType,
|
||||
runId: run.id,
|
||||
agent: {
|
||||
id: agent.id,
|
||||
name: agent.name,
|
||||
companyId: agent.companyId,
|
||||
},
|
||||
issue: issueRef,
|
||||
workspace: executionWorkspace,
|
||||
reports: adapterResult.runtimeServices,
|
||||
})
|
||||
: [];
|
||||
if (adapterManagedRuntimeServices.length > 0) {
|
||||
const combinedRuntimeServices = [
|
||||
...runtimeServices,
|
||||
...adapterManagedRuntimeServices,
|
||||
];
|
||||
context.paperclipRuntimeServices = combinedRuntimeServices;
|
||||
context.paperclipRuntimePrimaryUrl =
|
||||
combinedRuntimeServices.find((service) =>
|
||||
readNonEmptyString(service.url),
|
||||
)?.url ?? null;
|
||||
await db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
contextSnapshot: context,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, run.id));
|
||||
if (issueId) {
|
||||
try {
|
||||
await postWorkspaceReadyComment({
|
||||
issuesSvc,
|
||||
issueId,
|
||||
agentId: agent.id,
|
||||
runId: run.id,
|
||||
workspace: executionWorkspace,
|
||||
runtimeServices: adapterManagedRuntimeServices,
|
||||
});
|
||||
} catch (err) {
|
||||
await onLog(
|
||||
"stderr",
|
||||
`[paperclip] Failed to post adapter-managed runtime comment: ${err instanceof Error ? err.message : String(err)}\n`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
await recordHeartbeatAdapterRuntimeServices(db, {
|
||||
run, agent, issueId, issueRef, executionWorkspace, context, issuesSvc, onLog,
|
||||
adapterResult, runtimeServices,
|
||||
});
|
||||
await runCompletion.completeRun({
|
||||
run, agent, issueId, issueRef, signal: executionControl.controller.signal,
|
||||
output: runOutput,
|
||||
|
||||
@@ -0,0 +1,212 @@
|
||||
import { eq } from "drizzle-orm";
|
||||
import { heartbeatRuns, type Db } from "@paperclipai/db";
|
||||
import type { AdapterExecutionResult } from "@paperclipai/adapter-utils";
|
||||
import { parseObject } from "../../adapters/utils.js";
|
||||
import { assertLowTrustRuntimeServicesAllowed } from "../low-trust-runtime-containment.js";
|
||||
import {
|
||||
buildWorkspaceReadyComment, buildWorkspaceReadyMetadata, buildWorkspaceReadyPresentation,
|
||||
ensureRuntimeServicesForRun, persistAdapterManagedRuntimeServices,
|
||||
type RealizedExecutionWorkspace, type RuntimeServiceRef,
|
||||
} from "../workspace-runtime.js";
|
||||
|
||||
function readNonEmptyString(value: unknown): string | null {
|
||||
return typeof value === "string" && value.trim().length > 0 ? value : null;
|
||||
}
|
||||
|
||||
type WorkspaceReadyCommentWriter = {
|
||||
addComment: (
|
||||
issueId: string,
|
||||
body: string,
|
||||
actor: { agentId?: string; userId?: string; runId?: string | null },
|
||||
options?: {
|
||||
presentation?: ReturnType<typeof buildWorkspaceReadyPresentation>;
|
||||
metadata?: ReturnType<typeof buildWorkspaceReadyMetadata>;
|
||||
},
|
||||
) => Promise<unknown>;
|
||||
};
|
||||
|
||||
export function postWorkspaceReadyComment(input: {
|
||||
issuesSvc: WorkspaceReadyCommentWriter;
|
||||
issueId: string;
|
||||
agentId: string;
|
||||
runId: string;
|
||||
workspace: RealizedExecutionWorkspace;
|
||||
runtimeServices: RuntimeServiceRef[];
|
||||
}) {
|
||||
const workspaceReadyInput = {
|
||||
workspace: input.workspace,
|
||||
runtimeServices: input.runtimeServices,
|
||||
};
|
||||
return input.issuesSvc.addComment(
|
||||
input.issueId,
|
||||
buildWorkspaceReadyComment(workspaceReadyInput),
|
||||
{ agentId: input.agentId, runId: input.runId },
|
||||
{
|
||||
presentation: buildWorkspaceReadyPresentation(workspaceReadyInput),
|
||||
metadata: buildWorkspaceReadyMetadata(workspaceReadyInput),
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
type RuntimeInput = Parameters<typeof ensureRuntimeServicesForRun>[0];
|
||||
interface RuntimeServiceContext {
|
||||
run: { id: string };
|
||||
agent: RuntimeInput["agent"] & { id: string; adapterType: string };
|
||||
issueId: string | null;
|
||||
issueRef: (NonNullable<RuntimeInput["issue"]> & { executionWorkspaceId: string | null }) | null;
|
||||
executionWorkspace: RealizedExecutionWorkspace;
|
||||
context: Record<string, unknown>;
|
||||
issuesSvc: WorkspaceReadyCommentWriter;
|
||||
onLog: NonNullable<RuntimeInput["onLog"]>;
|
||||
}
|
||||
|
||||
/** Service processes and leases stay in workspace-runtime; the executor releases them. */
|
||||
export function prepareHeartbeatRuntimeServiceIntents(input: {
|
||||
hostExecutionWorkspaceConfig: Record<string, unknown>;
|
||||
context: Record<string, unknown>;
|
||||
trustPreset: Parameters<typeof assertLowTrustRuntimeServicesAllowed>[0]["resolution"];
|
||||
}) {
|
||||
const { hostExecutionWorkspaceConfig, context, trustPreset } = input;
|
||||
const runtimeServiceIntents = (() => {
|
||||
const runtimeConfig = parseObject(
|
||||
hostExecutionWorkspaceConfig.workspaceRuntime,
|
||||
);
|
||||
return Array.isArray(runtimeConfig.services)
|
||||
? runtimeConfig.services.filter(
|
||||
(value): value is Record<string, unknown> =>
|
||||
typeof value === "object" && value !== null,
|
||||
)
|
||||
: [];
|
||||
})();
|
||||
assertLowTrustRuntimeServicesAllowed({
|
||||
resolution: trustPreset,
|
||||
runtimeServiceCount: runtimeServiceIntents.length,
|
||||
});
|
||||
if (runtimeServiceIntents.length > 0) {
|
||||
context.paperclipRuntimeServiceIntents = runtimeServiceIntents;
|
||||
} else {
|
||||
delete context.paperclipRuntimeServiceIntents;
|
||||
}
|
||||
}
|
||||
|
||||
export async function startHeartbeatRuntimeServices(db: Db, input: RuntimeServiceContext & {
|
||||
persistedExecutionWorkspace: { id: string } | null;
|
||||
hostExecutionWorkspaceConfig: Record<string, unknown>;
|
||||
adapterEnv: RuntimeInput["adapterEnv"];
|
||||
workspaceOperationRecorder: RuntimeInput["recorder"];
|
||||
}) {
|
||||
const { run, agent, issueId, issueRef, executionWorkspace, context, issuesSvc, onLog,
|
||||
persistedExecutionWorkspace, hostExecutionWorkspaceConfig, adapterEnv, workspaceOperationRecorder } = input;
|
||||
const runtimeServices = await ensureRuntimeServicesForRun({
|
||||
db,
|
||||
runId: run.id,
|
||||
agent: {
|
||||
id: agent.id,
|
||||
name: agent.name,
|
||||
companyId: agent.companyId,
|
||||
},
|
||||
issue: issueRef,
|
||||
workspace: executionWorkspace,
|
||||
executionWorkspaceId:
|
||||
persistedExecutionWorkspace?.id ??
|
||||
issueRef?.executionWorkspaceId ??
|
||||
null,
|
||||
config: hostExecutionWorkspaceConfig,
|
||||
adapterEnv,
|
||||
onLog,
|
||||
recorder: workspaceOperationRecorder,
|
||||
});
|
||||
if (runtimeServices.length > 0) {
|
||||
context.paperclipRuntimeServices = runtimeServices;
|
||||
context.paperclipRuntimePrimaryUrl =
|
||||
runtimeServices.find((service) => readNonEmptyString(service.url))
|
||||
?.url ?? null;
|
||||
await db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
contextSnapshot: context,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, run.id));
|
||||
}
|
||||
if (
|
||||
issueId &&
|
||||
(executionWorkspace.created ||
|
||||
runtimeServices.some((service) => !service.reused))
|
||||
) {
|
||||
try {
|
||||
await postWorkspaceReadyComment({
|
||||
issuesSvc,
|
||||
issueId,
|
||||
agentId: agent.id,
|
||||
runId: run.id,
|
||||
workspace: executionWorkspace,
|
||||
runtimeServices,
|
||||
});
|
||||
} catch (err) {
|
||||
await onLog(
|
||||
"stderr",
|
||||
`[paperclip] Failed to post workspace-ready comment: ${err instanceof Error ? err.message : String(err)}\n`,
|
||||
);
|
||||
}
|
||||
}
|
||||
return runtimeServices;
|
||||
}
|
||||
|
||||
export async function recordHeartbeatAdapterRuntimeServices(db: Db, input: RuntimeServiceContext & {
|
||||
adapterResult: Pick<AdapterExecutionResult, "runtimeServices">;
|
||||
runtimeServices: RuntimeServiceRef[];
|
||||
}) {
|
||||
const { run, agent, issueId, issueRef, executionWorkspace, context, issuesSvc, onLog,
|
||||
adapterResult, runtimeServices } = input;
|
||||
const adapterManagedRuntimeServices = adapterResult.runtimeServices
|
||||
? await persistAdapterManagedRuntimeServices({
|
||||
db,
|
||||
adapterType: agent.adapterType,
|
||||
runId: run.id,
|
||||
agent: {
|
||||
id: agent.id,
|
||||
name: agent.name,
|
||||
companyId: agent.companyId,
|
||||
},
|
||||
issue: issueRef,
|
||||
workspace: executionWorkspace,
|
||||
reports: adapterResult.runtimeServices,
|
||||
})
|
||||
: [];
|
||||
if (adapterManagedRuntimeServices.length > 0) {
|
||||
const combinedRuntimeServices = [
|
||||
...runtimeServices,
|
||||
...adapterManagedRuntimeServices,
|
||||
];
|
||||
context.paperclipRuntimeServices = combinedRuntimeServices;
|
||||
context.paperclipRuntimePrimaryUrl =
|
||||
combinedRuntimeServices.find((service) =>
|
||||
readNonEmptyString(service.url),
|
||||
)?.url ?? null;
|
||||
await db
|
||||
.update(heartbeatRuns)
|
||||
.set({
|
||||
contextSnapshot: context,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(heartbeatRuns.id, run.id));
|
||||
if (issueId) {
|
||||
try {
|
||||
await postWorkspaceReadyComment({
|
||||
issuesSvc,
|
||||
issueId,
|
||||
agentId: agent.id,
|
||||
runId: run.id,
|
||||
workspace: executionWorkspace,
|
||||
runtimeServices: adapterManagedRuntimeServices,
|
||||
});
|
||||
} catch (err) {
|
||||
await onLog(
|
||||
"stderr",
|
||||
`[paperclip] Failed to post adapter-managed runtime comment: ${err instanceof Error ? err.message : String(err)}\n`,
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -34,4 +34,7 @@ state survives reload. Source edits remain intact and ignored private files are
|
||||
not copied. The worker also verifies that persisted task context contains the
|
||||
assignment, the full and compact prompt variants, and the description revision.
|
||||
Screenshots and failure traces go to `test-results/workspace-bootstrap`.
|
||||
The successful worker also requests a real ephemeral runtime service over HTTP.
|
||||
The browser checks its persisted URL and readiness notice, then verifies that
|
||||
the listener stops after run cleanup.
|
||||
Stop test-drive with Ctrl-C. Its disposable data directory remains for inspection.
|
||||
@@ -81,6 +81,12 @@ for (const persistent of [false, true]) {
|
||||
expect(completedRun.lastOutputSeq).toBeGreaterThan(0);
|
||||
expect(completedRun.lastOutputBytes).toBe(completedRun.logBytes);
|
||||
expect(completedLog.content).toContain("Workspace recovery verification complete.");
|
||||
const [runtimeService] = completedRun.contextSnapshot.paperclipRuntimeServices;
|
||||
expect(runtimeService).toMatchObject({ serviceName: "fixture-web", companyId: fixture.companyId, provider: "local_process" });
|
||||
const comments = await api(`/issues/${task.id}/comments`);
|
||||
expect(comments.some((comment: { body: string }) => comment.body.includes("## Workspace Ready") && comment.body.includes(runtimeService.url))).toBe(true);
|
||||
// Run status precedes final cleanup; wait for the owned ephemeral listener to stop.
|
||||
await expect.poll(async () => fetch(runtimeService.url, { signal: AbortSignal.timeout(2_000) }).then(() => true, () => false)).toBe(false);
|
||||
expect(await readFile(path.join(fixture.source, "README.md"), "utf8")).toBe("Existing uncommitted work\n");
|
||||
}
|
||||
await page.reload();
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
import { createServer } from "node:http";
|
||||
|
||||
createServer((_request, response) => {
|
||||
response.end("workspace runtime service ready");
|
||||
}).listen(Number(process.env.PORT), "127.0.0.1");
|
||||
@@ -32,7 +32,12 @@ export async function seedWorkspaceBootstrap(base, persistent = false) {
|
||||
await api(`/projects/${project.id}/workspaces`, "POST", { name: "Source copy", cwd: source, repoUrl: pathToFileURL(source).href, isPrimary: false });
|
||||
const agent = await api(`/companies/${company.id}/agents`, "POST", {
|
||||
name: `${persistent ? "Persistent failure worker" : "Recovery worker"} ${suffix}`, role: "engineer", adapterType: "process",
|
||||
adapterConfig: { command: process.execPath, args: [path.resolve(import.meta.dirname, "worker.mjs")], cwd: source },
|
||||
adapterConfig: { command: process.execPath, args: [path.resolve(import.meta.dirname, "worker.mjs")], cwd: source,
|
||||
workspaceRuntime: { services: [{ name: "fixture-web", lifecycle: "ephemeral", stopPolicy: { type: "on_run_finish" },
|
||||
command: `${JSON.stringify(process.execPath)} ${JSON.stringify(path.resolve(import.meta.dirname, "runtime-service.mjs"))}`,
|
||||
port: { type: "auto" }, readiness: { type: "http", urlTemplate: "http://127.0.0.1:{{port}}", timeoutSec: 10, intervalMs: 100 },
|
||||
}] },
|
||||
},
|
||||
runtimeConfig: { heartbeat: { enabled: false, wakeOnDemand: true } },
|
||||
});
|
||||
return { base, companyId: company.id, prefix: company.issuePrefix, projectId: project.id, agentId: agent.id, agentName: agent.name, projectName: project.name, source };
|
||||
|
||||
@@ -14,6 +14,11 @@ const api = async (route, method = "GET", body) => {
|
||||
};
|
||||
const run = await api(`/heartbeat-runs/${env.PAPERCLIP_RUN_ID}`);
|
||||
assert.equal(run.contextSnapshot.paperclipEnvironment.driver, "local");
|
||||
const services = run.contextSnapshot.paperclipRuntimeServices;
|
||||
assert.equal(services.length, 1);
|
||||
assert.equal(services[0].serviceName, "fixture-web");
|
||||
assert.equal(run.contextSnapshot.paperclipRuntimePrimaryUrl, services[0].url);
|
||||
assert.equal(await fetch(services[0].url).then(response => response.text()), "workspace runtime service ready");
|
||||
const issueId = run.contextSnapshot.issueId;
|
||||
const issue = await api(`/issues/${issueId}`);
|
||||
assert.equal(run.contextSnapshot.paperclipIssue.id, issue.id);
|
||||
|
||||
Reference in new issue
Block a user