import {  ApprovalRequestId,  EventId,  ProviderDriverKind,  ProviderInstanceId,  RuntimeRequestId,  RuntimeTaskId,  TurnId,  type AntigravitySettings,  type ProviderApprovalDecision,  type ProviderRuntimeEvent,  type ProviderSession,  type ProviderSetupError,  type ProviderUserInputAnswers,  type RuntimeTaskStatus,  type ThreadId,  type TurnCompletedPayload,} from "@t3tools/contracts";import * as Cause from "effect/Cause";import * as Crypto from "effect/Crypto";import * as DateTime from "effect/DateTime";import * as Deferred from "effect/Deferred";import * as Effect from "effect/Effect";import * as Exit from "effect/Exit";import * as Fiber from "effect/Fiber";import * as FileSystem from "effect/FileSystem";import * as Option from "effect/Option";import * as Path from "effect/Path";import * as PubSub from "effect/PubSub";import * as Schema from "effect/Schema";import * as Scope from "effect/Scope";import * as Semaphore from "effect/Semaphore";import * as Stream from "effect/Stream";import * as SynchronizedRef from "effect/SynchronizedRef";import * as EffectAcpErrors from "effect-acp/errors";import type * as EffectAcpSchema from "effect-acp/schema"; import { ServerConfig } from "../../config.ts";import { buildRuntimeInstructions } from "../RuntimeInstructions.ts";import * as McpProviderSession from "../../mcp/McpProviderSession.ts";import type { AntigravityAuth } from "../AntigravityAuth.ts";import {  ProviderAdapterRequestError,  ProviderAdapterSessionClosedError,  ProviderAdapterSessionNotFoundError,  ProviderAdapterValidationError,  type ProviderAdapterError,} from "../Errors.ts";import {  ANTIGRAVITY_SIGN_IN_REQUIRED_MESSAGE,  isAntigravitySignInRequiredError,} from "../antigravityAuthSupport.ts";import { mapAcpToAdapterError } from "../acp/AcpAdapterSupport.ts";import {  makeAcpAssistantItemEvent,  makeAcpContentDeltaEvent,  makeAcpPlanUpdatedEvent,  makeAcpRequestOpenedEvent,  makeAcpRequestResolvedEvent,  makeAcpToolCallEvent,} from "../acp/AcpCoreRuntimeEvents.ts";import { makeAcpNativeLoggerFactory } from "../acp/AcpNativeLogging.ts";import { parsePermissionRequest, type AcpToolCallState } from "../acp/AcpRuntimeModel.ts";import type * as AcpSessionRuntime from "../acp/AcpSessionRuntime.ts";import {  antigravityPermissionMode,  antigravityModelOptions,  applyAntigravityAcpModelSelection,  buildAntigravityPrompt,  type AntigravityAcpRuntimeInput,  resolveAntigravityModel,} from "../acp/AntigravityAcpSupport.ts";import {  antigravityApprovalOptions,  antigravitySubagentOutput,  classifyAntigravitySubagentToolCall,  extractAntigravityUserInputQuestion,  isAntigravityOpenCommand,  isAntigravitySubagentReplayStart,  isAntigravityUserInputRequest,  makeAntigravityUserInputResponse,  normalizeAntigravityToolCall,  sanitizeAntigravityToolPayload,  selectAntigravityPermissionOptionId,} from "../acp/AntigravityProtocol.ts";import type { ProviderAdapterShape } from "../Services/ProviderAdapter.ts";import type { EventNdjsonLogger } from "./EventNdjsonLogger.ts"; const PROVIDER = ProviderDriverKind.make("antigravity");const ResumeCursor = Schema.Struct({  schemaVersion: Schema.Literal(1),  sessionId: Schema.NonEmptyString,});const decodeResumeCursor = Schema.decodeUnknownOption(ResumeCursor);const isAcpError = Schema.is(EffectAcpErrors.AcpError); type Adapter = ProviderAdapterShape<ProviderAdapterError>;type Runtime = Pick<  AcpSessionRuntime.AcpSessionRuntime["Service"],  | "handleRequestPermission"  | "handleReadTextFile"  | "handleWriteTextFile"  | "start"  | "setMode"  | "setModel"  | "getConfigOptions"  | "getEvents"  | "drainEvents"  | "prompt"  | "cancel">;type NativePermission = EffectAcpSchema.RequestPermissionRequest;type NativePermissionResponse = EffectAcpSchema.RequestPermissionResponse; function mapAntigravityError(threadId: ThreadId, method: string, cause: EffectAcpErrors.AcpError) {  return isAntigravitySignInRequiredError(cause)    ? new ProviderAdapterRequestError({        provider: PROVIDER,        method,        detail: ANTIGRAVITY_SIGN_IN_REQUIRED_MESSAGE,        cause,      })    : mapAcpToAdapterError(PROVIDER, threadId, method, cause);} export interface AntigravityAdapterOptions {  readonly instanceId: ProviderInstanceId;  readonly makeRuntime: (    input: Omit<AntigravityAcpRuntimeInput, "spawn" | "childProcessSpawner" | "onAuthorizationUrl">,  ) => Effect.Effect<Runtime, EffectAcpErrors.AcpError | ProviderSetupError, Scope.Scope>;  readonly withProcess: AntigravityAuth["withProcess"];  readonly onSessionStarted?: (    started: AcpSessionRuntime.AcpSessionRuntimeStartResult,    cwd: string,  ) => Effect.Effect<void>;  readonly onAvailableCommands?: (    commands: ReadonlyArray<EffectAcpSchema.AvailableCommand>,    cwd: string,  ) => Effect.Effect<void>;  readonly onConfigOptionsUpdated?: (    configOptions: ReadonlyArray<EffectAcpSchema.SessionConfigOption>,  ) => Effect.Effect<void>;  readonly onAuthRequired?: Effect.Effect<void>;  /** Model the provider default alias selects, when the account offers it. */  readonly defaultModel?: Effect.Effect<string | undefined>;  readonly nativeEventLogger?: EventNdjsonLogger;} interface PendingApproval {  readonly request: NativePermission;  readonly response: Deferred.Deferred<{    readonly decision: ProviderApprovalDecision;    readonly result: NativePermissionResponse;  }>;} interface PendingQuestion {  readonly request: NativePermission;  readonly response: Deferred.Deferred<{    readonly answers: ProviderUserInputAnswers;    readonly result: NativePermissionResponse;  }>;} interface OpenCommand {  readonly toolCall: AcpToolCallState;  readonly turnId: TurnId | undefined;  readonly promoted: boolean;} interface OpenSubagent {  readonly turnId: TurnId | undefined;  readonly status: "pending" | "running" | undefined;  readonly description?: string;} function subagentLinkage(toolCallId: string) {  return {    taskId: RuntimeTaskId.make(toolCallId),    taskType: "subagent_batch",    toolUseId: toolCallId,    title: "Antigravity subagent batch",  };} interface TurnIntent {  readonly turnId: TurnId;  readonly generation: number;  settled: boolean;} interface SessionContext {  readonly threadId: ThreadId;  readonly cwd: string;  readonly nativeSessionId: string;  readonly scope: Scope.Closeable;  readonly runtime: Runtime;  readonly promptLock: Semaphore.Semaphore;  readonly stopLock: Semaphore.Semaphore;  readonly commandLock: Semaphore.Semaphore;  readonly approvals: Map<ApprovalRequestId, PendingApproval>;  readonly questions: Map<ApprovalRequestId, PendingQuestion>;  readonly commands: Map<string, OpenCommand>;  /** Keep only IDs after settlement or MCP exclusion so merged late updates cannot change identity. */  readonly subagents: Map<string, OpenSubagent | "finished" | "mcp">;  readonly turns: Array<{ id: TurnId; items: Array<unknown> }>;  session: ProviderSession;  activeTurnId: TurnId | undefined;  promptFiber: Fiber.Fiber<EffectAcpSchema.PromptResponse, EffectAcpErrors.AcpError> | undefined;  generation: number;  stopped: boolean;  closed: boolean;  disconnected: boolean;} const CLIENT_FILE_MAX_BYTES = 8 * 1024 * 1024; function isInsideRoot(path: Path.Path, root: string, candidate: string): boolean {  const relative = path.relative(root, candidate);  return relative === "" || (!relative.startsWith("..") && !path.isAbsolute(relative));} /** Resolves an agent-supplied path and rejects anything outside the session roots. */const resolveClientFilePath = Effect.fn("AntigravityAdapter.resolveClientFilePath")(  function* (input: {    readonly fileSystem: FileSystem.FileSystem;    readonly path: Path.Path;    readonly allowedRoots: ReadonlyArray<string>;    readonly requestPath: string;  }) {    const { path } = input;    const resolved = path.resolve(input.requestPath);    // Follow symlinks on the parent so a link out of the workspace cannot escape it.    const parent = yield* input.fileSystem      .realPath(path.dirname(resolved))      .pipe(Effect.orElseSucceed(() => path.dirname(resolved)));    const real = path.join(parent, path.basename(resolved));    const roots = yield* Effect.forEach(input.allowedRoots, (root) =>      input.fileSystem.realPath(root).pipe(Effect.orElseSucceed(() => root)),    );    if (!roots.some((root) => isInsideRoot(path, root, real))) {      return yield* EffectAcpErrors.AcpRequestError.invalidParams(        `Path '${input.requestPath}' is outside the session workspace.`,      );    }    return real;  },); const readClientTextFile = Effect.fn("AntigravityAdapter.readClientTextFile")(function* (input: {  readonly fileSystem: FileSystem.FileSystem;  readonly path: Path.Path;  readonly allowedRoots: ReadonlyArray<string>;  readonly request: EffectAcpSchema.ReadTextFileRequest;}): Effect.fn.Return<EffectAcpSchema.ReadTextFileResponse, EffectAcpErrors.AcpError> {  const filePath = yield* resolveClientFilePath({ ...input, requestPath: input.request.path });  const info = yield* input.fileSystem    .stat(filePath)    .pipe(      Effect.mapError(() =>        EffectAcpErrors.AcpRequestError.resourceNotFound(`File '${input.request.path}' not found.`),      ),    );  if (info.type !== "File" || Number(info.size) > CLIENT_FILE_MAX_BYTES) {    return yield* EffectAcpErrors.AcpRequestError.invalidParams(      `File '${input.request.path}' is not a readable text file under ${CLIENT_FILE_MAX_BYTES} bytes.`,    );  }  const text = yield* input.fileSystem    .readFileString(filePath)    .pipe(      Effect.mapError(() =>        EffectAcpErrors.AcpRequestError.internalError(`Could not read '${input.request.path}'.`),      ),    );  const line = input.request.line ?? undefined;  const limit = input.request.limit ?? undefined;  if (line === undefined && limit === undefined) {    return { content: text };  }  // ACP lines are 1-indexed. `limit` is a line count.  const lines = text.split("\n");  const start = Math.max(0, (line ?? 1) - 1);  const end = limit === undefined ? lines.length : Math.min(lines.length, start + limit);  return { content: lines.slice(start, end).join("\n") };}); const writeClientTextFile = Effect.fn("AntigravityAdapter.writeClientTextFile")(function* (input: {  readonly fileSystem: FileSystem.FileSystem;  readonly path: Path.Path;  readonly allowedRoots: ReadonlyArray<string>;  readonly request: EffectAcpSchema.WriteTextFileRequest;}): Effect.fn.Return<EffectAcpSchema.WriteTextFileResponse, EffectAcpErrors.AcpError> {  const filePath = yield* resolveClientFilePath({ ...input, requestPath: input.request.path });  yield* input.fileSystem.makeDirectory(input.path.dirname(filePath), { recursive: true }).pipe(    Effect.andThen(input.fileSystem.writeFileString(filePath, input.request.content)),    Effect.mapError(() =>      EffectAcpErrors.AcpRequestError.internalError(`Could not write '${input.request.path}'.`),    ),  );  return {};}); /** Keeps one official ACP process per thread and drains a cancelled prompt before steering. */export const makeAntigravityAdapter = Effect.fn("makeAntigravityAdapter")(function* (  settings: AntigravitySettings,  options: AntigravityAdapterOptions,) {  const crypto = yield* Crypto.Crypto;  const fileSystem = yield* FileSystem.FileSystem;  const path = yield* Path.Path;  const serverConfig = yield* ServerConfig;  const ownerScope = yield* Effect.scope;  const makeNativeLoggers = yield* makeAcpNativeLoggerFactory();  const sessions = new Map<ThreadId, SessionContext>();  const locks = yield* SynchronizedRef.make(new Map<ThreadId, Semaphore.Semaphore>());  const events = yield* PubSub.unbounded<ProviderRuntimeEvent>();  const nowIso = Effect.map(DateTime.now, DateTime.formatIso);  const randomId = crypto.randomUUIDv4.pipe(    Effect.mapError(      (cause) =>        new ProviderAdapterRequestError({          provider: PROVIDER,          method: "crypto/randomUUIDv4",          detail: "Could not create an Antigravity event ID.",          cause,        }),    ),  );  const stamp = Effect.all({    eventId: Effect.map(randomId, EventId.make),    createdAt: nowIso,  });  const emit = (event: ProviderRuntimeEvent) => PubSub.publish(events, event).pipe(Effect.asVoid);   const withThreadLock = <A, E, R>(threadId: ThreadId, task: Effect.Effect<A, E, R>) =>    SynchronizedRef.modifyEffect(locks, (current) => {      const existing = current.get(threadId);      if (existing) return Effect.succeed([existing, current] as const);      return Semaphore.make(1).pipe(        Effect.map((lock) => [lock, new Map(current).set(threadId, lock)] as const),      );    }).pipe(Effect.flatMap((lock) => lock.withPermit(task)));   const requireSession = (threadId: ThreadId) => {    const context = sessions.get(threadId);    return context && !context.stopped      ? Effect.succeed(context)      : Effect.fail(new ProviderAdapterSessionNotFoundError({ provider: PROVIDER, threadId }));  };   const cancelRequests = Effect.fn("AntigravityAdapter.cancelRequests")(function* (    context: SessionContext,  ) {    for (const pending of context.approvals.values()) {      yield* Deferred.succeed(pending.response, {        decision: "cancel",        result: { outcome: { outcome: "cancelled" } },      });    }    for (const pending of context.questions.values()) {      yield* Deferred.succeed(pending.response, {        answers: {},        result: { outcome: { outcome: "cancelled" } },      });    }  });   const finishBackgroundCommands = (context: SessionContext) =>    context.commandLock.withPermit(      Effect.gen(function* () {        for (const [id, command] of context.commands) {          if (!command.promoted) continue;          yield* emit({            type: "task.completed",            ...(yield* stamp),            provider: PROVIDER,            threadId: context.threadId,            turnId: command.turnId,            payload: {              taskId: RuntimeTaskId.make(id),              taskType: "local_bash",              toolUseId: id,              status: "stopped",            },          });        }        context.commands.clear();      }),    );   const finishSubagents = (    context: SessionContext,    status: Extract<RuntimeTaskStatus, "cancelled" | "failed" | "idle">,    error?: string,  ) =>    context.commandLock.withPermit(      Effect.gen(function* () {        for (const [id, subagent] of context.subagents) {          if (subagent === "finished" || subagent === "mcp") continue;          yield* emit({            type: "task.updated",            ...(yield* stamp),            provider: PROVIDER,            threadId: context.threadId,            turnId: subagent.turnId,            payload: {              ...subagentLinkage(id),              status,              ...(status === "idle"                ? {                    description: "Turn ended. Individual agent status is unavailable.",                    timelineBypass: true,                  }                : {}),              ...(error ? { error } : {}),            },          });          context.subagents.set(id, "finished");        }      }),    );   const stopContext = (context: SessionContext) =>    context.stopLock      .withPermit(        Effect.gen(function* () {          if (context.closed) return;          context.stopped = true;          yield* Effect.gen(function* () {            yield* cancelRequests(context);            if (context.promptFiber && !context.disconnected) {              yield* Effect.ignore(context.runtime.cancel);            }          }).pipe(Effect.ensuring(Scope.close(context.scope, Exit.void)));          context.closed = true;          if (sessions.get(context.threadId) === context) sessions.delete(context.threadId);          yield* finishBackgroundCommands(context);          yield* finishSubagents(            context,            context.disconnected ? "failed" : "cancelled",            context.disconnected ? "Antigravity process stopped." : undefined,          );          context.subagents.clear();          yield* emit({            type: "session.exited",            ...(yield* stamp),            provider: PROVIDER,            threadId: context.threadId,            payload: {              exitKind: context.disconnected ? "error" : "graceful",              ...(context.disconnected ? { reason: "Antigravity process stopped." } : {}),            },          });        }),      )      .pipe(Effect.uninterruptible);   const handlePermission = Effect.fn("AntigravityAdapter.handlePermission")(function* (    context: SessionContext,    request: NativePermission,  ): Effect.fn.Return<NativePermissionResponse, ProviderAdapterError> {    if (context.stopped || request.sessionId !== context.nativeSessionId) {      return { outcome: { outcome: "cancelled" } };    }    const requestId = ApprovalRequestId.make(yield* randomId);    const runtimeRequestId = RuntimeRequestId.make(requestId);    const turnId = context.activeTurnId;    const rawPayload = sanitizeAntigravityToolPayload(request);     if (isAntigravityUserInputRequest(request)) {      const question = extractAntigravityUserInputQuestion(request);      if (!question) return { outcome: { outcome: "cancelled" } };      const response = yield* Deferred.make<{        answers: ProviderUserInputAnswers;        result: NativePermissionResponse;      }>();      context.questions.set(requestId, { request, response });      return yield* Effect.gen(function* () {        yield* emit({          type: "user-input.requested",          ...(yield* stamp),          provider: PROVIDER,          threadId: context.threadId,          turnId,          requestId: runtimeRequestId,          payload: { questions: [question] },          raw: { source: "acp.jsonrpc", method: "session/request_permission", payload: rawPayload },        });        const answer = yield* Deferred.await(response);        yield* emit({          type: "user-input.resolved",          ...(yield* stamp),          provider: PROVIDER,          threadId: context.threadId,          turnId,          requestId: runtimeRequestId,          payload: { answers: answer.answers },        });        return answer.result;

Lines 1–500 of 1279. Load more