RemoteDelegation.ts

import {
  CommandId,
  type EnvironmentId,
  NodeId,
  OrchestratorMcpCapabilitiesInput,
  OrchestratorMcpCapabilitiesResult,
  OrchestratorMcpFailure,
  type OrchestratorMcpDelegateTaskInput,
  type OrchestratorMcpTaskCancelResult,
  type OrchestratorMcpThreadRun,
  type OrchestratorMcpThreadSendInput,
  type OrchestratorMcpThreadSendResult,
  type OrchestratorMcpThreadTimelineItem,
  type OrchestrationV2RemoteTaskChild,
  type OrchestrationV2Subagent,
  type ProjectId,
  type ProviderInstanceId,
  ProviderDriverKind,
  type RunId,
  type ThreadId,
} from "@t3tools/contracts";
import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
import * as KeyedLock from "@t3tools/shared/KeyedLock";
import * as Context from "effect/Context";
import * as Digest from "effect/crypto/Digest";
import * as SecureRandom from "effect/crypto/SecureRandom";
import * as DateTime from "effect/DateTime";
import * as Duration from "effect/Duration";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Schedule from "effect/Schedule";
import * as Scope from "effect/Scope";
import * as Stream from "effect/Stream";
import * as Tool from "effect/ai/Tool";
import { withExecutedModelNote } from "@t3tools/provider-core/server/subagentProjection";

import * as McpInvocationContext from "../mcp/McpInvocationContext.ts";
import { assertNotLinked } from "../mcp/linkOrigin.ts";
import { requestIdentity } from "../mcp/requestIdentity.ts";
import { readCaller, unavailable } from "../mcp/threadAccess.ts";
import { ProjectToolkit } from "../mcp/toolkits/project/tools.ts";
import {
  resolveInteractionMode,
  resolveRuntimeMode,
  taskPrompt,
} from "../mcp/delegatedTaskModes.ts";
import { followUpOwnerRun } from "../orchestration-v2/CommandPolicy.ts";
import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts";
import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts";
import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
import * as ProjectService from "../project/ProjectService.ts";
import * as RepositoryIdentityResolver from "../project/RepositoryIdentityResolver.ts";
import { forkParked } from "../serverActivation.ts";
import * as PeerForwarding from "./PeerForwarding.ts";
import * as PeerLinks from "./PeerLinks.ts";

/** How long one follow of a remote task waits on its thread there before checking again. */
const FOLLOW_WAIT_MS = 10 * 60 * 1_000;
/** A linked environment that stops answering is retried, backing off to this. */
const FOLLOW_MAX_BACKOFF = Duration.minutes(5);
/** How a follower retries; tests step it by hand. */
export const FollowRetrySchedule = Context.Reference<Schedule.Schedule<unknown>>(
  "t3/peer/RemoteDelegation/FollowRetrySchedule",
  {
    defaultValue: () =>
      Schedule.exponential(Duration.seconds(2)).pipe(
        Schedule.modifyDelay(({ duration }) =>
          Effect.succeed(Duration.min(duration, FOLLOW_MAX_BACKOFF)),
        ),
      ),
  },
);

type ThreadScope = McpInvocationContext.McpThreadInvocationScope;

/** `orchestrator_capabilities` there. Its toolkit imports this module, so it is described here. */
const RemoteCapabilitiesTool = Tool.make("orchestrator_capabilities", {
  parameters: OrchestratorMcpCapabilitiesInput,
  success: OrchestratorMcpCapabilitiesResult,
  failure: OrchestratorMcpFailure,
  failureMode: "return",
});

/**
 * `delegate_task` to a linked environment. The task's child is an ordinary
 * thread there, launched through the link; the parent here records the task
 * like a local one, without a child thread, and a follower completes it when
 * the thread there ends, so the parent wakes as it would for a local child.
 */
export class RemoteDelegation extends Context.Service<
  RemoteDelegation,
  {
    readonly delegate: (
      scope: ThreadScope,
      input: OrchestratorMcpDelegateTaskInput & {
        readonly target: NonNullable<OrchestratorMcpDelegateTaskInput["target"]> & {
          readonly environmentId: EnvironmentId;
        };
      },
    ) => Effect.Effect<{ readonly taskId: NodeId }, OrchestratorMcpFailure>;
    readonly sendToThread: (
      scope: McpInvocationContext.McpInvocationScope,
      input: OrchestratorMcpThreadSendInput & { readonly environmentId: EnvironmentId },
    ) => Effect.Effect<OrchestratorMcpThreadSendResult, OrchestratorMcpFailure>;
    /**
     * Stops the task's thread there, then completes it here as cancelled. A
     * task that already has its result keeps it; its thread there is still
     * stopped, as cancelling a local task stops follow-up work on its child.
     */
    readonly cancel: (
      scope: ThreadScope,
      task: OrchestrationV2Subagent,
      input: {
        readonly reason?: string | undefined;
        readonly clientRequestId?: string | undefined;
      },
    ) => Effect.Effect<OrchestratorMcpTaskCancelResult, OrchestratorMcpFailure>;
    /**
     * Follows every open remote task until it ends, including after a
     * restart, and a finished one again when a Stop here owes it a stop there.
     */
    readonly start: () => Effect.Effect<void, never, Scope.Scope>;
  }
