// apps/api/src/sandbox/channel-a.ts — the API-DIRECT Channel-A seam (P4.4).
//
// The structured services (FileSystem / Git / Terminal) are SYNCHRONOUS point
// queries served client -> API -> box IN-PROCESS. Each call:
//
// For a provider-backed home it acquires an exact direct-request lease holder,
// resumes the box by id, and releases the holder after the operation. For a
// Connected Machine home it follows the durable active pointer directly and
// uses its NATS request/reply control channel; it creates no phantom cloud lease.
// Both paths build one SandboxChannelAService, run the operation, and return the
// result inline.
//
// NO Temporal or worker RPC sits in this path. Provider-backed reads remain
// process-local; Connected Machine operations necessarily ride NATS. Side-effect
// notifications (fs.changed/git.changed/terminal.pty.*) ride A1.
//
// IMPORT DISCIPLINE: sandbox symbols come ONLY from @opengeni/runtime/sandbox
// (the agent-loop-free leaf) — enforced by sandbox-access-import-guard.test.ts.

import {
  applyGitAuthPointerEnvironment,
  hasGitCredentialRepositorySelection,
  hasGitHubRepositorySelection,
  sandboxLifecycleTransitionWaitMs,
  stableSandboxEnvironmentForRun,
  type Settings,
} from "@opengeni/config";
import { githubAppBotIdentity } from "@opengeni/github";
import type { Session } from "@opengeni/contracts";
import {
  acquireLease,
  getSandboxSessionEnvelope,
  getLiveEnrollmentConnection,
  getSandbox,
  getScheduledScopedRigVersionMetadata,
  touchLeaseHolder,
  loadWorkspaceEnvironmentForRun,
  markWarmLeaseInstanceLost,
  readActiveSandbox,
  readLease,
  releaseLeaseHolder,
  SandboxImageConflictError,
  SandboxProviderReadLockUnavailableError,
  SandboxRigConflictError,
  withSandboxProviderReadLock,
  type Database,
  type LeaseSnapshot,
} from "@opengeni/db";
import { appendAndPublishEvents, type EventBus } from "@opengeni/events";
import {
  recordTenancyCompatibilityLaneUse,
  sandboxLeaseTelemetryKey,
  sandboxOperationMetricObserver,
  type Observability,
} from "@opengeni/observability";
import { HTTPException } from "hono/http-exception";
import { ApiHttpError } from "../http/api-error";
import type { ObjectStorage } from "@opengeni/storage";

import {
  buildSelfhostedBackendSession,
  establishSandboxSessionFromEnvelope,
  isProviderSandboxNotFoundError,
  SandboxChannelAService,
  NatsControlRpc,
  NatsOpStreamTransport,
  SandboxResumeIdentityMismatchError,
  SandboxResumeIdentityUnavailableError,
  RoutingActiveRouteChangedError,
  RoutingWorkspaceRootChangedError,
  SelfhostedWorkspaceRootChangedError,
  ChannelAConflictError,
  ChannelAFileSystemRouteChangedError,
  ChannelANotFoundError,
  ChannelAUnsupportedError,
  ChannelAUnavailableError,
  ChannelAValidationError,
  BrowserControlRequestError,
  BrowserControlTransportError,
  SelfhostedControlError,
  agentErrorToControlError,
  codemodeTokenFileFromEnvironment,
  offlineAgentError,
  resolveConnectedMachineWorkspaceRoot,
  withCodemodeTokenSession,
  withRunCredentialsSession,
  type ChannelASession,
  type EstablishedSandboxSession,
  type RoutingSandboxSession,
} from "@opengeni/runtime/sandbox";
import {
  managedSessionGroupBackend,
  managedSessionGroupOs,
  providerSettingsForSessionSandboxRuntime,
  relayConfigFromSettings,
  resolveSessionSandboxRuntime,
  wrapChannelABoxWithRouting,
} from "@opengeni/core";
import { establishApiSandboxSpawner } from "./rematerialize";

export type ChannelAServices = {
  db: Database;
  settings: Settings;
  bus: EventBus;
  objectStorage?: ObjectStorage | null;
  observability?: Observability | undefined;
};

export type ChannelAOperation =
  | "fs.list"
  | "fs.list-batch"
  | "fs.read"
  | "artifact.publish"
  | "fs.write"
  | "fs.delete"
  | "fs.move"
  | "fs.mkdir"
  | "git.status"
  | "git.diff"
  | "git.read-batch"
  | "git.log"
  | "git.show"
  | "terminal.exec"
  | "terminal.pty.open"
  | "terminal.pty.write"
  | "terminal.pty.resize"
  | "terminal.pty.close"
  | "browser.create"
  | "browser.resume"
  | "browser.suspend"
  | "browser.end"
  | "browser.read"
  | "browser.action"
  | "browser.control"
  | "browser.download.save"
  | "browser.attach"
  | "computer.create"
  | "computer.end"
  | "computer.read"
  | "computer.action"
  | "computer.control"
  | "computer.attach";

export type ChannelAContext = {
  accountId: string;
  workspaceId: string;
  session: Session;
  // The principal that drives the op (for emit attribution + pty opened_by).
  subjectId: string;
  /** Cancel lifecycle waiting when the originating HTTP request disconnects. */
  waitSignal?: AbortSignal | undefined;
  /** Bounded route identity for metrics and safe operator diagnostics. */
  operation?: ChannelAOperation | undefined;
  /** The callback is an interaction-controller read or an exactly-once action.
   * A controller transport failure may therefore rebuild the exact fenced
   * provider handle and replay the request. Tab/lifecycle mutations that lack
   * a controller operation id must never opt into this recovery. */
  retryControllerTransport?: boolean | undefined;
};

export type ChannelAOperationFailureReason =
  | "request_cancelled"
  | "provider_read_busy"
  | "provider_unavailable"
  | "lifecycle_conflict"
  | "request_rejected"
  | "unexpected";

export type ChannelAOperationFailureDiagnostic = {
  reason: ChannelAOperationFailureReason;
  status: number;
  errorCode:
    | "sandbox_channel_a_cancelled"
    | "sandbox_channel_a_provider_busy"
    | "sandbox_channel_a_provider_unavailable"
    | "sandbox_channel_a_lifecycle_conflict"
    | "sandbox_channel_a_operation_failed";
};

// The live op surface handed to a route's callback: the service + the live lease
// (for the pty exec-session epoch fence + revision seeding).
export type ChannelAHandle = {
  service: SandboxChannelAService;
  /** Connected Machine homes deliberately have no cloud lease. Durable PTYs
   * require a real home-provider lease and reject this null case. */
  lease: LeaseSnapshot | null;
  /** Exact placement-home session established under this request's lease or
   * Connected Machine fence. Unlike routingSession, this never follows a later
   * active-sandbox pointer and is safe for placement-bound controllers. */
  homeSession: ChannelASession;
  routingSession: RoutingSandboxSession;
  requestId: string;
};

