Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 73 additions & 0 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -17902,6 +17902,79 @@ describe("ClaudeAdapterV2 background wake turns", () => {
),
);

it.effect("ends a goal check without re-prompting while T3 work will wake the thread", () =>
Effect.gen(function* () {
const harness = yield* makeWakeHarness;
const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
yield* mcpSessions.set({
environmentId: EnvironmentId.make("environment-goal-wake"),
threadId: harness.threadId,
providerSessionId: "mcp-session-goal-wake",
providerInstanceId: ProviderInstanceId.make("claudeAgent"),
endpoint: "http://127.0.0.1:43123/mcp",
authorizationHeader: "Bearer secret-claude-token",
browserToolsAvailable: true,
hasPendingWake: Effect.succeed(true),
});
const stop = Effect.promise(() =>
Promise.all(
(harness.getOpenedOptions()?.hooks?.Stop ?? []).flatMap((matcher) =>
matcher.hooks.map((hook) =>
hook(
{
hook_event_name: "Stop",
stop_hook_active: true,
session_id: WAKE_NATIVE_SESSION,
transcript_path: "/transcript.jsonl",
cwd: "/workspace",
},
undefined,
{ signal: new AbortController().signal },
),
),
),
),
);
const condition = "the delegated review is addressed";
yield* harness.runtime.startTurn(
makeClaudeTestTurnInput({
threadId: harness.threadId,
providerThread: harness.providerThread,
now: yield* DateTime.now,
attemptId: RunAttemptId.make("goal-wake-attempt"),
text: `/goal ${condition}`,
attachments: [],
}),
);
// Without a goal, Stop keeps Claude Code's own behavior.
assert.deepEqual(yield* stop, [{}]);
yield* harness.offerAndWait(syntheticFrame("goal-wake-set", `Goal set: ${condition}`));
assert.deepEqual(yield* stop, [{ continue: false }]);
// The turn ends without an error and the goal waits for the wake.
yield* Queue.offer(
harness.sdkMessages,
makeResultFrame({
uuid: "goal-wake-result",
result: "Waiting for the review.",
terminalReason: "stop_hook_prevented",
}),
);
const terminal = yield* Queue.take(harness.terminalReceipts);
assert.equal(terminal.status, "completed");
assert.equal(goalStatuses(harness.events).at(-1)?.status, "active");
}).pipe(
Effect.scoped,
Effect.provide(
Layer.mergeAll(
IdAllocator.layer,
McpProviderSessions.layer,
NodeServices.layer,
NodeCrypto.layer,
),
),
),
);

