mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 19:28:46 +01:00
fix(observability): time execution configuration steps (#16236)
Record optional execution-configuration timings before provider dispatch, with success, failure, sampling, cleanup, and silent-unconfigured controls. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
6cb6ac3b8e
commit
c80d93e14e
5 files changed
+97
-40
No files matched your search
@@ -1,4 +1,6 @@
|
||||
import { randomUUID } from "node:crypto";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { trace } from "@opentelemetry/api";
|
||||
import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace";
|
||||
import { access, readFile, mkdtemp, rm } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import path from "node:path";
|
||||
@@ -120,14 +122,36 @@ describePostgres("heartbeat execution configuration through adapter dispatch", (
|
||||
expect(JSON.parse(await readFile(path.join(providerHome, "auth.json"), "utf8"))).toEqual({ OPENAI_API_KEY: f.credential });
|
||||
await input.onMeta?.({ adapterType, command: "fixture", env, prompt: "fixture" });
|
||||
});
|
||||
const { finished, input } = await execute(f);
|
||||
expect(finished.status, finished.error ?? undefined).toBe("succeeded");
|
||||
expect(input?.context.aiConnection).toMatchObject({ accountName: "Fixture account", provider: "openai" });
|
||||
expect(providerHome).toEqual(expect.any(String));
|
||||
await expect(access(providerHome!)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, finished.id));
|
||||
expect(events.some(event => event.eventType === "adapter.invoke")).toBe(true);
|
||||
expect(JSON.stringify(events)).not.toContain(f.credential);
|
||||
const exporter = new InMemorySpanExporter();
|
||||
const provider = new TracerProvider({ spanProcessors: [new SimpleSpanProcessor({ exporter })] });
|
||||
expect(trace.setGlobalTracerProvider(provider)).toBe(true);
|
||||
try {
|
||||
const { finished, input } = await execute(f);
|
||||
expect(finished.status, finished.error ?? undefined).toBe("succeeded");
|
||||
expect(input?.context.aiConnection).toMatchObject({ accountName: "Fixture account", provider: "openai" });
|
||||
expect(providerHome).toEqual(expect.any(String));
|
||||
await expect(access(providerHome!)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
const events = await db.select().from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, finished.id));
|
||||
expect(events.some(event => event.eventType === "adapter.invoke")).toBe(true);
|
||||
expect(JSON.stringify(events)).not.toContain(f.credential);
|
||||
await provider.forceFlush();
|
||||
const runIdHash = createHash("sha256").update(finished.id).digest("hex").slice(0, 12);
|
||||
const steps = exporter.getFinishedSpans().filter(span => span.name.startsWith("heartbeat.prepare.configuration_")
|
||||
&& span.attributes["paperclip.task.run.run_id"] === runIdHash);
|
||||
expect(steps.map(span => span.name)).toEqual([
|
||||
"configuration_mentions", "configuration_github_identity", "configuration_secrets", "configuration_ai_runtime",
|
||||
"configuration_skills", "configuration_connector_assignments", "configuration_connector_skills",
|
||||
"configuration_connector_delivery", "configuration_connection_instructions", "configuration_agent_identity",
|
||||
].map(step => `heartbeat.prepare.${step}`));
|
||||
for (const span of steps) {
|
||||
expect(span.status).toEqual({ code: 0 });
|
||||
expect(span.attributes).toEqual({ "paperclip.task.run.run_id": runIdHash });
|
||||
expect(span.events).toEqual([]);
|
||||
}
|
||||
} finally {
|
||||
trace.disable();
|
||||
await provider.shutdown();
|
||||
}
|
||||
});
|
||||
|
||||
it("cleans acquired credentials and redacts diagnostics when later skill preparation fails", async () => {
|
||||
@@ -142,18 +166,36 @@ describePostgres("heartbeat execution configuration through adapter dispatch", (
|
||||
await access(acquiredHome!);
|
||||
return runtime;
|
||||
});
|
||||
const { finished, input } = await execute(f);
|
||||
expect(finished.status).toBe("failed");
|
||||
expect(input).toBeUndefined();
|
||||
expect(acquiredHome).toEqual(expect.any(String));
|
||||
await expect(access(acquiredHome!)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
expect(finished.error).toContain("Invalid fixture skill configuration");
|
||||
// The diagnostic reporter redacts runtime values before the external sink.
|
||||
// The legacy local run.error field does not apply this value-based redaction.
|
||||
await waitForPendingRunFailureReports();
|
||||
expect(capturedFailure).toHaveBeenCalled();
|
||||
expect(JSON.stringify(capturedFailure.mock.calls)).toContain("Invalid fixture skill configuration");
|
||||
expect(JSON.stringify(capturedFailure.mock.calls)).not.toContain(f.credential);
|
||||
const exporter = new InMemorySpanExporter();
|
||||
const provider = new TracerProvider({ spanProcessors: [new SimpleSpanProcessor({ exporter })] });
|
||||
expect(trace.setGlobalTracerProvider(provider)).toBe(true);
|
||||
try {
|
||||
const { finished, input } = await execute(f);
|
||||
expect(finished.status).toBe("failed");
|
||||
expect(input).toBeUndefined();
|
||||
expect(acquiredHome).toEqual(expect.any(String));
|
||||
await expect(access(acquiredHome!)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
expect(finished.error).toContain("Invalid fixture skill configuration");
|
||||
await waitForPendingRunFailureReports();
|
||||
expect(capturedFailure).toHaveBeenCalled();
|
||||
expect(JSON.stringify(capturedFailure.mock.calls)).toContain("Invalid fixture skill configuration");
|
||||
expect(JSON.stringify(capturedFailure.mock.calls)).not.toContain(f.credential);
|
||||
await provider.forceFlush();
|
||||
const runIdHash = createHash("sha256").update(finished.id).digest("hex").slice(0, 12);
|
||||
const steps = exporter.getFinishedSpans().filter(span => span.name.startsWith("heartbeat.prepare.configuration_")
|
||||
&& span.attributes["paperclip.task.run.run_id"] === runIdHash);
|
||||
expect(steps.find(span => span.name === "heartbeat.prepare.configuration_ai_runtime")?.status.code).toBe(0);
|
||||
expect(steps.find(span => span.name === "heartbeat.prepare.configuration_skills")?.status).toEqual({ code: 2 });
|
||||
expect(steps.some(span => span.name === "heartbeat.prepare.configuration_connector_assignments")).toBe(false);
|
||||
for (const span of steps) {
|
||||
expect(span.attributes).toEqual({ "paperclip.task.run.run_id": runIdHash });
|
||||
expect(span.events).toEqual([]);
|
||||
}
|
||||
expect(JSON.stringify(steps.map(span => ({ name: span.name, status: span.status, attributes: span.attributes, events: span.events })))).not.toContain(f.credential);
|
||||
} finally {
|
||||
trace.disable();
|
||||
await provider.shutdown();
|
||||
}
|
||||
});
|
||||
|
||||
it("defers a busy account without dispatch and preserves its scheduled retry", async () => {
|
||||
|
||||
@@ -25,6 +25,7 @@ import { skillVersionSelectionMap } from "../runtime-skill-selections.js";
|
||||
import type { selectHeartbeatEnvironment } from "./environment-selection.js";
|
||||
import type { createHeartbeatRetries } from "./retries.js";
|
||||
import type { createHeartbeatLifecycle } from "./run-lifecycle.js";
|
||||
import { createHeartbeatPreparationTiming } from "./preparation-timing.js";
|
||||
import {
|
||||
ConfigurationIncompleteFailure, applyRunScopedMentionedSkillKeys, resolveExecutionRunAdapterConfig,
|
||||
resolveRunScopedMentionedSkillKeys, type createHeartbeatRunPreparation, type resolveAcceptedPlanWakeRoutingDecision,
|
||||
@@ -73,6 +74,7 @@ export interface HeartbeatExecutionConfigurationInput {
|
||||
/** Assembles effective run configuration; the executor retains cleanup ownership. */
|
||||
export async function prepareHeartbeatExecutionConfiguration(db: Db, input: HeartbeatExecutionConfigurationInput) {
|
||||
const { run, agent, onFailureSecretsResolved, onManagedAiRuntimeAcquired, onIdentityRedactorPrepared } = input;
|
||||
const time = createHeartbeatPreparationTiming(run.id);
|
||||
const { issueId, context, responsibleUserId, acceptedPlanContinuationWake, acceptedPlanWakeRoutingDecision, taskProjectContext, routineEnvContext } = input.task;
|
||||
const { sourceWorkspaceConfig, aiBinding } = input.config;
|
||||
const { trustPreset, useIsolatedTaskDirectory, explicitTaskDirectory } = input.policy;
|
||||
@@ -94,19 +96,19 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear
|
||||
const skillActor: AuthorizationActor = { type: "agent", agentId: agent.id, companyId: agent.companyId,
|
||||
runId: run.id, source: "agent_jwt", onBehalfOfUserId: responsibleUserId === "local-board" ? null : responsibleUserId };
|
||||
const runScopedMentionedSkillKeys =
|
||||
await resolveRunScopedMentionedSkillKeys({
|
||||
await time("configuration_mentions", () => resolveRunScopedMentionedSkillKeys({
|
||||
db,
|
||||
companyId: agent.companyId,
|
||||
issueId,
|
||||
commentIds: extractWakeCommentIds(context),
|
||||
actor: skillActor,
|
||||
});
|
||||
}));
|
||||
const runScopedSkillKeys =
|
||||
acceptedPlanContinuationWake &&
|
||||
!acceptedPlanWakeRoutingDecision?.suppressAcceptedContinuation
|
||||
? [...runScopedMentionedSkillKeys, ACCEPTED_PLAN_CONVERSION_SKILL_KEY]
|
||||
: runScopedMentionedSkillKeys;
|
||||
const githubSelection = await resolveManagedGitHubIdentitySelection(
|
||||
const githubSelection = await time("configuration_github_identity", () => resolveManagedGitHubIdentitySelection(
|
||||
db,
|
||||
agent.companyId,
|
||||
{
|
||||
@@ -114,7 +116,7 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear
|
||||
responsibleUserId,
|
||||
allowStandingDelegation: false,
|
||||
},
|
||||
);
|
||||
));
|
||||
const useHostGitHub =
|
||||
!githubSelection.configured &&
|
||||
trustPreset.kind === "standard" &&
|
||||
@@ -123,7 +125,7 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear
|
||||
);
|
||||
|
||||
const { resolvedConfig, configuredTaskEnvironment, secretKeys, secretManifest } =
|
||||
await resolveExecutionRunAdapterConfig({
|
||||
await time("configuration_secrets", () => resolveExecutionRunAdapterConfig({
|
||||
managedAiCredentials: Boolean(aiBinding),
|
||||
managedGitHubCredentials: !useHostGitHub,
|
||||
companyId: agent.companyId,
|
||||
@@ -142,11 +144,11 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear
|
||||
routineEnv: aiBinding ? stripAiAuthBindings(routineEnvContext.env) : routineEnvContext.env,
|
||||
secretsSvc,
|
||||
trustPreset,
|
||||
});
|
||||
}));
|
||||
onFailureSecretsResolved(() => collectRunFailureSecretValues(resolvedConfig.env, secretKeys));
|
||||
if (aiBinding) {
|
||||
try {
|
||||
managedAiRuntime = await prepareManagedAiRuntime(db, { companyId: agent.companyId, agentId: agent.id, responsibleUserId, adapterType: agent.adapterType, binding: aiBinding, config: resolvedConfig });
|
||||
managedAiRuntime = await time("configuration_ai_runtime", () => prepareManagedAiRuntime(db, { companyId: agent.companyId, agentId: agent.id, responsibleUserId, adapterType: agent.adapterType, binding: aiBinding, config: resolvedConfig }));
|
||||
onManagedAiRuntimeAcquired(managedAiRuntime);
|
||||
} catch (error) {
|
||||
// Only fresh executions can receive a pre-provider wait receipt. A
|
||||
@@ -206,7 +208,7 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear
|
||||
const nativeRunnerPreparationSpans: NativeRunHistoricalSpan[] = [];
|
||||
const skillsPrepareStartedAtMs = Date.now();
|
||||
let skillDeliveryConfig = effectiveResolvedConfig;
|
||||
const runtimeSkillEntries = await (async () => {
|
||||
const runtimeSkillEntries = await time("configuration_skills", async () => {
|
||||
try {
|
||||
const storedVersions = parseObject(run.contextSnapshot?.skillVersions);
|
||||
const hasStoredVersions = Object.prototype.hasOwnProperty.call(run.contextSnapshot ?? {}, "skillVersions");
|
||||
@@ -261,26 +263,26 @@ export async function prepareHeartbeatExecutionConfiguration(db: Db, input: Hear
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
})();
|
||||
});
|
||||
nativeRunnerPreparationSpans.push({
|
||||
name: "skills.prepare",
|
||||
parentName: "task.prepare",
|
||||
startedAtMs: skillsPrepareStartedAtMs,
|
||||
endedAtMs: Date.now(),
|
||||
});
|
||||
const connectorAssignments = await resolveConnectorAssignments(db, { companyId: agent.companyId, agentId: agent.id, runId: run.id, issueId: typeof context.issueId === "string" ? context.issueId : undefined });
|
||||
const connectorSkillConfig = await applyConnectorSkills(skillDeliveryConfig, runtimeSkillEntries, connectorAssignments);
|
||||
const connectorAssignments = await time("configuration_connector_assignments", () => resolveConnectorAssignments(db, { companyId: agent.companyId, agentId: agent.id, runId: run.id, issueId: typeof context.issueId === "string" ? context.issueId : undefined }));
|
||||
const connectorSkillConfig = await time("configuration_connector_skills", () => applyConnectorSkills(skillDeliveryConfig, runtimeSkillEntries, connectorAssignments));
|
||||
// Both CLI adapters and native context materialization use the same resolved set.
|
||||
runtimeSkillEntries.splice(0, runtimeSkillEntries.length, ...connectorSkillConfig.paperclipRuntimeSkills);
|
||||
const connectorDelivery = await prepareConnectorSkillDelivery(connectorSkillConfig, agent.adapterType);
|
||||
const connectorDelivery = await time("configuration_connector_delivery", () => prepareConnectorSkillDelivery(connectorSkillConfig, agent.adapterType));
|
||||
// Always replace this runtime-only field; caller wake data cannot supply skills.
|
||||
context.paperclipWake = { ...parseObject(context.paperclipWake), connectorSkillInstructions: connectorDelivery.instructions };
|
||||
let runtimeConfig = await prepareConnectionInstructionDelivery({
|
||||
let runtimeConfig = await time("configuration_connection_instructions", () => prepareConnectionInstructionDelivery({
|
||||
resolve: () => resolveAssignedConnectionInstructionsForRun(db, { companyId: agent.companyId, agentId: agent.id, runId: run.id }),
|
||||
context, config: connectorDelivery.config, native: agent.adapterType === "paperclip_runner",
|
||||
});
|
||||
}));
|
||||
const agentIdentity = supportsManagedAgentIdentity(agent.adapterType, agent.adapterConfig, run.runtimeMode === "native" ? run.driverKind : undefined)
|
||||
? await agentIdentityService(db).ensureAgentIdentity(agent.companyId, agent.id)
|
||||
? await time("configuration_agent_identity", () => agentIdentityService(db).ensureAgentIdentity(agent.companyId, agent.id))
|
||||
: undefined;
|
||||
const identityRedactor = createAgentIdentityRedactor(agentIdentity?.privateKeyPem);
|
||||
onIdentityRedactorPrepared(identityRedactor);
|
||||
|
||||
@@ -77,7 +77,7 @@ it("honors an operator sampler that disables root traces even when the trigger i
|
||||
}
|
||||
});
|
||||
|
||||
it("stays silent and preserves values and errors without a tracer provider", async () => {
|
||||
it.each(["workspace", "configuration_ai_runtime", "configuration_skills", "configuration_agent_identity"] as const)("stays silent and preserves values and errors without a tracer provider for %s", async phase => {
|
||||
trace.disable();
|
||||
context.disable();
|
||||
const messages = ["log", "info", "warn", "error", "debug"] as const;
|
||||
@@ -85,12 +85,12 @@ it("stays silent and preserves values and errors without a tracer provider", asy
|
||||
try {
|
||||
const time = createHeartbeatPreparationTiming("unconfigured-run");
|
||||
const result = {};
|
||||
await expect(time("workspace", async () => {
|
||||
await expect(time(phase, async () => {
|
||||
expect(trace.getActiveSpan()?.isRecording()).not.toBe(true);
|
||||
return result;
|
||||
})).resolves.toBe(result);
|
||||
const error = new Error("original failure");
|
||||
await expect(time("instructions", async () => { throw error; })).rejects.toBe(error);
|
||||
await expect(time(phase, async () => { throw error; })).rejects.toBe(error);
|
||||
for (const spy of spies) expect(spy).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
for (const spy of spies) spy.mockRestore();
|
||||
|
||||
@@ -3,7 +3,11 @@ import { isSpanContextValid, ROOT_CONTEXT, SpanStatusCode, trace } from "@opente
|
||||
|
||||
type PreparationPhase = "environment_selection" | "execution_configuration" | "session"
|
||||
| "workspace" | "github_environment" | "github_launchers" | "run_output"
|
||||
| "workspace_validation" | "runtime_services" | "instructions" | "runtime_selection";
|
||||
| "workspace_validation" | "runtime_services" | "instructions" | "runtime_selection"
|
||||
| "configuration_mentions" | "configuration_github_identity" | "configuration_secrets"
|
||||
| "configuration_ai_runtime" | "configuration_skills" | "configuration_connector_assignments"
|
||||
| "configuration_connector_skills" | "configuration_connector_delivery"
|
||||
| "configuration_connection_instructions" | "configuration_agent_identity";
|
||||
|
||||
export function createHeartbeatPreparationTiming(runId: string) {
|
||||
const runIdHash = createHash("sha256").update(runId).digest("hex").slice(0, 12);
|
||||
|
||||
Reference in new issue
Block a user