mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-08 01:57:31 +01:00
fix(runner): retain bounded Cursor diagnostics in eval evidence
Validate the Cursor notice and active session/turn before retaining its closed observation fields. Preserve unknown token usage and cost, including malformed or stale diagnostics. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
a8df2064d6
commit
653bf7c37d
4 files changed
+260
-1
No files matched your search
@@ -0,0 +1,95 @@
|
||||
import type { CanonicalProviderEvent } from "../provider-events.js";
|
||||
|
||||
const SUMMARY = "Cursor reported partial native counters. These observations are not authoritative token usage or billing cost.";
|
||||
const PROVENANCE = "Cursor native turnEnded observations; partial, unsummed, unverified counter semantics";
|
||||
const REASONS = new Set([
|
||||
"native_counter_semantics_unverified", "native_counters_missing", "invalid_native_counter",
|
||||
"multiple_terminal_observations", "native_terminal_not_observed", "observation_limit_reached",
|
||||
"child_run_attribution_unverified",
|
||||
]);
|
||||
const COUNTERS = new Set(["inputTokens", "outputTokens", "cacheReadTokens", "cacheWriteTokens", "reasoningTokens"]);
|
||||
const object = (value: unknown): value is Record<string, unknown> =>
|
||||
value !== null && typeof value === "object" && !Array.isArray(value);
|
||||
const exact = (value: Record<string, unknown>, keys: readonly string[]): boolean =>
|
||||
Object.keys(value).length === keys.length && keys.every(key => Object.hasOwn(value, key));
|
||||
const positive = (value: unknown): value is number => Number.isSafeInteger(value) && Number(value) > 0;
|
||||
|
||||
/**
|
||||
* The eval transport wraps rich events in a Codex-shaped notification. Admit
|
||||
* only the closed Cursor diagnostic, never arbitrary canonical provider data.
|
||||
* This notice no longer contains the private prompt/request receipt IDs, so
|
||||
* validate its projected shape without inventing a persisted usage receipt.
|
||||
* Neither these unsummed observations nor their absence are accounting facts.
|
||||
*/
|
||||
export function cursorUsageNotice(
|
||||
raw: unknown,
|
||||
owner: { provider?: string; agent?: string; threadId: string; turnId: string | null },
|
||||
): CanonicalProviderEvent | null {
|
||||
try {
|
||||
if (owner.provider !== "acpx" || owner.agent !== "cursor" || !owner.threadId || !owner.turnId
|
||||
|| !object(raw) || !exact(raw, ["threadId", "turnId", "eventType", "itemId", "payload"])
|
||||
|| raw.threadId !== owner.threadId || raw.turnId !== owner.turnId
|
||||
|| raw.eventType !== "provider.notice.recorded" || typeof raw.itemId !== "string"
|
||||
|| !/^[A-Za-z0-9][A-Za-z0-9._:-]{0,159}$/.test(raw.itemId)) return null;
|
||||
const p = raw.payload;
|
||||
if (!object(p) || !exact(p, ["schema", "noticeId", "severity", "category", "scope", "recoverable", "userActionable", "summary", "details"])
|
||||
|| p.schema !== "paperclip.provider.notice.v1" || p.noticeId !== raw.itemId
|
||||
|| p.severity !== "info" || p.category !== "cursor_native_usage_observed" || p.scope !== "turn"
|
||||
|| p.recoverable !== true || p.userActionable !== false || p.summary !== SUMMARY
|
||||
|| !Array.isArray(p.details) || p.details.length < 3 || p.details.length > 11) return null;
|
||||
// Bound strings before JSON parsing/cloning. The collector's complete
|
||||
// envelope is <=16 KiB; its observation strings cannot exceed that budget.
|
||||
const details: Array<{ name: string; value: string }> = [];
|
||||
let bytes = 0;
|
||||
for (const detail of p.details) {
|
||||
if (!object(detail) || !exact(detail, ["name", "value"]) || typeof detail.name !== "string"
|
||||
|| detail.name.length > 64 || typeof detail.value !== "string" || detail.value.length > 4_000) return null;
|
||||
bytes += Buffer.byteLength(detail.value, "utf8");
|
||||
if (bytes > 16_384) return null;
|
||||
details.push({ name: detail.name, value: detail.value });
|
||||
}
|
||||
if (details[0]!.name !== "Provenance" || details[0]!.value !== PROVENANCE
|
||||
|| details[1]!.name !== "Partial reasons" || details[2]!.name !== "Collector truncated"
|
||||
|| !["true", "false"].includes(details[2]!.value)) return null;
|
||||
const reasons = details[1]!.value.split(", ");
|
||||
if (!reasons.includes("native_counter_semantics_unverified") || reasons.length > REASONS.size
|
||||
|| new Set(reasons).size !== reasons.length || !reasons.every(reason => REASONS.has(reason))) return null;
|
||||
const invocations = new Map<string, { role: string; nativeRun: number; sequence: number }>();
|
||||
let count = 0;
|
||||
const retained = details.slice(0, 3);
|
||||
for (let group = 3; group < details.length; group++) {
|
||||
const values: unknown = JSON.parse(details[group]!.value);
|
||||
if (!Array.isArray(values) || values.length < 1 || values.length > 8
|
||||
|| (group < details.length - 1 && values.length !== 8)) return null;
|
||||
if (details[group]!.name !== `Native observations ${count + 1}-${count + values.length}`) return null;
|
||||
const observations = [];
|
||||
for (const value of values) {
|
||||
if (!object(value) || !exact(value, ["invocationId", "role", "nativeRun", "sequence", "counters"])
|
||||
|| typeof value.invocationId !== "string" || !/^invocation-([1-9]|[1-5][0-9]|6[0-4])$/.test(value.invocationId)
|
||||
|| (value.role !== "parent" && value.role !== "child") || !positive(value.nativeRun)
|
||||
|| !positive(value.sequence) || !object(value.counters)) return null;
|
||||
const counters: Record<string, number> = {};
|
||||
for (const [key, number] of Object.entries(value.counters)) {
|
||||
if (!COUNTERS.has(key) || !Number.isSafeInteger(number) || Number(number) < 0) return null;
|
||||
counters[key] = Number(number);
|
||||
}
|
||||
const prior = invocations.get(value.invocationId);
|
||||
if (prior && (prior.role !== value.role || prior.nativeRun !== value.nativeRun || prior.sequence >= value.sequence)) return null;
|
||||
invocations.set(value.invocationId, { role: value.role, nativeRun: value.nativeRun, sequence: value.sequence });
|
||||
observations.push({ invocationId: value.invocationId, role: value.role, nativeRun: value.nativeRun, sequence: value.sequence, counters });
|
||||
}
|
||||
count += observations.length;
|
||||
if (count > 64) return null;
|
||||
retained.push({ name: details[group]!.name, value: JSON.stringify(observations) });
|
||||
}
|
||||
return {
|
||||
eventType: "provider.notice.recorded", itemId: raw.itemId,
|
||||
payload: { schema: "paperclip.provider.notice.v1", noticeId: raw.itemId, severity: "info",
|
||||
category: "cursor_native_usage_observed", scope: "turn", recoverable: true, userActionable: false,
|
||||
summary: SUMMARY, details: retained },
|
||||
};
|
||||
} catch {
|
||||
// Optional diagnostics, including malformed JSON, cannot poison settlement.
|
||||
return null;
|
||||
}
|
||||
}
|
||||
@@ -1,7 +1,7 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { createHash } from "node:crypto";
|
||||
import { existsSync } from "node:fs";
|
||||
import { mkdtemp, readFile, writeFile } from "node:fs/promises";
|
||||
import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
|
||||
import { tmpdir } from "node:os";
|
||||
import { join } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
@@ -23,6 +23,7 @@ import {
|
||||
} from "./live-session.js";
|
||||
import { DurableCapabilityLiveSessionStore } from "./durable-live-session-store.js";
|
||||
import { defaultCapabilityRunnerdBinary } from "./runnerd-codex-transport.js";
|
||||
import { persistedCursorUsageNotice } from "../drivers/acpx/usage-accounting.js";
|
||||
import { captureTurnRejection } from "../../test/capture-turn-rejection.js";
|
||||
import * as workspaceDiff from "./workspace-diff.js";
|
||||
|
||||
@@ -1164,6 +1165,139 @@ describe("Capability live runnerd and Codex session", () => {
|
||||
await service.shutdown(session.id);
|
||||
});
|
||||
|
||||
describe("Cursor partial usage notice capture", () => {
|
||||
function notice(threadId: string, turnId: string): CodexRpcNotification {
|
||||
const canonical = persistedCursorUsageNotice({ promptMessageIds: [] }, {
|
||||
lastRequestId: "request-1", promptMessageIds: ["prompt-1"], cursorPromptUsage: {
|
||||
request_id: "request-1", prompt_message_id: "prompt-1", receipt: {
|
||||
schema: "paperclip.cursor.native-usage.v1", source: "native_turn_ended",
|
||||
promptId: "12345678-1234-1234-1234-123456789abc", completeness: "partial",
|
||||
reasons: ["native_counter_semantics_unverified", "child_run_attribution_unverified"],
|
||||
observations: [
|
||||
{ invocationId: "invocation-1", role: "parent", nativeRun: 1, sequence: 1, counters: { inputTokens: 12, outputTokens: 3 } },
|
||||
{ invocationId: "invocation-2", role: "child", nativeRun: 1, sequence: 1, counters: { cacheReadTokens: 2, cacheWriteTokens: 4, reasoningTokens: 1 } },
|
||||
],
|
||||
limits: { maxObservations: 64, maxInvocations: 64, maxBytes: 16384 }, truncated: false,
|
||||
},
|
||||
},
|
||||
}, "request-1", "cursor", `${turnId}:cursor-native-usage`)!;
|
||||
return { method: "paperclip/canonicalProviderEvent", params: { threadId, turnId, ...canonical } };
|
||||
}
|
||||
const retained = (snapshot: CapabilityLiveSessionSnapshot) => snapshot.evidence.filter(entry =>
|
||||
entry.kind === "provider_event" && entry.data.canonical === true && entry.data.event === "provider.notice.recorded");
|
||||
|
||||
it("persists and reloads bounded parent/child observations without authoritative usage", async () => {
|
||||
const state = providerState(); state.omitReadUsage = true;
|
||||
state.onUsage = (queue, turnId) => {
|
||||
queue.push(notice(state.threadId, turnId));
|
||||
queue.push({ method: "turn/completed", params: { threadId: state.threadId, turn: { id: turnId, status: "completed" } } });
|
||||
};
|
||||
const directory = await mkdtemp(join(tmpdir(), "cursor-usage-checkpoint-"));
|
||||
const binding = { sessionId: "cursor-usage-session", runId: "cursor-usage-run", companyId: "company-1", actorId: "actor-1", taskId: "task-1" };
|
||||
const store = new DurableCapabilityLiveSessionStore({ directory, binding });
|
||||
const service = new CapabilityLiveSessionService({ store, transportFactory: fakeTransportFactory(state), transportOptions: { acpxCandidateProfile: "cursor" } });
|
||||
const session = await service.create({ ...binding, provider: "acpx", acpxAgent: "cursor", requestedModel: "exact-model" });
|
||||
try {
|
||||
const result = await session.sendMessage("Orient to this task.");
|
||||
expect(result.status).toBe("completed");
|
||||
const rows = retained(result.snapshot);
|
||||
expect(rows).toHaveLength(1);
|
||||
expect(rows[0]?.turnId).toBe(result.turnId);
|
||||
expect(rows[0]?.data.payload).toEqual(notice(state.threadId, result.turnId).params.payload);
|
||||
expect(result.snapshot.usageLedger).toEqual([]);
|
||||
expect(result.snapshot.usageUnavailable).toEqual([expect.objectContaining({ tokenUsage: null, costNanodollars: null, reason: "provider_did_not_report_usage" })]);
|
||||
expect(retained((await store.load(session.id))!)).toEqual(rows);
|
||||
await service.shutdown(session.id);
|
||||
const reloadedStore = new DurableCapabilityLiveSessionStore({ directory, binding });
|
||||
expect(retained((await reloadedStore.load(session.id))!)).toEqual(rows);
|
||||
} finally {
|
||||
await service.shutdown(session.id);
|
||||
await rm(directory, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
const malformedNotices: Array<(params: Record<string, any>, payload: Record<string, any>) => void> = [
|
||||
params => { params.threadId = "foreign-thread"; },
|
||||
params => { params.turnId = "previous-turn"; },
|
||||
params => { delete params.turnId; },
|
||||
params => { params.itemId = "raw private identity with spaces"; },
|
||||
params => { params.unexpected = "SECRET_CANARY"; },
|
||||
(_params, payload) => { payload.unexpected = "SECRET_CANARY"; },
|
||||
(_params, payload) => { payload.summary = "SECRET_CANARY"; },
|
||||
(_params, payload) => { payload.details.push({ name: "rawProviderText", value: "SECRET_CANARY" }); },
|
||||
(_params, payload) => { payload.details.push(payload.details[0]); },
|
||||
(_params, payload) => { payload.details[1].value = "native_counter_semantics_unverified, SECRET_CANARY"; },
|
||||
(_params, payload) => { payload.details[3].value = "{"; },
|
||||
(_params, payload) => { payload.details[3].value = "x".repeat(4001); },
|
||||
(_params, payload) => { payload.details[3].value = JSON.stringify([{ invocationId: "invocation-1", role: "parent", nativeRun: 1, sequence: 1, counters: { inputTokens: -1 } }]); },
|
||||
(_params, payload) => { payload.details[3].value = JSON.stringify([{ invocationId: "invocation-1", role: "parent", nativeRun: 1, sequence: 1, counters: { inputTokens: 1, rawText: "SECRET_CANARY" } }]); },
|
||||
(_params, payload) => { const values = JSON.parse(payload.details[3].value); values[1] = values[0]; payload.details[3].value = JSON.stringify(values); },
|
||||
(_params, payload) => { const values = JSON.parse(payload.details[3].value); values[1].invocationId = values[0].invocationId; values[1].sequence = 2; payload.details[3].value = JSON.stringify(values); },
|
||||
(_params, payload) => { payload.details[3].name = "Native observations 2-3"; },
|
||||
(_params, payload) => { payload.details = Array.from({ length: 12 }, () => payload.details[0]); },
|
||||
(_params, payload) => { const values = JSON.parse(payload.details[3].value); values[0].counters.inputTokens = Number.MAX_SAFE_INTEGER + 1; payload.details[3].value = JSON.stringify(values); },
|
||||
(_params, payload) => { const values = JSON.parse(payload.details[3].value); values[0].rawText = "SECRET_CANARY"; payload.details[3].value = JSON.stringify(values); },
|
||||
];
|
||||
it.each(malformedNotices.map((mutate, index) => ({ mutate, index })))("rejects malformed/foreign notice $index without poisoning settlement", async ({ mutate }) => {
|
||||
const state = providerState(); state.omitReadUsage = true;
|
||||
state.onUsage = (queue, turnId) => {
|
||||
const value = notice(state.threadId, turnId);
|
||||
mutate(value.params, value.params.payload as Record<string, any>);
|
||||
queue.push(value);
|
||||
queue.push({ method: "turn/completed", params: { threadId: state.threadId, turn: { id: turnId, status: "completed" } } });
|
||||
};
|
||||
const service = new CapabilityLiveSessionService({ transportFactory: fakeTransportFactory(state), transportOptions: { acpxCandidateProfile: "cursor" } });
|
||||
const session = await service.create({ provider: "acpx", acpxAgent: "cursor", requestedModel: "exact-model" });
|
||||
try {
|
||||
const result = await session.sendMessage("Orient to this task.");
|
||||
expect(result.status).toBe("completed"); expect(retained(result.snapshot)).toEqual([]);
|
||||
expect(JSON.stringify(result.snapshot.evidence)).not.toContain("SECRET_CANARY");
|
||||
expect(result.snapshot.usageLedger).toEqual([]); expect(result.snapshot.usageUnavailable).toHaveLength(1);
|
||||
} finally { await service.shutdown(session.id); }
|
||||
});
|
||||
|
||||
it.each([0, 64])("retains %i bounded observations, deduplicates notices and rejects stale turns", async count => {
|
||||
const state = providerState(); state.omitReadUsage = true;
|
||||
state.onUsage = (queue, turnId) => {
|
||||
queue.push(notice(state.threadId, "previous-turn"));
|
||||
const value = notice(state.threadId, turnId);
|
||||
const payload = value.params.payload as Record<string, any>;
|
||||
const observation = JSON.parse(payload.details[3].value)[0];
|
||||
payload.details = payload.details.slice(0, 3);
|
||||
if (count === 0) payload.details[1].value += ", native_terminal_not_observed";
|
||||
for (let offset = 0; offset < count; offset += 8) {
|
||||
payload.details.push({ name: `Native observations ${offset + 1}-${offset + 8}`, value: JSON.stringify(Array.from({ length: 8 }, (_, index) => ({ ...observation, invocationId: `invocation-${offset + index + 1}` }))) });
|
||||
}
|
||||
queue.push(value); queue.push(structuredClone(value));
|
||||
queue.push({ method: "turn/completed", params: { threadId: state.threadId, turn: { id: turnId, status: "completed" } } });
|
||||
};
|
||||
const service = new CapabilityLiveSessionService({ transportFactory: fakeTransportFactory(state), transportOptions: { acpxCandidateProfile: "cursor" } });
|
||||
const session = await service.create({ provider: "acpx", acpxAgent: "cursor", requestedModel: "exact-model" });
|
||||
try {
|
||||
const result = await session.sendMessage("Orient to this task.");
|
||||
expect(result.status).toBe("completed"); expect(retained(result.snapshot)).toHaveLength(1);
|
||||
const payload = retained(result.snapshot)[0]!.data.payload as Record<string, any>;
|
||||
expect(payload.details.slice(3).flatMap((detail: { value: string }) => JSON.parse(detail.value))).toHaveLength(count);
|
||||
expect(result.snapshot.usageLedger).toEqual([]); expect(result.snapshot.usageUnavailable).toHaveLength(1);
|
||||
} finally { await service.shutdown(session.id); }
|
||||
});
|
||||
|
||||
it.each(["pi", "copilot"] as const)("does not accept a Cursor receipt from %s", async acpxAgent => {
|
||||
const state = providerState(); state.omitReadUsage = true;
|
||||
state.onUsage = (queue, turnId) => {
|
||||
queue.push(notice(state.threadId, turnId));
|
||||
queue.push({ method: "turn/completed", params: { threadId: state.threadId, turn: { id: turnId, status: "completed" } } });
|
||||
};
|
||||
const service = new CapabilityLiveSessionService({ transportFactory: fakeTransportFactory(state), transportOptions: { acpxCandidateProfile: acpxAgent } });
|
||||
const session = await service.create({ provider: "acpx", acpxAgent, requestedModel: acpxAgent === "pi" ? "openrouter/deepseek/deepseek-v4-flash-0731" : "exact-model" });
|
||||
try {
|
||||
const result = await session.sendMessage("Orient to this task.");
|
||||
expect(result.status).toBe("completed"); expect(retained(result.snapshot)).toEqual([]);
|
||||
expect(result.snapshot.usageLedger).toEqual([]);
|
||||
} finally { await service.shutdown(session.id); }
|
||||
});
|
||||
});
|
||||
|
||||
it.each(["pi", "cursor", "copilot"] as const)("retains a completed %s result with explicit unavailable usage and no fabricated receipt", async (acpxAgent) => {
|
||||
const state = providerState();
|
||||
state.omitReadUsage = true;
|
||||
|
||||
@@ -1,3 +1,4 @@
|
||||
import { cursorUsageNotice } from "./cursor-usage-notice.js";
|
||||
import { liveRunResultFeedback } from "./run-result-feedback.js";
|
||||
import { createHash, randomUUID } from "node:crypto";
|
||||
import { resolve } from "node:path";
|
||||
@@ -2757,6 +2758,23 @@ export class CapabilityLiveSession {
|
||||
|
||||
async #handleNotification(notification: CodexRpcNotification): Promise<void> {
|
||||
const params = notification.params;
|
||||
if (notification.method === "paperclip/canonicalProviderEvent"
|
||||
&& params.eventType === "provider.notice.recorded"
|
||||
&& record(params.payload).category === "cursor_native_usage_observed") {
|
||||
const notice = cursorUsageNotice(params, {
|
||||
provider: this.#config.provider, agent: this.#config.acpxAgent,
|
||||
threadId: this.#providerThreadId, turnId: this.#activeTurnId,
|
||||
});
|
||||
if (notice && !this.#evidence.some(entry => entry.turnId === this.#activeTurnId
|
||||
&& entry.kind === "provider_event" && entry.data.canonical === true
|
||||
&& record(entry.data.payload).category === "cursor_native_usage_observed")) {
|
||||
this.#appendEvidence("provider_event", this.#activeTurnId, {
|
||||
canonicalEventType: notice.eventType, itemId: notice.itemId, payload: notice.payload,
|
||||
});
|
||||
await this.#persist();
|
||||
}
|
||||
return;
|
||||
}
|
||||
const turn = record(params.turn);
|
||||
const item = record(params.item);
|
||||
const turnId = text(params.turnId, text(turn.id));
|
||||
|
||||
Reference in new issue
Block a user