Open History to see the last commit
1import {
2 isProviderNativeSubagentThread,
3 OrchestrationV2GetNativeChildWorkError,
4 type OrchestrationV2GetNativeChildWorkInput,
5 type OrchestrationV2GetNativeChildWorkResult,
6} from "@t3tools/contracts";
7import * as Context from "effect/Context";
8import * as Digest from "effect/crypto/Digest";
9import * as Effect from "effect/Effect";
10import * as Layer from "effect/Layer";
11import * as Path from "effect/Path";
12
13import * as ServerConfig from "../config.ts";
14import { readAntigravityChildWork } from "../provider/acp/AntigravityChildWork.ts";
15import { resolveAntigravityInstanceDirectories } from "../provider/antigravityAuthSupport.ts";
16import * as ThreadManagement from "./ThreadManagementService.ts";
17
18/** Reads what a provider-native subagent's children did, on demand. Other providers stream real child sessions. */
19export class NativeChildWorkQuery extends Context.Service<
20 NativeChildWorkQuery,
21 {
22 readonly getNativeChildWork: (
23 input: OrchestrationV2GetNativeChildWorkInput,
24 ) => Effect.Effect<
25 OrchestrationV2GetNativeChildWorkResult,
26 OrchestrationV2GetNativeChildWorkError
27 >;
28 }
29>()("t3/orchestration-v2/NativeChildWorkQuery") {}
30
31const NO_CHILDREN: OrchestrationV2GetNativeChildWorkResult = { children: [] };
32
33const make = Effect.gen(function* () {
34 const threads = yield* ThreadManagement.ThreadManagementService;
35 const { stateDir } = yield* ServerConfig.ServerConfig;
36 const digest = yield* Digest.Digest;
37 const path = yield* Path.Path;
38
39 const getNativeChildWork = Effect.fn("NativeChildWorkQuery.getNativeChildWork")(
40 function* (input: OrchestrationV2GetNativeChildWorkInput) {
41 const { thread } = yield* threads.getThreadRecords(input.threadId, []);
42 const parentThreadId = thread.lineage.parentThreadId;
43 if (
44 !isProviderNativeSubagentThread(thread) ||
45 parentThreadId === null ||
46 thread.forkedFrom?.type !== "node"
47 ) {
48 return NO_CHILDREN;
49 }
50 const subagentId = thread.forkedFrom.nodeId;
51 const { subagents } = yield* threads.getThreadRecords(parentThreadId, ["subagents"]);
52 const subagent = subagents.find(
53 (candidate) => candidate.id === subagentId && candidate.childThreadId === thread.id,
54 );
55 const nativeTaskId = subagent?.nativeTaskRef?.nativeId;
56 if (
57 subagent?.driver !== "antigravity" ||
58 subagent.origin !== "provider_native" ||
59 subagent.runId === null ||
60 !nativeTaskId
61 ) {
62 return NO_CHILDREN;
63 }
64 const others = subagents.filter(
65 (candidate) =>
66 candidate.id !== subagent.id &&
67 candidate.driver === "antigravity" &&
68 candidate.origin === "provider_native" &&
69 candidate.providerInstanceId === subagent.providerInstanceId,
70 );
71 const { runs, providerThreads } = yield* threads.getThreadRecords(
72 parentThreadId,
73 ["runs", "providerThreads"],
74 { runIds: [subagent.runId, ...others.flatMap((other) => other.runId ?? [])] },
75 );
76 const providerThreadOf = (runId: string | null) => {
77 const providerThreadId = runs.find((run) => run.id === runId)?.providerThreadId;
78 return providerThreads.find((candidate) => candidate.id === providerThreadId);
79 };
80 const providerThread = providerThreadOf(subagent.runId);
81 const rootSessionId = providerThread?.nativeThreadRef?.nativeId;
82 if (!rootSessionId || providerThread?.providerInstanceId !== subagent.providerInstanceId) {
83 return NO_CHILDREN;
84 }
85 const directories = yield* resolveAntigravityInstanceDirectories(
86 stateDir,
87 subagent.providerInstanceId,
88 ).pipe(Effect.provideService(Digest.Digest, digest), Effect.provideService(Path.Path, path));
89 const children = yield* readAntigravityChildWork({
90 profileDirectory: directories.profile,
91 rootSessionId,
92 nativeTaskId,
93 launch: { title: subagent.title, prompt: subagent.prompt },
94 // Only batches from the same agent session can share its records.
95 siblings: others.flatMap((other) => {
96 const id = other.nativeTaskRef?.nativeId;
97 return id && providerThreadOf(other.runId)?.nativeThreadRef?.nativeId === rootSessionId
98 ? [{ nativeTaskId: id, title: other.title, prompt: other.prompt }]
99 : [];
100 }),
101 });
102 return { children };
103 },
104 (effect, input) =>
105 effect.pipe(
106 Effect.mapError(
107 (cause) =>
108 new OrchestrationV2GetNativeChildWorkError({ threadId: input.threadId, cause }),
109 ),
110 ),
111 );
112
113 return NativeChildWorkQuery.of({ getNativeChildWork });
114});
115
116export const layer = Layer.effect(NativeChildWorkQuery, make);