Open History to see the last commit
1import { ModelSelection } from "@t3tools/contracts";
2import type { UsageSpeed } from "@t3tools/provider-core/server/usage";
3import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
4import * as Context from "effect/Context";
5import * as Effect from "effect/Effect";
6import * as Layer from "effect/Layer";
7import * as Option from "effect/Option";
8import * as Schema from "effect/Schema";
9import * as SqlClient from "effect/sql/SqlClient";
10
11import { getCodexServiceTierOptionValue } from "../codexModelOptions.ts";
12
13export class CodexUsageHistoryReadError extends Schema.TaggedError<CodexUsageHistoryReadError>()(
14 "CodexUsageHistoryReadError",
15 { requestedTurnCount: Schema.Number, cause: Schema.Defect() },
16) {
17 override get message(): string {
18 return "Could not read historical Codex usage tiers.";
19 }
20}
21
22const decodeSelection = Schema.decodeUnknownOption(Schema.fromJsonString(ModelSelection));
23const encodeTurnIds = Schema.encodeSync(Schema.fromJsonString(Schema.Array(Schema.String)));
24
25export interface CodexUsageHistoryRequest {
26 readonly sessionId: string;
27 readonly nativeTurnId: string;
28 readonly model: string;
29}
30
31export class CodexUsageHistory extends Context.Service<
32 CodexUsageHistory,
33 {
34 readonly resolve: (
35 requests: readonly CodexUsageHistoryRequest[],
36 ) => Effect.Effect<
37 ReadonlyMap<CodexUsageHistoryRequest, UsageSpeed>,
38 CodexUsageHistoryReadError
39 >;
40 }
41>()("t3/usage/CodexUsageHistory") {}
42
43const make = Effect.gen(function* () {
44 const sql = yield* SqlClient.SqlClient;
45 const resolve = Effect.fn("CodexUsageHistory.resolve")(function* (
46 requests: readonly CodexUsageHistoryRequest[],
47 ) {
48 const resolved = new Map<CodexUsageHistoryRequest, UsageSpeed>();
49 if (requests.length === 0) return resolved;
50 const turnIdsJson = encodeTurnIds([
51 ...new Set(requests.map((request) => request.nativeTurnId)),
52 ]);
53
54 // The first running event for this attempt precedes any steering changes
55 // to the selection. A missing selection there must not fall through to a later event.
56 const rows = yield* sql<{
57 readonly native_thread_id: string | null;
58 readonly native_turn_id: string;
59 readonly selection_json: string | null;
60 }>`
61 SELECT json_extract(attempt.payload_json, '$.nativeThreadId') AS native_thread_id,
62 json_extract(turn.payload_json, '$.nativeTurnRef.nativeId') AS native_turn_id,
63 (
64 SELECT json_extract(event.payload_json, '$.modelSelection')
65 FROM orchestration_events AS event
66 WHERE event.aggregate_kind = 'thread'
67 AND event.application_event_version = 2
68 AND event.stream_id = attempt.thread_id
69 AND event.event_type = 'run.updated'
70 AND json_extract(event.payload_json, '$.id') = attempt.run_id
71 AND json_extract(event.payload_json, '$.activeAttemptId') = attempt.attempt_id
72 AND json_extract(event.payload_json, '$.status') = 'running'
73 ORDER BY event.sequence
74 LIMIT 1
75 ) AS selection_json
76 FROM orchestration_v2_projection_provider_turns AS turn
77 LEFT JOIN orchestration_v2_projection_run_attempts AS attempt
78 ON attempt.attempt_id = turn.run_attempt_id
79 AND attempt.thread_id = turn.thread_id
80 AND attempt.provider_thread_id = turn.provider_thread_id
81 WHERE json_extract(turn.payload_json, '$.nativeTurnRef.driver') = 'codex'
82 AND json_extract(turn.payload_json, '$.nativeTurnRef.nativeId') IN (
83 SELECT value FROM json_each(${turnIdsJson})
84 )
85 `.pipe(
86 NodeSqliteClient.readOnly,
87 Effect.mapError(
88 (cause) => new CodexUsageHistoryReadError({ requestedTurnCount: requests.length, cause }),
89 ),
90 );
91
92 // Keep duplicates ambiguous even when an attempt is missing or its destination differs.
93 const byTurn = new Map<string, (typeof rows)[number] | null>();
94 for (const row of rows) {
95 byTurn.set(row.native_turn_id, byTurn.has(row.native_turn_id) ? null : row);
96 }
97 for (const request of requests) {
98 const match = byTurn.get(request.nativeTurnId);
99 if (!match || match.native_thread_id !== request.sessionId) continue;
100 const selection = decodeSelection(match.selection_json);
101 if (Option.isNone(selection) || selection.value.model !== request.model) continue;
102 if (
103 selection.value.options?.some(
104 (option) => option.id === "serviceTier" && typeof option.value !== "string",
105 )
106 )
107 continue;
108 const tier = getCodexServiceTierOptionValue(selection.value);
109 if (tier === "fast" || tier === "priority") resolved.set(request, "fast");
110 else if (tier === "ultrafast") resolved.set(request, "ultrafast");
111 else if (tier === undefined || tier === "default" || tier === "standard") {
112 resolved.set(request, "standard");
113 }
114 }
115 return resolved;
116 });
117 return CodexUsageHistory.of({ resolve });
118});
119
120export const layer = Layer.effect(CodexUsageHistory, make);