/**
 * Provider handles are lightweight references to a lease-owned sandbox, but
 * reconstructing one is not free: Modal resume-by-id plus its first command can
 * dominate a small Git/files read. Workspace panels issue several independent
 * Channel-A requests together, so reuse the exact fenced handle briefly instead
 * of making every request reattach to the same warm instance.
 *
 * The key includes the session, lease epoch, and immutable provider instance id.
 * A rotation can therefore never inherit an old handle. Entries are bounded and
 * expire opportunistically; eviction only drops local references and never
 * terminates the lease-owned sandbox.
 */
// Read/viewer handles and process-capable handles deliberately have separate
// caches. The pinned Modal patch rotates a handle's command-router transport in
// place, while a typed read failure below still gets one fresh-handle fallback.
// Periodically rebuilding a healthy hot read handle would add multi-second
// stalls, so both caches keep the pre-existing five-minute IDLE lifetime.
//
// Modal and Unix/Docker SDK sessions retain yielded exec/PTY process objects in
// a process-local map; rebuilding the wrapper cannot reconstruct those objects
// from the numeric provider session id. Reads therefore never enter or evict
// the process cache merely to refresh their own transport.
const CHANNEL_A_READ_HANDLE_CACHE_IDLE_TTL_MS = 5 * 60_000;
const CHANNEL_A_PROCESS_HANDLE_CACHE_IDLE_TTL_MS = 5 * 60_000;
const CHANNEL_A_HANDLE_CACHE_MAX_ENTRIES = 64;
type CachedReadHandle = {
  promise: Promise<EstablishedSandboxSession>;
  lastUsedAtMonotonicMs: number;
};
type CachedProcessHandle = {
  promise: Promise<EstablishedSandboxSession>;
  lastUsedAtMonotonicMs: number;
};
export type EstablishedHandleCacheKind = "read" | "process" | "none";
const establishedReadHandleCache = new Map<string, CachedReadHandle>();
const establishedProcessHandleCache = new Map<string, CachedProcessHandle>();

function establishedHandleCacheKey(
  workspaceId: string,
  sessionId: string,
  lease: LeaseSnapshot,
): string {
  return [workspaceId, sessionId, lease.leaseEpoch, lease.instanceId ?? ""].join("\u0000");
}

export function isChannelAHandleCacheEntryFresh(
  lastUsedAtMonotonicMs: number,
  nowMonotonicMs: number,
  idleTtlMs = CHANNEL_A_READ_HANDLE_CACHE_IDLE_TTL_MS,
): boolean {
  return nowMonotonicMs - lastUsedAtMonotonicMs < idleTtlMs;
}

export function isChannelAProcessHandleCacheEntryFresh(
  lastUsedAtMonotonicMs: number,
  nowMonotonicMs: number,
  idleTtlMs = CHANNEL_A_PROCESS_HANDLE_CACHE_IDLE_TTL_MS,
): boolean {
  return nowMonotonicMs - lastUsedAtMonotonicMs < idleTtlMs;
}

function pruneEstablishedReadHandleCache(nowMonotonicMs: number): void {
  for (const [key, entry] of establishedReadHandleCache) {
    if (!isChannelAHandleCacheEntryFresh(entry.lastUsedAtMonotonicMs, nowMonotonicMs)) {
      establishedReadHandleCache.delete(key);
    }
  }
}

function pruneEstablishedProcessHandleCache(nowMonotonicMs: number): void {
  for (const [key, entry] of establishedProcessHandleCache) {
    if (!isChannelAProcessHandleCacheEntryFresh(entry.lastUsedAtMonotonicMs, nowMonotonicMs)) {
      establishedProcessHandleCache.delete(key);
    }
  }
}

function enforceEstablishedHandleCacheSize<T>(cache: Map<string, T>): void {
  while (cache.size > CHANNEL_A_HANDLE_CACHE_MAX_ENTRIES) {
    const oldestKey = cache.keys().next().value as string | undefined;
    if (oldestKey === undefined) break;
    cache.delete(oldestKey);
  }
}

async function establishCachedReadHandle(
  key: string,
  establish: () => Promise<EstablishedSandboxSession>,
): Promise<EstablishedSandboxSession> {
  const now = performance.now();
  pruneEstablishedReadHandleCache(now);
  const cached = establishedReadHandleCache.get(key);
  if (cached) {
    cached.lastUsedAtMonotonicMs = now;
    // Refresh insertion order so the bounded map evicts the least-recently used
    // exact lease identity first.
    establishedReadHandleCache.delete(key);
    establishedReadHandleCache.set(key, cached);
    return await cached.promise;
  }

  const promise = establish();
  const entry: CachedReadHandle = { promise, lastUsedAtMonotonicMs: now };
  establishedReadHandleCache.set(key, entry);
  enforceEstablishedHandleCacheSize(establishedReadHandleCache);
  try {
    return await promise;
  } catch (error) {
    if (establishedReadHandleCache.get(key) === entry) establishedReadHandleCache.delete(key);
    throw error;
  }
}

async function establishCachedProcessHandle(
  key: string,
  establish: () => Promise<EstablishedSandboxSession>,
): Promise<EstablishedSandboxSession> {
  const now = performance.now();
  pruneEstablishedProcessHandleCache(now);
  const cached = establishedProcessHandleCache.get(key);
  if (cached) {
    cached.lastUsedAtMonotonicMs = now;
    establishedProcessHandleCache.delete(key);
    establishedProcessHandleCache.set(key, cached);
    return await cached.promise;
  }

  const promise = establish();
  const entry: CachedProcessHandle = { promise, lastUsedAtMonotonicMs: now };
  establishedProcessHandleCache.set(key, entry);
  enforceEstablishedHandleCacheSize(establishedProcessHandleCache);
  try {
    return await promise;
  } catch (error) {
    if (establishedProcessHandleCache.get(key) === entry) {
      establishedProcessHandleCache.delete(key);
    }
    throw error;
  }
}

/** Reuse the exact lease-fenced provider handle across API-direct surfaces.
 * Stream capability negotiation and the first Files/Changes reads commonly run
 * back-to-back; sharing this handle avoids paying the same Modal resume twice. */
export async function establishCachedChannelAHandle(
  workspaceId: string,
  sessionId: string,
  lease: LeaseSnapshot,
  establish: () => Promise<EstablishedSandboxSession>,
): Promise<EstablishedSandboxSession> {
  return await establishCachedReadHandle(
    establishedHandleCacheKey(workspaceId, sessionId, lease),
    establish,
  );
}