it.effect(
"defers goal completion until an ambient subagent finishes and Claude checks again",
() =>
Expand Down
23 changes: 21 additions & 2 deletions apps/server/src/orchestration-v2/Adapters/ClaudeAdapterV2.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1539,6 +1539,7 @@ export function makeClaudeQueryOptions(input: {
readonly canUseTool?: CanUseTool;
readonly onUserDialog?: ClaudeQueryOptions["onUserDialog"];
readonly onElicitation?: ClaudeQueryOptions["onElicitation"];
readonly onStop?: HookCallback;
readonly supportedDialogKinds?: ClaudeQueryOptions["supportedDialogKinds"];
readonly allowDangerouslySkipPermissions?: boolean;
readonly appendSystemPromptFileText?: string;
Expand Down Expand Up @@ -1663,8 +1664,15 @@ export function makeClaudeQueryOptions(input: {
...launchAppendSystemPrompt,
].join("\n\n"),
},
...(deviceInstructions
? { hooks: { SubagentStart: [{ hooks: [subagentDeviceInstructions] }] } }
...(deviceInstructions || input.onStop !== undefined
? {
hooks: {
...(deviceInstructions
? { SubagentStart: [{ hooks: [subagentDeviceInstructions] }] }
: {}),
...(input.onStop === undefined ? {} : { Stop: [{ hooks: [input.onStop] }] }),
},
}
: {}),
...(Object.keys(extraArgs).length === 0 ? {} : { extraArgs }),
};
Expand Down Expand Up @@ -10383,6 +10391,16 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
) {
const queryPolicy = yield* queryPolicyForTurn(turnInput);
const mcpSession = yield* mcpSessions.read(turnInput.threadId);
// Claude Code defers its goal Stop hook only for its own background tasks, so a goal
// re-prompts in a loop while T3 work that will wake this thread is still running.
const onStop: HookCallback = () =>
runPromise(
Effect.gen(function* () {
if (goalsByNativeThread.get(nativeThreadId)?.status !== "active") return {};
const pendingWake = (yield* mcpSessions.read(turnInput.threadId))?.hasPendingWake;
return pendingWake !== undefined && (yield* pendingWake) ? { continue: false } : {};
}),
);
const mcpOverrides = claudeMcpQueryOverrides({
mcpSession,
readOnlySandbox:
Expand Down Expand Up @@ -10541,6 +10559,7 @@ export const makeClaudeAdapterV2 = Effect.fn("makeClaudeAdapterV2")(function* (
canUseTool,
onUserDialog,
onElicitation,
onStop,
supportedDialogKinds: ["resume_return"],
...(appendSystemPromptFileText === undefined ? {} : { appendSystemPromptFileText }),
...(toolSearchOverride === undefined ? {} : { toolSearchOverride }),
Expand Down
40 changes: 40 additions & 0 deletions apps/server/src/orchestration-v2/ProviderSessionManager.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ import {
ProviderDriverKind,
ProviderInstanceId,
type ProviderSessionId,
RunId,
ThreadId,
} from "@t3tools/contracts";
import * as Cause from "effect/Cause";
Expand Down Expand Up @@ -1769,6 +1770,43 @@ it.effect("ProviderSessionManagerV2 drains subscribers when the provider stops",
}),
);

it("ProviderSessionManagerV2 counts a delegated task as a wake only while its result can be delivered", () => {
const runId = RunId.make("run-delegating");
const thread = { archivedAt: null, deletedAt: null };
const task = {
origin: "app_owned" as const,
status: "running" as const,
runId,
completionDelivery: undefined,
};
const run = (disposition: "open" | "disposed") => ({
id: runId,
delegatedCompletion: { disposition, nextGeneration: 1, delivery: null },
});
assert.isTrue(
ProviderSessionManager.delegatedTaskWillWake({
thread,
runs: [run("open")],
subagents: [task],
}),
);
// An interrupted completion wake disposes the cohort, so the result never wakes the thread.
assert.isFalse(
ProviderSessionManager.delegatedTaskWillWake({
thread,
runs: [run("disposed")],
subagents: [task],
}),
);
assert.isFalse(
ProviderSessionManager.delegatedTaskWillWake({
thread,
runs: [run("open")],
subagents: [{ ...task, origin: "provider_native" }],
}),
);
});

it.effect(
"ProviderSessionManagerV2 issues MCP credentials before opening and revokes them on close",
() =>
Expand Down Expand Up @@ -1814,6 +1852,8 @@ it.effect(
new Set(["preview", "orchestration", "worktree", "pull-requests"]),
);

assert.isFalse(yield* captured!.hasPendingWake!);

yield* manager.close(providerSessionId);
assert.isUndefined(yield* mcpSessions.read(threadId));
assert.isUndefined(yield* registry.resolve(token!));
Expand Down
37 changes: 36 additions & 1 deletion apps/server/src/orchestration-v2/ProviderSessionManager.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,13 @@ import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
import {
isOrchestrationV2WorkActive,
ModelSelection,
type OrchestrationV2AppThread,
OrchestrationV2DomainEvent,
OrchestrationV2ProviderSession,
type OrchestrationV2ProviderThread,
type OrchestrationV2Run,
OrchestrationV2RuntimeRequest,
type OrchestrationV2Subagent,
ProviderInstanceId,
ProviderSessionId,
ThreadId,
Expand Down Expand Up @@ -71,6 +74,30 @@ const WORKING_RUN_STATUSES: ReadonlySet<string> = new Set([
"waiting",
]);

/** Matches planDelegatedCompletionDelivery: true while a delegated task's result still wakes the thread. */
export const delegatedTaskWillWake = (records: {
readonly thread: Pick<OrchestrationV2AppThread, "archivedAt" | "deletedAt">;
readonly runs: ReadonlyArray<Pick<OrchestrationV2Run, "id" | "delegatedCompletion">>;
readonly subagents: ReadonlyArray<
Pick<OrchestrationV2Subagent, "origin" | "status" | "runId" | "completionDelivery">
>;
}): boolean =>
records.thread.archivedAt === null &&
records.thread.deletedAt === null &&
records.subagents.some((task) => {
const run = records.runs.find((candidate) => candidate.id === task.runId);
const delivery = task.completionDelivery?.state;
return (
task.origin === "app_owned" &&
isOrchestrationV2WorkActive(task.status) &&
run !== undefined &&
(run.delegatedCompletion?.disposition ?? "open") === "open" &&
delivery !== "acknowledged" &&
delivery !== "delivered" &&
delivery !== "disposed"
);
});

export const ProviderSessionReleaseReason = Schema.Literals([
"idle_timeout",
"runtime_error",
Expand Down Expand Up @@ -497,7 +524,15 @@ export const layerWithOptions = (
capabilities,
subagent,
});
yield* mcpSessions.set(credential.config);
yield* mcpSessions.set({
...credential.config,
hasPendingWake: projectionStore
.getThreadRecords(threadId, ["runs", "subagents"])
.pipe(
Effect.map(delegatedTaskWillWake),
Effect.orElseSucceed(() => false),
),
});
Comment thread
coderabbitai[bot] marked this conversation as resolved.
Resolved
reserveMcpCredential(threadId, credential.config.providerSessionId);
return { mcpCredentialId: credential.config.providerSessionId, issued: true };
}),
Expand Down
3 changes: 3 additions & 0 deletions packages/provider-core/src/server/mcpSession.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import type { EnvironmentId, ProviderInstanceId, ThreadId } from "@t3tools/contracts";
import type * as Effect from "effect/Effect";

export interface McpProviderSessionConfig {
readonly environmentId: EnvironmentId;
Expand All @@ -11,6 +12,8 @@ export interface McpProviderSessionConfig {
readonly browserToolsAvailable: boolean;
readonly capabilities?: ReadonlySet<string>;
readonly agentDeviceEnvironment?: Readonly<Record<string, string>>;
/** True while a delegated task this thread started will wake it with its result. */
readonly hasPendingWake?: Effect.Effect<boolean>;
}

export function withAgentDeviceEnvironment(
Expand Down
Loading