diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index 6d09b6b1b3a68..af569e902c027 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -25,6 +25,7 @@ import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; @@ -644,6 +645,8 @@ const makeLaptop = ( >, /** The laptop's links from before a restart, which keep their sessions there. */ kept?: Context.Context>>, + /** Runs as each follow attempt here starts, on its follower's fiber. */ + beforeFollow: Effect.Effect = Effect.void, ) => Effect.gen(function* () { const linking = kept ?? (yield* layerLinkingEnvironment(laptop).pipe(Layer.build)); @@ -672,6 +675,19 @@ const makeLaptop = ( Layer.provideMerge(McpHttpServer.layerInProcessMcpServer), Layer.provideMerge( RemoteDelegation.layer.pipe( + Layer.provide( + Layer.effect( + ThreadManagement.ThreadManagementService, + Effect.gen(function* () { + const threads = yield* ThreadManagement.ThreadManagementService; + return { + ...threads, + getThreadEventSequence: (threadId: ThreadId) => + beforeFollow.pipe(Effect.andThen(threads.getThreadEventSequence(threadId))), + }; + }), + ), + ), Layer.provideMerge(PeerForwarding.layer), Layer.provide( Layer.mock(ProjectService.ProjectService)({ @@ -1784,6 +1800,108 @@ const steppedRetries = Effect.gen(function* () { return { failed, retry, schedule }; }); +it.effect("a follower that keeps hitting defects fails the task and wakes the parent", () => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const retries = yield* steppedRetries; + const attempts = yield* Ref.make(0); + // The defect carries internal detail that the task's result must not. + const beforeFollow = Ref.update(attempts, (count) => count + 1).pipe( + Effect.andThen(Effect.die(new Error("A follow attempt hit a bug: defect-detail-4f1c9e"))), + ); + const a = yield* makeLaptop(database, offers, undefined, beforeFollow).pipe( + Effect.provideService(RemoteDelegation.FollowRetrySchedule, retries.schedule), + ); + yield* linkTo(a.links, b, "full-access"); + yield* seedParent(a); + yield* Queue.take(retries.failed).pipe( + Effect.andThen(Queue.offer(retries.retry, undefined)), + Effect.forever, + Effect.forkScoped, + ); + + const delegated = yield* delegateToBox(a); + const task = yield* taskResult(a, delegated.taskId); + expect(yield* Ref.get(attempts)).toBe(3); + expect(task?.status).toBe("failed"); + expect(task?.result).toContain(delegated.remoteChild.threadId); + expect(task?.result).not.toContain("defect-detail-4f1c9e"); + const offered = yield* SubscriptionRef.changes(offers).pipe( + Stream.filter((seen) => seen.some((offer) => offer.threadId === parentThreadId)), + Stream.runHead, + ); + expect(Option.isSome(offered)).toBe(true); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +it.effect.each(["a still-running wait", "a failed attempt"] as const)( + "defects separated by %s do not add up to giving up", + (between) => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const retries = yield* steppedRetries; + const attempts = yield* Ref.make(0); + const defect = Effect.die(new Error("A follow attempt hit a bug.")); + const script = [ + defect, + between === "a still-running wait" + ? Effect.void + : Effect.fail( + new Orchestrator.OrchestratorProjectionError({ threadId: parentThreadId }), + ), + defect, + between === "a still-running wait" + ? Effect.void + : Effect.fail( + new Orchestrator.OrchestratorProjectionError({ threadId: parentThreadId }), + ), + defect, + ]; + // Past the script, the run there has ended, so the next attempt completes the task. + const beforeFollow = Ref.getAndUpdate(attempts, (count) => count + 1).pipe( + Effect.flatMap( + (count) => + script[count] ?? + Ref.set(thread.reply, "All green.").pipe( + Effect.andThen(Ref.set(thread.status, "completed")), + ), + ), + ); + // Each wait there answers at once that its run is still running. + yield* Deferred.succeed(thread.finished, undefined); + const a = yield* makeLaptop(database, offers, undefined, beforeFollow).pipe( + Effect.provideService(RemoteDelegation.FollowRetrySchedule, retries.schedule), + ); + yield* linkTo(a.links, b, "full-access"); + yield* seedParent(a); + yield* Queue.take(retries.failed).pipe( + Effect.andThen(Queue.offer(retries.retry, undefined)), + Effect.forever, + Effect.forkScoped, + ); + + const delegated = yield* delegateToBox(a); + expect(yield* taskResult(a, delegated.taskId)).toMatchObject({ + status: "completed", + result: "All green.", + }); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + /** Waits until the task here records that the `stopCount`th Stop after its result reached there. */ const stoppedThere = ( laptopEnv: Effect.Success>, @@ -1813,6 +1931,87 @@ const stoppedThere = ( ), ); +it.effect("a follower that keeps hitting defects leaves a stop it owes owed", () => + Effect.scoped( + Effect.gen(function* () { + const thread = yield* makeBoxThread; + const b = yield* serveBox(thread); + const offers = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); + const retries = yield* steppedRetries; + // Follow attempts hit a defect only once the task has its result. + const broken = yield* Ref.make(false); + const follower = yield* Deferred.make>(); + const beforeFollow = Ref.get(broken).pipe( + Effect.flatMap((isBroken) => + isBroken + ? Effect.withFiber((fiber) => Deferred.succeed(follower, fiber)).pipe( + Effect.andThen(Effect.die(new Error("A follow attempt hit a bug."))), + ) + : Effect.void, + ), + ); + const a = yield* makeLaptop(database, offers, undefined, beforeFollow).pipe( + Effect.provideService(RemoteDelegation.FollowRetrySchedule, retries.schedule), + ); + yield* a.remote.start(); + yield* linkTo(a.links, b, "full-access"); + yield* seedParent(a); + const delegated = yield* delegateToBox(a); + yield* Ref.set(thread.reply, "All green."); + yield* Ref.set(thread.status, "completed"); + yield* Deferred.succeed(thread.finished, undefined); + yield* taskResult(a, delegated.taskId); + yield* Ref.set(broken, true); + yield* Queue.take(retries.failed).pipe( + Effect.andThen(Queue.offer(retries.retry, undefined)), + Effect.forever, + Effect.forkScoped, + ); + + // The Stop owes one stop there, which every attempt to send fails on. + yield* a.threads.stopDelegatedTasks({ + threadId: parentThreadId, + commandId: CommandId.make("command:stop:laptop-parent-defective-follower"), + }); + const owedAgain = a.orchestrator + .streamStoredEventsFrom({ + threadId: parentThreadId, + afterSequence: 0, + eventType: "subagent.updated", + }) + .pipe( + Stream.filter( + (stored) => + stored.event.type === "subagent.updated" && + stored.event.payload.id === delegated.taskId && + (stored.event.payload.remoteChild?.stopCount ?? 0) > 1, + ), + Stream.runHead, + Effect.flatMap((owed) => + Option.isSome(owed) ? Effect.succeed("owed another stop" as const) : Effect.never, + ), + ); + const ended = yield* Deferred.await(follower).pipe( + Effect.flatMap(Fiber.await), + Effect.as("follower ended" as const), + Effect.raceFirst(owedAgain), + ); + expect(ended).toBe("follower ended"); + const { subagents } = yield* a.threads.getThreadRecords(parentThreadId, ["subagents"], { + subagentIds: [delegated.taskId], + }); + expect(subagents[0]).toMatchObject({ + status: "completed", + result: "All green.", + remoteChild: { stoppedThere: false, stopCount: 1 }, + }); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + it.effect("stopping the parent stops follow-up work on a task there that has its result", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/peer/RemoteDelegation.ts b/apps/server/src/peer/RemoteDelegation.ts index 3b01c48b01db0..c9e38b4bc73d4 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -19,6 +19,7 @@ import { type RunId, type ThreadId, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as SecureRandom from "effect/crypto/SecureRandom"; import * as DateTime from "effect/DateTime"; @@ -53,6 +54,8 @@ import * as PeerLinks from "./PeerLinks.ts"; 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); +/** A follower that hits this many defects in a row fails its task instead of retrying. */ +const FOLLOW_MAX_DEFECTS = 3; /** How a follower retries; tests step it by hand. */ export const FollowRetrySchedule = Context.Reference>( "t3/peer/RemoteDelegation/FollowRetrySchedule", @@ -673,26 +676,54 @@ const make = Effect.gen(function* () { /** * Waits for the task's run there to end, completes the task here, and stops it there if it * ended here first. A revoked, expired or forgotten link, or a thread or run gone there, - * fails the task with nothing left to stop; anything else (the other machine asleep, a - * dropped route) is retried with back-off, and so is completing. + * fails the task with nothing left to stop; repeated defects here fail it too. Anything else + * (the other machine asleep, a dropped route) is retried with back-off, and so is completing. */ const follow = (parentThreadId: ThreadId, taskId: NodeId): Effect.Effect => - 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.followFailed")(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.suspend(() => { + let defects = 0; + const noDefect = Effect.sync(() => { + defects = 0; + }); + return followOnce(parentThreadId, taskId).pipe( + Effect.tap(noDefect), + // `false` means it is still running there: follow again straight away. + Effect.repeat({ while: (done) => !done }), + Effect.catchTags({ + OrchestratorMcpFailure: Effect.fn("RemoteDelegation.followFailed")(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 }), - ); + Effect.tapError(() => noDefect), + Effect.tapDefect(() => + Effect.sync(() => { + defects += 1; + }), + ), + failOnDefect, + Effect.retry({ schedule: followRetry, while: () => defects < FOLLOW_MAX_DEFECTS }), + Effect.catch( + Effect.fn("RemoteDelegation.followGaveUp")( + function* () { + const { task } = yield* readTask(parentThreadId, taskId); + const remote = task?.remoteChild; + // A task with its result keeps it; completing it again would owe another stop there. + if (remote === undefined || task?.result !== null) return; + const result = `Following the thread ${remote.threadId} in ${remote.label} failed repeatedly, so it may still run there (task_cancel stops it).`; + yield* complete(parentThreadId, taskId, "failed", result, undefined); + }, + failOnDefect, + Effect.retry(followRetry), + ), + ), + Effect.ignoreCause({ log: true }), + ); + }); /** * One wait on the task's thread there; true once the task here is done. @@ -976,6 +1007,14 @@ const make = Effect.gen(function* () { return RemoteDelegation.of({ delegate, followUp, cancel, start }); }); +/** A defect, logged, as a failure, so a retry covers it. */ +const failOnDefect = (effect: Effect.Effect) => + Effect.catchDefect(effect, (defect) => + Effect.logWarning("A remote task's follower hit a defect", Cause.die(defect)).pipe( + Effect.andThen(Effect.fail(new Cause.UnknownError(defect, "A follower hit a defect."))), + ), + ); + const isTerminal = (status: string) => status === "completed" || status === "failed" ||