async function establishCachedChannelAProcessHandle(
  workspaceId: string,
  sessionId: string,
  lease: LeaseSnapshot,
  establish: () => Promise<EstablishedSandboxSession>,
): Promise<EstablishedSandboxSession> {
  return await establishCachedProcessHandle(
    establishedHandleCacheKey(workspaceId, sessionId, lease),
    establish,
  );
}

/**
 * Run independent, side-effect-free Channel-A reads concurrently without
 * releasing the direct-request holder while sibling provider commands are
 * still settling. A typed temporary-unavailable failure is retried exactly
 * once after every first attempt has settled; validation, conflict, not-found,
 * and unknown failures are never replayed.
 */
export async function runConcurrentChannelAReads<T>(
  operations: readonly (() => Promise<T>)[],
): Promise<T[]> {
  const values = new Array<T>(operations.length);
  const first = await Promise.allSettled(
    operations.map((operation) => Promise.resolve().then(operation)),
  );
  const retryIndexes: number[] = [];

  for (const [index, result] of first.entries()) {
    if (result.status === "fulfilled") {
      values[index] = result.value;
      continue;
    }
    if (!(result.reason instanceof ChannelAUnavailableError)) {
      throw result.reason;
    }
    retryIndexes.push(index);
  }

  if (retryIndexes.length === 0) return values;

  const retried = await Promise.allSettled(
    retryIndexes.map((index) => Promise.resolve().then(operations[index]!)),
  );
  for (const [retryIndex, result] of retried.entries()) {
    if (result.status === "rejected") throw result.reason;
    values[retryIndexes[retryIndex]!] = result.value;
  }
  return values;
}

type ChannelAReadRecoveryOptions = {
  /** Modal may expose one more stale command-router route after the first
   * successful handle rebuild. Keep this closed and statically bounded. */
  maxFreshHandleRetries?: 1 | 2;
  /** Never start another provider attempt after the originating request ends. */
  waitSignal?: AbortSignal | undefined;
  /** Additional callback-specific failure that is safe to replay. */
  retryableError?: ((error: unknown) => boolean) | undefined;
};

/** Retry a side-effect-free Channel-A read only after the caller has discarded
 * and freshly re-established its provider handle. The ordinary provider-neutral
 * contract allows one retry; Modal opts into one additional rebuild because a
 * command-router rollover can outlive the first replacement handle. Provider
 * commands are never replayed for validation/conflict/unknown errors, mutation
 * routes never call this helper, and request cancellation stops recovery before
 * another provider command begins. */
export async function runChannelAReadWithFreshHandleRetry<T>(
  run: () => Promise<T>,
  refreshHandle: (attempt: 1 | 2) => Promise<void>,
  options: ChannelAReadRecoveryOptions = {},
): Promise<T> {
  const maxFreshHandleRetries = options.maxFreshHandleRetries ?? 1;
  for (let retries = 0; ; retries += 1) {
    options.waitSignal?.throwIfAborted();
    try {
      return await run();
    } catch (error) {
      const retryable =
        error instanceof ChannelAUnavailableError || options.retryableError?.(error) === true;
      if (!retryable || retries >= maxFreshHandleRetries) {
        throw error;
      }
      options.waitSignal?.throwIfAborted();
      const attempt = retries === 0 ? 1 : 2;
      await refreshHandle(attempt);
    }
  }
}

export function shouldEvictChannelAHandleAfterError(
  error: unknown,
  cacheKind: EstablishedHandleCacheKind,
): boolean {
  return (
    cacheKind === "read" &&
    (error instanceof ChannelAUnavailableError || isRetryableControllerTransport(error))
  );
}

function isRetryableControllerTransport(error: unknown): boolean {
  return (
    error instanceof BrowserControlTransportError ||
    (error instanceof BrowserControlRequestError && error.retryable)
  );
}

function evictEstablishedHandle(key: string, cacheKind: EstablishedHandleCacheKind): void {
  if (cacheKind === "read") establishedReadHandleCache.delete(key);
  if (cacheKind === "process") establishedProcessHandleCache.delete(key);
}

function evictAllEstablishedHandles(key: string): void {
  establishedReadHandleCache.delete(key);
  establishedProcessHandleCache.delete(key);
}

function rememberEstablishedHandle(
  key: string,
  established: EstablishedSandboxSession,
  cacheKind: Exclude<EstablishedHandleCacheKind, "none">,
): void {
  const now = performance.now();
  if (cacheKind === "read") {
    pruneEstablishedReadHandleCache(now);
    establishedReadHandleCache.delete(key);
    establishedReadHandleCache.set(key, {
      promise: Promise.resolve(established),
      lastUsedAtMonotonicMs: now,
    });
    enforceEstablishedHandleCacheSize(establishedReadHandleCache);
    return;
  }
  pruneEstablishedProcessHandleCache(now);
  establishedProcessHandleCache.delete(key);
  establishedProcessHandleCache.set(key, {
    promise: Promise.resolve(established),
    lastUsedAtMonotonicMs: now,
  });
  enforceEstablishedHandleCacheSize(establishedProcessHandleCache);
}

/**
 * Run a Channel-A op against a live box, API-direct. Acquires an exact direct holder
 * (warming the box when cold), resumes by id, builds the service, runs `fn`, and
 * ALWAYS releases the holder + drops the handle in `finally`. Maps the service's
 * typed errors to HTTP status (the route never sees a raw ChannelA*Error).
 *
 * Gated behind sandboxOwnershipEnabled at the route (the lease is dormant
 * otherwise). A `backend:none` session has no box -> 409 before touching it.
 */
export async function withChannelA<T>(
  services: ChannelAServices,
  ctx: ChannelAContext,
  fn: (handle: ChannelAHandle) => Promise<T>,
): Promise<T> {
  return await withChannelAOperation(services, ctx, false, fn);
}

/** Read-only API-direct seam. Separate requests for the same exact live Modal
 * instance are serialized across API replicas; each request's batched reads
 * remain concurrent behind that one distributed boundary. A typed temporary
 * provider-channel failure gets one retry only after rebuilding the exact
 * lease-fenced handle. */
export async function withChannelARead<T>(
  services: ChannelAServices,
  ctx: ChannelAContext,
  fn: (handle: ChannelAHandle) => Promise<T>,
): Promise<T> {
  return await withChannelAOperation(services, ctx, true, fn);
}

