Skip to content

Commit d81afa0

Browse files
t3dotggclaude
andauthored
fix(muse): workflow subagents no longer stall on hidden approvals (pingdotgg#17329)
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
1 parent 48b71f0 commit d81afa0

3 files changed

Lines changed: 187 additions & 66 deletions

File tree

‎apps/server/src/orchestration-v2/Adapters/MuseAdapterV2.test.ts‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -809,6 +809,65 @@ describe("MuseAdapterV2", () => {
809809
}).pipe(Effect.scoped, Effect.provide(testLayer)),
810810
);
811811

812+
it.effect("approves a workflow child's approval in full access after its turn ended", () =>
813+
Effect.gen(function* () {
814+
const fake = yield* makeFakeMuse();
815+
const harness = yield* makeHarness(fake);
816+
const { nativeId } = yield* startConversation(harness, fake);
817+
yield* fake.emit("turn/completed", { turnId: nativeId, terminal: "completed" });
818+
yield* harness.takeEvent("turn.terminal");
819+
// Muse runs workflow children without the session's allowAll mode.
820+
const childApproval = { ...approval("child-run-1"), subagentOrigin: { subagentId: "a7" } };
821+
yield* fake.emit("approval/requested", childApproval);
822+
const asked = yield* harness.takeEvent("runtime_request.updated");
823+
assert.strictEqual(asked.runtimeRequest.status, "pending");
824+
assert.isNull(asked.runtimeRequest.providerTurnId);
825+
const decision = yield* fake.takeCall("approval/decide");
826+
assert.strictEqual(decision.params.choiceId, "once");
827+
yield* fake.emit("approval/resolved", childApproval);
828+
const resolved = yield* harness.takeEvent("runtime_request.updated");
829+
assert.strictEqual(resolved.runtimeRequest.status, "resolved");
830+
assert.strictEqual(resolved.runtimeRequest.decision, "accept");
831+
}).pipe(Effect.scoped, Effect.provide(testLayer)),
832+
);
833+
834+
it.effect("keeps a workflow child's approval pending for the user past the turn's end", () =>
835+
Effect.gen(function* () {
836+
const fake = yield* makeFakeMuse();
837+
const policy = ProviderAdapterV2RuntimePolicy.make({
838+
...runtimePolicy,
839+
runtimeMode: "approval-required",
840+
});
841+
const harness = yield* makeHarness(
842+
fake,
843+
INSTANCE_ID,
844+
undefined,
845+
undefined,
846+
undefined,
847+
policy,
848+
);
849+
const { nativeId } = yield* startConversation(harness, fake);
850+
const childApproval = { ...approval("child-run-1"), subagentOrigin: { subagentId: "a7" } };
851+
yield* fake.emit("approval/requested", childApproval);
852+
const asked = yield* harness.takeEvent("runtime_request.updated");
853+
assert.isNull(asked.runtimeRequest.providerTurnId);
854+
yield* fake.emit("turn/completed", { turnId: nativeId, terminal: "completed" });
855+
yield* harness.takeEvent("turn.terminal");
856+
assert.isFalse(
857+
harness.allEvents.some(
858+
(event) =>
859+
event.type === "runtime_request.updated" && event.runtimeRequest.status !== "pending",
860+
),
861+
);
862+
yield* harness.runtime.respondToRuntimeRequest({
863+
requestId: asked.runtimeRequest.id,
864+
decision: "decline",
865+
});
866+
const decision = yield* fake.takeCall("approval/decide");
867+
assert.strictEqual(decision.params.choiceId, "deny");
868+
}).pipe(Effect.scoped, Effect.provide(testLayer)),
869+
);
870+
812871
it.effect("validates structured answers and preserves selected labels with a custom note", () =>
813872
Effect.gen(function* () {
814873
const fake = yield* makeFakeMuse();

‎apps/server/src/orchestration-v2/Adapters/MuseAdapterV2.ts‎

Lines changed: 126 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import {
1818
type ProviderInstanceId,
1919
type PlanId,
2020
type OrchestrationV2PlanStep,
21+
type ProviderApprovalDecision,
2122
type RuntimeRequestId,
2223
type ServerProviderModel,
2324
} from "@t3tools/contracts";
@@ -157,7 +158,7 @@ const MuseProviderCapabilitiesV2 = {
157158
supportsApplyPatchApproval: false,
158159
approvalsHaveNativeRequestIds: true,
159160
approvalCallbacksAreLiveOnly: true,
160-
approvalsCanOriginateFromSubagents: false,
161+
approvalsCanOriginateFromSubagents: true,
161162
},
162163
planning: {
163164
emitsPlanUpdated: true,
@@ -664,6 +665,8 @@ export function makeMuseAdapterV2(options: MuseAdapterV2Options): ProviderAdapte
664665
}
665666
observedChildren.clear();
666667
if (thread?.pendingBackgroundTasks?.length) yield* syncBackground();
668+
// Only workflow children's requests outlive a turn; they end with the workflow.
669+
for (const entry of pending.values()) yield* resolvePending(entry, "cancelled");
667670
});
668671
const resolvePending = Effect.fnUntraced(function* (
669672
entry: PendingRequest,
@@ -718,7 +721,9 @@ export function makeMuseAdapterV2(options: MuseAdapterV2Options): ProviderAdapte
718721
yield* publishItem(turn, item, item.status === "inProgress" ? status : undefined);
719722
}
720723
turn.dirty.clear();
721-
for (const entry of pending.values()) yield* resolvePending(entry, "cancelled");
724+
// A workflow child's request is runless and keeps waiting after the turn ends.
725+
for (const entry of pending.values())
726+
if (entry.request.providerTurnId !== null) yield* resolvePending(entry, "cancelled");
722727
turn.providerTurn = { ...turn.providerTurn, status, completedAt };
723728
lastProviderTurn = turn.providerTurn;
724729
yield* emit({
@@ -801,28 +806,54 @@ export function makeMuseAdapterV2(options: MuseAdapterV2Options): ProviderAdapte
801806
});
802807
const owns = (turn: ActiveTurn, turnId: string) =>
803808
turnId === turn.nativeId || turn.joined.has(turnId);
809+
/** Sends T3's decision for a pending approval to Muse. */
810+
const decideApproval = Effect.fnUntraced(function* (
811+
entry: PendingRequest,
812+
decision: ProviderApprovalDecision | undefined,
813+
) {
814+
if (entry.native.type !== "approval") return;
815+
const approval = entry.native.value;
816+
const choice = decision && museApprovalChoices(approval).get(decision);
817+
if (!choice) return yield* protocolError("Muse did not offer this approval decision");
818+
// Recorded first: Muse can settle the approval before it acknowledges the command.
819+
entry.response = { decision };
820+
yield* request("approval/decide", {
821+
approvalId: approval.approvalId,
822+
requirementId: approval.currentRequirementId,
823+
choiceId: choice.choiceId,
824+
}).pipe(Effect.tapError(() => Effect.sync(() => delete entry.response)));
825+
});
826+
/**
827+
* Shows a Muse approval or question as a T3 request. A workflow child's approval
828+
* is runless: the turn that started the workflow may end long before the child
829+
* asks, and T3 keeps a runless request on the thread until Muse settles it.
830+
*/
804831
const publishRequest = Effect.fnUntraced(function* (native: PendingRequest["native"]) {
805-
const turn = active;
832+
const child = native.type === "approval" && native.value.subagentOrigin !== undefined;
833+
const turn = child ? undefined : active;
806834
if (
807-
!turn ||
835+
!thread ||
836+
(!child && !turn) ||
808837
native.value.sessionId !== nativeSessionId ||
809-
(native.value.turnId && !owns(turn, native.value.turnId))
838+
(turn && native.value.turnId && !owns(turn, native.value.turnId))
810839
)
811840
return;
812841
const nativeId =
813842
native.type === "approval" ? native.value.approvalId : native.value.userInputId;
814-
if (turn.settledRequests.has(`${native.type}:${nativeId}`)) return;
843+
if (turn?.settledRequests.has(`${native.type}:${nativeId}`)) return;
815844
const previous = [...pending.values()].find(
816845
(entry) =>
817846
entry.native.type === native.type &&
818847
entry.request.nativeRequestRef?.nativeId === nativeId,
819848
);
849+
const providerTurnId = turn?.providerTurn.id ?? null;
850+
const threadId = turn?.input.threadId ?? thread.appThreadId ?? input.threadId;
820851
const requestId =
821852
previous?.request.id ??
822853
(yield* idAllocator.allocate
823854
.runtimeRequest({
824855
driver: MUSE_PROVIDER,
825-
providerTurnId: turn.providerTurn.id,
856+
...(providerTurnId ? { providerTurnId } : {}),
826857
nativeRequestId: nativeId,
827858
})
828859
.pipe(
@@ -843,7 +874,7 @@ export function makeMuseAdapterV2(options: MuseAdapterV2Options): ProviderAdapte
843874
const runtimeRequest: OrchestrationV2RuntimeRequest = {
844875
id: requestId,
845876
nodeId,
846-
providerTurnId: turn.providerTurn.id,
877+
providerTurnId,
847878
nativeRequestRef: nativeRef(nativeId),
848879
kind,
849880
status: "pending",
@@ -853,23 +884,37 @@ export function makeMuseAdapterV2(options: MuseAdapterV2Options): ProviderAdapte
853884
};
854885
const node: OrchestrationV2ExecutionNode = {
855886
id: nodeId,
856-
threadId: turn.input.threadId,
857-
runId: turn.input.runId,
858-
parentNodeId: turn.input.rootNodeId,
859-
rootNodeId: turn.input.rootNodeId,
887+
threadId,
888+
runId: turn?.input.runId ?? null,
889+
parentNodeId: turn?.input.rootNodeId ?? null,
890+
rootNodeId: turn?.input.rootNodeId ?? nodeId,
860891
kind: native.type === "approval" ? "approval_request" : "user_input_request",
861892
status: "waiting",
862893
countsForRun: false,
863-
providerThreadId: turn.input.providerThread.id,
864-
providerTurnId: turn.providerTurn.id,
894+
providerThreadId: thread.id,
895+
providerTurnId,
865896
nativeItemRef: nativeRef(nativeId),
866897
runtimeRequestId: requestId,
867898
checkpointScopeId: null,
868899
startedAt: runtimeRequest.createdAt,
869900
completedAt: null,
870901
};
902+
const latestTurn = active?.providerTurn ?? lastProviderTurn;
871903
const base = {
872-
...baseItem(turn, `request:${nativeId}`, time),
904+
...(turn
905+
? baseItem(turn, `request:${nativeId}`, time)
906+
: {
907+
threadId,
908+
runId: null,
909+
providerThreadId: thread.id,
910+
providerTurnId: null,
911+
nativeItemRef: nativeRef(nativeId),
912+
parentItemId: null,
913+
// Sorts after the latest turn's items.
914+
ordinal: previous?.item.ordinal ?? ((latestTurn?.ordinal ?? 0) + 1) * 100,
915+
startedAt: runtimeRequest.createdAt,
916+
updatedAt: time,
917+
}),
873918
id: idAllocator.derive.approvalTurnItem({ requestId }),
874919
nodeId,
875920
status: "waiting" as const,
@@ -909,16 +954,71 @@ export function makeMuseAdapterV2(options: MuseAdapterV2Options): ProviderAdapte
909954
required: true,
910955
})),
911956
};
912-
pending.set(requestId, { native, request: runtimeRequest, node, item });
957+
const entry: PendingRequest = { native, request: runtimeRequest, node, item };
958+
pending.set(requestId, entry);
913959
yield* emit({
914960
type: "runtime_request.updated",
915961
driver: MUSE_PROVIDER,
916-
threadId: turn.input.threadId,
962+
threadId,
917963
runtimeRequest,
918964
});
919965
yield* emit({ type: "node.updated", driver: MUSE_PROVIDER, node });
920966
yield* emit({ type: "turn_item.updated", driver: MUSE_PROVIDER, turnItem: item });
921967
yield* updateSession("waiting");
968+
// Muse does not give workflow children the session's allowAll mode, so in full
969+
// access T3 approves for them. If Muse rejects that, the request waits for the user.
970+
if (child && input.runtimePolicy.runtimeMode === "full-access")
971+
yield* decideApproval(entry, "accept").pipe(Effect.ignore, Effect.forkIn(scope));
972+
});
973+
/** Approval and question events; `turn` is undefined for a workflow child's approval. */
974+
const handleRequestEvent = Effect.fnUntraced(function* (
975+
method: string,
976+
params: Record<string, unknown>,
977+
turn: ActiveTurn | undefined,
978+
) {
979+
switch (method) {
980+
case "approval/requested":
981+
case "approval/updated":
982+
if (
983+
method === "approval/updated" &&
984+
![...pending.values()].some(
985+
(entry) =>
986+
entry.native.type === "approval" &&
987+
entry.native.value.approvalId === params.approvalId,
988+
)
989+
)
990+
return;
991+
return yield* publishRequest({
992+
type: "approval",
993+
value: yield* decode(MuseApproval, params),
994+
});
995+
case "userInput/requested":
996+
return yield* publishRequest({
997+
type: "question",
998+
value: yield* decode(MuseUserInput, params),
999+
});
1000+
case "approval/resolved":
1001+
case "userInput/settled": {
1002+
const id = method === "approval/resolved" ? params.approvalId : params.userInputId;
1003+
if (typeof id === "string")
1004+
turn?.settledRequests.add(
1005+
`${method === "approval/resolved" ? "approval" : "question"}:${id}`,
1006+
);
1007+
const entry = [...pending.values()].find(
1008+
(candidate) => candidate.request.nativeRequestRef?.nativeId === id,
1009+
);
1010+
if (entry) {
1011+
// An approval settled outside T3 still shows Muse's own decision.
1012+
const native =
1013+
method === "approval/resolved" && typeof params.decision === "string"
1014+
? museApprovalDecision({ decision: params.decision, scope: "once" })
1015+
: undefined;
1016+
if (!entry.response && native) entry.response = { decision: native };
1017+
yield* resolvePending(entry, "resolved");
1018+
}
1019+
if (!pending.size) yield* updateSession(active ? "running" : "ready");
1020+
}
1021+
}
9221022
});
9231023
const handleNotification = Effect.fnUntraced(function* (method: string, data: unknown) {
9241024
const params = yield* decode(recordSchema, data);
@@ -962,6 +1062,10 @@ export function makeMuseAdapterV2(options: MuseAdapterV2Options): ProviderAdapte
9621062
return;
9631063
}
9641064
}
1065+
// A workflow child's approval comes under the child's own run id, during a turn
1066+
// or after it, so it skips turn routing.
1067+
if (params.subagentOrigin !== undefined && method.startsWith("approval/"))
1068+
return yield* handleRequestEvent(method, params, undefined);
9651069
// Muse starts a turn on its own when a workflow finishes, to report its result.
9661070
// During a run, that run shows it. Otherwise hold it and ask the orchestrator for
9671071
// a run; that run takes the held events, unless a user turn takes them first.
@@ -1097,45 +1201,11 @@ export function makeMuseAdapterV2(options: MuseAdapterV2Options): ProviderAdapte
10971201
}
10981202
case "approval/requested":
10991203
case "approval/updated":
1100-
if (
1101-
method === "approval/updated" &&
1102-
![...pending.values()].some(
1103-
(entry) =>
1104-
entry.native.type === "approval" &&
1105-
entry.native.value.approvalId === params.approvalId,
1106-
)
1107-
)
1108-
break;
1109-
yield* publishRequest({ type: "approval", value: yield* decode(MuseApproval, params) });
1110-
break;
1111-
case "userInput/requested":
1112-
yield* publishRequest({
1113-
type: "question",
1114-
value: yield* decode(MuseUserInput, params),
1115-
});
1116-
break;
11171204
case "approval/resolved":
1118-
case "userInput/settled": {
1119-
const id = method === "approval/resolved" ? params.approvalId : params.userInputId;
1120-
if (typeof id === "string")
1121-
turn.settledRequests.add(
1122-
`${method === "approval/resolved" ? "approval" : "question"}:${id}`,
1123-
);
1124-
const entry = [...pending.values()].find(
1125-
(candidate) => candidate.request.nativeRequestRef?.nativeId === id,
1126-
);
1127-
if (entry) {
1128-
// An approval settled outside T3 still shows Muse's own decision.
1129-
const native =
1130-
method === "approval/resolved" && typeof params.decision === "string"
1131-
? museApprovalDecision({ decision: params.decision, scope: "once" })
1132-
: undefined;
1133-
if (!entry.response && native) entry.response = { decision: native };
1134-
yield* resolvePending(entry, "resolved");
1135-
}
1136-
if (!pending.size) yield* updateSession("running");
1205+
case "userInput/requested":
1206+
case "userInput/settled":
1207+
yield* handleRequestEvent(method, params, turn);
11371208
break;
1138-
}
11391209
case "session/tokenUsage": {
11401210
const usage = yield* decode(MuseTokenUsageEvent, params);
11411211
const previous = turn.providerTurn.turnTokenUsage;
@@ -1892,18 +1962,8 @@ export function makeMuseAdapterV2(options: MuseAdapterV2Options): ProviderAdapte
18921962
Effect.gen(function* () {
18931963
const entry = pending.get(args.requestId);
18941964
if (!entry) return yield* protocolError("This Muse request is no longer pending");
1895-
if (entry.native.type === "approval") {
1896-
const approval = entry.native.value;
1897-
const choice = args.decision && museApprovalChoices(approval).get(args.decision);
1898-
if (!choice) return yield* protocolError("Muse did not offer this approval decision");
1899-
// Recorded first: Muse can settle the approval before it acknowledges the command.
1900-
entry.response = { decision: args.decision };
1901-
yield* request("approval/decide", {
1902-
approvalId: approval.approvalId,
1903-
requirementId: approval.currentRequirementId,
1904-
choiceId: choice.choiceId,
1905-
}).pipe(Effect.tapError(() => Effect.sync(() => delete entry.response)));
1906-
} else {
1965+
if (entry.native.type === "approval") yield* decideApproval(entry, args.decision);
1966+
else {
19071967
const questionRequest = entry.native.value;
19081968
const answers = [];
19091969
for (const question of questionRequest.questions) {

‎apps/server/src/provider/museProtocol.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -91,6 +91,8 @@ export const MuseApproval = Schema.Struct({
9191
itemId: Schema.optional(Schema.String),
9292
toolName: Schema.optional(Schema.String),
9393
rawArgs: Schema.optional(Schema.String),
94+
/** Set when a workflow child asks; `turnId` is then the child's run, not a session turn. */
95+
subagentOrigin: Schema.optional(Schema.Struct({ subagentId: NonEmptyString })),
9496
currentRequirementId: Schema.Struct({ approvalId: NonEmptyString, sourceIndex: Schema.Int }),
9597
availableChoices: Schema.Array(
9698
Schema.Struct({

0 commit comments

Comments
 (0)