Skip to content
19 changes: 18 additions & 1 deletion apps/server/scripts/acp-mock-agent.ts
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,10 @@ const emitStaleXAiPromptCompleteBeforeSecondHang =
const emitOverlappingXAiPromptCompleteOutOfOrder =
process.env.T3_ACP_EMIT_OVERLAPPING_XAI_PROMPT_COMPLETE_OUT_OF_ORDER === "1";
const failPrompt = process.env.T3_ACP_FAIL_PROMPT === "1";
const failPromptOrdinal = Number(process.env.T3_ACP_FAIL_PROMPT_ORDINAL);
const compactionPromptOrdinals = new Set(
(process.env.T3_ACP_COMPACTION_PROMPT_ORDINALS ?? "").split(",").map(Number),
);
const failSetConfigOption = process.env.T3_ACP_FAIL_SET_CONFIG_OPTION === "1";
const failSetModeConfigOption = process.env.T3_ACP_FAIL_SET_MODE_CONFIG_OPTION === "1";
const exitOnSetConfigOption = process.env.T3_ACP_EXIT_ON_SET_CONFIG_OPTION === "1";
Expand Down Expand Up @@ -1299,7 +1303,20 @@ const program = Effect.gen(function* () {
yield* Effect.sleep(`${promptDelayMs} millis`);
}

if (failPrompt) {
if (compactionPromptOrdinals.has(promptCount)) {
for (const status of ["in_progress", "completed"] as const) {
yield* agent.client.sessionUpdate({
sessionId: requestedSessionId,
update: {
sessionUpdate: "compaction_update",
compactionId: `compact-${promptCount}`,
status,
},
});
}
}

if (failPrompt || promptCount === failPromptOrdinal) {
return yield* AcpError.AcpRequestError.internalError("Mock prompt failure");
}

Expand Down
140 changes: 140 additions & 0 deletions apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -538,6 +538,100 @@ const pollProtocolMethods = (events: Queue.Queue<EffectAcpProtocol.AcpProtocolLo
return methods;
});

const makeInstructionCompactionHarness = Effect.fnUntraced(function* (
environment: Readonly<Record<string, string>>,
) {
const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner;
const path = yield* Path.Path;
const selfInvocation = yield* resolveSelfInvocation();
const mockAgentPath = yield* path.fromFileUrl(
new URL("../../../scripts/acp-mock-agent.ts", import.meta.url),
);
const protocolEvents = yield* Queue.unbounded<EffectAcpProtocol.AcpProtocolLogEvent>();
const instanceId = ProviderInstanceId.make("acp-test-instruction-compaction");
const threadId = ThreadId.make("thread-acp-instruction-compaction");
yield* (yield* McpProviderSessions.McpProviderSessions).set({
environmentId: EnvironmentId.make("environment-acp-instruction-compaction"),
threadId,
providerSessionId: "mcp-session-acp-instruction-compaction",
providerInstanceId: instanceId,
endpoint: "http://127.0.0.1:43123/mcp",
authorizationHeader: "Bearer instruction-compaction-token",
browserToolsAvailable: true,
});
const adapter = yield* makeAcpAdapterV2({
instanceId,
flavor: {
driver: ACP_TEST_DRIVER,
capabilities: AcpProviderCapabilitiesV2,
supportsCompaction: true,
makeRuntime: makeMockRuntime({
childProcessSpawner,
mockAgentPath,
protocolEvents,
environment,
wrapRuntime: (runtime) => ({
...runtime,
// Make compaction during prompt completion deterministic at the adapter boundary.
prompt: (payload, options) =>
runtime.prompt(payload, options).pipe(Effect.tap(() => runtime.drainEvents)),
}),
}),
},
selfInvocation,
});
const runtimePolicy = ProviderAdapter.ProviderAdapterV2RuntimePolicy.make({
runtimeMode: "full-access",
interactionMode: "default",
cwd: process.cwd(),
});
const modelSelection = { instanceId, model: "default" } as const;
const runtime = yield* adapter.openSession({
threadId,
providerSessionId: ProviderSessionId.make("provider-session-acp-instruction-compaction"),
modelSelection,
runtimePolicy,
});
const providerThread = yield* runtime.ensureThread({ threadId, modelSelection, runtimePolicy });
const now = yield* DateTime.now;
let ordinal = 0;
return Effect.fnUntraced(function* (messageText: string, expectedStatus = "completed") {
yield* runtime.startTurn(
makeTurnInput({
threadId,
providerThread,
instanceId,
runtimePolicy,
now,
ordinal: ++ordinal,
messageText,
}),
);
const terminal = yield* runtime.events.pipe(
Stream.filter((event) => event.type === "turn.terminal"),
Stream.runHead,
);
assert.equal(Option.getOrThrow(terminal).status, expectedStatus);
while (true) {
const event = yield* Queue.take(protocolEvents);
if (event.direction === "outgoing" && rawProtocolMethod(event) === "session/prompt") {
return rawProtocolPromptTextBlocks(event);
}
}
});
});

function assertReinjectedInstructions(
blocks: Array<{ type: "text"; text: string }>,
request: string,
) {
const text = blocks[0]?.text ?? "";
assert.include(text, "T3 Code interaction mode: Default");
assert.include(text, "T3 Code collaborative browser");
assert.include(text, "T3 Code orchestration");
assert.include(text, `<user_request>\n${request}\n</user_request>`);
}

describe("AcpAdapterV2", () => {
it.live.each(["failed", "recovered", "completed", "cancelled"] as const)(
"projects Mistral retry notices and their %s outcome",
Expand Down Expand Up @@ -827,6 +921,52 @@ describe("AcpAdapterV2", () => {
}).pipe(Effect.provide(layerTest), Effect.scoped),
);

it.live(
"reinjects ACP instructions after protocol compaction, including during reinjection",
() =>
Effect.gen(function* () {
const runTurn = yield* makeInstructionCompactionHarness({
T3_ACP_COMPACTION_PROMPT_ORDINALS: "2,4",
});
yield* runTurn("Start work.");
yield* runTurn("Continue until context compacts.");
assert.deepEqual(yield* runTurn(" /repro-skill"), [
{ type: "text", text: " /repro-skill" },
]);
assertReinjectedInstructions(
yield* runTurn("Continue after compaction."),
"Continue after compaction.",
);
assertReinjectedInstructions(
yield* runTurn("Continue after the second compaction."),
"Continue after the second compaction.",
);
assert.equal((yield* runTurn("Finish work."))[0]?.text, "Finish work.");
}).pipe(Effect.provide(layerTest), Effect.scoped),
);

it.live(
"reinjects ACP instructions after successful native compact, but not failed compact",
() =>
Effect.gen(function* () {
const runTurn = yield* makeInstructionCompactionHarness({
T3_ACP_FAIL_PROMPT_ORDINAL: "2",
});
yield* runTurn("Start work.");
assert.deepEqual(yield* runTurn("/compact", "failed"), [
{ type: "text", text: "/compact" },
]);
assert.equal(
(yield* runTurn("Continue after failure."))[0]?.text,
"Continue after failure.",
);
assert.deepEqual(yield* runTurn("/compact"), [{ type: "text", text: "/compact" }]);
assert.deepEqual(yield* runTurn("/repro-skill"), [{ type: "text", text: "/repro-skill" }]);
assertReinjectedInstructions(yield* runTurn("Continue work."), "Continue work.");
assert.equal((yield* runTurn("Finish work."))[0]?.text, "Finish work.");
}).pipe(Effect.provide(layerTest), Effect.scoped),
);

it.live("reapplies a selection after a switch away from it failed partway", () =>
Effect.gen(function* () {
const childProcessSpawner = yield* ChildProcessSpawner.ChildProcessSpawner;
Expand Down
36 changes: 34 additions & 2 deletions packages/provider-acp/src/server/adapter.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1574,6 +1574,8 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
const activeSelection = yield* Ref.make<ModelSelection | null>(null);
const activeInteractionMode = yield* Ref.make<ProviderInteractionMode | null>(null);
const promptInstructionStates = yield* Ref.make(new Map<string, T3AcpInstructionState>());
// Unique markers keep delayed prompt completions from clearing later compactions.
const compactionMarkers = yield* Ref.make(new Map<string, symbol>());
const runtimeRestartRequired = yield* Ref.make(false);
const runtimeTeardownState = yield* Ref.make<AcpRuntimeTeardownState>({ _tag: "Idle" });
const runtimeCallbackGeneration = yield* Ref.make(0);
Expand Down Expand Up @@ -3623,6 +3625,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
updated.delete(context.nativeThreadId);
return updated;
});
yield* Ref.update(compactionMarkers, (current) => {
const updated = new Map(current);
updated.set(context.nativeThreadId, Symbol());
return updated;
});
}
} else if (mcpIdentity !== undefined) {
turnItem = {
Expand Down Expand Up @@ -6826,6 +6833,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
settledStatus === "completed"
) {
const nativeItemId = `${context.nativeTurnId}:compaction`;
yield* Ref.update(compactionMarkers, (current) => {
const updated = new Map(current);
updated.set(context.nativeThreadId, Symbol());
return updated;
});
yield* emitProviderEvent({
type: "turn_item.updated",
driver,
Expand Down Expand Up @@ -6991,6 +7003,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
hasT3Mcp: (yield* readMcpContext(turnInput.threadId)).servers.length > 0,
} satisfies T3AcpInstructionState;
const previousInstructionState = (yield* Ref.get(promptInstructionStates)).get(sessionId);
const compactionMarker = (yield* Ref.get(compactionMarkers)).get(sessionId);
const messageText = providerMessageTextWithAttachmentPaths({
text: turnInput.message.text,
attachments: turnInput.message.attachments,
Expand All @@ -7000,7 +7013,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
const text = t3AcpPromptWithInstructions({
prompt: messageText,
state: instructionState,
...(previousInstructionState === undefined
...(previousInstructionState === undefined || compactionMarker !== undefined
? {}
: { previousState: previousInstructionState }),
});
Expand Down Expand Up @@ -7056,7 +7069,11 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
}),
});
}
return { prompt, instructionState: text === messageText ? undefined : instructionState };
return {
prompt,
instructionState: text === messageText ? undefined : instructionState,
compactionMarker,
};
});

const restartRuntimeAfterTeardownIfRequired = Effect.fnUntraced(function* (
Expand Down Expand Up @@ -7370,6 +7387,20 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
return updated;
}),
),
Effect.tap(() =>
Ref.update(compactionMarkers, (current) => {
if (promptParts?.instructionState === undefined) return current;
const observedMarker = promptParts.compactionMarker;
if (
observedMarker === undefined ||
current.get(requestedSessionId) !== observedMarker
)
return current;
const updated = new Map(current);
updated.delete(requestedSessionId);
return updated;
Comment thread
coderabbitai[bot] marked this conversation as resolved.
}),
),
// Wire settlement precedes the completion callback's permit request so
// settled-soft classification can observe the native return even when
// the completion fiber has not yet set promptSettled under the permit.
Expand Down Expand Up @@ -8163,6 +8194,7 @@ export const makeAcpAdapterV2 = Effect.fn("makeAcpAdapterV2")(function* (
yield* Ref.set(activeSelection, null);
yield* Ref.set(activeInteractionMode, null);
yield* Ref.set(promptInstructionStates, new Map());
yield* Ref.set(compactionMarkers, new Map());
yield* Ref.set(providerTurns, new Map());
yield* Ref.set(snapshot, {
order: [],
Expand Down
Loading