1/**
2 * Record one anonymous `mcp.delegated_task.finished` event per app-owned task reaching final status.
3 *
4 * @module DelegatedTaskAnalytics
5 */
6import type { OrchestrationV2DomainEvent } from "@t3tools/contracts";
7import * as DateTime from "effect/DateTime";
8import * as Effect from "effect/Effect";
9import * as Layer from "effect/Layer";
10import * as Stream from "effect/Stream";
11
12import * as Orchestrator from "../orchestration-v2/Orchestrator.ts";
13import * as ProviderRegistry from "../provider/ProviderRegistry.ts";
14import { forkParked } from "../serverActivation.ts";
15import * as AnalyticsService from "./AnalyticsService.ts";
16import { crossProvider, providerDimensions } from "./ProviderDimensions.ts";
17
18type SubagentEvent = Extract<OrchestrationV2DomainEvent, { readonly type: "subagent.updated" }>;
19
20const FINAL_STATUSES: ReadonlySet<string> = new Set([
21 "completed",
22 "failed",
23 "cancelled",
24 "interrupted",
25]);
26
27const make = Effect.gen(function* () {
28 const orchestrator = yield* Orchestrator.OrchestratorV2;
29 const analytics = yield* AnalyticsService.AnalyticsService;
30 const registry = yield* ProviderRegistry.ProviderRegistry;
31 // A task can be re-reported with the same final status; count it once per process.
32 const reported = new Set<string>();
33
34 // Runs for every subagent update; most return early, so avoid a span per event.
35 const onSubagent = Effect.fnUntraced(
36 function* (event: SubagentEvent) {
37 const task = event.payload;
38 if (task.origin !== "app_owned" || !FINAL_STATUSES.has(task.status)) return;
39 if (reported.has(task.id)) return;
40 const parent = yield* orchestrator.getThreadShell(task.threadId);
41 const providers = yield* registry.getProviders;
42 const parentDimensions =
43 parent === null
44 ? { provider: "unknown" }
45 : providerDimensions(providers, parent.modelSelection);
46 const child = providerDimensions(providers, {
47 instanceId: task.providerInstanceId,
48 model: task.model ?? "",
49 });
50 const cross = crossProvider(parentDimensions.provider, child.provider);
51 const durationSeconds =
52 task.startedAt !== null && task.completedAt !== null
53 ? Math.max(
54 0,
55 Math.round(
56 (DateTime.toEpochMillis(task.completedAt) -
57 DateTime.toEpochMillis(task.startedAt)) /
58 1000,
59 ),
60 )
61 : undefined;
62 yield* analytics.record("mcp.delegated_task.finished", {
63 status: task.status,
64 callerProvider: parentDimensions.provider,
65 ...(parentDimensions.model === undefined ? {} : { callerModel: parentDimensions.model }),
66 targetProvider: child.provider,
67 ...(child.model === undefined ? {} : { targetModel: child.model }),
68 ...(cross === undefined ? {} : { crossProvider: cross }),
69 ...(task.completionWake === undefined ? {} : { completionWake: task.completionWake }),
70 ...(durationSeconds === undefined ? {} : { durationSeconds }),
71 });
72 // Only after recording: a failed lookup leaves the next final update to report it.
73 reported.add(task.id);
74 },
75 (effect) => effect.pipe(Effect.ignoreCause),
76 );
77
78 yield* forkParked(
79 Stream.runForEach(orchestrator.streamDomainEvents, (event) =>
80 event.type === "subagent.updated" ? onSubagent(event) : Effect.void,
81 ),
82 );
83});
84
85export const layer = Layer.effectDiscard(make);