async function withChannelAOperation<T>(
  services: ChannelAServices,
  ctx: ChannelAContext,
  readOnly: boolean,
  fn: (handle: ChannelAHandle) => Promise<T>,
): Promise<T> {
  const { db, settings, bus } = services;
  const onSandboxOperation = services.observability
    ? sandboxOperationMetricObserver(services.observability)
    : undefined;
  const { accountId, workspaceId, session } = ctx;

  if (session.sandboxBackend === "none") {
    throw new HTTPException(409, { message: "sandbox not available" });
  }

  const sandboxGroupId = session.sandboxGroupId;
  const requestId = crypto.randomUUID();
  const holderId = `direct:${requestId}`;
  const operationStartedAt = performance.now();
  const operation = ctx.operation ?? (readOnly ? "read" : "mutation");
  const leaseTtlMs = settings.sandboxLeaseTtlMs;

  // The STABLE run-environment used by both a cloud home and a machine home.
  // It also carries the per-session Codemode pointer selected below.
  const workspaceEnvironmentValues: Record<string, string> = {};
  const rigVersion =
    session.rigId && session.rigVersionId
      ? await getScheduledScopedRigVersionMetadata(
          db,
          {
            accountId,
            workspaceId,
            subjectId: ctx.subjectId ?? "session-attach",
          },
          session.rigId,
          session.rigVersionId,
        )
      : null;
  for (const variableSetId of rigVersion?.version.defaultVariableSetIds ?? []) {
    const workspaceEnvironment = await loadWorkspaceEnvironmentForRun(db, settings, {
      accountId,
      workspaceId,
      variableSetId,
      authority: { kind: "session_attach", sessionId: session.id, subjectId: ctx.subjectId },
    });
    Object.assign(workspaceEnvironmentValues, workspaceEnvironment?.values ?? {});
  }
  for (const variableSetId of session.variableSetIds) {
    const workspaceEnvironment = await loadWorkspaceEnvironmentForRun(db, settings, {
      accountId,
      workspaceId,
      variableSetId,
      authority: { kind: "session_attach", sessionId: session.id, subjectId: ctx.subjectId },
    });
    Object.assign(workspaceEnvironmentValues, workspaceEnvironment?.values ?? {});
  }
  const settingsForSession =
    session.sandboxBackend !== settings.sandboxBackend
      ? { ...settings, sandboxBackend: session.sandboxBackend }
      : settings;
  const environment = stableSandboxEnvironmentForRun(
    settingsForSession,
    workspaceEnvironmentValues,
    { workspaceId },
  );
  if (hasGitCredentialRepositorySelection(session.resources)) {
    applyGitAuthPointerEnvironment(
      environment,
      hasGitHubRepositorySelection(session.resources) ? githubAppBotIdentity(settings) : null,
    );
  }

  const runEstablished = async (
    routed: EstablishedSandboxSession,
    lease: LeaseSnapshot | null,
    homeSession: ChannelASession,
  ): Promise<T> => {
    const emit = async (events: { type: string; payload: unknown }[]): Promise<void> => {
      await appendAndPublishEvents(
        db,
        bus,
        workspaceId,
        session.id,
        events.map((e) => ({ type: e.type as never, payload: e.payload })),
      );
    };
    const routingSession = routed.session as RoutingSandboxSession;
    const credentialSession = withRunCredentialsSession(routingSession as object, session.id);
    const scopedSession = environment.OPENGENI_CODEMODE_TOKEN_FILE
      ? withCodemodeTokenSession(
          credentialSession,
          codemodeTokenFileFromEnvironment(environment, session.id),
        )
      : credentialSession;
    const fileSystemAuthority = await routingSession.fileSystemAuthority();
    const fileSystemEpoch =
      fileSystemAuthority.backendKind === "selfhosted"
        ? fileSystemAuthority.activeEpoch
        : (lease?.leaseEpoch ?? fileSystemAuthority.activeEpoch);
    const service = new SandboxChannelAService({
      session: scopedSession as ChannelASession,
      workspaceRoot: fileSystemAuthority.root,
      leaseEpoch: fileSystemEpoch,
      ...(fileSystemAuthority.backendKind === "selfhosted"
        ? { providerPathMode: "workspace-relative" as const, fileReadScope: "machine" as const }
        : {}),
      emit,
    });
    const result = await fn({
      service,
      lease,
      homeSession,
      routingSession,
      requestId,
    });
    // The direct request has accepted the result in memory. Finalize every
    // Connected Machine backend the routing proxy reached so a mid-request
    // route transition cannot leave completed output retained until TTL.
    await routingSession.finalizeOpStreamOps().catch(() => undefined);
    return result;
  };

  // A machine-targeted top-level session has an honest selfhosted HOME label.
  // It has no cloud provider box and therefore must not acquire or establish a
  // phantom home lease before following its active machine pointer.
  if (session.sandboxBackend === "selfhosted") {
    let established: EstablishedSandboxSession | undefined;
    try {
      ctx.waitSignal?.throwIfAborted();
      const pointer = await readActiveSandbox(db, workspaceId, session.id);
      if (!pointer?.activeSandboxId) {
        // A machine-home session with no selected machine uses the deployment's
        // managed group box, exactly like the worker turn path. Keep the durable
        // home label honest; only this request's effective backend changes.
        const groupBackend = managedSessionGroupBackend(
          settings.sandboxBackend,
          session.sandboxBackend,
        );
        if (groupBackend) {
          return await withChannelAOperation(
            services,
            {
              ...ctx,
              session: {
                ...session,
                sandboxBackend: groupBackend,
                sandboxOs: managedSessionGroupOs(session.sandboxBackend, session.sandboxOs),
              },
            },
            readOnly,
            fn,
          );
        }
        throw new HTTPException(409, {
          message: "machine-home session has no active Connected Machine or managed sandbox",
        });
      }
      const sandbox = await getSandbox(
        db,
        { accountId: ctx.accountId, workspaceId, subjectId: ctx.subjectId },
        pointer.activeSandboxId,
      );
      if (sandbox?.kind !== "selfhosted" || !sandbox.enrollmentId) {
        throw new HTTPException(409, {
          message: "machine-home session points to an unavailable Connected Machine",
        });
      }
      const originWorkspaceId = sandbox.workspaceId;
      const enrollment = await getLiveEnrollmentConnection(
        db,
        { accountId, workspaceId, subjectId: ctx.subjectId },
        sandbox.enrollmentId,
      );
      if (!enrollment?.connectionInstanceId) {
        // Preserve causal machine liveness through Channel-A. Generic callers
        // retain the established 409 mapping below; Browser/Computer callers can
        // project the same bounded `agent_offline` contract as control RPC.
        throw agentErrorToControlError(
          offlineAgentError("Connected Machine has no live runner connection", true),
        );
      }
      if (!enrollment.workspaceRoot) {
        throw new HTTPException(409, {
          message:
            "Connected Machine has not reported an absolute workspace root; reconnect it with a current agent",
        });
      }
      const built = await buildSelfhostedBackendSession({
        workspaceId: originWorkspaceId,
        agentId: sandbox.enrollmentId,
        connectionInstanceId: enrollment.connectionInstanceId,
        workspaceRoot: resolveConnectedMachineWorkspaceRoot(
          enrollment.workspaceRoot,
          pointer.workingDir,
        ),
        relay: relayConfigFromSettings(settings),
        controlRpcFactory: () => new NatsControlRpc(async () => bus.getRequestConnection()),
        epoch: pointer.activeEpoch,
        environment,
        timeoutMs: settings.sandboxSelfhostedControlTimeoutMs,
        execTimeoutMs: settings.sandboxSelfhostedExecTimeoutMs,
        operationResourcePolicy: enrollment.operationPolicy,
        operationResourcePolicySupported:
          enrollment.agentCapabilities.operationResourcePolicy === true,
        operationCpuQuotaSupported: enrollment.agentCapabilities.operationCpuQuota === true,
        transactionalFsWriteSupported: enrollment.agentCapabilities.transactionalFsWrite === true,
        ...(settings.agentOpStreamEnabled === true &&
        enrollment?.opStream === true &&
        bus.getOpStreamConnection
          ? {
              opStream: {
                transport: new NatsOpStreamTransport(
                  async () => bus.getOpStreamConnection?.() ?? null,
                ),
              },
            }
          : {}),
      });
      established = {
        client: built.client,
        session: built.session,
        sessionState: { agentId: sandbox.enrollmentId },
        instanceId: sandbox.enrollmentId,
        backendId: "selfhosted",
      };
      const routed = wrapChannelABoxWithRouting(
        {
          db,
          settings,
          bus,
          ...(onSandboxOperation ? { onSandboxOperation } : {}),
          ...(ctx.waitSignal ? { waitSignal: ctx.waitSignal } : {}),
        },
        {
          accountId,
          workspaceId,
          sessionId: session.id,
          resourceSubjectId: ctx.subjectId,
          pinnedSelfhosted: {
            sandboxId: sandbox.id,
            epoch: pointer.activeEpoch,
          },
          directRequest: { requestId, holderId },
        },
        established,
      );
      return await runEstablished(routed, null, established.session as ChannelASession);
    } catch (error) {
      observeChannelAOperationFailure(services, {
        workspaceId,
        sandboxGroupId,
        backend: session.sandboxBackend,
        operation,
        durationMs: performance.now() - operationStartedAt,
        error,
        waitSignal: ctx.waitSignal,
      });
      throw mapChannelAError(error, ctx.waitSignal);
    } finally {
      await dropEstablishedHandle(established);
    }
  }

  // One session has one logical runtime across turns and every API-direct
  // surface. Without this, Terminal/Files/Browser/Computer/viewers could rearm
  // a stale deployment image after the worker had resolved a newer Pack/Rig or
  // deployment image for the same durable sandbox group.
  const sandboxRuntime = await resolveSessionSandboxRuntime(db, settings, session);

  const release = async (): Promise<void> => {
    await releaseLeaseHolder(db, {
      accountId,
      workspaceId,
      sandboxGroupId,
      kind: "direct",
      holderId,
      idleGraceMs: settings.sandboxIdleGraceMs,
    });
  };

  // Acquire exact request authority; the cold->warming CAS spawns the box when
  // cold. This wait is request-abort aware and must pass through the same typed
  // cancellation/diagnostic seam as provider execution below.
  let acquired: Awaited<ReturnType<typeof acquireLease>>;
  let acquisitionMayHaveCommitted = false;
  try {
    ctx.waitSignal?.throwIfAborted();
    acquisitionMayHaveCommitted = true;
    acquired = await acquireLease(db, {
      accountId,
      workspaceId,
      sandboxGroupId,
      kind: "direct",
      holderId,
      subjectId: session.id,
      backend: session.sandboxBackend,
      os: session.sandboxOs,
      image: sandboxRuntime.image,
      rigVersionId: session.rigVersionId,
      leaseTtlMs,
      warmingLeaseTtlMs: settings.sandboxWarmingTimeoutMs,
      captureWaitMs: sandboxLifecycleTransitionWaitMs(settings),
      ...(ctx.waitSignal ? { waitSignal: ctx.waitSignal } : {}),
    });
    // Close the commit/abort race before any provider handle is established.
    // The catch below drops an exact holder committed just before disconnect.
    ctx.waitSignal?.throwIfAborted();

    if (acquired.role === "blocked") {
      throw new HTTPException(409, {
        message: `sandbox recovery ${acquired.lease.recovery.restore.status} at epoch ${acquired.lease.leaseEpoch}`,
      });
    }
    if (acquired.role === "fenced") {
      throw new HTTPException(409, {
        message:
          acquired.reason === "superseded"
            ? `sandbox lease superseded (epoch ${acquired.lease.leaseEpoch}); retry`
            : `sandbox lifecycle transition in progress (${acquired.reason}, epoch ${acquired.lease.leaseEpoch}, backend ${acquired.lease.backend}, instance ${acquired.lease.instanceId ?? "none"}); retry`,
      });
    }
  } catch (error) {
    let mappedError: unknown = error;
    if (error instanceof SandboxImageConflictError) {
      // A durable Browser/Computer holder can outlive its provider container.
      // Runtime drift is checked before a new direct holder is admitted, so a
      // dead old container would otherwise strand the lease behind a permanent
      // conflict. Probe the exact old identity without replacement; only an
      // authoritative provider-missing result may retire it and unblock the
      // next request's cold successor election.
      const live = await readLease(db, workspaceId, sandboxGroupId).catch(() => null);
      if (live?.liveness === "warm" && live.instanceId !== null && live.resumeState !== null) {
        let probe: EstablishedSandboxSession | undefined;
        try {
          probe = await establishSandboxSessionFromEnvelope(
            sandboxRuntime.settings,
            live.resumeState,
            {
              sessionId: session.id,
              recovery: "resume-only",
              backendOverride: session.sandboxBackend,
              environment,
            },
          );
        } catch (probeError) {
          if (isProviderSandboxNotFoundError(session.sandboxBackend, probeError)) {
            const marked = await markWarmLeaseInstanceLost(db, {
              accountId,
              workspaceId,
              sandboxGroupId,
              expectedEpoch: live.leaseEpoch,
              expectedInstanceId: live.instanceId,
            }).catch(() => null);
            if (marked?.status === "marked") {
              await appendAndPublishEvents(db, bus, workspaceId, session.id, [
                {
                  type: "sandbox.box.lost",
                  payload: { sandboxId: live.instanceId },
                },
              ]).catch(() => undefined);
              mappedError = new ApiHttpError(409, {
                code: "conflict",
                message: "sandbox instance was lost; retry to restore it",
                retryable: true,
              });
            }
          }
        } finally {
          await dropEstablishedHandle(probe);
        }
      }
    }
    // Release is idempotent. If acquisition committed before the request was
    // cancelled, this removes that exact direct holder; a transient DB failure
    // must not overwrite the original structural error (holder TTL is the final
    // cleanup fence).
    if (acquisitionMayHaveCommitted) await release().catch(() => undefined);
    observeChannelAOperationFailure(services, {
      workspaceId,
      sandboxGroupId,
      backend: session.sandboxBackend,
      operation,
      durationMs: performance.now() - operationStartedAt,
      error: mappedError,
      waitSignal: ctx.waitSignal,
    });
    throw mapChannelAError(mappedError, ctx.waitSignal);
  }

  // Keep the exact direct-request owner visible for the full operation. A
  // yielded command is also tracked by a non-TTL process holder, but the
  // retained-process reconciler must not poll that provider session while this
  // request is still its active owner. Stale direct holders remain crash-
  // recoverable through the ordinary holder TTL reaper.
  const directHolderHeartbeat = setInterval(() => {
    void touchLeaseHolder(db, {
      accountId,
      workspaceId,
      sandboxGroupId,
      kind: "direct",
      holderId,
    }).catch(() => undefined);
  }, 10_000);
  directHolderHeartbeat.unref?.();

  let established: EstablishedSandboxSession | undefined;
  let leaseSnapshot: LeaseSnapshot = acquired.lease;
  let establishedCacheKey: string | null = null;
  let establishedCacheKind: EstablishedHandleCacheKind = "none";
  const requestedCacheKind: Exclude<EstablishedHandleCacheKind, "none"> = readOnly
    ? "read"
    : "process";

  const establishAttachedLiveHandle = async (
    live: LeaseSnapshot,
    cacheKind: EstablishedHandleCacheKind,
  ): Promise<{ established: EstablishedSandboxSession; cacheKey: string }> => {
    const cacheKey = establishedHandleCacheKey(workspaceId, session.id, live);
    const establish = () =>
      establishSandboxSessionFromEnvelope(sandboxRuntime.settings, live.resumeState, {
        sessionId: session.id,
        recovery: "resume-only",
        backendOverride: session.sandboxBackend,
        environment,
      });
    try {
      const attached =
        cacheKind === "read"
          ? await establishCachedChannelAHandle(workspaceId, session.id, live, establish)
          : cacheKind === "process"
            ? await establishCachedChannelAProcessHandle(workspaceId, session.id, live, establish)
            : await establish();
      return { established: attached, cacheKey };
    } catch (error) {
      if (!isProviderSandboxNotFoundError(session.sandboxBackend, error)) throw error;
      // The exact provider instance is definitively gone, so neither a read nor
      // a process wrapper for that lease identity may survive locally.
      evictAllEstablishedHandles(cacheKey);
      const marked = await markWarmLeaseInstanceLost(db, {
        accountId,
        workspaceId,
        sandboxGroupId,
        expectedEpoch: live.leaseEpoch,
        expectedInstanceId: live.instanceId!,
      });
      if (marked.status === "marked") {
        await appendAndPublishEvents(db, bus, workspaceId, session.id, [
          {
            type: "sandbox.box.lost",
            payload: { sandboxId: live.instanceId },
          },
        ]);
      }
      throw new ApiHttpError(409, {
        code: "conflict",
        message: `sandbox instance was lost; retry to restore it`,
        retryable: true,
      });
    }
  };

  try {
    const envelope = await getSandboxSessionEnvelope(db, workspaceId, session.id);
    if (acquired.role === "spawner") {
      // We won the cold->warming CAS: establish the box from the envelope, then
      // commit warm. The established handle IS our live handle for the op.
      const expectedEpoch = acquired.lease.leaseEpoch;
      // Prefer the COLD lease's preserved resume_state when it carries a persisted
      // /workspace snapshot (confirmDrainCold keeps a minimal archive-only envelope
      // across draining->cold for exactly this re-warm). establishSandboxSessionFromEnvelope
      // cold-creates a fresh box and replays the archive via hydrateWorkspace, so
      // /workspace survives the box churn (sandbox-file-persistence). No archive ->
      // the bare session envelope (a never-warmed cold start). The order matters:
      // resume_state is the lease's authoritative box descriptor; the session
      // `_sandbox` envelope is only the per-session fallback.
      try {
        const providerSettings = await providerSettingsForSessionSandboxRuntime(
          sandboxRuntime,
          session.sandboxBackend,
        );
        const result = await establishApiSandboxSpawner({
          db,
          settings: providerSettings,
          accountId,
          workspaceId,
          sandboxGroupId,
          sessionId: session.id,
          backend: session.sandboxBackend,
          environment,
          expectedEpoch,
          acquiredLease: acquired.lease,
          fallbackEnvelope: envelope,
          dataPlaneUrl: acquired.lease.dataPlaneUrl,
          ...(services.objectStorage !== undefined
            ? { objectStorage: services.objectStorage }
            : {}),
        });
        established = result.established;
        leaseSnapshot = result.lease;
        establishedCacheKey = establishedHandleCacheKey(workspaceId, session.id, leaseSnapshot);
        establishedCacheKind = requestedCacheKind;
        rememberEstablishedHandle(establishedCacheKey, established, establishedCacheKind);
      } catch (error) {
        throw new HTTPException(409, {
          message: `sandbox not available (${error instanceof Error ? error.message : "spawn failed"})`,
        });
      }
    } else {
      // ATTACHED / REARMED: the box is live. Read the lease to get the
      // authoritative resume_state, then resume by id for this op.
      const live = await readLease(db, workspaceId, sandboxGroupId);
      if (
        !live ||
        live.liveness !== "warm" ||
        live.leaseEpoch !== acquired.lease.leaseEpoch ||
        live.instanceId === null
      ) {
        throw new HTTPException(409, {
          message: `sandbox lease is not attachable; retry`,
        });
      }
      leaseSnapshot = live;
      const attached = await establishAttachedLiveHandle(live, requestedCacheKind);
      established = attached.established;
      establishedCacheKey = attached.cacheKey;
      establishedCacheKind = requestedCacheKind;
    }

    const runProviderOperation = async (): Promise<T> => {
      // Route every call through the same proxy, even when hot-swap is disabled:
      // routing may be dormant, but its direct mutation admission is mandatory for
      // every persistable provider write.
      const routed = wrapChannelABoxWithRouting(
        {
          db,
          settings,
          bus,
          ...(onSandboxOperation ? { onSandboxOperation } : {}),
          ...(ctx.waitSignal ? { waitSignal: ctx.waitSignal } : {}),
        },
        {
          accountId,
          workspaceId,
          sessionId: session.id,
          resourceSubjectId: ctx.subjectId,
          homeLease: {
            sandboxGroupId,
            leaseEpoch: leaseSnapshot.leaseEpoch,
            instanceId: leaseSnapshot.instanceId!,
            backend: session.sandboxBackend,
          },
          directRequest: { requestId, holderId },
        },
        established!,
      );
      const run = async () =>
        await runEstablished(routed, leaseSnapshot, established!.session as ChannelASession);
      return readOnly && session.sandboxBackend === "modal"
        ? await withSandboxProviderReadLock(
            db,
            {
              workspaceId,
              sandboxGroupId,
              leaseEpoch: leaseSnapshot.leaseEpoch,
              instanceId: leaseSnapshot.instanceId!,
            },
            ctx.waitSignal,
            run,
          )
        : await run();
    };

    // A failed attempt leaves the advisory-lock transaction before refresh;
    // the retry acquires a new transaction/lock against the same fenced lease.
    return readOnly
      ? await runChannelAReadWithFreshHandleRetry(
          runProviderOperation,
          async (attempt) => {
            const refreshStartedAt = performance.now();
            const observeRefresh = (outcome: "ok" | "failed"): void => {
              if (!services.observability) return;
              const attributes = {
                sandboxLeaseKey: sandboxLeaseTelemetryKey(workspaceId, sandboxGroupId),
                backend: session.sandboxBackend,
                reason: "provider_handle_unavailable",
                outcome,
                attempt,
                durationMs: Math.max(0, Math.round(performance.now() - refreshStartedAt)),
              };
              try {
                services.observability.incrementCounter({
                  name: "opengeni_channel_a_handle_refresh_total",
                  help: "Channel-A provider handles rebuilt after a typed temporary-unavailable read.",
                  labels: { backend: session.sandboxBackend, outcome },
                });
              } catch {
                // Metrics can never alter lease or provider authority.
              }
              try {
                if (outcome === "ok") {
                  services.observability.info(
                    "Channel-A provider handle refresh completed",
                    attributes,
                  );
                } else {
                  services.observability.warn(
                    "Channel-A provider handle refresh failed",
                    attributes,
                  );
                }
              } catch {
                // Logs can never alter lease or provider authority.
              }
            };
            try {
              if (establishedCacheKey) {
                evictEstablishedHandle(establishedCacheKey, establishedCacheKind);
              }
              await dropEstablishedHandle(established);
              // This request still owns its direct holder, so the exact live identity
              // should be stable. Revalidate it before rebuilding the provider handle.
              const live = await readLease(db, workspaceId, sandboxGroupId);
              if (
                !live ||
                live.liveness !== "warm" ||
                live.leaseEpoch !== leaseSnapshot.leaseEpoch ||
                live.instanceId !== leaseSnapshot.instanceId
              ) {
                throw new HTTPException(409, {
                  message: `sandbox lease changed while refreshing its provider handle; retry`,
                });
              }
              const refreshed = await establishAttachedLiveHandle(live, "none");
              established = refreshed.established;
              establishedCacheKey = refreshed.cacheKey;
              establishedCacheKind = "read";
              leaseSnapshot = live;
              rememberEstablishedHandle(refreshed.cacheKey, refreshed.established, "read");
              observeRefresh("ok");
            } catch (error) {
              observeRefresh("failed");
              throw error;
            }
          },
          {
            maxFreshHandleRetries: session.sandboxBackend === "modal" ? 2 : 1,
            ...(ctx.waitSignal ? { waitSignal: ctx.waitSignal } : {}),
            ...(ctx.retryControllerTransport
              ? { retryableError: isRetryableControllerTransport }
              : {}),
          },
        )
      : await runProviderOperation();
  } catch (error) {
    // A read wrapper carries no yielded process state and is safe to discard.
    // A mutation/terminal wrapper may own the SDK's only local process object;
    // retain it after an ambiguous transport failure so a later control call
    // can use the in-place provider transport recovery without losing the PTY.
    if (establishedCacheKey && shouldEvictChannelAHandleAfterError(error, establishedCacheKind)) {
      evictEstablishedHandle(establishedCacheKey, establishedCacheKind);
    }
    observeChannelAOperationFailure(services, {
      workspaceId,
      sandboxGroupId,
      backend: session.sandboxBackend,
      operation,
      durationMs: performance.now() - operationStartedAt,
      error,
      waitSignal: ctx.waitSignal,
    });
    throw mapChannelAError(error, ctx.waitSignal);
  } finally {
    clearInterval(directHolderHeartbeat);
    await release();
    await dropEstablishedHandle(established);
  }
}

