Merge origin/master runtime service extraction

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-12 10:43:34 -05:00
commit e7d07f00a3
9 files changed
+405 -168

No files matched your search

@@ -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");
});
});
+10 -167
View File
@@ -1,4 +1,5 @@
import { nativeChatWorkspaceCwd } from "./native-runtime/native-chat-workspace.js";
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";
@@ -402,14 +403,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";
@@ -513,10 +507,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";
@@ -777,40 +767,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";
@@ -3177,26 +3134,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 (
!issueContext && executionWorkspace.projectId &&
!readNonEmptyString(context.projectId)
@@ -3340,59 +3278,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 ||
@@ -3739,56 +3628,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`,
);
}
}
}
}