Skip to content

Commit 867112c

Browse files
SunkenInTimeclaudejuliusmarminge
authored
fix(server): agent questions stay answerable after a server restart (#19229)
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: Julius Marminge <julius0216@outlook.com>
1 parent cb7e381 commit 867112c

7 files changed

Lines changed: 718 additions & 147 deletions

‎apps/server/src/orchestration-v2/DelegatedCompletionDelivery.test.ts‎

Lines changed: 172 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,11 +15,14 @@ import {
1515
ProviderInstanceId,
1616
ProviderThreadId,
1717
RunId,
18+
RuntimeRequestId,
1819
ThreadId,
1920
TurnItemId,
2021
} from "@t3tools/contracts";
2122
import * as DateTime from "effect/DateTime";
2223
import * as Effect from "effect/Effect";
24+
import * as Deferred from "effect/Deferred";
25+
import * as Fiber from "effect/Fiber";
2326
import * as Layer from "effect/Layer";
2427
import * as Stream from "effect/Stream";
2528

@@ -40,6 +43,7 @@ import * as EventSink from "./EventSink.ts";
4043
import * as Orchestrator from "./Orchestrator.ts";
4144
import { continueRestartedRun } from "./RestartContinuation.ts";
4245
import * as RuntimeLayer from "./runtimeLayer.ts";
46+
import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
4347
import * as ProviderTurnStartServiceTestkit from "./ProviderTurnStartService.testkit.ts";
4448
import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
4549
import * as ProviderContinuationRequests from "@t3tools/provider-core/server/ProviderContinuationRequests";
@@ -104,7 +108,11 @@ const layerTestProviderInstanceRegistry = Layer.succeed(
104108
},
105109
);
106110