/** Map the service's typed errors to HTTP status (the §5.3 matrix). Re-throws an
 *  already-HTTPException unchanged. */
export function mapChannelAError(error: unknown, waitSignal?: AbortSignal): unknown {
  if (error instanceof HTTPException) return error;
  if (isChannelARequestCancellation(error, waitSignal))
    return new HTTPException(499 as never, {
      message: "request cancelled",
      cause: error,
    });
  if (
    error instanceof SandboxResumeIdentityMismatchError ||
    error instanceof SandboxResumeIdentityUnavailableError
  )
    return new HTTPException(409, { message: error.message });
  if (
    error instanceof RoutingActiveRouteChangedError ||
    error instanceof RoutingWorkspaceRootChangedError ||
    error instanceof SelfhostedWorkspaceRootChangedError ||
    error instanceof ChannelAFileSystemRouteChangedError
  )
    return new ApiHttpError(409, {
      code: "conflict",
      message: error.message,
      retryable: true,
    });
  if (error instanceof SandboxProviderReadLockUnavailableError)
    return new HTTPException(503, { message: error.message });
  if (error instanceof SandboxImageConflictError || error instanceof SandboxRigConflictError)
    return new ApiHttpError(409, {
      code: "conflict",
      message: "sandbox runtime changed while this session still has active operations; retry",
      retryable: true,
    });
  if (error instanceof SelfhostedControlError && error.agentOffline)
    return new HTTPException(409, { message: error.message, cause: error });
  if (error instanceof ChannelAUnavailableError)
    return new HTTPException(503, { message: error.message });
  if (error instanceof ChannelAValidationError)
    return new HTTPException(400, { message: error.message });
  if (error instanceof ChannelANotFoundError)
    return new HTTPException(404, { message: error.message });
  if (error instanceof ChannelAConflictError)
    return new HTTPException(409, { message: error.message });
  if (error instanceof ChannelAUnsupportedError)
    return new HTTPException(409, { message: error.message });
  const authorityFence = workspaceMutationAuthorityFenceCode(error);
  // A revoked grant is an authorization outcome, not a server fault; an
  // unattributed pre-0277 writer is a conflict the caller resolves by starting
  // fresh work. Both must be visible instead of surfacing as a 500.
  if (authorityFence === "authority_revoked")
    return new HTTPException(403, { message: (error as Error).message });
  if (authorityFence === "authority_unattributed")
    return new HTTPException(409, { message: (error as Error).message });
  return error;
}

