1/** Native Pi session usage, including calls billed outside the main conversation. */
2import * as Effect from "effect/Effect";
3import * as Option from "effect/Option";
4import * as Path from "effect/Path";
5import * as Schema from "effect/Schema";
6
7import { expandHomePath } from "@t3tools/provider-core/server/pathExpansion";
8import {
9 parseTimestampMs,
10 tokenCount,
11 totalTokens,
12 type ProviderUsageReader,
13 type TranscriptUsageFormat,
14 type UsageRecord,
15} from "@t3tools/provider-core/server/usage";
16import * as HostProcess from "@t3tools/shared/HostProcess";
17import type { PiSettings } from "../settings.ts";
18
19const PiScanState = Schema.Struct({ sessionId: Schema.mutableKey(Schema.String) });
20const Usage = Schema.Struct({
21 input: Schema.optional(Schema.Unknown),
22 output: Schema.optional(Schema.Unknown),
23 cacheRead: Schema.optional(Schema.Unknown),
24 cacheWrite: Schema.optional(Schema.Unknown),
25 reasoning: Schema.optional(Schema.Unknown),
26 cost: Schema.optional(Schema.Struct({ total: Schema.Unknown })),
27});
28const Identity = { id: Schema.NonEmptyString, timestamp: Schema.String };
29const Model = { provider: Schema.NonEmptyString, model: Schema.NonEmptyString };
30const Entry = Schema.Union([
31 Schema.Struct({ type: Schema.Literal("session"), id: Schema.NonEmptyString }),
32 Schema.Struct({
33 ...Identity,
34 type: Schema.Literal("message"),
35 message: Schema.Union([
36 Schema.Struct({
37 ...Model,
38 role: Schema.Literal("assistant"),
39 responseModel: Schema.optional(Schema.NonEmptyString),
40 usage: Usage,
41 }),
42 Schema.Struct({ role: Schema.Literal("toolResult"), usage: Usage }),
43 ]),
44 }),
45 Schema.Struct({ ...Identity, ...Model, type: Schema.Literal("usage"), usage: Usage }),
46 Schema.Struct({
47 ...Identity,
48 type: Schema.Literals(["compaction", "branch_summary"]),
49 usage: Usage,
50 }),
51]);
52const decodeEntry = Schema.decodeUnknownOption(Entry);
53
54function parseRecord(parsed: unknown, state: typeof PiScanState.Type): readonly UsageRecord[] {
55 const decoded = decodeEntry(parsed);
56 if (Option.isNone(decoded)) return [];
57 const entry = decoded.value;
58 if (entry.type === "session") {
59 state.sessionId = entry.id;
60 return [];
61 }
62 const timestampMs = parseTimestampMs(entry.timestamp);
63 if (timestampMs === null) return [];
64 const source = entry.type === "message" ? entry.message : entry;
65 const usage = source.usage;
66 // Pi's auxiliary tool and summary records carry no model attribution. Match
67 // Pi's own breakdown rather than assigning them to the conversation model.
68 const rateModel =
69 "model" in source
70 ? "responseModel" in source
71 ? (source.responseModel ?? source.model)
72 : source.model
73 : undefined;
74 const model = "provider" in source ? `${source.provider}/${rateModel}` : "Tools/summaries";
75 const outputTokens = tokenCount(usage.output);
76 const totals = {
77 // Pi's input/cache counters are already disjoint.
78 uncachedInputTokens: tokenCount(usage.input),
79 cachedInputTokens: tokenCount(usage.cacheRead),
80 cacheCreationTokens: tokenCount(usage.cacheWrite),
81 outputTokens,
82 reasoningTokens: Math.min(outputTokens, tokenCount(usage.reasoning)),
83 };
84 const cost = usage.cost?.total;
85 // Pi defaults omitted catalog rates to zero. Let shared pricing estimate
86 // API-equivalent cost when the transcript has no positive cost estimate.
87 const reportedCostUsd =
88 typeof cost === "number" && Number.isFinite(cost) && cost > 0 ? cost : null;
89 if (totalTokens(totals) === 0 && reportedCostUsd === null) return [];
90 return [
91 {
92 provider: "pi",
93 timestampMs,
94 model,
95 ...(rateModel === undefined ? {} : { rateModel }),
96 sessionId: state.sessionId,
97 totals,
98 reportedCostUsd,
99 speed: "standard",
100 // Forks copy entries verbatim, including ID/timestamp. Entry IDs are only
101 // eight random hex digits, so retain timestamp/model to avoid collisions
102 // across unrelated sessions. Parent IDs may change when Pi branches.
103 dedupeKey: JSON.stringify(["pi", entry.id, entry.timestamp, model]),
104 },
105 ];
106}
107
108export const piUsageFormat: TranscriptUsageFormat<typeof PiScanState.Type> = {
109 selectFields: {
110 type: true,
111 id: true,
112 timestamp: true,
113 provider: true,
114 model: true,
115 usage: true,
116 message: { role: true, provider: true, model: true, responseModel: true, usage: true },
117 },
118 usageMarkers: ['"usage"', '"session"'],
119 parseLine: (line, state) => {
120 let parsed: unknown;
121 try {
122 parsed = JSON.parse(line);
123 } catch {
124 return [];
125 }
126 return parseRecord(parsed, state);
127 },
128 parseProjected: parseRecord,
129 state: { initial: () => ({ sessionId: "" }), schema: PiScanState },
130};
131
132export const piUsageReader: ProviderUsageReader<PiSettings, Path.Path> = {
133 kind: "transcripts",
134 provider: "pi",
135 format: piUsageFormat,
136 directories: Effect.fn("piUsageReader.directories")(function* ({ environment }) {
137 const path = yield* Path.Path;
138 const home = yield* HostProcess.HomeDirectory;
139 const sessionDir = environment.PI_CODING_AGENT_SESSION_DIR?.trim();
140 if (sessionDir) return [{ dir: path.resolve(expandHomePath(sessionDir, home)) }];
141 const agentDir = environment.PI_CODING_AGENT_DIR?.trim();
142 return [
143 {
144 dir: path.resolve(
145 agentDir ? expandHomePath(agentDir, home) : path.join(home, ".pi", "agent"),
146 "sessions",
147 ),
148 },
149 ];
150 }),
151};