107-
const layerTest = Layer.mergeAll(RuntimeLayer.layer, RuntimeLayer.layerEventSink).pipe(
111+
const layerTest = Layer.mergeAll(
112+
RuntimeLayer.layer,
113+
RuntimeLayer.layerEventSink,
114+
ThreadCommandExecutor.layer,
115+
).pipe(
108116
Layer.provideMerge(RuntimeLayer.layerProjectService),
109117
Layer.provide(
110118
Layer.mock(WorkspacePaths.WorkspacePaths)({
@@ -1980,6 +1988,169 @@ it.layer(Layer.merge(layerTest, ProviderContinuationRequests.layer))(
19801988
}
19811989
}),
19821990
);
1991+
1992+
it.effect("serializes a recovered child answer with its parent and tracks the follow-up", () =>
1993+
Effect.gen(function* () {
1994+
const orchestrator = yield* Orchestrator.OrchestratorV2;
1995+
const eventSink = yield* EventSink.EventSinkV2;
1996+
const now = yield* DateTime.now;
1997+
const parentThreadId = ThreadId.make("thread:question-answer-parent");
1998+
const projectId = ProjectId.make("project:question-answer");
1999+
const parentRunId = RunId.make("run:question-answer-parent");
2000+
const rootNodeId = NodeId.make("node:question-answer-parent");
2001+
yield* seedParentWithTerminalTask({
2002+
threadId: parentThreadId,
2003+
projectId,
2004+
runId: parentRunId,
2005+
rootNodeId,
2006+
taskId: NodeId.make("node:question-answer-seed"),
2007+
deliveryState: "acknowledged",
2008+
now,
2009+
});
2010+
const child = yield* seedRestartCancelledChild({
2011+
parentThreadId,
2012+
projectId,
2013+
parentRunId,
2014+
rootNodeId,
2015+
name: "question-answer-child",
2016+
completionWake: "always",
2017+
continuationPending: false,
2018+
now,
2019+
});
2020+
// The cancelled result is published before anyone answers the question.
2021+
yield* orchestrator.recoverDelegatedTasks;
2022+
assert.isNotNull(
2023+
(yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find(
2024+
(task) => task.id === child.taskId,
2025+
)?.result,
2026+
);
2027+
const parentRun = (yield* orchestrator.getThreadProjection(parentThreadId)).runs.find(
2028+
(run) => run.id === parentRunId,
2029+
)!;
2030+
yield* eventSink.write({
2031+
events: [
2032+
{
2033+
id: EventId.make("event:question-answer-parent-completed"),
2034+
type: "run.updated",
2035+
threadId: parentThreadId,
2036+
runId: parentRunId,
2037+
occurredAt: now,
2038+
payload: { ...parentRun, status: "completed", completedAt: now },
2039+
},
2040+
],
2041+
});
2042+
const requestId = RuntimeRequestId.make("request:question-answer");
2043+
const nodeId = NodeId.make("node:question-answer-request");
2044+
yield* eventSink.write({
2045+
commandId: CommandId.make("command:seed-question-answer"),
2046+
events: [
2047+
{
2048+
id: EventId.make("event:question-answer-request"),
2049+
type: "runtime-request.updated",
2050+
threadId: child.childThreadId,
2051+
nodeId,
2052+
occurredAt: now,
2053+
payload: {
2054+
id: requestId,
2055+
nodeId,
2056+
providerTurnId: null,
2057+
nativeRequestRef: null,
2058+
kind: "user_input",
2059+
status: "pending",
2060+
responseCapability: { type: "message" },
2061+
createdAt: now,
2062+
resolvedAt: null,
2063+
},
2064+
},
2065+
{
2066+
id: EventId.make("event:question-answer-item"),
2067+
type: "turn-item.updated",
2068+
threadId: child.childThreadId,
2069+
nodeId,
2070+
occurredAt: now,
2071+
payload: {
2072+
id: TurnItemId.make("turn-item:question-answer"),
2073+
type: "user_input_request",
2074+
threadId: child.childThreadId,
2075+
runId: child.childRunId,
2076+
nodeId,
2077+
providerThreadId: null,
2078+
providerTurnId: null,
2079+
nativeItemRef: null,
2080+
parentItemId: null,
2081+
ordinal: 0,
2082+
status: "waiting",
2083+
title: null,
2084+
startedAt: now,
2085+
completedAt: null,
2086+
updatedAt: now,
2087+
requestId,
2088+
questions: [
2089+
{ id: "color", header: "Color", question: "Which color?", options: [] },
2090+
],
2091+
},
2092+
},
2093+
],
2094+
});
2095+
2096+
const executor = yield* ThreadCommandExecutor.ThreadCommandExecutor;
2097+
const parentHeld = yield* Deferred.make<void>();
2098+
const releaseParent = yield* Deferred.make<void>();
2099+
const answerWaiting = yield* Deferred.make<void>();
2100+
const withLock = executor.withLock;
2101+
const parentMutation = yield* withLock(
2102+
parentThreadId,
2103+
Deferred.succeed(parentHeld, undefined).pipe(
2104+
Effect.andThen(Deferred.await(releaseParent)),
2105+
),
2106+
).pipe(Effect.forkScoped({ startImmediately: true }));
2107+
yield* Deferred.await(parentHeld);
2108+
const observeLock: ThreadCommandExecutor.ThreadCommandExecutor["Service"]["withLock"] = (
2109+
key,
2110+
effect,
2111+
) =>
2112+
key === parentThreadId
2113+
? Deferred.succeed(answerWaiting, undefined).pipe(Effect.andThen(withLock(key, effect)))
2114+
: withLock(key, effect);
2115+
const spy = vi.spyOn(executor, "withLock").mockImplementation(observeLock);
2116+
yield* Effect.gen(function* () {
2117+
const answer = yield* orchestrator
2118+
.dispatch({
2119+
type: "runtime-request.respond",
2120+
commandId: CommandId.make("command:question-answer"),
2121+
threadId: child.childThreadId,
2122+
requestId,
2123+
answers: { color: "Blue" },
2124+
})
2125+
.pipe(Effect.forkScoped({ startImmediately: true }));
2126+
yield* Effect.raceFirst(Deferred.await(answerWaiting), Fiber.join(answer));
2127+
const waiting = yield* orchestrator.getThreadProjection(child.childThreadId);
2128+
assert.equal(
2129+
waiting.runtimeRequests.find((row) => row.id === requestId)?.status,
2130+
"pending",
2131+
);
2132+
assert.isFalse((yield* executor.activeKeys).includes(child.childThreadId));
2133+
yield* Deferred.succeed(releaseParent, undefined);
2134+
yield* Fiber.join(parentMutation);
2135+
yield* Fiber.join(answer);
2136+
}).pipe(Effect.ensuring(Effect.sync(() => spy.mockRestore())));
2137+
2138+
const answerRun = (yield* orchestrator.getThreadProjection(child.childThreadId)).runs.find(
2139+
(run) => run.userMessageId === MessageId.make(`async-answer:${requestId}`),
2140+
);
2141+
assert.isDefined(answerRun?.delegatedTaskId);
2142+
const followup = (yield* orchestrator.getThreadProjection(parentThreadId)).subagents.find(
2143+
(task) => task.id === answerRun?.delegatedTaskId,
2144+
);
2145+
assert.equal(followup?.childThreadId, child.childThreadId);
2146+
assert.isNull(followup?.result);
2147+
assert.isTrue(
2148+
(yield* orchestrator.getThreadShell(parentThreadId))?.pendingBackgroundTasks?.some(
2149+
(task) => task.kind === "subagent" && task.childThreadId === child.childThreadId,
2150+
),
2151+
);
2152+
}),
2153+
);
19832154
},
19842155
);
19852156

0 commit comments

Comments
 (0)