/** Structural check: the fence type lives in `@opengeni/db` and is raised deep
 * inside admission, so match it the same way the worker does. */
function workspaceMutationAuthorityFenceCode(error: unknown): string | null {
  if (
    !(error instanceof Error) ||
    (error.name !== "SandboxWorkspaceMutationFencedError" &&
      error.name !== "SandboxRetainedProcessPromotionFencedError")
  ) {
    return null;
  }
  const code = (error as Error & { code?: unknown }).code;
  return typeof code === "string" ? code : null;
}

export function isChannelARequestCancellation(error: unknown, waitSignal?: AbortSignal): boolean {
  if (waitSignal?.aborted !== true) return false;
  const isSignalReason = waitSignal.reason !== undefined && error === waitSignal.reason;
  const isAbortError =
    (error instanceof DOMException && error.name === "AbortError") ||
    (error instanceof Error && error.name === "AbortError");
  return isSignalReason || isAbortError;
}

/** Structural classification only: exact provider exception text, codes, URLs,
 * and identifiers never cross the telemetry boundary. */
export function channelAOperationFailureDiagnostic(
  error: unknown,
  waitSignal?: AbortSignal,
): ChannelAOperationFailureDiagnostic {
  if (isChannelARequestCancellation(error, waitSignal)) {
    return {
      reason: "request_cancelled",
      status: 499,
      errorCode: "sandbox_channel_a_cancelled",
    };
  }
  if (error instanceof SandboxProviderReadLockUnavailableError) {
    return {
      reason: "provider_read_busy",
      status: 503,
      errorCode: "sandbox_channel_a_provider_busy",
    };
  }
  if (error instanceof SandboxImageConflictError || error instanceof SandboxRigConflictError) {
    return {
      reason: "lifecycle_conflict",
      status: 409,
      errorCode: "sandbox_channel_a_lifecycle_conflict",
    };
  }
  if (error instanceof ChannelAUnavailableError || error instanceof BrowserControlTransportError) {
    return {
      reason: "provider_unavailable",
      status: 503,
      errorCode: "sandbox_channel_a_provider_unavailable",
    };
  }
  if (
    error instanceof SandboxResumeIdentityMismatchError ||
    error instanceof SandboxResumeIdentityUnavailableError ||
    error instanceof RoutingActiveRouteChangedError ||
    error instanceof RoutingWorkspaceRootChangedError ||
    error instanceof SelfhostedWorkspaceRootChangedError ||
    error instanceof ChannelAFileSystemRouteChangedError ||
    (error instanceof HTTPException && error.status === 409)
  ) {
    return {
      reason: "lifecycle_conflict",
      status: 409,
      errorCode: "sandbox_channel_a_lifecycle_conflict",
    };
  }
  if (
    error instanceof ChannelAValidationError ||
    error instanceof ChannelANotFoundError ||
    error instanceof ChannelAConflictError ||
    error instanceof ChannelAUnsupportedError
  ) {
    const mapped = mapChannelAError(error, waitSignal);
    return {
      reason: "request_rejected",
      status: mapped instanceof HTTPException ? mapped.status : 500,
      errorCode: "sandbox_channel_a_operation_failed",
    };
  }
  if (error instanceof HTTPException) {
    return {
      reason: error.status >= 500 ? "unexpected" : "request_rejected",
      status: error.status,
      errorCode: "sandbox_channel_a_operation_failed",
    };
  }
  // An authority fence is a deliberate rejection, not a fault: keep it out of
  // the unexpected-failure signal that operators page on.
  const authorityFence = workspaceMutationAuthorityFenceCode(error);
  if (authorityFence === "authority_revoked" || authorityFence === "authority_unattributed") {
    return {
      reason: "request_rejected",
      status: authorityFence === "authority_revoked" ? 403 : 409,
      errorCode: "sandbox_channel_a_operation_failed",
    };
  }
  return {
    reason: "unexpected",
    status: 500,
    errorCode: "sandbox_channel_a_operation_failed",
  };
}

