From 04a2e60dabbb24c8c8ebef6cb9a0b6b3a2f49a28 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 7 Nov 2026 03:32:21 +0000 Subject: [PATCH 1/7] test(server): reproduce remote task followers dropped by a defect (#32278) These tests reproduce #32278: a defect in one follow attempt of a task delegated to a linked environment ends its follower instead of retrying, so the task stays running on the parent and the parent is never woken. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/peer/RemoteDelegation.test.ts | 114 ++++++++++++++++++ 1 file changed, 114 insertions(+) diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index 00b7b5d6f6963..43d69eb9d7e2e 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -23,6 +23,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"; @@ -540,6 +541,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)); @@ -568,6 +571,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)({ @@ -1637,6 +1653,104 @@ const steppedRetries = Effect.gen(function* () { return { failed, retry, schedule }; }); +/** + * Fails each follow attempt here with a defect, `times` times. Returns, once + * the first has, whether its follower retried rather than ending. + */ +const defectiveFollows = (times: number) => + Effect.gen(function* () { + const follower = yield* Deferred.make>(); + const defects = yield* Ref.make(0); + const beforeFollow = Effect.withFiber((fiber) => Deferred.succeed(follower, fiber)).pipe( + Effect.andThen(Ref.getAndUpdate(defects, (count) => count + 1)), + Effect.flatMap((count) => + count < times ? Effect.die(new Error("A follow attempt hit a bug.")) : Effect.void, + ), + ); + const retriedOrEnded = (retries: Effect.Success) => + Queue.take(retries.failed).pipe( + Effect.as("retried" as const), + Effect.raceFirst( + Deferred.await(follower).pipe( + Effect.flatMap(Fiber.await), + Effect.as("follower ended" as const), + ), + ), + ); + return { beforeFollow, retriedOrEnded }; + }); + +it.effect("a follower that hits a defect retries instead of dropping the task", () => + 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 defects = yield* defectiveFollows(1); + const a = yield* makeLaptop(database, offers, undefined, defects.beforeFollow).pipe( + Effect.provideService(RemoteDelegation.FollowRetrySchedule, retries.schedule), + ); + yield* linkTo(a.links, b, "full-access"); + yield* seedParent(a); + + const delegated = yield* delegateToBox(a); + expect(yield* defects.retriedOrEnded(retries)).toBe("retried"); + + yield* Ref.set(thread.reply, "All 412 tests pass on the box."); + yield* Ref.set(thread.status, "completed"); + yield* Deferred.succeed(thread.finished, undefined); + yield* Queue.offer(retries.retry, undefined); + expect(yield* taskResult(a, delegated.taskId)).toMatchObject({ + status: "completed", + result: "All 412 tests pass on the box.", + }); + }), + ).pipe(Effect.provide(NodeHttpServer.layerTest)), +); + +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 defects = yield* defectiveFollows(Number.POSITIVE_INFINITY); + const a = yield* makeLaptop(database, offers, undefined, defects.beforeFollow).pipe( + Effect.provideService(RemoteDelegation.FollowRetrySchedule, retries.schedule), + ); + yield* linkTo(a.links, b, "full-access"); + yield* seedParent(a); + + const delegated = yield* delegateToBox(a); + expect(yield* defects.retriedOrEnded(retries)).toBe("retried"); + // Every later attempt hits it too; each is let through to retry. + yield* Queue.offer(retries.retry, undefined); + yield* Queue.take(retries.failed).pipe( + Effect.andThen(Queue.offer(retries.retry, undefined)), + Effect.forever, + Effect.forkScoped, + ); + + const task = yield* taskResult(a, delegated.taskId); + expect(task?.status).toBe("failed"); + expect(task?.result).toContain(delegated.remoteChild.threadId); + 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)), +); + /** Waits until the task here records that the `stopCount`th Stop after its result reached there. */ const stoppedThere = ( laptopEnv: Effect.Success>, From af35e71f038d865ee96fa6f2d93572ac1dca1ab6 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 7 Nov 2026 03:41:36 +0000 Subject: [PATCH 2/7] fix(server): remote task followers retry defects instead of dropping the task A defect in one follow attempt of a task delegated to a linked environment ended its follower, because the retry only covered failures. The task stayed running on the parent until the next server start and the parent never woke. A defect is now retried on the same schedule as a failure. A follower that hits three defects in a row fails the task, naming the thread there, so the parent wakes. Interruption still just ends the follower. Fixes pingdotgg/t3code#32278 Co-Authored-By: Claude Opus 5.5 --- apps/server/src/peer/RemoteDelegation.ts | 34 ++++++++++++++++++++---- 1 file changed, 29 insertions(+), 5 deletions(-) diff --git a/apps/server/src/peer/RemoteDelegation.ts b/apps/server/src/peer/RemoteDelegation.ts index 89917026ff8f7..69fd7f6b2a71a 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -14,6 +14,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 Duration from "effect/Duration"; @@ -45,6 +46,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", @@ -508,11 +511,16 @@ 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, as do repeated defects here; 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( + const follow = (parentThreadId: ThreadId, taskId: NodeId): Effect.Effect => { + 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({ @@ -525,9 +533,25 @@ const make = Effect.gen(function* () { yield* complete(parentThreadId, taskId, "failed", reason, "stopped", stopCount); }), }), - Effect.retry(followRetry), + Effect.tapError(() => noDefect), + Effect.catchDefect((defect) => { + defects += 1; + const message = defect instanceof Error ? defect.message : String(defect); + return Effect.fail(new Cause.UnknownError(defect, message)); + }), + Effect.retry({ schedule: followRetry, while: () => defects < FOLLOW_MAX_DEFECTS }), + Effect.catch( + Effect.fn("RemoteDelegation.followGaveUp")(function* (error) { + const { task } = yield* readTask(parentThreadId, taskId); + const remote = task?.remoteChild; + if (remote === undefined) return; + const result = `Following the thread ${remote.threadId} in ${remote.label} failed: ${error.message}`; + yield* complete(parentThreadId, taskId, "failed", result, undefined); + }), + ), Effect.ignoreCause({ log: true }), ); + }; /** * One wait on the task's thread there; true once the task here is done. From 70aed7d40b535e4c0b700816402f8586ad79a673 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 7 Nov 2026 03:59:51 +0000 Subject: [PATCH 3/7] fix(server): a follower giving up leaves a finished task's owed stop alone Completing a task that already has its result records another stop owed there, which started the follower again with its defect count already at the cap: a loop of stop-owed events. A follower that gives up now leaves such a task as it is, and each follow starts its count afresh. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/peer/RemoteDelegation.test.ts | 81 +++++++++++++++++++ apps/server/src/peer/RemoteDelegation.ts | 74 ++++++++--------- 2 files changed, 119 insertions(+), 36 deletions(-) diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index 43d69eb9d7e2e..a45d60fc5a701 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -1780,6 +1780,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 69fd7f6b2a71a..21bed73f7e372 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -514,44 +514,46 @@ const make = Effect.gen(function* () { * fails the task with nothing left to stop, as do repeated defects here; 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 => { - 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); + const follow = (parentThreadId: ThreadId, taskId: NodeId): Effect.Effect => + 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.tapError(() => noDefect), - Effect.catchDefect((defect) => { - defects += 1; - const message = defect instanceof Error ? defect.message : String(defect); - return Effect.fail(new Cause.UnknownError(defect, message)); - }), - Effect.retry({ schedule: followRetry, while: () => defects < FOLLOW_MAX_DEFECTS }), - Effect.catch( - Effect.fn("RemoteDelegation.followGaveUp")(function* (error) { - const { task } = yield* readTask(parentThreadId, taskId); - const remote = task?.remoteChild; - if (remote === undefined) return; - const result = `Following the thread ${remote.threadId} in ${remote.label} failed: ${error.message}`; - yield* complete(parentThreadId, taskId, "failed", result, undefined); + Effect.tapError(() => noDefect), + Effect.catchDefect((defect) => { + defects += 1; + const message = defect instanceof Error ? defect.message : String(defect); + return Effect.fail(new Cause.UnknownError(defect, message)); }), - ), - Effect.ignoreCause({ log: true }), - ); - }; + Effect.retry({ schedule: followRetry, while: () => defects < FOLLOW_MAX_DEFECTS }), + Effect.catch( + Effect.fn("RemoteDelegation.followGaveUp")(function* (error) { + 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: ${error.message}`; + yield* complete(parentThreadId, taskId, "failed", result, undefined); + }), + ), + Effect.ignoreCause({ log: true }), + ); + }); /** * One wait on the task's thread there; true once the task here is done. From a470988cd241660462a929964249500311fe0ff7 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 7 Nov 2026 04:21:24 +0000 Subject: [PATCH 4/7] fix(server): a follower that gives up retries recording it Recording the failure after repeated defects ran after the retry, so a transient write or read failure there dropped it and left the task running. It now backs off and retries like completing does elsewhere, and its result says the thread there may still run and how to stop it. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/peer/RemoteDelegation.test.ts | 26 ++++++++++++++++--- apps/server/src/peer/RemoteDelegation.ts | 8 +++--- 2 files changed, 26 insertions(+), 8 deletions(-) diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index a45d60fc5a701..f98fbae98fe88 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -543,6 +543,8 @@ const makeLaptop = ( kept?: Context.Context>>, /** Runs as each follow attempt here starts, on its follower's fiber. */ beforeFollow: Effect.Effect = Effect.void, + /** Runs before each read of a thread's records here, as the follower makes them. */ + beforeRead: Effect.Effect = Effect.void, ) => Effect.gen(function* () { const linking = kept ?? (yield* layerLinkingEnvironment(laptop).pipe(Layer.build)); @@ -580,6 +582,10 @@ const makeLaptop = ( ...threads, getThreadEventSequence: (threadId: ThreadId) => beforeFollow.pipe(Effect.andThen(threads.getThreadEventSequence(threadId))), + getThreadRecords: (threadId, fields, filter) => + beforeRead.pipe( + Effect.andThen(threads.getThreadRecords(threadId, fields, filter)), + ), }; }), ), @@ -1677,7 +1683,14 @@ const defectiveFollows = (times: number) => ), ), ); - return { beforeFollow, retriedOrEnded }; + // Once it has hit one, the follower's only read here is the one that records giving up. + const failedRecords = yield* Ref.make(1); + const beforeRead = Effect.gen(function* () { + if ((yield* Ref.get(defects)) === 0) return; + if ((yield* Ref.getAndUpdate(failedRecords, (count) => count - 1)) <= 0) return; + return yield* new Orchestrator.OrchestratorProjectionError({ threadId: parentThreadId }); + }); + return { beforeFollow, beforeRead, retriedOrEnded }; }); it.effect("a follower that hits a defect retries instead of dropping the task", () => @@ -1723,9 +1736,14 @@ it.effect("a follower that keeps hitting defects fails the task and wakes the pa const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); const retries = yield* steppedRetries; const defects = yield* defectiveFollows(Number.POSITIVE_INFINITY); - const a = yield* makeLaptop(database, offers, undefined, defects.beforeFollow).pipe( - Effect.provideService(RemoteDelegation.FollowRetrySchedule, retries.schedule), - ); + // Recording that it gave up fails once too, and is retried. + const a = yield* makeLaptop( + database, + offers, + undefined, + defects.beforeFollow, + defects.beforeRead, + ).pipe(Effect.provideService(RemoteDelegation.FollowRetrySchedule, retries.schedule)); yield* linkTo(a.links, b, "full-access"); yield* seedParent(a); diff --git a/apps/server/src/peer/RemoteDelegation.ts b/apps/server/src/peer/RemoteDelegation.ts index 21bed73f7e372..db691bc889811 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -511,8 +511,8 @@ 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, as do repeated defects here; 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 => Effect.suspend(() => { @@ -547,9 +547,9 @@ const make = Effect.gen(function* () { 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: ${error.message}`; + const result = `Following the thread ${remote.threadId} in ${remote.label} failed, so it may still run there (task_cancel stops it): ${error.message}`; yield* complete(parentThreadId, taskId, "failed", result, undefined); - }), + }, Effect.retry(followRetry)), ), Effect.ignoreCause({ log: true }), ); From ef69b3b1644630c5a586b8652f148f38b75b356d Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 7 Nov 2026 04:46:09 +0000 Subject: [PATCH 5/7] fix(server): a defect while recording a give-up is retried too Co-Authored-By: Claude Opus 5.5 --- apps/server/src/peer/RemoteDelegation.test.ts | 4 +- apps/server/src/peer/RemoteDelegation.ts | 39 ++++++++++++------- 2 files changed, 28 insertions(+), 15 deletions(-) diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index f98fbae98fe88..7f93b89d68f15 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -1688,7 +1688,7 @@ const defectiveFollows = (times: number) => const beforeRead = Effect.gen(function* () { if ((yield* Ref.get(defects)) === 0) return; if ((yield* Ref.getAndUpdate(failedRecords, (count) => count - 1)) <= 0) return; - return yield* new Orchestrator.OrchestratorProjectionError({ threadId: parentThreadId }); + return yield* Effect.die(new Error("Recording it hit a bug.")); }); return { beforeFollow, beforeRead, retriedOrEnded }; }); @@ -1736,7 +1736,7 @@ it.effect("a follower that keeps hitting defects fails the task and wakes the pa const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); const retries = yield* steppedRetries; const defects = yield* defectiveFollows(Number.POSITIVE_INFINITY); - // Recording that it gave up fails once too, and is retried. + // Recording that it gave up hits a defect once too, and is retried. const a = yield* makeLaptop( database, offers, diff --git a/apps/server/src/peer/RemoteDelegation.ts b/apps/server/src/peer/RemoteDelegation.ts index db691bc889811..023889e81c9ad 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -535,21 +535,26 @@ const make = Effect.gen(function* () { }), }), Effect.tapError(() => noDefect), - Effect.catchDefect((defect) => { - defects += 1; - const message = defect instanceof Error ? defect.message : String(defect); - return Effect.fail(new Cause.UnknownError(defect, message)); - }), + Effect.tapDefect(() => + Effect.sync(() => { + defects += 1; + }), + ), + failOnDefect, Effect.retry({ schedule: followRetry, while: () => defects < FOLLOW_MAX_DEFECTS }), Effect.catch( - Effect.fn("RemoteDelegation.followGaveUp")(function* (error) { - 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, so it may still run there (task_cancel stops it): ${error.message}`; - yield* complete(parentThreadId, taskId, "failed", result, undefined); - }, Effect.retry(followRetry)), + Effect.fn("RemoteDelegation.followGaveUp")( + function* (error) { + 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, so it may still run there (task_cancel stops it): ${error.message}`; + yield* complete(parentThreadId, taskId, "failed", result, undefined); + }, + failOnDefect, + Effect.retry(followRetry), + ), ), Effect.ignoreCause({ log: true }), ); @@ -814,6 +819,14 @@ const make = Effect.gen(function* () { return RemoteDelegation.of({ delegate, cancel, start }); }); +/** A defect as a failure, so a retry covers it. */ +const failOnDefect = (effect: Effect.Effect) => + Effect.catchDefect(effect, (defect) => + Effect.fail( + new Cause.UnknownError(defect, defect instanceof Error ? defect.message : String(defect)), + ), + ); + const isTerminal = (status: string) => status === "completed" || status === "failed" || From 834d3432fed9de9d8a281aa4bcb7caf54b0cee11 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 16:53:41 +0000 Subject: [PATCH 6/7] fix(server): keep defect text out of a given-up remote task's result and log handled defects A follower that gave up after repeated defects put the raw defect message into the task result, which agents see and the server persists. The result is now a fixed message naming the thread there and its environment; the defect stays only as the error's cause. Each defect the follower turns into a retry is logged once with its cause, since recovery used to swallow it. Tests pin the give-up threshold at three attempts, check the result leaves out the defect's text, and show that defects separated by a still-running wait or a failed attempt do not add up to giving up. AI-assisted; verified with the tests and tools listed above. --- apps/server/src/peer/RemoteDelegation.test.ts | 74 ++++++++++++++++++- apps/server/src/peer/RemoteDelegation.ts | 10 +-- 2 files changed, 76 insertions(+), 8 deletions(-) diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index 618b67606008d..649ccf6aba31a 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -553,7 +553,7 @@ 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, + beforeFollow: Effect.Effect = Effect.void, /** Runs before each read of a thread's records here, as the follower makes them. */ beforeRead: Effect.Effect = Effect.void, ) => @@ -1673,6 +1673,9 @@ const steppedRetries = Effect.gen(function* () { return { failed, retry, schedule }; }); +/** Internal detail in a follower's defect, which no task result may carry. */ +const defectDetail = "defect-detail-4f1c9e"; + /** * Fails each follow attempt here with a defect, `times` times. Returns, once * the first has, whether its follower retried rather than ending. @@ -1684,7 +1687,9 @@ const defectiveFollows = (times: number) => const beforeFollow = Effect.withFiber((fiber) => Deferred.succeed(follower, fiber)).pipe( Effect.andThen(Ref.getAndUpdate(defects, (count) => count + 1)), Effect.flatMap((count) => - count < times ? Effect.die(new Error("A follow attempt hit a bug.")) : Effect.void, + count < times + ? Effect.die(new Error(`A follow attempt hit a bug: ${defectDetail}`)) + : Effect.void, ), ); const retriedOrEnded = (retries: Effect.Success) => @@ -1704,7 +1709,7 @@ const defectiveFollows = (times: number) => if ((yield* Ref.getAndUpdate(failedRecords, (count) => count - 1)) <= 0) return; return yield* Effect.die(new Error("Recording it hit a bug.")); }); - return { beforeFollow, beforeRead, retriedOrEnded }; + return { attempts: defects, beforeFollow, beforeRead, retriedOrEnded }; }); it.effect("a follower that hits a defect retries instead of dropping the task", () => @@ -1772,8 +1777,10 @@ it.effect("a follower that keeps hitting defects fails the task and wakes the pa ); const task = yield* taskResult(a, delegated.taskId); + expect(yield* Ref.get(defects.attempts)).toBe(3); expect(task?.status).toBe("failed"); expect(task?.result).toContain(delegated.remoteChild.threadId); + expect(task?.result).not.toContain(defectDetail); const offered = yield* SubscriptionRef.changes(offers).pipe( Stream.filter((seen) => seen.some((offer) => offer.threadId === parentThreadId)), Stream.runHead, @@ -1783,6 +1790,67 @@ it.effect("a follower that keeps hitting defects fails the task and wakes the pa ).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>, diff --git a/apps/server/src/peer/RemoteDelegation.ts b/apps/server/src/peer/RemoteDelegation.ts index 943e2bcc6f2d6..47d3656a422a5 100644 --- a/apps/server/src/peer/RemoteDelegation.ts +++ b/apps/server/src/peer/RemoteDelegation.ts @@ -552,12 +552,12 @@ const make = Effect.gen(function* () { Effect.retry({ schedule: followRetry, while: () => defects < FOLLOW_MAX_DEFECTS }), Effect.catch( Effect.fn("RemoteDelegation.followGaveUp")( - function* (error) { + 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, so it may still run there (task_cancel stops it): ${error.message}`; + 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, @@ -827,11 +827,11 @@ const make = Effect.gen(function* () { return RemoteDelegation.of({ delegate, cancel, start }); }); -/** A defect as a failure, so a retry covers it. */ +/** A defect, logged, as a failure, so a retry covers it. */ const failOnDefect = (effect: Effect.Effect) => Effect.catchDefect(effect, (defect) => - Effect.fail( - new Cause.UnknownError(defect, defect instanceof Error ? defect.message : String(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."))), ), ); From 52170e378e823fb012a95f84007c0e9f73bd9429 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 20:47:23 +0000 Subject: [PATCH 7/7] test(server): cut remote follower defect tests to the regressions AI-assisted; verified with the tests and tools listed above. --- apps/server/src/peer/RemoteDelegation.test.ts | 102 ++---------------- 1 file changed, 10 insertions(+), 92 deletions(-) diff --git a/apps/server/src/peer/RemoteDelegation.test.ts b/apps/server/src/peer/RemoteDelegation.test.ts index 649ccf6aba31a..a414ec08781d4 100644 --- a/apps/server/src/peer/RemoteDelegation.test.ts +++ b/apps/server/src/peer/RemoteDelegation.test.ts @@ -554,8 +554,6 @@ const makeLaptop = ( kept?: Context.Context>>, /** Runs as each follow attempt here starts, on its follower's fiber. */ beforeFollow: Effect.Effect = Effect.void, - /** Runs before each read of a thread's records here, as the follower makes them. */ - beforeRead: Effect.Effect = Effect.void, ) => Effect.gen(function* () { const linking = kept ?? (yield* layerLinkingEnvironment(laptop).pipe(Layer.build)); @@ -593,10 +591,6 @@ const makeLaptop = ( ...threads, getThreadEventSequence: (threadId: ThreadId) => beforeFollow.pipe(Effect.andThen(threads.getThreadEventSequence(threadId))), - getThreadRecords: (threadId, fields, filter) => - beforeRead.pipe( - Effect.andThen(threads.getThreadRecords(threadId, fields, filter)), - ), }; }), ), @@ -1673,46 +1667,7 @@ const steppedRetries = Effect.gen(function* () { return { failed, retry, schedule }; }); -/** Internal detail in a follower's defect, which no task result may carry. */ -const defectDetail = "defect-detail-4f1c9e"; - -/** - * Fails each follow attempt here with a defect, `times` times. Returns, once - * the first has, whether its follower retried rather than ending. - */ -const defectiveFollows = (times: number) => - Effect.gen(function* () { - const follower = yield* Deferred.make>(); - const defects = yield* Ref.make(0); - const beforeFollow = Effect.withFiber((fiber) => Deferred.succeed(follower, fiber)).pipe( - Effect.andThen(Ref.getAndUpdate(defects, (count) => count + 1)), - Effect.flatMap((count) => - count < times - ? Effect.die(new Error(`A follow attempt hit a bug: ${defectDetail}`)) - : Effect.void, - ), - ); - const retriedOrEnded = (retries: Effect.Success) => - Queue.take(retries.failed).pipe( - Effect.as("retried" as const), - Effect.raceFirst( - Deferred.await(follower).pipe( - Effect.flatMap(Fiber.await), - Effect.as("follower ended" as const), - ), - ), - ); - // Once it has hit one, the follower's only read here is the one that records giving up. - const failedRecords = yield* Ref.make(1); - const beforeRead = Effect.gen(function* () { - if ((yield* Ref.get(defects)) === 0) return; - if ((yield* Ref.getAndUpdate(failedRecords, (count) => count - 1)) <= 0) return; - return yield* Effect.die(new Error("Recording it hit a bug.")); - }); - return { attempts: defects, beforeFollow, beforeRead, retriedOrEnded }; - }); - -it.effect("a follower that hits a defect retries instead of dropping the task", () => +it.effect("a follower that keeps hitting defects fails the task and wakes the parent", () => Effect.scoped( Effect.gen(function* () { const thread = yield* makeBoxThread; @@ -1722,65 +1677,28 @@ it.effect("a follower that hits a defect retries instead of dropping the task", >([]); const database = yield* SqlitePersistence.layerMemory.pipe(Layer.build); const retries = yield* steppedRetries; - const defects = yield* defectiveFollows(1); - const a = yield* makeLaptop(database, offers, undefined, defects.beforeFollow).pipe( + 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); - - const delegated = yield* delegateToBox(a); - expect(yield* defects.retriedOrEnded(retries)).toBe("retried"); - - yield* Ref.set(thread.reply, "All 412 tests pass on the box."); - yield* Ref.set(thread.status, "completed"); - yield* Deferred.succeed(thread.finished, undefined); - yield* Queue.offer(retries.retry, undefined); - expect(yield* taskResult(a, delegated.taskId)).toMatchObject({ - status: "completed", - result: "All 412 tests pass on the box.", - }); - }), - ).pipe(Effect.provide(NodeHttpServer.layerTest)), -); - -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 defects = yield* defectiveFollows(Number.POSITIVE_INFINITY); - // Recording that it gave up hits a defect once too, and is retried. - const a = yield* makeLaptop( - database, - offers, - undefined, - defects.beforeFollow, - defects.beforeRead, - ).pipe(Effect.provideService(RemoteDelegation.FollowRetrySchedule, retries.schedule)); - yield* linkTo(a.links, b, "full-access"); - yield* seedParent(a); - - const delegated = yield* delegateToBox(a); - expect(yield* defects.retriedOrEnded(retries)).toBe("retried"); - // Every later attempt hits it too; each is let through to retry. - yield* Queue.offer(retries.retry, undefined); 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(defects.attempts)).toBe(3); + expect(yield* Ref.get(attempts)).toBe(3); expect(task?.status).toBe("failed"); expect(task?.result).toContain(delegated.remoteChild.threadId); - expect(task?.result).not.toContain(defectDetail); + 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,