Skip to content
Closed
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
3 changes: 2 additions & 1 deletion apps/server/src/mcp/McpSessionRegistry.ts
Original file line number Diff line number Diff line change
Expand Up @@ -244,7 +244,8 @@ export const issueActiveMcpCredential = (

/**
* Refreshes the liveness of a thread's MCP credential. Called on every provider
* turn so an active session is never mistaken for an abandoned one.
* turn, and hourly while its event stream is open, so an active session is
* never mistaken for an abandoned one.
*/
export const touchActiveMcpThread = (threadId: ThreadId): Effect.Effect<void> =>
activeMcpSessionRegistry ? activeMcpSessionRegistry.touch(threadId) : Effect.void;
Expand Down
81 changes: 81 additions & 0 deletions apps/server/src/orchestration-v2/RunExecutionService.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@ import {
TurnItemId,
} from "@t3tools/contracts";
import * as Cause from "effect/Cause";
import * as Clock from "effect/Clock";
import * as Exit from "effect/Exit";
import * as DateTime from "effect/DateTime";
import * as Deferred from "effect/Deferred";
Expand All @@ -36,6 +37,7 @@ import * as Layer from "effect/Layer";
import * as Option from "effect/Option";
import * as Ref from "effect/Ref";
import * as Stream from "effect/Stream";
import * as TestClock from "effect/testing/TestClock";

import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts";
import * as ServerSettings from "../serverSettings.ts";
Expand Down Expand Up @@ -744,6 +746,85 @@ it.effect("refreshes MCP credential liveness before calling the provider", () =>
}).pipe(Effect.provide(RunExecutionTestLayer)),
);

for (const [scenario, events, keepsAlive] of [
["while the turn waits on the user", Stream.never, true],
["only until the provider event stream ends", Stream.empty, false],
] as const) {
it.effect(`refreshes MCP credential liveness ${scenario}`, () =>
Effect.gen(function* () {
const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
const lastTouchAt = yield* Ref.make(0);
const threadId = ThreadId.make("thread:run-execution-mcp-heartbeat");
const touchActiveMcpThread = vi
.spyOn(McpSessionRegistry, "touchActiveMcpThread")
.mockImplementation(() =>
Clock.currentTimeMillis.pipe(Effect.flatMap((now) => Ref.set(lastTouchAt, now))),
);

yield* Effect.gen(function* () {
yield* runExecution.startRootRun({
commandId: CommandId.make("command:run-execution-mcp-heartbeat"),
appThread: { id: threadId } as OrchestrationV2AppThread,
providerSessionId: ProviderSessionId.make("session:run-execution-mcp-heartbeat"),
session: {
events,
startTurn: () => Effect.void,
} as unknown as ProviderAdapterV2SessionRuntime,
run: {
id: RunId.make("run:run-execution-mcp-heartbeat"),
threadId,
ordinal: 1,
providerInstanceId: ProviderInstanceId.make("codex"),
} as OrchestrationV2Run,
rootNode: {
id: NodeId.make("node:run-execution-mcp-heartbeat"),
} as OrchestrationV2ExecutionNode,
checkpointScope: {
id: CheckpointScopeId.make("checkpoint-scope:run-execution-mcp-heartbeat"),
} as OrchestrationV2CheckpointScope,
providerThread: {
id: ProviderThreadId.make("provider-thread:run-execution-mcp-heartbeat"),
driver,
} as OrchestrationV2ProviderThread,
attempt: {
id: RunAttemptId.make("attempt:run-execution-mcp-heartbeat"),
providerTurnId: null,
} as OrchestrationV2RunAttempt,
attemptId: RunAttemptId.make("attempt:run-execution-mcp-heartbeat"),
providerTurnOrdinal: 1,
message: {
messageId: MessageId.make("message:run-execution-mcp-heartbeat"),
text: "Ask me something and wait.",
attachments: [],
createdBy: "user",
creationSource: "web",
},
modelSelection: {
instanceId: ProviderInstanceId.make("codex"),
model: "gpt-5.4",
},
runtimePolicy: {
runtimeMode: "full-access",
interactionMode: "default",
cwd: process.cwd(),
approvalPolicy: "never",
sandboxPolicy: {
type: "readOnly",
access: { type: "fullAccess" },
networkAccess: false,
},
},
});
// The report in #14076 answered a question 44 hours into the turn.
yield* TestClock.adjust("44 hours");
}).pipe(Effect.ensuring(Effect.sync(() => touchActiveMcpThread.mockRestore())));

const sinceLastTouch = (yield* Clock.currentTimeMillis) - (yield* Ref.get(lastTouchAt));
assert.equal(sinceLastTouch < 24 * 60 * 60 * 1_000, keepsAlive);
}).pipe(Effect.provide(RunExecutionTestLayer)),
);
}

it.effect("starts the provider when checkpoint baseline capture fails", () =>
Effect.gen(function* () {
const threadId = ThreadId.make("thread:run-execution-baseline-failure");
Expand Down
8 changes: 8 additions & 0 deletions apps/server/src/orchestration-v2/RunExecutionService.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1167,6 +1167,13 @@ export const layer: Layer.Layer<
return true;
});
const filterAssistantEvent = makeAssistantStreamingFilter(responseStreamingMode);
// A turn can wait days on a user answer or approval without any MCP
// traffic. Keep the credential alive while its event stream is open;
// the stream ends when the provider exits, so a dead one still expires.
// A failed touch must not end the race and abort event ingestion.
const keepMcpCredentialAlive = Effect.suspend(() =>
McpSessionRegistry.touchActiveMcpThread(input.run.threadId),
).pipe(Effect.ignoreCause({ log: true }), Effect.delay("1 hour"), Effect.forever);
const providerEventFiber = yield* eventSubscription.events.pipe(
Stream.filterEffect((event) =>
Ref.modify(eventRouting, (state) => routeProviderEvent(event, routeIdentity, state)),
Expand Down Expand Up @@ -1325,6 +1332,7 @@ export const layer: Layer.Layer<
),
),
Effect.ensuring(eventSubscription.close),
Effect.raceFirst(keepMcpCredentialAlive),
Effect.forkDetach,
);

Expand Down
Loading