/** Exported for the telemetry contract tests; not a route surface. */
export function observeChannelAOperationFailure(
  services: ChannelAServices,
  input: {
    workspaceId: string;
    sandboxGroupId: string;
    backend: string;
    operation: string;
    durationMs: number;
    error: unknown;
    waitSignal?: AbortSignal | undefined;
  },
): void {
  if (!services.observability) return;
  // A writer with no recorded authority is a compatibility lane, not a fault:
  // count the lane itself so an operator can see whether it is still live,
  // separately from the structural failure signal operators page on below.
  // Lane name only - never the workspace, sandbox group, backend, or operation.
  if (workspaceMutationAuthorityFenceCode(input.error) === "authority_unattributed") {
    recordTenancyCompatibilityLaneUse(services.observability, "workspace_writer_unattributed");
  }
  const diagnostic = channelAOperationFailureDiagnostic(input.error, input.waitSignal);
  const attributes = {
    sandboxLeaseKey: sandboxLeaseTelemetryKey(input.workspaceId, input.sandboxGroupId),
    backend: input.backend,
    op: input.operation,
    outcome: "failed",
    reason: diagnostic.reason,
    status: diagnostic.status,
    durationMs: Math.max(0, Math.round(input.durationMs)),
    errorClass: "SandboxChannelAOperationError",
    errorCode: diagnostic.errorCode,
    origin: "api",
  } as const;
  try {
    services.observability.incrementCounter({
      name: "opengeni_channel_a_operation_failures_total",
      help: "Channel-A failures by bounded operation and structural reason.",
      labels: {
        backend: input.backend,
        op: input.operation,
        reason: diagnostic.reason,
        status: String(diagnostic.status),
      },
    });
  } catch {
    // Metrics can never alter request or lease settlement.
  }
  try {
    if (diagnostic.reason === "request_cancelled") {
      services.observability.info("Channel-A request cancelled", attributes);
    } else if (diagnostic.reason === "request_rejected") {
      services.observability.debug("Channel-A request rejected", attributes);
    } else {
      services.observability.warn("Channel-A operation failed", attributes);
    }
  } catch {
    // Logs can never alter request or lease settlement.
  }
}

// Drop a transiently-established, NON-OWNED handle WITHOUT terminating the box.
// The box is owned by the LEASE (resumed by id); this handle is incidental.
//
// CRITICAL (deployed-integration bug, prove-it D2): a provider session's
// `close()` is NOT a neutral local-resource free — Modal's session.close() calls
// sandbox.terminate(), KILLING THE BOX. Calling it after each Channel-A op
// destroyed the box mid-flight, so a subsequent fs.read/git/exec hit a different
// (cold-restored) box and 404'd. We DO NOT close the session; only the reaper
// (provider stop at refcount 0) terminates a box.
async function dropEstablishedHandle(
  established: EstablishedSandboxSession | undefined,
): Promise<void> {
  // No-op beyond dropping the reference: the lease owns lifecycle, the reaper
  // owns teardown. Never session.close()/terminate() a non-owned handle here.
  void established;
}