mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-10-24 01:41:42 +02:00
fix: integrate reviewed adapter and quota contracts into accounting
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Devin FoleyandPaperclip committed
commit
c0aa8f879d
19 files changed
+831
-543
No files matched your search
@@ -0,0 +1,73 @@
|
||||
import { describe, expect, it, vi } from "vitest";
|
||||
import { createUsageCheckpointLog } from "./usage-checkpoint.js";
|
||||
import { runChildProcess } from "./server-utils.js";
|
||||
import type { AdapterUsageCheckpoint } from "./types.js";
|
||||
|
||||
const createParser = () => {
|
||||
let costUsd = 0;
|
||||
return (stdout: string): AdapterUsageCheckpoint => {
|
||||
const events = stdout.trim().split("\n").map(line => JSON.parse(line));
|
||||
costUsd += events.reduce((sum, event) => sum + event.usage.costUsd, 0);
|
||||
return { costUsd, complete: events.some(event => event.type === "result") };
|
||||
};
|
||||
};
|
||||
const event = JSON.stringify({ type: "usage", usage: { inputTokens: 2, costUsd: 0.25 }, text: "private content" });
|
||||
|
||||
describe("usage checkpoint stream", () => {
|
||||
it("flushes an unterminated Pi price and closes each attempt independently", async () => {
|
||||
const saved = vi.fn(); const parser = vi.fn(createParser());
|
||||
const first = createUsageCheckpointLog(vi.fn(), saved, parser);
|
||||
await first("stdout", event);
|
||||
expect(saved).not.toHaveBeenCalled();
|
||||
await first.flush({ complete: true });
|
||||
const second = createUsageCheckpointLog(vi.fn(), saved, createParser());
|
||||
await second("stdout", event + "\n");
|
||||
await second.flush({ complete: true });
|
||||
expect(saved.mock.calls[0][0]).toMatchObject({ costUsd: 0.25, complete: true });
|
||||
expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 0.25, complete: true });
|
||||
expect(saved.mock.calls.at(-1)![0].attemptId).not.toBe(saved.mock.calls[0][0].attemptId);
|
||||
expect(parser.mock.calls[0][0]).not.toContain("private content");
|
||||
});
|
||||
|
||||
it("surfaces a persistence failure outside the local process's best-effort log path", async () => {
|
||||
const failure = new Error("Receipt storage unavailable");
|
||||
const output = vi.fn(); const saved = vi.fn().mockRejectedValue(failure);
|
||||
const log = createUsageCheckpointLog(output, saved, createParser());
|
||||
const result = await runChildProcess("checkpoint-test", process.execPath, ["-e", `process.stdout.write(${JSON.stringify(event + "\n")})`], {
|
||||
cwd: process.cwd(), env: {}, timeoutSec: 5, graceSec: 1, onLog: log,
|
||||
});
|
||||
expect(result.exitCode).toBe(0);
|
||||
expect(output).toHaveBeenCalledWith("stdout", event + "\n");
|
||||
await expect(log.flush()).rejects.toBe(failure);
|
||||
});
|
||||
|
||||
it("does not promote partial usage to complete merely because the stream was flushed", async () => {
|
||||
const saved = vi.fn(); const log = createUsageCheckpointLog(vi.fn(), saved, createParser());
|
||||
await log("stdout", event); await log.flush();
|
||||
expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 0.25, complete: false });
|
||||
});
|
||||
|
||||
it("persists each changed total and the final receipt before publishing its chunk", async () => {
|
||||
const order: string[] = [];
|
||||
const saved = vi.fn(async (receipt: AdapterUsageCheckpoint) => { order.push(`saved:${receipt.costUsd}:${receipt.complete}`); });
|
||||
const log = createUsageCheckpointLog(async () => { order.push("log"); }, saved, createParser());
|
||||
await log("stdout", event + "\n");
|
||||
await log("stdout", event + "\n");
|
||||
await log("stdout", JSON.stringify({ type: "result", usage: { costUsd: 0 } }) + "\n");
|
||||
expect(order).toEqual(["saved:0.25:false", "log", "saved:0.5:false", "log", "saved:0.5:true", "log"]);
|
||||
});
|
||||
|
||||
it("saves every new receipt before logging with linear parsing work", async () => {
|
||||
const parser = vi.fn(createParser()); const saved = vi.fn();
|
||||
const log = createUsageCheckpointLog(vi.fn(), saved, parser);
|
||||
for (let i = 0; i < 1024; i++) {
|
||||
await log("stdout", event + "\n");
|
||||
expect(saved.mock.calls.at(-1)![0].costUsd).toBe((i + 1) * 0.25);
|
||||
}
|
||||
await log.flush({ complete: true });
|
||||
expect(saved.mock.calls.at(-1)![0]).toMatchObject({ costUsd: 256, complete: true });
|
||||
expect(saved).toHaveBeenCalledTimes(1025); // all updates plus completion
|
||||
expect(parser).toHaveBeenCalledTimes(1024);
|
||||
expect(parser.mock.calls.reduce((sum, [input]) => sum + input.length, 0)).toBeLessThanOrEqual(1024 * (event.length + 1));
|
||||
});
|
||||
});
|
||||
@@ -9,7 +9,7 @@ const fields = new Set([
|
||||
"cache_read_input_tokens", "cache_creation_input_tokens", "inputTokens", "outputTokens", "cacheReadInputTokens",
|
||||
"cacheCreationInputTokens", "cachedInputTokens", "promptTokenCount", "candidatesTokenCount", "cachedContentTokenCount",
|
||||
"reasoning", "cacheRead", "cacheWrite", "prompt", "candidates", "totalTokenCount", "total_tokens", "toolUsePromptTokenCount", "messages",
|
||||
"thoughtsTokenCount", "inputTokens", "totalTokens", "cached", "thoughts", "total_cost_usd", "cost_usd", "costUSD",
|
||||
"thoughtsTokenCount", "inputTokens", "totalTokens", "cached", "thoughts", "total_cost_usd", "cost_usd", "costUSD", "costUsd",
|
||||
]);
|
||||
const stringFields = new Set(["type", "subtype", "role", "id", "session_id", "sessionID", "thread_id", "model", "modelID", "providerID"]);
|
||||
function project(value: unknown, depth = 0, field = ""): unknown {
|
||||
@@ -24,36 +24,65 @@ function project(value: unknown, depth = 0, field = ""): unknown {
|
||||
: project(entry, depth + 1, key)]));
|
||||
}
|
||||
|
||||
/** A fresh instance belongs to one CLI attempt. Retries must not concatenate
|
||||
* cumulative usage from different attempts. Callback durability is awaited
|
||||
* before ordinary log publication. */
|
||||
/** The consumer incrementally parses only new records (never the full history).
|
||||
* A fresh instance belongs to one CLI attempt. Call flush() after its process
|
||||
* exits, before retrying or final result handling. Checkpoint failures are
|
||||
* retained outside runChildProcess's best-effort onLog error handler. */
|
||||
export function createUsageCheckpointLog(
|
||||
onLog: (stream: "stdout" | "stderr", chunk: string) => Promise<void>,
|
||||
onUsage: ((receipt: AdapterUsageCheckpoint) => Promise<void>) | undefined,
|
||||
parse: (stdout: string) => AdapterUsageCheckpoint | null,
|
||||
consume: (records: string) => AdapterUsageCheckpoint | null,
|
||||
) {
|
||||
if (!onUsage) return onLog;
|
||||
const attemptId = randomUUID();
|
||||
let remainder = "", accounting = "", previous = "";
|
||||
return async (stream: "stdout" | "stderr", chunk: string) => {
|
||||
if (stream === "stdout") {
|
||||
remainder += chunk;
|
||||
const lines = remainder.split("\n"); remainder = lines.pop() ?? "";
|
||||
if (remainder.length > 8 * 1024 * 1024) throw new Error("Accounting protocol line exceeds 8 MiB");
|
||||
for (const line of lines) {
|
||||
let raw: unknown;
|
||||
try { raw = JSON.parse(line); } catch { continue; }
|
||||
const compact = JSON.stringify(project(raw));
|
||||
if (!compact || !/usage|tokens|cost|"result"|"turn.completed"|"agent_end"/.test(compact)) continue;
|
||||
accounting += compact + "\n";
|
||||
if (accounting.length > 8 * 1024 * 1024) throw new Error("Accounting checkpoint history exceeds 8 MiB");
|
||||
const receipt = parse(accounting);
|
||||
if (!receipt) continue;
|
||||
const serialized = JSON.stringify(receipt);
|
||||
if (serialized !== previous) { await onUsage({ ...receipt, attemptId }); previous = serialized; }
|
||||
let snapshot: AdapterUsageCheckpoint | null = null;
|
||||
let failure: unknown;
|
||||
let failed = false;
|
||||
function retain(line: string) {
|
||||
let raw: unknown;
|
||||
try { raw = JSON.parse(line); } catch { return; }
|
||||
const compact = JSON.stringify(project(raw));
|
||||
if (!compact || !/usage|tokens|cost|"result"|"turn.completed"|"agent_end"|"model"|"modelID"/.test(compact)) return;
|
||||
accounting += compact + "\n";
|
||||
if (accounting.length > 8 * 1024 * 1024) throw new Error("Accounting checkpoint chunk exceeds 8 MiB");
|
||||
}
|
||||
async function publish(complete = false) {
|
||||
if (!onUsage) return;
|
||||
if (accounting) {
|
||||
snapshot = consume(accounting);
|
||||
accounting = "";
|
||||
}
|
||||
const parsed = snapshot;
|
||||
if (!parsed) return;
|
||||
if (!parsed.complete && parsed.costUsd == null && parsed.costUsdExact == null &&
|
||||
!Object.values(parsed.usage ?? {}).some(value => typeof value === "number" && value > 0)) return;
|
||||
const receipt = { ...parsed, complete: parsed.complete || complete };
|
||||
const serialized = JSON.stringify(receipt);
|
||||
if (serialized !== previous) { await onUsage({ ...receipt, attemptId }); previous = serialized; }
|
||||
}
|
||||
const log = async (stream: "stdout" | "stderr", chunk: string) => {
|
||||
if (onUsage && !failed && stream === "stdout") {
|
||||
try {
|
||||
remainder += chunk;
|
||||
const lines = remainder.split("\n"); remainder = lines.pop() ?? "";
|
||||
if (remainder.length > 8 * 1024 * 1024) throw new Error("Accounting protocol line exceeds 8 MiB");
|
||||
for (const line of lines) retain(line);
|
||||
await publish();
|
||||
} catch (error) {
|
||||
failed = true; failure = error; remainder = ""; accounting = "";
|
||||
}
|
||||
}
|
||||
// Receipt persistence and ordinary logs have separate failure contracts.
|
||||
// A failed receipt must not hide the provider's diagnostic output.
|
||||
await onLog(stream, chunk);
|
||||
};
|
||||
return Object.assign(log, {
|
||||
async flush(options: { complete?: boolean } = {}) {
|
||||
if (failed) throw failure;
|
||||
if (!onUsage) return;
|
||||
if (remainder) { retain(remainder); remainder = ""; }
|
||||
await publish(options.complete ?? false);
|
||||
},
|
||||
});
|
||||
}
|
||||
import { randomUUID } from "node:crypto";
|
||||
@@ -65,7 +65,7 @@ import {
|
||||
import {
|
||||
claudeModelUsageTotals,
|
||||
claudeModelReceipts,
|
||||
parseClaudeStreamJson,
|
||||
parseClaudeStreamJson, createClaudeStreamParser,
|
||||
describeClaudeFailure,
|
||||
detectClaudeLoginRequired,
|
||||
extractClaudeRetryNotBefore,
|
||||
@@ -964,6 +964,14 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
});
|
||||
}
|
||||
|
||||
const consumeAccounting = createClaudeStreamParser();
|
||||
const accountingLog = createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
|
||||
const parsed = consumeAccounting(stdout);
|
||||
return { usage: parsed.usage ?? undefined, usageBasis: "per_run", costUsd: parsed.costUsd,
|
||||
provider: "anthropic", biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic",
|
||||
billingType, model: Object.keys(parseObject(parsed.resultJson?.modelUsage)).length > 1 ? "mixed" : parsed.model || model,
|
||||
usageByModel: claudeModelReceipts(parsed.resultJson?.modelUsage), complete: parsed.resultJson !== null };
|
||||
});
|
||||
const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, {
|
||||
onProcessStopped: providerStop.beginInvocation(),
|
||||
cwd,
|
||||
@@ -973,12 +981,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog: createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
|
||||
const parsed = parseClaudeStreamJson(stdout);
|
||||
return { usage: parsed.usage ?? undefined, usageBasis: "per_run", costUsd: parsed.costUsd,
|
||||
provider: "anthropic", biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic",
|
||||
billingType, model: parsed.model || model, complete: parsed.resultJson !== null };
|
||||
}),
|
||||
onLog: accountingLog,
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
terminalResultCleanup: {
|
||||
@@ -987,6 +990,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
},
|
||||
localProcessSandbox,
|
||||
});
|
||||
await accountingLog.flush();
|
||||
|
||||
const parsedStream = parseClaudeStreamJson(proc.stdout);
|
||||
const parsed = parsedStream.resultJson ?? parseJson(proc.stdout);
|
||||
@@ -1019,11 +1023,13 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: proc.exitCode,
|
||||
signal: proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: parsedStream.resultJson !== null,
|
||||
usage: parsedStream.usage ?? undefined,
|
||||
usageBasis: "per_run",
|
||||
provider: "anthropic",
|
||||
biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic",
|
||||
model: parsedStream.model || model,
|
||||
model: Object.keys(parseObject(parsedStream.resultJson?.modelUsage)).length > 1 ? "mixed" : parsedStream.model || model,
|
||||
usageByModel: claudeModelReceipts(parsedStream.resultJson?.modelUsage),
|
||||
billingType,
|
||||
costUsd: parsedStream.costUsd,
|
||||
errorMessage: `Timed out after ${timeoutSec}s`,
|
||||
@@ -1087,11 +1093,13 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: proc.exitCode,
|
||||
signal: proc.signal,
|
||||
timedOut: false,
|
||||
usageComplete: parsedStream.resultJson !== null,
|
||||
usage: parsedStream.usage ?? undefined,
|
||||
usageBasis: "per_run",
|
||||
provider: "anthropic",
|
||||
biller: isBedrockAuth(effectiveEnv) ? "aws_bedrock" : "anthropic",
|
||||
model: parsedStream.model || model,
|
||||
model: Object.keys(parseObject(parsedStream.resultJson?.modelUsage)).length > 1 ? "mixed" : parsedStream.model || model,
|
||||
usageByModel: claudeModelReceipts(parsedStream.resultJson?.modelUsage),
|
||||
billingType,
|
||||
costUsd: parsedStream.costUsd,
|
||||
errorMessage: fallbackErrorMessage,
|
||||
@@ -1256,6 +1264,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: proc.exitCode,
|
||||
signal: proc.signal,
|
||||
timedOut: false,
|
||||
usageComplete: parsedStream.resultJson !== null || parsed.type === "result",
|
||||
errorMessage,
|
||||
errorCode: resolvedErrorCode,
|
||||
errorFamily,
|
||||
|
||||
@@ -67,93 +67,102 @@ export function claudeModelReceipts(modelUsage: unknown) {
|
||||
}
|
||||
|
||||
export function parseClaudeStreamJson(stdout: string) {
|
||||
return createClaudeStreamParser()(stdout);
|
||||
}
|
||||
|
||||
/** Consume complete JSONL records once, retaining protocol accounting state. */
|
||||
export function createClaudeStreamParser() {
|
||||
let sessionId: string | null = null;
|
||||
let model = "";
|
||||
let finalResult: Record<string, unknown> | null = null;
|
||||
const assistantTexts: string[] = [];
|
||||
const messageUsage = new Map<string, UsageSummary>();
|
||||
let anonymousMessage = 0;
|
||||
const observedTotals = { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0 };
|
||||
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = rawLine.trim();
|
||||
if (!line) continue;
|
||||
const event = parseJson(line);
|
||||
if (!event) continue;
|
||||
return (stdout: string) => {
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = rawLine.trim();
|
||||
if (!line) continue;
|
||||
const event = parseJson(line);
|
||||
if (!event) continue;
|
||||
|
||||
const type = asString(event.type, "");
|
||||
if (type === "system" && asString(event.subtype, "") === "init") {
|
||||
sessionId = asString(event.session_id, sessionId ?? "") || sessionId;
|
||||
model = asString(event.model, model);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "assistant") {
|
||||
sessionId = asString(event.session_id, sessionId ?? "") || sessionId;
|
||||
const message = parseObject(event.message);
|
||||
const observed = parseObject(message.usage);
|
||||
if (Object.keys(observed).length > 0) {
|
||||
const key = asString(message.id, "") || `anonymous:${anonymousMessage++}`;
|
||||
const previous = messageUsage.get(key);
|
||||
messageUsage.set(key, {
|
||||
inputTokens: Math.max(previous?.inputTokens ?? 0, asNumber(observed.input_tokens, 0) + asNumber(observed.cache_creation_input_tokens, 0)),
|
||||
cachedInputTokens: Math.max(previous?.cachedInputTokens ?? 0, asNumber(observed.cache_read_input_tokens, 0)),
|
||||
outputTokens: Math.max(previous?.outputTokens ?? 0, asNumber(observed.output_tokens, 0)),
|
||||
});
|
||||
const type = asString(event.type, "");
|
||||
if (type === "system" && asString(event.subtype, "") === "init") {
|
||||
sessionId = asString(event.session_id, sessionId ?? "") || sessionId;
|
||||
model = asString(event.model, model);
|
||||
continue;
|
||||
}
|
||||
const content = Array.isArray(message.content) ? message.content : [];
|
||||
for (const entry of content) {
|
||||
if (typeof entry !== "object" || entry === null || Array.isArray(entry)) continue;
|
||||
const block = entry as Record<string, unknown>;
|
||||
if (asString(block.type, "") === "text") {
|
||||
const text = asString(block.text, "");
|
||||
if (text) assistantTexts.push(text);
|
||||
|
||||
if (type === "assistant") {
|
||||
sessionId = asString(event.session_id, sessionId ?? "") || sessionId;
|
||||
const message = parseObject(event.message);
|
||||
const observed = parseObject(message.usage);
|
||||
if (Object.keys(observed).length > 0) {
|
||||
const key = asString(message.id, "") || `anonymous:${anonymousMessage++}`;
|
||||
const previous = messageUsage.get(key);
|
||||
const next = {
|
||||
inputTokens: Math.max(previous?.inputTokens ?? 0, asNumber(observed.input_tokens, 0) + asNumber(observed.cache_creation_input_tokens, 0)),
|
||||
cachedInputTokens: Math.max(previous?.cachedInputTokens ?? 0, asNumber(observed.cache_read_input_tokens, 0)),
|
||||
outputTokens: Math.max(previous?.outputTokens ?? 0, asNumber(observed.output_tokens, 0)),
|
||||
};
|
||||
observedTotals.inputTokens += next.inputTokens - (previous?.inputTokens ?? 0);
|
||||
observedTotals.outputTokens += next.outputTokens - (previous?.outputTokens ?? 0);
|
||||
observedTotals.cachedInputTokens += next.cachedInputTokens - (previous?.cachedInputTokens ?? 0);
|
||||
messageUsage.set(key, next);
|
||||
}
|
||||
const content = Array.isArray(message.content) ? message.content : [];
|
||||
for (const entry of content) {
|
||||
if (typeof entry !== "object" || entry === null || Array.isArray(entry)) continue;
|
||||
const block = entry as Record<string, unknown>;
|
||||
if (asString(block.type, "") === "text") {
|
||||
const text = asString(block.text, "");
|
||||
if (text) assistantTexts.push(text);
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "result") {
|
||||
finalResult = event;
|
||||
sessionId = asString(event.session_id, sessionId ?? "") || sessionId;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "result") {
|
||||
finalResult = event;
|
||||
sessionId = asString(event.session_id, sessionId ?? "") || sessionId;
|
||||
if (!finalResult) {
|
||||
return {
|
||||
sessionId,
|
||||
model,
|
||||
costUsd: null as number | null,
|
||||
usage: messageUsage.size ? { ...observedTotals } : null,
|
||||
usageBasis: "per_run" as const,
|
||||
summary: assistantTexts.join("\n\n").trim(),
|
||||
resultJson: null as Record<string, unknown> | null,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
if (!finalResult) {
|
||||
const modelUsageTotals = claudeModelUsageTotals(finalResult.modelUsage);
|
||||
const usageObj = parseObject(finalResult.usage);
|
||||
const usage: UsageSummary = modelUsageTotals ?? {
|
||||
inputTokens: asNumber(usageObj.input_tokens, 0) + asNumber(usageObj.cache_creation_input_tokens, 0),
|
||||
cachedInputTokens: asNumber(usageObj.cache_read_input_tokens, 0),
|
||||
outputTokens: asNumber(usageObj.output_tokens, 0),
|
||||
};
|
||||
const costRaw = finalResult.total_cost_usd;
|
||||
const costUsd = typeof costRaw === "number" && Number.isFinite(costRaw) ? costRaw : null;
|
||||
const summary = asString(finalResult.result, assistantTexts.join("\n\n")).trim();
|
||||
|
||||
return {
|
||||
sessionId,
|
||||
model,
|
||||
costUsd: null as number | null,
|
||||
usage: messageUsage.size ? [...messageUsage.values()].reduce<UsageSummary>((total, part) => ({
|
||||
inputTokens: total.inputTokens + part.inputTokens, outputTokens: total.outputTokens + part.outputTokens,
|
||||
cachedInputTokens: (total.cachedInputTokens ?? 0) + (part.cachedInputTokens ?? 0),
|
||||
}), { inputTokens: 0, outputTokens: 0, cachedInputTokens: 0 }) : null,
|
||||
costUsd,
|
||||
usage,
|
||||
// modelUsage covers exactly this CLI invocation, so mark it per-run to
|
||||
// keep the server from applying its session-cumulative delta heuristic.
|
||||
usageBasis: "per_run" as const,
|
||||
summary: assistantTexts.join("\n\n").trim(),
|
||||
resultJson: null as Record<string, unknown> | null,
|
||||
summary,
|
||||
resultJson: finalResult,
|
||||
};
|
||||
}
|
||||
|
||||
const modelUsageTotals = claudeModelUsageTotals(finalResult.modelUsage);
|
||||
const usageObj = parseObject(finalResult.usage);
|
||||
const usage: UsageSummary = modelUsageTotals ?? {
|
||||
inputTokens: asNumber(usageObj.input_tokens, 0) + asNumber(usageObj.cache_creation_input_tokens, 0),
|
||||
cachedInputTokens: asNumber(usageObj.cache_read_input_tokens, 0),
|
||||
outputTokens: asNumber(usageObj.output_tokens, 0),
|
||||
};
|
||||
const costRaw = finalResult.total_cost_usd;
|
||||
const costUsd = typeof costRaw === "number" && Number.isFinite(costRaw) ? costRaw : null;
|
||||
const summary = asString(finalResult.result, assistantTexts.join("\n\n")).trim();
|
||||
|
||||
return {
|
||||
sessionId,
|
||||
model,
|
||||
costUsd,
|
||||
usage,
|
||||
// modelUsage covers exactly this CLI invocation, so mark it per-run to
|
||||
// keep the server from applying its session-cumulative delta heuristic.
|
||||
usageBasis: "per_run" as const,
|
||||
summary,
|
||||
resultJson: finalResult,
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -62,7 +62,7 @@ import {
|
||||
type LocalProcessSandboxOptions,
|
||||
} from "@paperclipai/adapter-utils/local-process-sandbox";
|
||||
import {
|
||||
parseCodexJsonl,
|
||||
parseCodexJsonl, createCodexJsonlParser,
|
||||
classifyCodexAuthRefreshFailure,
|
||||
extractCodexRetryNotBefore,
|
||||
isCodexHarnessCrash,
|
||||
@@ -1319,8 +1319,9 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
}
|
||||
};
|
||||
|
||||
const consumeAccounting = createCodexJsonlParser();
|
||||
const accountingLog = createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
|
||||
const parsed = parseCodexJsonl(stdout);
|
||||
const parsed = consumeAccounting(stdout);
|
||||
return { usage: parsed.usage, usageBasis: "per_run", provider: "openai", biller: resolveCodexBiller(effectiveEnv, billingType), billingType, model, costUsd: null, complete: parsed.sawProtocolTerminalEvent };
|
||||
});
|
||||
try {
|
||||
@@ -1347,6 +1348,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
localProcessSandbox,
|
||||
});
|
||||
await accountingLog.flush();
|
||||
const cleanedStderr = stripCodexRolloutNoise(proc.stderr);
|
||||
return {
|
||||
proc: {
|
||||
@@ -1396,6 +1398,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: null,
|
||||
signal: attempt.monitor.terminationSignal ?? attempt.proc.signal,
|
||||
timedOut: false,
|
||||
usageComplete: attempt.parsed.sawProtocolTerminalEvent,
|
||||
errorMessage,
|
||||
errorCode: "codex_output_inactivity_monitor",
|
||||
errorFamily: null,
|
||||
@@ -1428,6 +1431,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: attempt.parsed.sawProtocolTerminalEvent,
|
||||
usage: attempt.parsed.usage,
|
||||
usageBasis: "per_run",
|
||||
provider: "openai",
|
||||
@@ -1514,6 +1518,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: false,
|
||||
usageComplete: attempt.parsed.sawProtocolTerminalEvent,
|
||||
errorMessage:
|
||||
(attempt.proc.exitCode ?? 0) === 0
|
||||
? null
|
||||
|
||||
@@ -28,6 +28,11 @@ export type CodexAuthRefreshFailureClass =
|
||||
| "refresh_token_invalidated";
|
||||
|
||||
export function parseCodexJsonl(stdout: string) {
|
||||
return createCodexJsonlParser()(stdout);
|
||||
}
|
||||
|
||||
/** Consume complete JSONL records once, retaining protocol accounting state. */
|
||||
export function createCodexJsonlParser() {
|
||||
let sessionId: string | null = null;
|
||||
let finalMessage: string | null = null;
|
||||
let errorMessage: string | null = null;
|
||||
@@ -39,62 +44,64 @@ export function parseCodexJsonl(stdout: string) {
|
||||
outputTokens: 0,
|
||||
};
|
||||
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = rawLine.trim();
|
||||
if (!line) continue;
|
||||
return (stdout: string) => {
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = rawLine.trim();
|
||||
if (!line) continue;
|
||||
|
||||
const event = parseJson(line);
|
||||
if (!event) continue;
|
||||
const event = parseJson(line);
|
||||
if (!event) continue;
|
||||
|
||||
const type = asString(event.type, "");
|
||||
if (type) sawProtocolEvent = true;
|
||||
if (type === "error" || type === "turn.completed" || type === "turn.failed") {
|
||||
sawProtocolTerminalEvent = true;
|
||||
}
|
||||
if (type === "thread.started") {
|
||||
sessionId = asString(event.thread_id, sessionId ?? "") || sessionId;
|
||||
continue;
|
||||
}
|
||||
const type = asString(event.type, "");
|
||||
if (type) sawProtocolEvent = true;
|
||||
if (type === "error" || type === "turn.completed" || type === "turn.failed") {
|
||||
sawProtocolTerminalEvent = true;
|
||||
}
|
||||
if (type === "thread.started") {
|
||||
sessionId = asString(event.thread_id, sessionId ?? "") || sessionId;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "error") {
|
||||
const msg = asString(event.message, "").trim();
|
||||
if (msg) errorMessage = msg;
|
||||
continue;
|
||||
}
|
||||
if (type === "error") {
|
||||
const msg = asString(event.message, "").trim();
|
||||
if (msg) errorMessage = msg;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "item.completed") {
|
||||
const item = parseObject(event.item);
|
||||
if (asString(item.type, "") === "agent_message") {
|
||||
const text = asString(item.text, "");
|
||||
if (text) finalMessage = text;
|
||||
if (type === "item.completed") {
|
||||
const item = parseObject(event.item);
|
||||
if (asString(item.type, "") === "agent_message") {
|
||||
const text = asString(item.text, "");
|
||||
if (text) finalMessage = text;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "turn.completed") {
|
||||
const usageObj = parseObject(event.usage);
|
||||
usage.inputTokens = asNumber(usageObj.input_tokens, usage.inputTokens);
|
||||
usage.cachedInputTokens = asNumber(usageObj.cached_input_tokens, usage.cachedInputTokens);
|
||||
usage.outputTokens = asNumber(usageObj.output_tokens, usage.outputTokens);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "turn.failed") {
|
||||
const err = parseObject(event.error);
|
||||
const msg = asString(err.message, "").trim();
|
||||
if (msg) errorMessage = msg;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "turn.completed") {
|
||||
const usageObj = parseObject(event.usage);
|
||||
usage.inputTokens = asNumber(usageObj.input_tokens, usage.inputTokens);
|
||||
usage.cachedInputTokens = asNumber(usageObj.cached_input_tokens, usage.cachedInputTokens);
|
||||
usage.outputTokens = asNumber(usageObj.output_tokens, usage.outputTokens);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "turn.failed") {
|
||||
const err = parseObject(event.error);
|
||||
const msg = asString(err.message, "").trim();
|
||||
if (msg) errorMessage = msg;
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
sessionId,
|
||||
summary: finalMessage?.trim() ?? "",
|
||||
// Codex includes cache hits in input_tokens; Paperclip stores them separately.
|
||||
usage: { ...usage, inputTokens: Math.max(0, usage.inputTokens - usage.cachedInputTokens) },
|
||||
usageBasis: "per_run" as const,
|
||||
errorMessage,
|
||||
sawProtocolEvent,
|
||||
sawProtocolTerminalEvent,
|
||||
return {
|
||||
sessionId,
|
||||
summary: finalMessage?.trim() ?? "",
|
||||
// Codex includes cache hits in input_tokens; Paperclip stores them separately.
|
||||
usage: { ...usage, inputTokens: Math.max(0, usage.inputTokens - usage.cachedInputTokens) },
|
||||
usageBasis: "per_run" as const,
|
||||
errorMessage,
|
||||
sawProtocolEvent,
|
||||
sawProtocolTerminalEvent,
|
||||
};
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -54,7 +54,7 @@ import {
|
||||
joinPromptSections,
|
||||
} from "@paperclipai/adapter-utils/server-utils";
|
||||
import { DEFAULT_CURSOR_LOCAL_MODEL, SANDBOX_INSTALL_COMMAND } from "../index.js";
|
||||
import { firstCursorDiagnosticLine, parseCursorJsonl, isCursorUnknownSessionError } from "./parse.js";
|
||||
import { firstCursorDiagnosticLine, parseCursorJsonl, createCursorJsonlParser, isCursorUnknownSessionError } from "./parse.js";
|
||||
import { prepareCursorSandboxCommand } from "./remote-command.js";
|
||||
import { normalizeCursorStreamLine } from "../shared/stream.js";
|
||||
import { hasCursorTrustBypassArg } from "../shared/trust.js";
|
||||
@@ -627,10 +627,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
});
|
||||
}
|
||||
|
||||
const consumeAccounting = createCursorJsonlParser();
|
||||
const accountingLog = createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
|
||||
const parsed = parseCursorJsonl(stdout);
|
||||
const parsed = consumeAccounting(stdout);
|
||||
const provider = resolveProviderFromModel(model);
|
||||
return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveCursorBiller(effectiveEnv, billingType, provider), billingType, model, complete: false };
|
||||
return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveCursorBiller(effectiveEnv, billingType, provider), billingType, model, complete: parsed.sawResult };
|
||||
});
|
||||
let stdoutLineBuffer = "";
|
||||
const emitNormalizedStdoutLine = async (rawLine: string) => {
|
||||
@@ -676,6 +677,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
});
|
||||
await flushStdoutChunk("", true);
|
||||
await accountingLog.flush();
|
||||
|
||||
return {
|
||||
proc,
|
||||
@@ -704,6 +706,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: attempt.parsed.sawResult,
|
||||
usage: attempt.parsed.usage,
|
||||
usageBasis: "per_run",
|
||||
provider: providerFromModel,
|
||||
@@ -743,6 +746,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: false,
|
||||
usageComplete: attempt.parsed.sawResult,
|
||||
usageBasis: "per_run",
|
||||
errorMessage:
|
||||
(attempt.proc.exitCode ?? 0) === 0
|
||||
|
||||
@@ -61,7 +61,13 @@ function readSessionId(event: Record<string, unknown>): string | null {
|
||||
}
|
||||
|
||||
export function parseCursorJsonl(stdout: string) {
|
||||
return createCursorJsonlParser()(stdout);
|
||||
}
|
||||
|
||||
/** Consume complete JSONL records once, retaining protocol accounting state. */
|
||||
export function createCursorJsonlParser() {
|
||||
let sessionId: string | null = null;
|
||||
let sawResult = false;
|
||||
const messages: string[] = [];
|
||||
let errorMessage: string | null = null;
|
||||
let totalCostUsd = 0;
|
||||
@@ -79,92 +85,96 @@ export function parseCursorJsonl(stdout: string) {
|
||||
outputTokens: 0,
|
||||
};
|
||||
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = normalizeCursorStreamLine(rawLine).line;
|
||||
if (!line) continue;
|
||||
return (stdout: string) => {
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = normalizeCursorStreamLine(rawLine).line;
|
||||
if (!line) continue;
|
||||
|
||||
const event = parseJson(line);
|
||||
if (!event) continue;
|
||||
const event = parseJson(line);
|
||||
if (!event) continue;
|
||||
|
||||
const foundSession = readSessionId(event);
|
||||
if (foundSession) sessionId = foundSession;
|
||||
const foundSession = readSessionId(event);
|
||||
if (foundSession) sessionId = foundSession;
|
||||
|
||||
const type = asString(event.type, "").trim();
|
||||
const type = asString(event.type, "").trim();
|
||||
|
||||
if (type === "assistant") {
|
||||
messages.push(...collectAssistantText(event.message));
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "result") {
|
||||
const usageObj = parseObject(event.usage);
|
||||
usage.inputTokens += asNumber(
|
||||
usageObj.input_tokens,
|
||||
asNumber(usageObj.inputTokens, 0),
|
||||
);
|
||||
usage.cachedInputTokens += asNumber(
|
||||
usageObj.cached_input_tokens,
|
||||
asNumber(usageObj.cachedInputTokens, asNumber(usageObj.cache_read_input_tokens, 0)),
|
||||
);
|
||||
usage.outputTokens += asNumber(
|
||||
usageObj.output_tokens,
|
||||
asNumber(usageObj.outputTokens, 0),
|
||||
);
|
||||
addCost(event.total_cost_usd ?? event.cost_usd ?? event.cost);
|
||||
|
||||
const isError = event.is_error === true || asString(event.subtype, "").toLowerCase() === "error";
|
||||
const resultText = asString(event.result, "").trim();
|
||||
if (resultText && messages.length === 0) {
|
||||
messages.push(resultText);
|
||||
if (type === "assistant") {
|
||||
messages.push(...collectAssistantText(event.message));
|
||||
continue;
|
||||
}
|
||||
if (isError) {
|
||||
const resultError = asErrorText(event.error ?? event.message ?? event.result).trim();
|
||||
if (resultError) errorMessage = resultError;
|
||||
|
||||
if (type === "result") {
|
||||
sawResult = true;
|
||||
const usageObj = parseObject(event.usage);
|
||||
usage.inputTokens += asNumber(
|
||||
usageObj.input_tokens,
|
||||
asNumber(usageObj.inputTokens, 0),
|
||||
);
|
||||
usage.cachedInputTokens += asNumber(
|
||||
usageObj.cached_input_tokens,
|
||||
asNumber(usageObj.cachedInputTokens, asNumber(usageObj.cache_read_input_tokens, 0)),
|
||||
);
|
||||
usage.outputTokens += asNumber(
|
||||
usageObj.output_tokens,
|
||||
asNumber(usageObj.outputTokens, 0),
|
||||
);
|
||||
addCost(event.total_cost_usd ?? event.cost_usd ?? event.cost);
|
||||
|
||||
const isError = event.is_error === true || asString(event.subtype, "").toLowerCase() === "error";
|
||||
const resultText = asString(event.result, "").trim();
|
||||
if (resultText && messages.length === 0) {
|
||||
messages.push(resultText);
|
||||
}
|
||||
if (isError) {
|
||||
const resultError = asErrorText(event.error ?? event.message ?? event.result).trim();
|
||||
if (resultError) errorMessage = resultError;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "error") {
|
||||
const message = asErrorText(event.message ?? event.error ?? event.detail).trim();
|
||||
if (message) errorMessage = message;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "system") {
|
||||
const subtype = asString(event.subtype, "").trim().toLowerCase();
|
||||
if (subtype === "error") {
|
||||
if (type === "error") {
|
||||
const message = asErrorText(event.message ?? event.error ?? event.detail).trim();
|
||||
if (message) errorMessage = message;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "system") {
|
||||
const subtype = asString(event.subtype, "").trim().toLowerCase();
|
||||
if (subtype === "error") {
|
||||
const message = asErrorText(event.message ?? event.error ?? event.detail).trim();
|
||||
if (message) errorMessage = message;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
// Compatibility with older stream-json shapes.
|
||||
if (type === "text") {
|
||||
const part = parseObject(event.part);
|
||||
const text = asString(part.text, "").trim();
|
||||
if (text) messages.push(text);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "step_finish") {
|
||||
const part = parseObject(event.part);
|
||||
const tokens = parseObject(part.tokens);
|
||||
const cache = parseObject(tokens.cache);
|
||||
usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0);
|
||||
usage.cachedInputTokens += asNumber(cache.read, 0);
|
||||
usage.outputTokens += asNumber(tokens.output, 0);
|
||||
addCost(part.cost);
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
// Compatibility with older stream-json shapes.
|
||||
if (type === "text") {
|
||||
const part = parseObject(event.part);
|
||||
const text = asString(part.text, "").trim();
|
||||
if (text) messages.push(text);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "step_finish") {
|
||||
const part = parseObject(event.part);
|
||||
const tokens = parseObject(part.tokens);
|
||||
const cache = parseObject(tokens.cache);
|
||||
usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0);
|
||||
usage.cachedInputTokens += asNumber(cache.read, 0);
|
||||
usage.outputTokens += asNumber(tokens.output, 0);
|
||||
addCost(part.cost);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
sessionId,
|
||||
summary: messages.join("\n\n").trim(),
|
||||
usage,
|
||||
costUsd: reportedCost && !missingCost ? totalCostUsd : null,
|
||||
errorMessage,
|
||||
return {
|
||||
sawResult,
|
||||
sessionId,
|
||||
summary: messages.join("\n\n").trim(),
|
||||
usage: { ...usage },
|
||||
costUsd: reportedCost && !missingCost ? totalCostUsd : null,
|
||||
errorMessage,
|
||||
};
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -63,7 +63,7 @@ import {
|
||||
isGeminiTransientNetworkError,
|
||||
isGeminiTurnLimitResult,
|
||||
isGeminiSessionUnrecoverableError,
|
||||
parseGeminiJsonl,
|
||||
parseGeminiJsonl, createGeminiJsonlParser,
|
||||
} from "./parse.js";
|
||||
import { firstNonEmptyLine } from "./utils.js";
|
||||
import {
|
||||
@@ -643,6 +643,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
});
|
||||
}
|
||||
|
||||
const consumeAccounting = createGeminiJsonlParser();
|
||||
const accountingLog = createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
|
||||
const parsed = consumeAccounting(stdout);
|
||||
return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider: "google", biller: "google", billingType, model, complete: parsed.resultEvent !== null };
|
||||
});
|
||||
const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, {
|
||||
onProcessStopped: providerStop.beginInvocation(),
|
||||
cwd,
|
||||
@@ -651,13 +656,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog: createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
|
||||
const parsed = parseGeminiJsonl(stdout);
|
||||
return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider: "google", biller: "google", billingType, model, complete: parsed.resultEvent !== null };
|
||||
}),
|
||||
onLog: accountingLog,
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
});
|
||||
await accountingLog.flush();
|
||||
return {
|
||||
proc,
|
||||
parsed: parseGeminiJsonl(proc.stdout),
|
||||
@@ -691,6 +694,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: attempt.parsed.resultEvent !== null,
|
||||
usage: attempt.parsed.usage,
|
||||
usageBasis: "per_run",
|
||||
provider: "google",
|
||||
@@ -754,6 +758,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: false,
|
||||
usageComplete: attempt.parsed.resultEvent !== null,
|
||||
usageBasis: "per_run",
|
||||
errorMessage: failed ? fallbackErrorMessage : null,
|
||||
// Forward the transport-level error code from the run-disposition seam
|
||||
|
||||
@@ -74,6 +74,11 @@ function accumulateUsage(
|
||||
}
|
||||
|
||||
export function parseGeminiJsonl(stdout: string) {
|
||||
return createGeminiJsonlParser()(stdout);
|
||||
}
|
||||
|
||||
/** Consume complete JSONL records once, retaining protocol accounting state. */
|
||||
export function createGeminiJsonlParser() {
|
||||
let sessionId: string | null = null;
|
||||
const messages: string[] = [];
|
||||
let errorMessage: string | null = null;
|
||||
@@ -86,116 +91,118 @@ export function parseGeminiJsonl(stdout: string) {
|
||||
outputTokens: 0,
|
||||
};
|
||||
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = rawLine.trim();
|
||||
if (!line) continue;
|
||||
return (stdout: string) => {
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = rawLine.trim();
|
||||
if (!line) continue;
|
||||
|
||||
const event = parseJson(line);
|
||||
if (!event) continue;
|
||||
const event = parseJson(line);
|
||||
if (!event) continue;
|
||||
|
||||
const foundSessionId = readSessionId(event);
|
||||
if (foundSessionId) sessionId = foundSessionId;
|
||||
const foundSessionId = readSessionId(event);
|
||||
if (foundSessionId) sessionId = foundSessionId;
|
||||
|
||||
const type = asString(event.type, "").trim();
|
||||
const type = asString(event.type, "").trim();
|
||||
|
||||
if (type === "assistant") {
|
||||
messages.push(...collectMessageText(event.message));
|
||||
const messageObj = parseObject(event.message);
|
||||
const content = Array.isArray(messageObj.content) ? messageObj.content : [];
|
||||
for (const partRaw of content) {
|
||||
const part = parseObject(partRaw);
|
||||
if (asString(part.type, "").trim() === "question") {
|
||||
question = {
|
||||
prompt: asString(part.prompt, "").trim(),
|
||||
choices: (Array.isArray(part.choices) ? part.choices : []).map((choiceRaw) => {
|
||||
const choice = parseObject(choiceRaw);
|
||||
return {
|
||||
key: asString(choice.key, "").trim(),
|
||||
label: asString(choice.label, "").trim(),
|
||||
description: asString(choice.description, "").trim() || undefined,
|
||||
};
|
||||
}),
|
||||
};
|
||||
break; // only one question per message
|
||||
if (type === "assistant") {
|
||||
messages.push(...collectMessageText(event.message));
|
||||
const messageObj = parseObject(event.message);
|
||||
const content = Array.isArray(messageObj.content) ? messageObj.content : [];
|
||||
for (const partRaw of content) {
|
||||
const part = parseObject(partRaw);
|
||||
if (asString(part.type, "").trim() === "question") {
|
||||
question = {
|
||||
prompt: asString(part.prompt, "").trim(),
|
||||
choices: (Array.isArray(part.choices) ? part.choices : []).map((choiceRaw) => {
|
||||
const choice = parseObject(choiceRaw);
|
||||
return {
|
||||
key: asString(choice.key, "").trim(),
|
||||
label: asString(choice.label, "").trim(),
|
||||
description: asString(choice.description, "").trim() || undefined,
|
||||
};
|
||||
}),
|
||||
};
|
||||
break; // only one question per message
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
// Gemini CLI v0.38+ stream-json schema emits assistant turns as:
|
||||
// {"type":"message","role":"assistant","content":"...","delta":true}
|
||||
// These are discrete final messages (one per assistant turn), not
|
||||
// cumulative streaming tokens, so collecting all of them produces the
|
||||
// expected concatenated turn-by-turn summary rather than duplicated text.
|
||||
if (type === "message") {
|
||||
const role = asString(event.role, "").trim().toLowerCase();
|
||||
if (role === "assistant") {
|
||||
messages.push(...collectMessageText(event.content));
|
||||
// Gemini CLI v0.38+ stream-json schema emits assistant turns as:
|
||||
// {"type":"message","role":"assistant","content":"...","delta":true}
|
||||
// These are discrete final messages (one per assistant turn), not
|
||||
// cumulative streaming tokens, so collecting all of them produces the
|
||||
// expected concatenated turn-by-turn summary rather than duplicated text.
|
||||
if (type === "message") {
|
||||
const role = asString(event.role, "").trim().toLowerCase();
|
||||
if (role === "assistant") {
|
||||
messages.push(...collectMessageText(event.content));
|
||||
}
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "result") {
|
||||
resultEvent = event;
|
||||
accumulateUsage(usage, event.usage ?? event.usageMetadata ?? event.stats);
|
||||
const resultText =
|
||||
asString(event.result, "").trim() ||
|
||||
asString(event.text, "").trim() ||
|
||||
asString(event.response, "").trim();
|
||||
if (resultText && messages.length === 0) messages.push(resultText);
|
||||
const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0);
|
||||
if (typeof reportedCost === "number") costUsd = reportedCost;
|
||||
const status = asString(event.status, "").toLowerCase();
|
||||
const isError =
|
||||
event.is_error === true ||
|
||||
asString(event.subtype, "").toLowerCase() === "error" ||
|
||||
status === "error" ||
|
||||
status === "failed";
|
||||
if (isError) {
|
||||
const text = asErrorText(event.error ?? event.message ?? event.result).trim();
|
||||
if (text) errorMessage = text;
|
||||
if (type === "result") {
|
||||
resultEvent = event;
|
||||
accumulateUsage(usage, event.usage ?? event.usageMetadata ?? event.stats);
|
||||
const resultText =
|
||||
asString(event.result, "").trim() ||
|
||||
asString(event.text, "").trim() ||
|
||||
asString(event.response, "").trim();
|
||||
if (resultText && messages.length === 0) messages.push(resultText);
|
||||
const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0);
|
||||
if (typeof reportedCost === "number") costUsd = reportedCost;
|
||||
const status = asString(event.status, "").toLowerCase();
|
||||
const isError =
|
||||
event.is_error === true ||
|
||||
asString(event.subtype, "").toLowerCase() === "error" ||
|
||||
status === "error" ||
|
||||
status === "failed";
|
||||
if (isError) {
|
||||
const text = asErrorText(event.error ?? event.message ?? event.result).trim();
|
||||
if (text) errorMessage = text;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "error") {
|
||||
const text = asErrorText(event.error ?? event.message ?? event.detail).trim();
|
||||
if (text) errorMessage = text;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "system") {
|
||||
const subtype = asString(event.subtype, "").trim().toLowerCase();
|
||||
if (subtype === "error") {
|
||||
if (type === "error") {
|
||||
const text = asErrorText(event.error ?? event.message ?? event.detail).trim();
|
||||
if (text) errorMessage = text;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "system") {
|
||||
const subtype = asString(event.subtype, "").trim().toLowerCase();
|
||||
if (subtype === "error") {
|
||||
const text = asErrorText(event.error ?? event.message ?? event.detail).trim();
|
||||
if (text) errorMessage = text;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "text") {
|
||||
const part = parseObject(event.part);
|
||||
const text = asString(part.text, "").trim();
|
||||
if (text) messages.push(text);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "step_finish" || event.usage || event.usageMetadata) {
|
||||
accumulateUsage(usage, event.usage ?? event.usageMetadata);
|
||||
const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0);
|
||||
if (typeof reportedCost === "number") costUsd = reportedCost;
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "text") {
|
||||
const part = parseObject(event.part);
|
||||
const text = asString(part.text, "").trim();
|
||||
if (text) messages.push(text);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "step_finish" || event.usage || event.usageMetadata) {
|
||||
accumulateUsage(usage, event.usage ?? event.usageMetadata);
|
||||
const reportedCost = [event.total_cost_usd, event.cost_usd, event.cost].find((value) => typeof value === "number" && Number.isFinite(value) && value >= 0);
|
||||
if (typeof reportedCost === "number") costUsd = reportedCost;
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
sessionId,
|
||||
summary: messages.join("\n\n").trim(),
|
||||
usage,
|
||||
costUsd,
|
||||
errorMessage,
|
||||
resultEvent,
|
||||
question,
|
||||
return {
|
||||
sessionId,
|
||||
summary: messages.join("\n\n").trim(),
|
||||
usage: { ...usage },
|
||||
costUsd,
|
||||
errorMessage,
|
||||
resultEvent,
|
||||
question,
|
||||
};
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -648,6 +648,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: false,
|
||||
usageBasis: "per_run",
|
||||
provider: "moonshot",
|
||||
biller: "moonshot",
|
||||
|
||||
@@ -53,7 +53,7 @@ import {
|
||||
readPaperclipIssueWorkModeFromContext,
|
||||
resolveLegacyPaperclipDesiredSkillNames,
|
||||
} from "@paperclipai/adapter-utils/server-utils";
|
||||
import { isOpenCodeUnknownSessionError, parseOpenCodeJsonl } from "./parse.js";
|
||||
import { isOpenCodeUnknownSessionError, parseOpenCodeJsonl, createOpenCodeJsonlParser } from "./parse.js";
|
||||
import {
|
||||
ensureOpenCodeModelConfiguredAndAvailable,
|
||||
isTruthyEnvFlag,
|
||||
@@ -646,6 +646,12 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
});
|
||||
}
|
||||
|
||||
const consumeAccounting = createOpenCodeJsonlParser();
|
||||
const accountingLog = createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
|
||||
const parsed = consumeAccounting(stdout);
|
||||
const provider = parseModelProvider(model || null);
|
||||
return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveOpenCodeBiller(runtimeEnv, provider), billingType: "unknown", model, complete: false };
|
||||
});
|
||||
const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, {
|
||||
onProcessStopped: providerStop.beginInvocation(),
|
||||
cwd,
|
||||
@@ -655,14 +661,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog: createUsageCheckpointLog(onLog, ctx.onUsage, stdout => {
|
||||
const parsed = parseOpenCodeJsonl(stdout);
|
||||
const provider = parseModelProvider(model || null);
|
||||
return { usage: parsed.usage, costUsd: parsed.costUsd, usageBasis: "per_run", provider, biller: resolveOpenCodeBiller(runtimeEnv, provider), billingType: "unknown", model, complete: false };
|
||||
}),
|
||||
onLog: accountingLog,
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
});
|
||||
await accountingLog.flush({ complete: proc.exitCode === 0 && !proc.timedOut && !proc.signal });
|
||||
return {
|
||||
proc,
|
||||
rawStderr: proc.stderr,
|
||||
@@ -683,6 +686,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: false,
|
||||
usage: attempt.parsed.usage,
|
||||
usageBasis: "per_run",
|
||||
provider: parseModelProvider(model || null),
|
||||
@@ -727,6 +731,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: synthesizedExitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: false,
|
||||
usageComplete: attempt.proc.exitCode === 0 && !attempt.proc.signal,
|
||||
usageBasis: "per_run",
|
||||
errorMessage: (synthesizedExitCode ?? 0) === 0 ? null : fallbackErrorMessage,
|
||||
// Forward the transport-level error code from the run-disposition seam.
|
||||
|
||||
@@ -20,6 +20,11 @@ function errorText(value: unknown): string {
|
||||
}
|
||||
|
||||
export function parseOpenCodeJsonl(stdout: string) {
|
||||
return createOpenCodeJsonlParser()(stdout);
|
||||
}
|
||||
|
||||
/** Consume complete JSONL records once, retaining protocol accounting state. */
|
||||
export function createOpenCodeJsonlParser() {
|
||||
let sessionId: string | null = null;
|
||||
const messages: string[] = [];
|
||||
const errors: string[] = [];
|
||||
@@ -32,61 +37,63 @@ export function parseOpenCodeJsonl(stdout: string) {
|
||||
let costUsd: number | null = null;
|
||||
let missingCost = false;
|
||||
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = rawLine.trim();
|
||||
if (!line) continue;
|
||||
return (stdout: string) => {
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = rawLine.trim();
|
||||
if (!line) continue;
|
||||
|
||||
const event = parseJson(line);
|
||||
if (!event) continue;
|
||||
const event = parseJson(line);
|
||||
if (!event) continue;
|
||||
|
||||
const currentSessionId = asString(event.sessionID, "").trim();
|
||||
if (currentSessionId) sessionId = currentSessionId;
|
||||
const currentSessionId = asString(event.sessionID, "").trim();
|
||||
if (currentSessionId) sessionId = currentSessionId;
|
||||
|
||||
const type = asString(event.type, "");
|
||||
const type = asString(event.type, "");
|
||||
|
||||
if (type === "text") {
|
||||
const part = parseObject(event.part);
|
||||
const text = asString(part.text, "").trim();
|
||||
if (text) messages.push(text);
|
||||
continue;
|
||||
}
|
||||
if (type === "text") {
|
||||
const part = parseObject(event.part);
|
||||
const text = asString(part.text, "").trim();
|
||||
if (text) messages.push(text);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "step_finish") {
|
||||
const part = parseObject(event.part);
|
||||
const tokens = parseObject(part.tokens);
|
||||
const cache = parseObject(tokens.cache);
|
||||
usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0);
|
||||
usage.cachedInputTokens += asNumber(cache.read, 0);
|
||||
usage.outputTokens += asNumber(tokens.output, 0) + asNumber(tokens.reasoning, 0);
|
||||
if (typeof part.cost === "number" && Number.isFinite(part.cost) && part.cost >= 0) costUsd = (costUsd ?? 0) + part.cost;
|
||||
else missingCost = true;
|
||||
continue;
|
||||
}
|
||||
if (type === "step_finish") {
|
||||
const part = parseObject(event.part);
|
||||
const tokens = parseObject(part.tokens);
|
||||
const cache = parseObject(tokens.cache);
|
||||
usage.inputTokens += asNumber(tokens.input, 0) + asNumber(cache.write, 0);
|
||||
usage.cachedInputTokens += asNumber(cache.read, 0);
|
||||
usage.outputTokens += asNumber(tokens.output, 0) + asNumber(tokens.reasoning, 0);
|
||||
if (typeof part.cost === "number" && Number.isFinite(part.cost) && part.cost >= 0) costUsd = (costUsd ?? 0) + part.cost;
|
||||
else missingCost = true;
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "tool_use") {
|
||||
const part = parseObject(event.part);
|
||||
const state = parseObject(part.state);
|
||||
if (asString(state.status, "") === "error") {
|
||||
const text = asString(state.error, "").trim();
|
||||
if (text) toolErrors.push(text);
|
||||
if (type === "tool_use") {
|
||||
const part = parseObject(event.part);
|
||||
const state = parseObject(part.state);
|
||||
if (asString(state.status, "") === "error") {
|
||||
const text = asString(state.error, "").trim();
|
||||
if (text) toolErrors.push(text);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "error") {
|
||||
const text = errorText(event.error ?? event.message).trim();
|
||||
if (text) errors.push(text);
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (type === "error") {
|
||||
const text = errorText(event.error ?? event.message).trim();
|
||||
if (text) errors.push(text);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
sessionId,
|
||||
summary: messages.join("\n\n").trim(),
|
||||
usage,
|
||||
costUsd: missingCost ? null : costUsd,
|
||||
errorMessage: errors.length > 0 ? errors.join("\n") : null,
|
||||
toolErrors,
|
||||
return {
|
||||
sessionId,
|
||||
summary: messages.join("\n\n").trim(),
|
||||
usage: { ...usage },
|
||||
costUsd: missingCost ? null : costUsd,
|
||||
errorMessage: errors.length > 0 ? errors.join("\n") : null,
|
||||
toolErrors,
|
||||
};
|
||||
};
|
||||
}
|
||||
|
||||
|
||||
@@ -55,7 +55,7 @@ import {
|
||||
runChildProcess,
|
||||
} from "@paperclipai/adapter-utils/server-utils";
|
||||
import { shellQuote } from "@paperclipai/adapter-utils/ssh";
|
||||
import { isPiUnknownSessionError, parsePiJsonl } from "./parse.js";
|
||||
import { isPiUnknownSessionError, parsePiJsonl, createPiJsonlParser } from "./parse.js";
|
||||
import { ensurePiModelConfiguredAndAvailable } from "./models.js";
|
||||
import { preparePiRuntimeConfig } from "./runtime-config.js";
|
||||
import { SANDBOX_INSTALL_COMMAND } from "../index.js";
|
||||
@@ -737,6 +737,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
}
|
||||
};
|
||||
|
||||
const consumeAccounting = createPiJsonlParser();
|
||||
const accountingLog = createUsageCheckpointLog(bufferedOnLog, ctx.onUsage, stdout => {
|
||||
const parsed = consumeAccounting(stdout);
|
||||
return { usage: parsed.usage, costUsd: parsed.usage.costUsd, usageBasis: "per_run", provider, biller: resolvePiBiller(runtimeEnv, provider), billingType: "unknown", model, complete: parsed.sawAgentEnd };
|
||||
});
|
||||
const proc = await runAdapterExecutionTargetProcess(runId, runtimeExecutionTarget, command, args, {
|
||||
onProcessStopped: providerStop.beginInvocation(),
|
||||
cwd,
|
||||
@@ -745,13 +750,11 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
graceSec,
|
||||
onSpawn,
|
||||
onRuntimeProgress: ctx.onRuntimeProgress,
|
||||
onLog: createUsageCheckpointLog(bufferedOnLog, ctx.onUsage, stdout => {
|
||||
const parsed = parsePiJsonl(stdout);
|
||||
return { usage: parsed.usage, costUsd: parsed.usage.costUsd, usageBasis: "per_run", provider, biller: resolvePiBiller(runtimeEnv, provider), billingType: "unknown", model, complete: false };
|
||||
}),
|
||||
onLog: accountingLog,
|
||||
runLogTail: paperclipBridge?.runLogTail,
|
||||
settleRunDisposition: paperclipBridge?.settleRunDisposition,
|
||||
});
|
||||
await accountingLog.flush();
|
||||
|
||||
// Flush any remaining buffer content
|
||||
if (stdoutBuffer) {
|
||||
@@ -778,6 +781,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: attempt.proc.exitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: true,
|
||||
usageComplete: attempt.parsed.sawAgentEnd,
|
||||
usage: attempt.parsed.usage, usageBasis: "per_run", provider, biller: resolvePiBiller(runtimeEnv, provider), model, billingType: "unknown", costUsd: attempt.parsed.usage.costUsd,
|
||||
errorMessage: `Timed out after ${timeoutSec}s`,
|
||||
clearSession: clearSessionOnMissingSession,
|
||||
@@ -810,6 +814,7 @@ export async function execute(ctx: AdapterExecutionContext): Promise<AdapterExec
|
||||
exitCode: effectiveExitCode,
|
||||
signal: attempt.proc.signal,
|
||||
timedOut: false,
|
||||
usageComplete: attempt.parsed.sawAgentEnd,
|
||||
usageBasis: "per_run",
|
||||
errorMessage: (effectiveExitCode ?? 0) === 0 ? null : fallbackErrorMessage,
|
||||
// Forward the transport-level error code from the run-disposition seam.
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { asNumber, asString, parseJson, parseObject } from "@paperclipai/adapter-utils/server-utils";
|
||||
|
||||
interface ParsedPiOutput {
|
||||
sawAgentEnd: boolean;
|
||||
sessionId: string | null;
|
||||
messages: string[];
|
||||
errors: string[];
|
||||
@@ -28,8 +29,14 @@ function extractTextContent(content: string | Array<{ type: string; text?: strin
|
||||
.join("");
|
||||
}
|
||||
|
||||
export function parsePiJsonl(stdout: string): ParsedPiOutput {
|
||||
export function parsePiJsonl(stdout: string) {
|
||||
return createPiJsonlParser()(stdout);
|
||||
}
|
||||
|
||||
/** Consume complete JSONL records once, retaining protocol accounting state. */
|
||||
export function createPiJsonlParser() {
|
||||
const result: ParsedPiOutput = {
|
||||
sawAgentEnd: false,
|
||||
sessionId: null,
|
||||
messages: [],
|
||||
errors: [],
|
||||
@@ -50,185 +57,188 @@ export function parsePiJsonl(stdout: string): ParsedPiOutput {
|
||||
}
|
||||
let currentToolCall: { toolCallId: string; toolName: string; args: unknown } | null = null;
|
||||
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = rawLine.trim();
|
||||
if (!line) continue;
|
||||
return (stdout: string) => {
|
||||
for (const rawLine of stdout.split(/\r?\n/)) {
|
||||
const line = rawLine.trim();
|
||||
if (!line) continue;
|
||||
|
||||
const event = parseJson(line);
|
||||
if (!event) continue;
|
||||
const event = parseJson(line);
|
||||
if (!event) continue;
|
||||
|
||||
const eventType = asString(event.type, "");
|
||||
const eventType = asString(event.type, "");
|
||||
|
||||
// Pi can exit successfully after a provider failure. The terminal assistant
|
||||
// message carries that failure in both message_end and turn_end envelopes.
|
||||
const terminalMessages = eventType === "agent_end"
|
||||
? (Array.isArray(event.messages) ? event.messages : [])
|
||||
: eventType === "message_end" || eventType === "turn_end"
|
||||
? [event.message]
|
||||
: [];
|
||||
for (const rawMessage of terminalMessages) {
|
||||
const message = asRecord(rawMessage);
|
||||
if (message?.role !== "assistant" || message.stopReason !== "error") continue;
|
||||
const error = asString(message.errorMessage, "").trim() || "Pi provider request failed.";
|
||||
if (!result.errors.includes(error)) result.errors.push(error);
|
||||
}
|
||||
|
||||
// RPC protocol messages - skip these (internal implementation detail)
|
||||
if (eventType === "response" || eventType === "extension_ui_request" || eventType === "extension_ui_response" || eventType === "extension_error") {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Agent lifecycle
|
||||
if (eventType === "agent_start") {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (eventType === "agent_end") {
|
||||
const messages = event.messages as Array<Record<string, unknown>> | undefined;
|
||||
if (messages && messages.length > 0) {
|
||||
const lastMessage = messages[messages.length - 1];
|
||||
if (lastMessage?.role === "assistant") {
|
||||
const content = lastMessage.content as string | Array<{ type: string; text?: string }>;
|
||||
result.finalMessage = extractTextContent(content);
|
||||
}
|
||||
// Pi can exit successfully after a provider failure. The terminal assistant
|
||||
// message carries that failure in both message_end and turn_end envelopes.
|
||||
const terminalMessages = eventType === "agent_end"
|
||||
? (Array.isArray(event.messages) ? event.messages : [])
|
||||
: eventType === "message_end" || eventType === "turn_end"
|
||||
? [event.message]
|
||||
: [];
|
||||
for (const rawMessage of terminalMessages) {
|
||||
const message = asRecord(rawMessage);
|
||||
if (message?.role !== "assistant" || message.stopReason !== "error") continue;
|
||||
const error = asString(message.errorMessage, "").trim() || "Pi provider request failed.";
|
||||
if (!result.errors.includes(error)) result.errors.push(error);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (eventType === "auto_retry_end") {
|
||||
const succeeded = event.success === true;
|
||||
if (!succeeded) {
|
||||
const finalError = asString(event.finalError, "").trim();
|
||||
result.errors.push(finalError || "Pi exhausted automatic retries without producing a response.");
|
||||
// RPC protocol messages - skip these (internal implementation detail)
|
||||
if (eventType === "response" || eventType === "extension_ui_request" || eventType === "extension_ui_response" || eventType === "extension_error") {
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
// Turn lifecycle
|
||||
if (eventType === "turn_start") {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (eventType === "turn_end") {
|
||||
const message = asRecord(event.message);
|
||||
if (message) {
|
||||
const content = message.content as string | Array<{ type: string; text?: string }>;
|
||||
const text = extractTextContent(content);
|
||||
if (text) {
|
||||
result.finalMessage = text;
|
||||
result.messages.push(text);
|
||||
}
|
||||
|
||||
// Extract usage and cost from assistant message
|
||||
const usage = asRecord(message.usage);
|
||||
if (usage) {
|
||||
result.usage.inputTokens += asNumber(usage.input, 0) + asNumber(usage.cacheWrite, 0);
|
||||
result.usage.outputTokens += asNumber(usage.output, 0);
|
||||
result.usage.cachedInputTokens += asNumber(usage.cacheRead, 0);
|
||||
|
||||
// Pi stores cost in usage.cost.total (and broken down in usage.cost.input, etc.)
|
||||
const cost = asRecord(usage.cost);
|
||||
addCost(cost?.total);
|
||||
}
|
||||
// Agent lifecycle
|
||||
if (eventType === "agent_start") {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Tool results are in toolResults array
|
||||
const toolResults = event.toolResults as Array<Record<string, unknown>> | undefined;
|
||||
if (toolResults) {
|
||||
for (const tr of toolResults) {
|
||||
const toolCallId = asString(tr.toolCallId, "");
|
||||
const content = tr.content;
|
||||
const isError = tr.isError === true;
|
||||
|
||||
// Find matching tool call by toolCallId
|
||||
const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId);
|
||||
if (existingCall) {
|
||||
existingCall.result = typeof content === "string" ? content : JSON.stringify(content);
|
||||
existingCall.isError = isError;
|
||||
|
||||
if (eventType === "agent_end") {
|
||||
result.sawAgentEnd = true;
|
||||
const messages = event.messages as Array<Record<string, unknown>> | undefined;
|
||||
if (messages && messages.length > 0) {
|
||||
const lastMessage = messages[messages.length - 1];
|
||||
if (lastMessage?.role === "assistant") {
|
||||
const content = lastMessage.content as string | Array<{ type: string; text?: string }>;
|
||||
result.finalMessage = extractTextContent(content);
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
// Message updates (streaming)
|
||||
if (eventType === "message_update") {
|
||||
const assistantEvent = asRecord(event.assistantMessageEvent);
|
||||
if (assistantEvent) {
|
||||
const msgType = asString(assistantEvent.type, "");
|
||||
if (msgType === "text_delta") {
|
||||
const delta = asString(assistantEvent.delta, "");
|
||||
if (delta) {
|
||||
// Append to last message or create new
|
||||
if (result.messages.length === 0) {
|
||||
result.messages.push(delta);
|
||||
} else {
|
||||
result.messages[result.messages.length - 1] += delta;
|
||||
if (eventType === "auto_retry_end") {
|
||||
const succeeded = event.success === true;
|
||||
if (!succeeded) {
|
||||
const finalError = asString(event.finalError, "").trim();
|
||||
result.errors.push(finalError || "Pi exhausted automatic retries without producing a response.");
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
// Turn lifecycle
|
||||
if (eventType === "turn_start") {
|
||||
continue;
|
||||
}
|
||||
|
||||
if (eventType === "turn_end") {
|
||||
const message = asRecord(event.message);
|
||||
if (message) {
|
||||
const content = message.content as string | Array<{ type: string; text?: string }>;
|
||||
const text = extractTextContent(content);
|
||||
if (text) {
|
||||
result.finalMessage = text;
|
||||
result.messages.push(text);
|
||||
}
|
||||
|
||||
// Extract usage and cost from assistant message
|
||||
const usage = asRecord(message.usage);
|
||||
if (usage) {
|
||||
result.usage.inputTokens += asNumber(usage.input, 0) + asNumber(usage.cacheWrite, 0);
|
||||
result.usage.outputTokens += asNumber(usage.output, 0);
|
||||
result.usage.cachedInputTokens += asNumber(usage.cacheRead, 0);
|
||||
|
||||
// Pi stores cost in usage.cost.total (and broken down in usage.cost.input, etc.)
|
||||
const cost = asRecord(usage.cost);
|
||||
addCost(cost?.total);
|
||||
}
|
||||
}
|
||||
|
||||
// Tool results are in toolResults array
|
||||
const toolResults = event.toolResults as Array<Record<string, unknown>> | undefined;
|
||||
if (toolResults) {
|
||||
for (const tr of toolResults) {
|
||||
const toolCallId = asString(tr.toolCallId, "");
|
||||
const content = tr.content;
|
||||
const isError = tr.isError === true;
|
||||
|
||||
// Find matching tool call by toolCallId
|
||||
const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId);
|
||||
if (existingCall) {
|
||||
existingCall.result = typeof content === "string" ? content : JSON.stringify(content);
|
||||
existingCall.isError = isError;
|
||||
}
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
if (eventType === "error") {
|
||||
const message = asString(event.message, "").trim();
|
||||
if (message) {
|
||||
result.errors.push(message);
|
||||
// Message updates (streaming)
|
||||
if (eventType === "message_update") {
|
||||
const assistantEvent = asRecord(event.assistantMessageEvent);
|
||||
if (assistantEvent) {
|
||||
const msgType = asString(assistantEvent.type, "");
|
||||
if (msgType === "text_delta") {
|
||||
const delta = asString(assistantEvent.delta, "");
|
||||
if (delta) {
|
||||
// Append to last message or create new
|
||||
if (result.messages.length === 0) {
|
||||
result.messages.push(delta);
|
||||
} else {
|
||||
result.messages[result.messages.length - 1] += delta;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
continue;
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
// Tool execution
|
||||
if (eventType === "tool_execution_start") {
|
||||
const toolCallId = asString(event.toolCallId, "");
|
||||
const toolName = asString(event.toolName, "");
|
||||
const args = event.args;
|
||||
currentToolCall = { toolCallId, toolName, args };
|
||||
result.toolCalls.push({
|
||||
toolCallId,
|
||||
toolName,
|
||||
args,
|
||||
result: null,
|
||||
isError: false,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
if (eventType === "tool_execution_end") {
|
||||
const toolCallId = asString(event.toolCallId, "");
|
||||
const toolName = asString(event.toolName, "");
|
||||
const toolResult = event.result;
|
||||
const isError = event.isError === true;
|
||||
|
||||
// Find the tool call by toolCallId (not toolName, to handle multiple calls to same tool)
|
||||
const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId);
|
||||
if (existingCall) {
|
||||
existingCall.result = typeof toolResult === "string" ? toolResult : JSON.stringify(toolResult);
|
||||
existingCall.isError = isError;
|
||||
if (eventType === "error") {
|
||||
const message = asString(event.message, "").trim();
|
||||
if (message) {
|
||||
result.errors.push(message);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
currentToolCall = null;
|
||||
continue;
|
||||
}
|
||||
|
||||
// Usage tracking if available in the event (fallback for standalone usage events)
|
||||
if (eventType === "usage" || event.usage) {
|
||||
const usage = asRecord(event.usage);
|
||||
if (usage) {
|
||||
// Support both Pi format (input/output/cacheRead) and generic format (inputTokens/outputTokens/cachedInputTokens)
|
||||
result.usage.inputTokens += asNumber(usage.inputTokens ?? usage.input, 0) + asNumber(usage.cacheWrite, 0);
|
||||
result.usage.outputTokens += asNumber(usage.outputTokens ?? usage.output, 0);
|
||||
result.usage.cachedInputTokens += asNumber(usage.cachedInputTokens ?? usage.cacheRead, 0);
|
||||
|
||||
// Cost may be in usage.costUsd (direct) or usage.cost.total (Pi format)
|
||||
const cost = asRecord(usage.cost);
|
||||
addCost(cost?.total ?? usage.costUsd);
|
||||
// Tool execution
|
||||
if (eventType === "tool_execution_start") {
|
||||
const toolCallId = asString(event.toolCallId, "");
|
||||
const toolName = asString(event.toolName, "");
|
||||
const args = event.args;
|
||||
currentToolCall = { toolCallId, toolName, args };
|
||||
result.toolCalls.push({
|
||||
toolCallId,
|
||||
toolName,
|
||||
args,
|
||||
result: null,
|
||||
isError: false,
|
||||
});
|
||||
continue;
|
||||
}
|
||||
|
||||
if (eventType === "tool_execution_end") {
|
||||
const toolCallId = asString(event.toolCallId, "");
|
||||
const toolName = asString(event.toolName, "");
|
||||
const toolResult = event.result;
|
||||
const isError = event.isError === true;
|
||||
|
||||
// Find the tool call by toolCallId (not toolName, to handle multiple calls to same tool)
|
||||
const existingCall = result.toolCalls.find((tc) => tc.toolCallId === toolCallId);
|
||||
if (existingCall) {
|
||||
existingCall.result = typeof toolResult === "string" ? toolResult : JSON.stringify(toolResult);
|
||||
existingCall.isError = isError;
|
||||
}
|
||||
currentToolCall = null;
|
||||
continue;
|
||||
}
|
||||
|
||||
// Usage tracking if available in the event (fallback for standalone usage events)
|
||||
if (eventType === "usage" || event.usage) {
|
||||
const usage = asRecord(event.usage);
|
||||
if (usage) {
|
||||
// Support both Pi format (input/output/cacheRead) and generic format (inputTokens/outputTokens/cachedInputTokens)
|
||||
result.usage.inputTokens += asNumber(usage.inputTokens ?? usage.input, 0) + asNumber(usage.cacheWrite, 0);
|
||||
result.usage.outputTokens += asNumber(usage.outputTokens ?? usage.output, 0);
|
||||
result.usage.cachedInputTokens += asNumber(usage.cachedInputTokens ?? usage.cacheRead, 0);
|
||||
|
||||
// Cost may be in usage.costUsd (direct) or usage.cost.total (Pi format)
|
||||
const cost = asRecord(usage.cost);
|
||||
addCost(cost?.total ?? usage.costUsd);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (missingCost) result.usage.costUsd = null;
|
||||
return result;
|
||||
if (missingCost) result.usage.costUsd = null;
|
||||
return { ...result, usage: { ...result.usage }, messages: [...result.messages], errors: [...result.errors], toolCalls: [...result.toolCalls] };
|
||||
};
|
||||
}
|
||||
|
||||
export function isPiUnknownSessionError(stdout: string, stderr: string): boolean {
|
||||
|
||||
@@ -31,6 +31,61 @@ const cases: Array<{ name: string; execute: () => Promise<Execute>; event: unkno
|
||||
describe("CLI adapter accounting on timeout", () => {
|
||||
const directories: string[] = [];
|
||||
afterEach(async () => { vi.clearAllMocks(); for (const dir of directories.splice(0)) await rm(dir, { recursive: true, force: true }); });
|
||||
it.each(cases.filter(fixture => fixture.name !== "kimi"))("$name flushes its attempt and surfaces checkpoint persistence failures", async (fixture) => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paperclip-accounting-adapter-")); directories.push(dir);
|
||||
const command = join(dir, "runtime"); await writeFile(command, "#!/bin/sh\nprintf 'openai test\\n'\n", { mode: 0o755 });
|
||||
const stdout = JSON.stringify(fixture.event);
|
||||
processResult.mockImplementation(async (_run, _target, _command, _args, options) => {
|
||||
// Local process logging catches callback failures. Finalization must
|
||||
// independently enforce receipt durability despite that contract.
|
||||
await options.onLog("stdout", stdout).catch(() => {});
|
||||
return { exitCode: 0, signal: null, timedOut: false, stdout, stderr: "", pid: 123, startedAt: new Date().toISOString() };
|
||||
});
|
||||
const onUsage = vi.fn();
|
||||
const context: AdapterExecutionContext = {
|
||||
runId: "test-run", agent: { id: "test-agent", companyId: "test-company", name: "Accounting", adapterType: `${fixture.name}_local`, adapterConfig: {} },
|
||||
runtime: { sessionId: null, sessionParams: null, sessionDisplayId: null, taskKey: null },
|
||||
config: { engine: "cli", command, cwd: dir, model: ["opencode", "pi"].includes(fixture.name) ? "openai/test" : fixture.name === "cursor" ? "gpt-5" : "test", paperclipRuntimeSkills: [],
|
||||
env: { OPENAI_API_KEY: "test-placeholder", ANTHROPIC_API_KEY: "test-placeholder", GEMINI_API_KEY: "test-placeholder", OPENCODE_ALLOW_ALL_MODELS: "1", CLAUDE_CONFIG_DIR: dir } },
|
||||
context: {}, onLog: async () => {}, onUsage,
|
||||
};
|
||||
const execute = await fixture.execute();
|
||||
const initial = await execute(context);
|
||||
const complete = !["claude", "pi"].includes(fixture.name);
|
||||
expect(initial.usageComplete).toBe(complete);
|
||||
expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ usage: fixture.tokens, costUsd: fixture.price, complete });
|
||||
if (["claude", "pi", "opencode"].includes(fixture.name)) {
|
||||
processResult.mockImplementationOnce(async (_run, _target, _command, _args, options) => {
|
||||
await options.onLog("stdout", stdout).catch(() => {});
|
||||
return { exitCode: 1, signal: null, timedOut: false, stdout, stderr: "process failed", pid: 123, startedAt: new Date().toISOString() };
|
||||
});
|
||||
expect((await execute(context)).usageComplete).toBe(false);
|
||||
expect(onUsage.mock.calls.at(-1)![0].complete).toBe(false);
|
||||
}
|
||||
if (fixture.name === "claude") {
|
||||
const mixed = [
|
||||
{ type: "system", subtype: "init", model: "actual-model" },
|
||||
{ type: "result", total_cost_usd: 0.003, usage: { input_tokens: 3, output_tokens: 3 }, modelUsage: {
|
||||
"model-a": { inputTokens: 1, outputTokens: 1, costUSD: 0.001 },
|
||||
"model-b": { inputTokens: 2, outputTokens: 2, costUSD: 0.002 },
|
||||
} },
|
||||
];
|
||||
processResult.mockImplementation(async (_run, _target, _command, _args, options) => {
|
||||
const output = mixed.map(item => JSON.stringify(item)).join("\n");
|
||||
await options.onLog("stdout", output).catch(() => {});
|
||||
return { exitCode: 0, signal: null, timedOut: false, stdout: output, stderr: "", pid: 123, startedAt: new Date().toISOString() };
|
||||
});
|
||||
await execute(context);
|
||||
expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ model: "mixed", costUsd: 0.003, complete: true, usageByModel: [
|
||||
{ model: "model-a", costUsd: 0.001 }, { model: "model-b", costUsd: 0.002 },
|
||||
] });
|
||||
delete (mixed[1] as { modelUsage?: unknown }).modelUsage;
|
||||
await execute(context);
|
||||
expect(onUsage.mock.calls.at(-1)![0]).toMatchObject({ model: "actual-model", complete: true });
|
||||
}
|
||||
onUsage.mockRejectedValue(new Error("Receipt persistence failed"));
|
||||
await expect(execute(context)).rejects.toThrow("Receipt persistence failed");
|
||||
});
|
||||
it.each(cases)("$name retains observed accounting when the process times out", async (fixture) => {
|
||||
const dir = await mkdtemp(join(tmpdir(), "paperclip-accounting-adapter-")); directories.push(dir);
|
||||
const command = join(dir, "runtime"); await writeFile(command, "#!/bin/sh\nprintf 'openai test\\n'\n", { mode: 0o755 });
|
||||
@@ -44,6 +99,7 @@ describe("CLI adapter accounting on timeout", () => {
|
||||
context: {}, onLog: async () => {},
|
||||
});
|
||||
expect(result.timedOut).toBe(true);
|
||||
expect(result.usageComplete).toBe(["codex", "cursor", "gemini"].includes(fixture.name));
|
||||
expect(result.usageBasis).toBe("per_run");
|
||||
expect(result.provider).toBeTruthy();
|
||||
expect(result.billingType).toBeTruthy();
|
||||
@@ -51,3 +107,36 @@ describe("CLI adapter accounting on timeout", () => {
|
||||
expect(result.costUsd).toBe(fixture.price);
|
||||
});
|
||||
});
|
||||
|
||||
|
||||
describe("incremental protocol accounting", () => {
|
||||
it("keeps Claude message updates idempotent and preserves immutable snapshots", async () => {
|
||||
const { createClaudeStreamParser } = await import("../../../packages/adapters/claude-local/src/server/parse.js");
|
||||
const consume = createClaudeStreamParser();
|
||||
consume(JSON.stringify({ type: "system", subtype: "init", model: "actual-model" }));
|
||||
const message = (id: string, output: number) => JSON.stringify({ type: "assistant", message: { id, usage: { input_tokens: 2, output_tokens: output } } });
|
||||
const first = consume(message("one", 1));
|
||||
expect(first.usage).toMatchObject({ inputTokens: 2, outputTokens: 1 });
|
||||
expect(consume(message("one", 3)).usage).toMatchObject({ inputTokens: 2, outputTokens: 3 });
|
||||
expect(consume(message("two", 2)).usage).toMatchObject({ inputTokens: 4, outputTokens: 5 });
|
||||
expect(first.usage).toMatchObject({ inputTokens: 2, outputTokens: 1 });
|
||||
const final = consume(JSON.stringify({ type: "result", total_cost_usd: 1, usage: { input_tokens: 9, output_tokens: 8 } }));
|
||||
expect(final).toMatchObject({ model: "actual-model", costUsd: 1, usage: { inputTokens: 9, outputTokens: 8 } });
|
||||
});
|
||||
it.each(cases.filter(item => !["claude", "kimi"].includes(item.name)))("updates $name totals once per newly received record", async fixture => {
|
||||
const factories = {
|
||||
codex: async () => (await import("../../../packages/adapters/codex-local/src/server/parse.js")).createCodexJsonlParser(),
|
||||
cursor: async () => (await import("../../../packages/adapters/cursor-local/src/server/parse.js")).createCursorJsonlParser(),
|
||||
gemini: async () => (await import("../../../packages/adapters/gemini-local/src/server/parse.js")).createGeminiJsonlParser(),
|
||||
pi: async () => (await import("../../../packages/adapters/pi-local/src/server/parse.js")).createPiJsonlParser(),
|
||||
opencode: async () => (await import("../../../packages/adapters/opencode-local/src/server/parse.js")).createOpenCodeJsonlParser(),
|
||||
};
|
||||
const consume = await factories[fixture.name as keyof typeof factories]();
|
||||
const line = JSON.stringify(fixture.event);
|
||||
const first = consume(line);
|
||||
const second = consume(line);
|
||||
expect(first.usage).toMatchObject(fixture.tokens!);
|
||||
const multiplier = fixture.name === "codex" ? 1 : 2;
|
||||
expect(second.usage).toMatchObject(Object.fromEntries(Object.entries(fixture.tokens!).map(([key, count]) => [key, count * multiplier])));
|
||||
});
|
||||
});
|
||||
@@ -8,7 +8,7 @@ import { mkdtemp, rm, access, readFile, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { and, eq, sql } from "drizzle-orm";
|
||||
import { createDb, companies, agents, heartbeatRuns, companyMemberships, connectionGrants, connectionGrantDelegations, connectionGrantMembers, toolConnections, toolConnectionInstalls, aiConnectionDefaults, aiProviderDefaults, adapterAuthSessions, environments, issues, issueThreadInteractions, issueRecoveryActions, connectionIntentDeliveries, agentWakeupRequests, companySecrets, companySecretVersions, principalPermissionGrants } from "@paperclipai/db";
|
||||
import { createDb, companies, agents, heartbeatRuns, companyMemberships, connectionGrants, connectionGrantDelegations, connectionGrantMembers, toolConnections, toolConnectionInstalls, aiConnectionDefaults, aiProviderDefaults, adapterAuthSessions, environments, issues, issueThreadInteractions, issueRecoveryActions, connectionIntentDeliveries, agentWakeupRequests, companySecrets, companySecretVersions, userSecretDefinitions, principalPermissionGrants } from "@paperclipai/db";
|
||||
import { startEmbeddedPostgresTestDatabase } from "@paperclipai/db/test-embedded-postgres";
|
||||
import { fetchCompanyQuotaWindows } from "../services/quota-windows.js";
|
||||
import { aiConnectionService } from "../services/ai-connections.js";
|
||||
@@ -69,6 +69,13 @@ describe("managed AI connections", () => {
|
||||
expect(rotated[0].accountKey).not.toBe(first[0].accountKey);
|
||||
expect(fetchSpy).toHaveBeenCalledTimes(2);
|
||||
expect(fetchSpy).toHaveBeenLastCalledWith(expect.any(String), expect.objectContaining({ headers: expect.objectContaining({ Authorization: "Bearer quota-cache-rotated" }) }));
|
||||
await db.update(userSecretDefinitions).set({ status: "disabled" }).where(eq(userSecretDefinitions.id, before.userSecretDefinitionId!));
|
||||
const disabled = await fetchCompanyQuotaWindows(db, companyId, owner);
|
||||
expect(disabled[0]).toMatchObject({ ok: false, errorFamily: "credentials_unavailable", windows: [] });
|
||||
expect(disabled[0].accountKey).not.toBe(rotated[0].accountKey);
|
||||
expect(fetchSpy).toHaveBeenCalledTimes(2);
|
||||
await db.update(userSecretDefinitions).set({ status: "active" }).where(eq(userSecretDefinitions.id, before.userSecretDefinitionId!));
|
||||
expect((await fetchCompanyQuotaWindows(db, companyId, owner))[0].ok).toBe(true);
|
||||
await db.update(companySecretVersions).set({ status: "disabled", revokedAt: new Date() }).where(and(eq(companySecretVersions.secretId, secretId), eq(companySecretVersions.version, 2)));
|
||||
const revoked = await fetchCompanyQuotaWindows(db, companyId, owner);
|
||||
expect(revoked[0].ok).toBe(false);
|
||||
|
||||
@@ -28,6 +28,7 @@ let revision = 1;
|
||||
const db = {
|
||||
select: () => ({
|
||||
from: () => ({
|
||||
leftJoin: () => ({
|
||||
leftJoin: () => ({
|
||||
where: async () => [
|
||||
{
|
||||
@@ -38,6 +39,7 @@ const db = {
|
||||
},
|
||||
],
|
||||
}),
|
||||
}),
|
||||
}),
|
||||
}),
|
||||
} as unknown as Db;
|
||||
|
||||
@@ -4,7 +4,7 @@ import { listServerAdapters } from "../adapters/registry.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import { createHash } from "node:crypto";
|
||||
import { eq, inArray, and } from "drizzle-orm";
|
||||
import { companySecrets, companySecretVersions, type Db } from "@paperclipai/db";
|
||||
import { companySecrets, companySecretVersions, userSecretDefinitions, type Db } from "@paperclipai/db";
|
||||
import { fetchCodexQuota } from "@paperclipai/adapter-codex-local/server";
|
||||
import { fetchClaudeQuota } from "@paperclipai/adapter-claude-local/server";
|
||||
import { aiConnectionService } from "./ai-connections.js";
|
||||
@@ -25,15 +25,19 @@ export async function fetchCompanyQuotaWindows(db: Db, companyId: string, userId
|
||||
// Resolve revision metadata before consulting the cache, so rotation and
|
||||
// revocation cannot serve windows associated with an old credential.
|
||||
const secretIds = accounts.flatMap(row => row.grant.credentialSecretRefs.map(ref => ref.secretId));
|
||||
const revisions = secretIds.length ? await db.select({ id: companySecrets.id, latestVersion: companySecrets.latestVersion, status: companySecrets.status, versionStatus: companySecretVersions.status, revokedAt: companySecretVersions.revokedAt })
|
||||
.from(companySecrets).leftJoin(companySecretVersions, and(eq(companySecretVersions.secretId, companySecrets.id), eq(companySecretVersions.version, companySecrets.latestVersion))).where(and(eq(companySecrets.companyId, companyId), inArray(companySecrets.id, secretIds))) : [];
|
||||
const revisions = secretIds.length ? await db.select({ id: companySecrets.id, latestVersion: companySecrets.latestVersion, status: companySecrets.status, versionStatus: companySecretVersions.status, revokedAt: companySecretVersions.revokedAt, definitionId: companySecrets.userSecretDefinitionId, definitionStatus: userSecretDefinitions.status, definitionDeletedAt: userSecretDefinitions.deletedAt })
|
||||
.from(companySecrets).leftJoin(companySecretVersions, and(eq(companySecretVersions.secretId, companySecrets.id), eq(companySecretVersions.version, companySecrets.latestVersion))).leftJoin(userSecretDefinitions, and(eq(userSecretDefinitions.id, companySecrets.userSecretDefinitionId), eq(userSecretDefinitions.companyId, companyId))).where(and(eq(companySecrets.companyId, companyId), inArray(companySecrets.id, secretIds))) : [];
|
||||
async function readAccount(row: (typeof accounts)[number]): Promise<ProviderQuotaResult> {
|
||||
const accountKey = createHash("sha256").update(JSON.stringify([companyId, row.connection.id, row.grant.id,
|
||||
row.connection.updatedAt, row.grant.updatedAt,
|
||||
row.grant.credentialSecretRefs.map(ref => [ref.secretId, revisions.find(r => r.id === ref.secretId)]),
|
||||
])).digest("hex");
|
||||
const base = { provider: row.summary.provider, accountKey, accountLabel: row.summary.name, source: "managed-connection" };
|
||||
if (row.summary.status !== "connected") {
|
||||
const definitionUnavailable = row.grant.credentialSecretRefs.some(ref => {
|
||||
const revision = revisions.find(value => value.id === ref.secretId);
|
||||
return revision?.definitionId && (revision.definitionStatus !== "active" || revision.definitionDeletedAt !== null);
|
||||
});
|
||||
if (row.summary.status !== "connected" || definitionUnavailable) {
|
||||
return { ...base, ok: false, errorFamily: "credentials_unavailable", error: publicError, windows: [] };
|
||||
}
|
||||
const key = `${userId}:${accountKey}`;
|
||||
|
||||
Reference in new issue
Block a user