>()("t3/peer/RemoteDelegation") {}

/**
 * Refusals that following again cannot change: the link was revoked, expired
 * or forgotten, or the thread or run there is gone. Anything else (the other
 * machine asleep, a dropped route) is worth retrying.
 */
/** The other side refused because the thread there runs above this caller's modes. */
const isModeRefusal = (error: OrchestratorMcpFailure) =>
  error.code === "runtime_mode_escalation_denied" ||
  error.code === "interaction_mode_escalation_denied";

const isFinal = (error: OrchestratorMcpFailure) =>
  error.code === "capability_denied" ||
  error.code === "thread_not_found" ||
  error.code === "run_not_found";

const make = Effect.gen(function* () {
  const followRetry = yield* FollowRetrySchedule;
  const forwarding = yield* PeerForwarding.PeerForwarding;
  const links = yield* PeerLinks.PeerLinks;
  const threads = yield* ThreadManagement.ThreadManagementService;
  const projections = yield* ProjectionStore.ProjectionStoreV2;
  const projects = yield* ProjectService.ProjectService;
  const repositoryIdentities = yield* RepositoryIdentityResolver.RepositoryIdentityResolver;
  const crypto = yield* Digest.Digest;
  const random = yield* SecureRandom.SecureRandom;
  const launchLocks = yield* KeyedLock.make<CommandId>();
  const hereId = yield* ServerEnvironment.ServerEnvironment.pipe(
    Effect.flatMap((environment) => environment.getEnvironmentId),
  );
  // Followers live as long as this service, so shutting down stops them.
  const followers = yield* Scope.Scope;

  // One follower runs per open task and holds what it read across a wait of
  // up to FOLLOW_WAIT_MS, so it reads its own task, never the parent's whole
  // task list: a parent with thousands of tasks made every restart decode
  // that list once per open task at the same time and ran out of heap.
  const readTask = (parentThreadId: ThreadId, taskId: NodeId) =>
    threads
      .getThreadRecords(parentThreadId, ["subagents"], { subagentIds: [taskId] })
      .pipe(Effect.map(({ thread, subagents }) => ({ thread, task: subagents[0] })));

  /** The peer's project with this thread's repository, or the one the caller named. */
  const resolveRemoteProject = Effect.fn("RemoteDelegation.resolveRemoteProject")(function* (
    scope: ThreadScope,
    environmentId: EnvironmentId,
    parentProjectId: ProjectId,
    requested: ProjectId | undefined,
  ) {
    if (requested !== undefined) return requested;
    const here = yield* projects.getById(parentProjectId).pipe(
      Effect.map(Option.getOrUndefined),
      Effect.orElseSucceed(() => undefined),
    );
    // The project's own identity is a one-minute cache that reads blank when
    // cold, so resolve it here, as the other side's t3_project_list does.
    const key =
      here === undefined
        ? undefined
        : (yield* repositoryIdentities.resolve(here.workspaceRoot))?.canonicalKey;
    if (key === undefined) {
      return yield* new OrchestratorMcpFailure({
        code: "target_required",
        message: "Pass target.projectId: this thread's project has no repository to match there.",
      });
    }
    const matches: Array<{ readonly id: ProjectId; readonly title: string }> = [];
    let cursor: number | null = 0;
    while (cursor !== null) {
      const listed: Tool.Success<typeof ProjectToolkit.tools.t3_project_list> =
        yield* forwarding.call(scope, ProjectToolkit.tools.t3_project_list, environmentId, {
          cursor,
          limit: 100,
        });
      matches.push(
        ...listed.projects.filter((project) => project.repositoryIdentity?.canonicalKey === key),
      );
      cursor = listed.nextCursor;
    }
    if (matches.length === 1) return matches[0]!.id;
    return yield* new OrchestratorMcpFailure({
      code: "invalid_request",
      message:
        matches.length === 0
          ? `No project there has this repository (${key}). Pass target.projectId; t3_project_list with this environmentId lists them. To bring the repository there, use t3_project_clone and then t3_project_create with this environmentId.`
          : `Several projects there have this repository: ${matches.map((project) => `${project.title} (${project.id})`).join(", ")}. Pass target.projectId.`,
    });
  });

  /** The instance there of the driver the caller named, preferring the one this thread runs on, as a task here does. */
  const resolveRemoteInstance = Effect.fn("RemoteDelegation.resolveRemoteInstance")(function* (
    scope: ThreadScope,
    environmentId: EnvironmentId,
    requested: {
      readonly driverKind: ProviderDriverKind;
      readonly providerInstanceId: ProviderInstanceId | undefined;
      readonly inheritedInstanceId: ProviderInstanceId;
    },
  ) {
    const { providers } = yield* forwarding.call(scope, RemoteCapabilitiesTool, environmentId, {});
    if (requested.providerInstanceId !== undefined) {
      const named = providers.find(
        (provider) => provider.providerInstanceId === requested.providerInstanceId,
      );
      if (named !== undefined && named.driverKind !== requested.driverKind) {
        return yield* new OrchestratorMcpFailure({
          code: "invalid_request",
          message: `Provider instance ${named.providerInstanceId} there uses driver ${named.driverKind}, not ${requested.driverKind}.`,
        });
      }
      return requested.providerInstanceId;
    }
    const candidates = providers.filter(
      (provider) => provider.driverKind === requested.driverKind && provider.canRunChildTask,
    );
    const chosen =
      candidates.find(
        (provider) => provider.providerInstanceId === requested.inheritedInstanceId,
      ) ?? candidates[0];
    if (chosen === undefined) {
      return yield* new OrchestratorMcpFailure({
        code: "provider_unavailable",
        message: `No instance of driver ${requested.driverKind} there can run a child task. orchestrator_capabilities with this environmentId lists its providers.`,
      });
    }
    return chosen.providerInstanceId;
  });

  const launchAndRecord = Effect.fn("RemoteDelegation.launchAndRecord")(function* (
    scope: ThreadScope,
    input: Parameters<RemoteDelegation["Service"]["delegate"]>[1],
    identity: string,
    commandId: CommandId,
  ) {
    const caller = yield* readCaller().pipe(
      Effect.provideService(McpInvocationContext.McpInvocationContext, scope),
      Effect.provideService(ThreadManagement.ThreadManagementService, threads),
    );
    yield* assertNotLinked(caller, "use this environment's links");
    const parent = yield* threads.getThreadRecords(scope.thread.threadId, ["runs"]).pipe(
      Effect.mapError(
        () =>
          new OrchestratorMcpFailure({
            code: "thread_not_found",
            message: "The calling thread was not found.",
          }),
      ),
    );
    const parentRun = parent.runs
      .filter(ThreadManagement.isActiveRun)
      .toSorted((left, right) => right.ordinal - left.ordinal)[0];
    if (
      parentRun === undefined ||
      parentRun.rootNodeId === null ||
      parentRun.providerInstanceId !== scope.thread.providerInstanceId
    ) {
      return yield* new OrchestratorMcpFailure({
        code: "parent_not_active",
        message: "Delegated tasks require an active run owned by this MCP provider session.",
      });
    }
    const parentNodeId = parentRun.rootNodeId;
    const { target } = input;
    // A worktree there would branch from a commit this checkout has and the
    // other may not, so a remote task runs in that project's checkout.
    if (input.isolation === "worktree") {
      return yield* new OrchestratorMcpFailure({
        code: "invalid_request",
        message:
          'A task in a linked environment runs in that project\'s checkout; isolation "worktree" is not supported there. Omit isolation, or use t3_thread_launch with environmentId and a worktree workspaceStrategy.',
      });
    }
    const runtimeMode =
      input.runtimeMode === undefined || input.runtimeMode === "inherit"
        ? undefined
        : yield* resolveRuntimeMode(parent.thread.runtimeMode, input.runtimeMode);
    const interactionMode =
      input.interactionMode === undefined || input.interactionMode === "inherit"
        ? undefined
        : yield* resolveInteractionMode(parent.thread.interactionMode, input.interactionMode);
    const recordedTaskId = IdAllocator.derive.delegatedTaskNode({ commandId });
    const recordedTask = yield* readTask(scope.thread.threadId, recordedTaskId).pipe(
      Effect.map(({ task }) => task),
      Effect.mapError(
        () =>
          new OrchestratorMcpFailure({
            code: "orchestration_error",
            message: "Unable to read the recorded delegated task.",
          }),
      ),
    );
    if (recordedTask !== undefined) {
      if (recordedTask.remoteChild?.environmentId !== target.environmentId) {
        return yield* new OrchestratorMcpFailure({
          code: "invalid_request",
          message:
            "This clientRequestId already belongs to a task in another environment. Retry with the original arguments, or use a new clientRequestId for new work.",
        });
      }
      // The accepted task owns its project and model even if catalogs/defaults changed or the peer is offline.
      yield* followOne(scope.thread.threadId, recordedTask.id).pipe(Effect.forkIn(followers));
      return { taskId: recordedTask.id };
    }
    // The provider and model inherit as they do for a task here. Another
    // provider there has no model to inherit, so the caller names one.
    const inherited = parent.thread.modelSelection;
    const instanceId =
      target.driverKind === undefined
        ? (target.providerInstanceId ?? inherited.instanceId)
        : yield* resolveRemoteInstance(scope, target.environmentId, {
            driverKind: target.driverKind,
            providerInstanceId: target.providerInstanceId,
            inheritedInstanceId: inherited.instanceId,
          });
    const model =
      target.model ?? (instanceId === inherited.instanceId ? inherited.model : undefined);
    if (model === undefined) {
      return yield* new OrchestratorMcpFailure({
        code: "target_required",
        message: "Pass target.model from orchestrator_capabilities with this environmentId.",
      });
    }
    const options =
      target.options ??
      (instanceId === inherited.instanceId && model === inherited.model
        ? inherited.options
        : undefined);
    const projectId = yield* resolveRemoteProject(
      scope,
      target.environmentId,
      parent.thread.projectId,
      target.projectId,
    );
    const task = taskPrompt(input);
    const title = input.title ?? parent.thread.title;
    const launched = yield* forwarding.call(
      scope,
      ProjectToolkit.tools.t3_thread_launch,
      target.environmentId,
      {
        projectId,
        title,
        modelSelection: { instanceId, model, ...(options === undefined ? {} : { options }) },
        ...(runtimeMode === undefined ? {} : { runtimeMode }),
        ...(interactionMode === undefined ? {} : { interactionMode }),
        message: task,
        clientRequestId: `delegate:${identity}`,
        // The thread there shows it is this thread's subagent, with a way back.
        delegatedFrom: {
          environmentId: hereId,
          threadId: parent.thread.id,
          title: parent.thread.title,
        },
      },
    );
    const link = yield* links.get(target.environmentId).pipe(
      Effect.map(Option.getOrUndefined),
      Effect.orElseSucceed(() => undefined),
    );
    const remoteChild: OrchestrationV2RemoteTaskChild = {
      environmentId: target.environmentId,
      threadId: launched.threadId,
      label: link?.label ?? target.environmentId,
      title,
      ...(launched.runId === null ? {} : { runId: launched.runId }),
    };
    const recorded = yield* threads
      .dispatch({
        type: "delegated_task.remote.request",
        commandId,
        parentThreadId: scope.thread.threadId,
        parentRunId: parentRun.id,
        parentNodeId,
        task,
        ...(input.title === undefined ? {} : { title: input.title }),
        // The driver runs there; this side only labels the task with it.
        driver: ProviderDriverKind.make("remote"),
        modelSelection: launched.modelSelection,
        remoteChild,
        completionWake: input.mode === "wait" ? "settled_only" : "always",
      })
      .pipe(
        // The cause goes to the log; the caller learns only that recording failed.
        Effect.tapError((cause) =>
          Effect.logWarning("remote-delegation.record-failed", {
            parentThreadId: scope.thread.threadId,
            cause,
          }),
        ),
        Effect.mapError(
          () =>
            new OrchestratorMcpFailure({
              code: "orchestration_error",
              message: "Unable to record the delegated task.",
            }),
        ),
        // Nothing here follows a thread the task was never recorded for, so stop it there.
        // A crash between the launch and this record still leaves it running there
        // until a retry with the same clientRequestId records it.
        Effect.tapError(() =>
          stopThere(followScope(scope.thread, remoteChild), remoteChild, "stop").pipe(
            Effect.ignoreCause({ log: true }),
          ),
        ),
      );
    const taskEvent = recorded.storedEvents.find(
      (stored) => stored.event.type === "subagent.updated",
    );
    const taskId =
      taskEvent?.event.type === "subagent.updated"
        ? taskEvent.event.payload.id
        : // A replayed request recorded the task the first time.
          yield* threads.getThreadRecords(scope.thread.threadId, ["subagents"]).pipe(
            Effect.map(
              ({ subagents }) =>
                subagents.find((task) => task.remoteChild?.threadId === launched.threadId)?.id,
            ),
            Effect.orElseSucceed(() => undefined),
          );
    if (taskId === undefined) {
      return yield* new OrchestratorMcpFailure({
        code: "orchestration_error",
        message: "The delegated task was not recorded.",
      });
    }
    yield* followOne(scope.thread.threadId, taskId).pipe(Effect.forkIn(followers));
    return { taskId };
  });

  const delegate: RemoteDelegation["Service"]["delegate"] = Effect.fn("RemoteDelegation.delegate")(
    function* (scope, input) {
      const key = input.clientRequestId ?? (yield* random.uuidV4());
      const identity = yield* requestIdentity(scope, "remote-delegate", key).pipe(
        Effect.provideService(Digest.Digest, crypto),
      );
      const commandId = CommandId.make(`command:mcp:remote-delegate:${identity}`);
      return yield* launchLocks.withLock(
        commandId,
        launchAndRecord(scope, input, identity, commandId),
      );
    },
  );

  /**
   * Each report has a fresh command ID so a failed attempt cannot block its retry.
   * Already-settled tasks keep their result; stoppedThere can still acknowledge an owed stop.
   */
  const complete = Effect.fn("RemoteDelegation.complete")(function* (
    parentThreadId: ThreadId,
    taskId: NodeId,
    status: "completed" | "failed" | "cancelled" | "interrupted",
    result: string,
    stoppedThere: "ran-out" | "stopped" | undefined,
    stopCount?: number,
  ) {
    const attempt = yield* random.uuidV4();
    yield* threads
      .dispatch({
        type: "delegated_task.remote.complete",
        commandId: CommandId.make(`command:remote-task-complete:${taskId}:${attempt}`),
        parentThreadId,
        taskId,
        status,
        result,
        ...(stoppedThere === undefined ? {} : { stoppedThere }),
        ...(stopCount === undefined ? {} : { stopCount }),
      })
      .pipe(
        Effect.catch((error) =>
          readTask(parentThreadId, taskId).pipe(
            Effect.flatMap(({ task }) => (task?.result != null ? Effect.void : Effect.fail(error))),
          ),
        ),
      );
  });

  /**
   * How this environment reaches a task's thread there: as the parent
   * thread, under one namespace per parent, so a stop sent again after a
   * restart or a dropped route is recorded there once.
   */
  const followScope = (
    parent: { readonly threadId: ThreadId; readonly providerInstanceId: ProviderInstanceId },
    remote: OrchestrationV2RemoteTaskChild,
  ): ThreadScope => ({
    environmentId: hereId,
    requestNamespace: `remote-task:${parent.threadId}`,
    thread: {
      threadId: parent.threadId,
      providerSessionId: `remote-task:${remote.threadId}`,
      providerInstanceId: parent.providerInstanceId,
    },
    client: undefined,
    capabilities: new Set(["orchestration"]),
    issuedAt: 0,
  });

  /**
   * A remote Stop also cancels queued messages, watches, and delegated tasks on that thread.
   * Each key identifies one stop, so retries do not stop later work again.
   */
  const stopThere = Effect.fn("RemoteDelegation.stopThere")(function* (
    scope: ThreadScope,
    remote: OrchestrationV2RemoteTaskChild,
    key: string,
    reason?: string,
  ) {
    const identity = yield* requestIdentity(scope, `remote-stop:${remote.threadId}`, key).pipe(
      Effect.provideService(Digest.Digest, crypto),
    );
    yield* forwarding
      .interruptThread(scope, remote.environmentId, {
        threadId: remote.threadId,
        stop: true,
        ...(reason === undefined ? {} : { reason }),
        clientRequestId: `stop:${identity}`,
      })
      .pipe(Effect.asVoid);
  });

  /**
   * The last reply `runId` gave on the task's thread there, which is its
   * result, read whole however long it is. A failed run prefers its error
   * from the activity view over an earlier reply, without taking another run's.
   * Led by a note when its provider reported running another model than it asked for.
   */
  const lastReply = Effect.fn("RemoteDelegation.lastReply")(function* (
    scope: ThreadScope,
    remote: OrchestrationV2RemoteTaskChild,
    runId: RunId | null,
    failed = false,
  ) {
    let reply: OrchestratorMcpThreadTimelineItem | undefined;
    let error: OrchestratorMcpThreadTimelineItem | undefined;
    let afterPosition: number | undefined;
    let run: OrchestratorMcpThreadRun | undefined;
    while (true) {
      const read = yield* forwarding.readThread(scope, remote.environmentId, {
        threadId: remote.threadId,
        view: failed ? "activity" : "messages",
        limit: 100,
        runLimit: 50,
        ...(afterPosition === undefined ? {} : { afterPosition }),
      });
      run ??= read.recentRuns.find((candidate) => candidate.runId === runId);
      reply =
        read.items.findLast(
          (item) => item.type === "assistant_message" && (runId === null || item.runId === runId),
        ) ?? reply;
      if (failed) {
        error =
          read.items.findLast(
            (item) => item.type === "error" && (runId === null || item.runId === runId),
          ) ?? error;
      }
      if (!read.hasMore || read.nextPosition === null) break;
      afterPosition = read.nextPosition;
    }
    reply = error ?? reply;
    if (reply === undefined) return undefined;
    let text = reply.text ?? "";
    let textOffset = reply.textTruncated ? (reply.nextTextOffset ?? null) : null;
    while (textOffset !== null) {
      const rest = yield* forwarding.readThread(scope, remote.environmentId, {
        threadId: remote.threadId,
        itemId: reply.itemId,
        textOffset,
        maxCharsPerItem: 50_000,
      });
      const part = rest.items[0];
      if (part === undefined) break;
      text += part.text ?? "";
      textOffset = part.textTruncated ? (part.nextTextOffset ?? null) : null;
    }
    return run === undefined ? text : withExecutedModelNote(text, run.model, run.executedModel);
  });

  /**
   * Why following a task can never succeed, if that is what `error` means.
   * The far side's message already says what was refused and what to do; a
   * link forgotten here only reads as a bad request, so it is named.
   */
  const finalReason = Effect.fn("RemoteDelegation.finalReason")(
    function* (parentThreadId: ThreadId, taskId: NodeId, error: OrchestratorMcpFailure) {
      if (isFinal(error)) return error.message;
      if (error.code !== "invalid_request") return undefined;
      const { task } = yield* readTask(parentThreadId, taskId);
      const environmentId = task?.remoteChild?.environmentId;
      if (environmentId === undefined) return error.message;
      const link = yield* links.get(environmentId);
      return Option.isSome(link) ? undefined : error.message;
    },
    Effect.orElseSucceed(() => undefined),
  );

  /** A task's thread there that ended without a reply. */
  const noReply = "(The thread in the linked environment ended without a reply.)";

  /**
   * Missing threads, runs, or usable links terminate the task; transient failures retry with backoff.
   * A task ended locally still owes a remote stop, and recording completion must also be retried.
   */
  const follow = (parentThreadId: ThreadId, taskId: NodeId): Effect.Effect<void> =>
    followOnce(parentThreadId, taskId).pipe(
      // `false` means it is still running there: follow again straight away.
      Effect.repeat({ while: (done) => !done }),
      Effect.catchTags({
        OrchestratorMcpFailure: Effect.fn("RemoteDelegation.followFailure")(function* (error) {
          const reason = yield* finalReason(parentThreadId, taskId, error);
          if (reason === undefined) return yield* error;
          // Gone there, or no link to reach it by: nothing left to stop there.
          const { task } = yield* readTask(parentThreadId, taskId);
          const stopCount = task?.remoteChild?.stopCount;
          yield* complete(parentThreadId, taskId, "failed", reason, "stopped", stopCount);
        }),
      }),
      Effect.retry(followRetry),
      Effect.ignoreCause({ log: true }),
    );

  /**
   * One wait on the task's thread there; true once the task here is done.
   * A task completed here first (cancelled, or its parent stopped) is
   * stopped there, and done once that stop is recorded here.
   */
  const followOnce = Effect.fn("RemoteDelegation.followOnce")(function* (
    parentThreadId: ThreadId,
    taskId: NodeId,
  ) {
    const sequence = yield* threads.getThreadEventSequence(parentThreadId);
    const { thread, task } = yield* readTask(parentThreadId, taskId);
    if (task?.remoteChild === undefined) return true;
    const remote = task.remoteChild;
    const scope = followScope(
      { threadId: parentThreadId, providerInstanceId: thread.providerInstanceId },
      remote,
    );
    // Ended here, maybe while this follower backed off from an unreachable
    // machine or before a restart: it still has to be stopped there. So
    // does one its parent here was stopped after, for what followed there.
    if (task.result !== null) {
      if (
        remote.stoppedThere === false ||
        ((task.status === "cancelled" || task.status === "interrupted") &&
          remote.stoppedThere !== true)
      ) {
        const key = `${task.id}:${remote.stopCount === undefined ? "stop" : `stop-${remote.stopCount}`}`;
        yield* stopThere(scope, remote, key).pipe(
          // Its thread there now runs above what this side may touch, say
          // because the parent here was lowered since; no retry changes that.
          Effect.catchIf(isModeRefusal, (error) =>
            Effect.logWarning("A linked environment refused to stop a delegated task", {
              taskId,
              reason: error.message,
            }),
          ),
        );
        yield* complete(
          parentThreadId,
          taskId,
          terminalStatus(task.status),
          task.result,
          "stopped",
          remote.stopCount,
        );
        return yield* settled(parentThreadId, taskId);
      }
      return true;
    }
    const endedHere = threads
      .streamStoredEventsFrom({
        threadId: parentThreadId,
        afterSequence: sequence,
        eventType: "subagent.updated",
      })
      .pipe(
        Stream.filter(
          (stored) =>
            stored.event.type === "subagent.updated" &&
            stored.event.payload.id === taskId &&
            stored.event.payload.result !== null,
        ),
        Stream.runHead,
        Effect.flatMap((ended) =>
          Option.isSome(ended) ? Effect.succeed({ endedHere: true } as const) : Effect.never,
        ),
      );
    const waited = yield* forwarding
      .waitForThread(scope, remote.environmentId, {
        threadId: remote.threadId,
        ...(remote.runId === undefined ? {} : { runId: remote.runId }),
        timeoutMs: FOLLOW_WAIT_MS,
        // A thread there whose run ended may still be waiting on its own
        // tasks to wake it for the answer, which a task here waits for too.
        // An older peer ignores this and reports the run alone.
        untilDelegatedWorkSettled: true,
      })
      .pipe(Effect.raceFirst(endedHere));
    // The next look reads the result and stops it there.
    if ("endedHere" in waited) return false;
    if (waited.timedOut) return false;
    // A task recorded without its run waits on the thread's latest; a
    // thread there with none has nothing left to wait for.
    if (waited.runId === null) {
      const reply = yield* lastReply(scope, remote, null);
      yield* complete(
        parentThreadId,
        taskId,
        reply === undefined ? "failed" : "completed",
        reply ?? "The thread in the linked environment has no run to follow.",
        "ran-out",
      );
      return yield* settled(parentThreadId, taskId);
    }
    if (!isTerminal(waited.status)) return false;
    // A later run's answer there may be another open follow-up's, or its restart, which no run id
    // here names; concurrent sends may record here out of their order there. Then this task's
    // answer is its own run's latest restart, the one waited on.
    const sharedThread =
      waited.resultRun !== undefined &&
      waited.resultRun.runId !== waited.runId &&
      (yield* threads.getThreadRecords(parentThreadId, ["subagents"]).pipe(
        Effect.map(({ subagents }) => {
          const recordedAt = subagents.find((task) => task.id === taskId)?.startedAt ?? null;
          return subagents.some(
            (other) =>
              other.id !== taskId &&
              other.remoteChild?.environmentId === remote.environmentId &&
              other.remoteChild.threadId === remote.threadId &&
              // One that settled before this was recorded ran before it there.
              (recordedAt === null ||
                other.completedAt === null ||
                DateTime.isGreaterThan(other.completedAt, recordedAt)),
          );
        }),
      ));
    const resultRun =
      waited.resultRun !== undefined && !sharedThread
        ? waited.resultRun
        : { runId: waited.runId, status: waited.status };
    const result = yield* lastReply(scope, remote, resultRun.runId, resultRun.status === "failed");
    yield* complete(
      parentThreadId,
      taskId,
      terminalStatus(resultRun.status),
      result ?? noReply,
      "ran-out",
    );
    return yield* settled(parentThreadId, taskId);
  });

  /**
   * Whether the task is done here after the follower reported its run ran
   * out there. A cancel or Stop that landed while the reply was read keeps
   * its result, and still owes a stop there for what may follow the run;
   * the next look sends it.
   */
  const settled = (parentThreadId: ThreadId, taskId: NodeId) =>
    readTask(parentThreadId, taskId).pipe(
      Effect.map(({ task }) => task?.remoteChild?.stoppedThere === true),
    );

  const following = new Map<NodeId, { again: boolean }>();

  /**
   * `follow`, at most once per task. Asked again while it runs, say by a
   * Stop here on a task that already has its result, it looks once more
   * before it ends, so that stop is not missed.
   */
  const followOne = (parentThreadId: ThreadId, taskId: NodeId) =>
    Effect.suspend(() => {
      const running = following.get(taskId);
      if (running !== undefined) {
        running.again = true;
        return Effect.void;
      }
      const state = { again: false };
      following.set(taskId, state);
      return follow(parentThreadId, taskId).pipe(
        // Checked and released in one step, so a request between them is never lost.
        Effect.repeat({
          while: () => {
            if (state.again) {
              state.again = false;
              return true;
            }
            following.delete(taskId);
            return false;
          },
        }),
        Effect.ensuring(
          Effect.sync(() => {
            if (following.get(taskId) === state) following.delete(taskId);
          }),
        ),
      );
    });

  const cancel: RemoteDelegation["Service"]["cancel"] = Effect.fn("RemoteDelegation.cancel")(
    function* (scope, task, input) {
      const remote = task.remoteChild;
      if (remote === undefined) {
        return yield* new OrchestratorMcpFailure({
          code: "task_not_found",
          message: `Delegated task ${task.id} is not remote.`,
        });
      }
      const { reason } = input;
      const there = followScope(scope.thread, remote);
      // A later cancel stops what started there since; a retry of this one, nothing new.
      const requestId = input.clientRequestId ?? (yield* random.uuidV4());
      const key = `cancel:${yield* requestIdentity(scope, "remote-cancel", requestId).pipe(
        Effect.provideService(Digest.Digest, crypto),
      )}`;
      if (task.result !== null) {
        // A replayed cancel cannot acknowledge a later parent Stop; its follower owns that receipt.
        yield* stopThere(there, remote, key, reason).pipe(Effect.ignoreCause({ log: true }));
        return {
          taskId: task.id,
          status:
            task.status === "completed" || task.status === "failed" || task.status === "interrupted"
              ? task.status
              : "cancelled",
        } satisfies OrchestratorMcpTaskCancelResult;
      }
      // A link that is gone or a machine that is down still cancels the task
      // here; its follower keeps trying to stop it there, across restarts.
      const unstopped = yield* stopThere(there, remote, key, reason).pipe(
        Effect.as(undefined),
        Effect.catch((error) => Effect.succeed(error.message)),
      );
      const cancelled = reason === undefined ? "Cancelled." : `Cancelled: ${reason}`;
      yield* complete(
        task.threadId,
        task.id,
        "cancelled",
        unstopped === undefined
          ? cancelled
          : `${cancelled} It could not be stopped in ${remote.label} and may still run there: ${unstopped}`,
        unstopped === undefined ? "stopped" : undefined,
        remote.stopCount,
      ).pipe(
        Effect.tapError((cause) =>
          Effect.logWarning("remote-delegation.cancel-record-failed", { taskId: task.id, cause }),
        ),
        Effect.mapError(
          () =>
            new OrchestratorMcpFailure({
              code: "orchestration_error",
              message: "Unable to record the cancellation.",
            }),
        ),
      );
      return {
        taskId: task.id,
        status: "cancel_requested",
      } satisfies OrchestratorMcpTaskCancelResult;
    },
  );

  const start: RemoteDelegation["Service"]["start"] = () =>
    forkParked(
      Effect.gen(function* () {
        // A finished task's follower has ended, so a Stop here that owes its
        // thread there a stop starts it again. Started before the scan below;
        // a Stop that slips between the two stays owed for the next start.
        yield* Stream.runForEach(threads.streamDomainEvents, (event) =>
          event.type === "subagent.updated" && event.payload.remoteChild?.stoppedThere === false
            ? followOne(event.threadId, event.payload.id).pipe(Effect.forkIn(followers))
            : Effect.void,
        ).pipe(
          Effect.ignoreCause({ log: true }),
          Effect.forkIn(followers, { startImmediately: true }),
        );
        const open = yield* projections.getOpenRemoteDelegatedTasks;
        yield* Effect.forEach(
          open,
          ({ parentThreadId, taskId }) => followOne(parentThreadId, taskId),
          {
            concurrency: "unbounded",
            discard: true,
          },
        );
      }).pipe(Effect.ignoreCause({ log: true })),
    );

  const sendToThread: RemoteDelegation["Service"]["sendToThread"] = Effect.fn(
    "RemoteDelegation.sendToThread",
  )(function* (scope, input) {
    const caller = scope.thread;
    const filter = {
      subagentRemoteThread: { environmentId: input.environmentId, threadId: input.threadId },
      runStatuses: ["preparing", "queued", "starting", "running", "waiting"] as const,
    };
    const parent =
      caller === undefined
        ? undefined
        : yield* threads.getThreadRecords(caller.threadId, ["runs", "subagents"], filter).pipe(
            Effect.mapError(
              () =>
                new OrchestratorMcpFailure({
                  code: "orchestration_error",
                  message: "Unable to read the calling thread's delegated tasks.",
                }),
            ),
          );
    const original = parent?.subagents.find((task) => task.origin === "app_owned");
    const remote = original?.remoteChild;
    if (remote === undefined || caller === undefined || parent === undefined) {
      return yield* forwarding.sendToThread(scope, input.environmentId, input);
    }
    const key = input.clientRequestId ?? (yield* random.uuidV4());
    const commandId = CommandId.make(
      `command:mcp:remote-followup:${yield* requestIdentity(scope, "remote-followup", key).pipe(
        Effect.provideService(Digest.Digest, crypto),
      )}`,
    );
    return yield* launchLocks.withLock(
      commandId,
      Effect.gen(function* () {
        // A follow-up to a finished task binds as one here does: a background caller's
        // foreground run may have ended, so it falls back to the run that last ran.
        const owner =
          parent.runs
            .filter(ThreadManagement.isActiveRun)
            .toSorted((left, right) => right.ordinal - left.ordinal)[0] ??
          (parent.subagents.some((task) => task.result !== null)
            ? followUpOwnerRun(
                (yield* threads
                  .getThreadRecords(caller.threadId, ["runs"])
                  .pipe(Effect.mapError(unavailable))).runs,
              )
            : undefined);
        if (owner?.rootNodeId == null || owner.providerInstanceId !== caller.providerInstanceId) {
          return yield* new OrchestratorMcpFailure({
            code: "parent_not_active",
            message:
              "Delegated follow-ups require an active run owned by this MCP provider session.",
          });
        }
        const taskId = IdAllocator.derive.delegatedTaskNode({ commandId });
        const recorded = yield* readTask(parent.thread.id, taskId).pipe(
          Effect.mapError(unavailable),
        );
        if (
          recorded.task !== undefined &&
          (recorded.task.remoteChild?.environmentId !== input.environmentId ||
            recorded.task.remoteChild.threadId !== input.threadId)
        ) {
          return yield* new OrchestratorMcpFailure({
            code: "invalid_request",
            message: "This clientRequestId already belongs to another delegated follow-up.",
          });
        }
        const sent = yield* forwarding.sendToThread(scope, input.environmentId, {
          ...input,
          clientRequestId: key,
        });
        const existing = parent.subagents.find(
          (task) =>
            task.result === null &&
            task.remoteChild?.environmentId === input.environmentId &&
            task.remoteChild.threadId === input.threadId &&
            task.remoteChild.runId === sent.runId,
        );
        const tracked = recorded.task ?? existing;
        if (tracked !== undefined) {
          yield* followOne(parent.thread.id, tracked.id).pipe(Effect.forkIn(followers));
          return { ...sent, taskId: tracked.id };
        }
        if (sent.delivery === "steered") return sent;
        const target = yield* forwarding.readThread(scope, input.environmentId, {
          threadId: input.threadId,
          limit: 1,
          maxCharsPerItem: 1,
          runLimit: 1,
        });
        yield* threads
          .dispatch({
            type: "delegated_task.remote.request",
            commandId,
            parentThreadId: parent.thread.id,
            parentRunId: owner.id,
            parentNodeId: owner.rootNodeId,
            task: input.message,
            title: target.thread.title,
            driver: ProviderDriverKind.make("remote"),
            modelSelection: {
              instanceId: target.thread.providerInstanceId,
              model: target.thread.model,
            },
            remoteChild: {
              environmentId: input.environmentId,
              threadId: input.threadId,
              label: remote.label,
              title: target.thread.title,
              runId: sent.runId,
            },
            completionWake: "always",
          })
          .pipe(
            Effect.mapError(
              () =>
                new OrchestratorMcpFailure({
                  code: "orchestration_error",
                  message:
                    "The message was sent, but its delegated follow-up could not be recorded. Retry with the same clientRequestId.",
                }),
            ),
          );
        yield* followOne(parent.thread.id, taskId).pipe(Effect.forkIn(followers));
        return { ...sent, taskId };
      }),
    );
  });

  return RemoteDelegation.of({ delegate, sendToThread, cancel, start });
});

const isTerminal = (status: string) =>
  status === "completed" ||
  status === "failed" ||
  status === "cancelled" ||
  status === "interrupted" ||
  status === "rolled_back";

const terminalStatus = (status: string): "completed" | "failed" | "cancelled" | "interrupted" =>
  status === "completed" || status === "failed" || status === "interrupted" ? status : "cancelled";

export const layer = Layer.effect(RemoteDelegation, make);