fix(server): support Prime Agent 0.8.0 (#71)

Author: rynfar

apps/server/scripts/acp-mock-agent.ts

@@ -42,6 +42,20 @@ const failSetConfigOption = process.env.T3_ACP_FAIL_SET_CONFIG_OPTION === "1";
 const exitOnSetConfigOption = process.env.T3_ACP_EXIT_ON_SET_CONFIG_OPTION === "1";
 const promptResponseText = process.env.T3_ACP_PROMPT_RESPONSE_TEXT;
 const promptDelayMs = Number(process.env.T3_ACP_PROMPT_DELAY_MS ?? "0");
+const primeTerminalQuiescenceDelayMs = Number(
+  process.env.T3_ACP_PRIME_TERMINAL_QUIESCENCE_DELAY_MS ?? "0",
+);
+const primeTerminalQuiescenceOutcome =
+  process.env.T3_ACP_PRIME_TERMINAL_QUIESCENCE_OUTCOME === "error" ? "error" : "result";
+if (process.env.T3_ACP_ASSERT_TOP_LEVEL_ENV === "1") {
+  const inheritedInternal = Object.keys(process.env).find(
+    (name) => name.startsWith("PRIME_AGENT_INTERNAL_") || name === "RLM_DEPTH",
+  );
+  if (inheritedInternal !== undefined || process.env.RLM_MAX_DEPTH !== "4") {
+    throw new Error("Mock ACP process inherited private Prime Agent worker context.");
+  }
+}
+
 const permissionOptionIds = {
   allowOnce: process.env.T3_ACP_ALLOW_ONCE_OPTION_ID ?? "allow-once",
   allowAlways: process.env.T3_ACP_ALLOW_ALWAYS_OPTION_ID ?? "allow-always",
@@ -467,6 +481,59 @@ const program = Effect.gen(function* () {
         return yield* AcpError.AcpRequestError.internalError("Mock prompt failure");
       }
 
+      if (Number.isFinite(primeTerminalQuiescenceDelayMs) && primeTerminalQuiescenceDelayMs > 0) {
+        const namespace = "ai.primeintellect.prime-agent";
+        writeJsonRpcNotification("session/update", {
+          sessionId: requestedSessionId,
+          update: {
+            sessionUpdate: "agent_message_chunk",
+            content: { type: "text", text: "before terminal quiescence" },
+          },
+        });
+        writeJsonRpcNotification("session/update", {
+          sessionId: requestedSessionId,
+          update: {
+            sessionUpdate: "session_info_update",
+            _meta: {
+              [namespace]: {
+                promptTurnId: promptCount,
+                eventSequence: promptCount * 2 - 1,
+                phase: "responseBoundary",
+                outcome: primeTerminalQuiescenceOutcome,
+                terminalQuiescenceExpected: true,
+              },
+            },
+          },
+        });
+        const terminalPromptTurnId = promptCount;
+        yield* Effect.sleep(primeTerminalQuiescenceDelayMs).pipe(
+          Effect.andThen(
+            Effect.sync(() =>
+              writeJsonRpcNotification("session/update", {
+                sessionId: requestedSessionId,
+                update: {
+                  sessionUpdate: "session_info_update",
+                  _meta: {
+                    [namespace]: {
+                      promptTurnId: terminalPromptTurnId,
+                      eventSequence: terminalPromptTurnId * 2,
+                      phase: "terminalQuiescence",
+                      outcome: primeTerminalQuiescenceOutcome,
+                      quiescence: {
+                        outstandingSubagents: 0,
+                        remainingAutonomousContinuations: 0,
+                      },
+                    },
+                  },
+                },
+              }),
+            ),
+          ),
+          Effect.forkDetach,
+        );
+        return { stopReason: "end_turn" };
+      }
+
       if (emitStaleXAiPromptCompleteBeforeSecondHang && promptCount === 1) {
         return {
           stopReason: "end_turn",

apps/server/src/provider/Layers/PrimeAgentAdapter.test.ts

@@ -11,6 +11,7 @@ import * as Effect from "effect/Effect";
 import * as Exit from "effect/Exit";
 import * as Fiber from "effect/Fiber";
 import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
 import * as Schema from "effect/Schema";
 import * as Stream from "effect/Stream";
 
@@ -774,3 +775,65 @@ exec ${process.execPath} ${mockAgentPath} "$@"
       yield* Fiber.interrupt(cancelledEventFiber);
     }).pipe(Effect.scoped, Effect.provide(testLayer)),
 );
+
+it.effect("waits for Prime Agent 0.8 terminal quiescence before settling ACP fallback turns", () =>
+  Effect.gen(function* () {
+    const tempDir = yield* Effect.promise(() =>
+      NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "prime-agent-acp-quiescence-")),
+    );
+    const wrapperPath = NodePath.join(tempDir, "fake-prime-agent.sh");
+    const requestLogPath = NodePath.join(tempDir, "requests.ndjson");
+    yield* Effect.promise(() =>
+      NodeFSP.writeFile(
+        wrapperPath,
+        `#!/bin/sh
+exec ${process.execPath} ${mockAgentPath} "$@"
+`,
+        "utf8",
+      ),
+    );
+    yield* Effect.promise(() => NodeFSP.chmod(wrapperPath, 0o755));
+
+    const adapter = yield* makePrimeAgentAdapter(decodeSettings({ binaryPath: wrapperPath }), {
+      instanceId: ProviderInstanceId.make("primeAgent-quiescence"),
+      environment: {
+        ...process.env,
+        T3_ACP_PRIME_TERMINAL_QUIESCENCE_DELAY_MS: "500",
+        T3_ACP_PRIME_TERMINAL_QUIESCENCE_OUTCOME: "error",
+        T3_ACP_REQUEST_LOG_PATH: requestLogPath,
+        T3_ACP_ASSERT_TOP_LEVEL_ENV: "1",
+        PRIME_AGENT_INTERNAL_TEST_WORKER: "nested",
+        RLM_DEPTH: "3",
+        RLM_MAX_DEPTH: "4",
+      },
+    });
+    const threadId = ThreadId.make("terminal-quiescence");
+    const terminalFiber = yield* adapter.streamEvents.pipe(
+      Stream.filter(
+        (event): event is TurnCompletedEvent =>
+          event.threadId === threadId && event.type === "turn.completed",
+      ),
+      Stream.runHead,
+      Effect.forkChild,
+    );
+    yield* Effect.yieldNow;
+    yield* adapter.startSession({
+      threadId,
+      provider: ProviderDriverKind.make("primeAgent"),
+      cwd: process.cwd(),
+      runtimeMode: "full-access",
+    });
+
+    yield* adapter.sendTurn({
+      threadId,
+      input: "wait for descendants",
+      attachments: [],
+    });
+    const terminal = yield* Fiber.join(terminalFiber);
+    assert.isTrue(Option.isSome(terminal));
+    if (Option.isSome(terminal)) {
+      assert.equal(terminal.value.payload.state, "failed");
+    }
+    yield* adapter.stopSession(threadId);
+  }).pipe(Effect.scoped, Effect.provide(testLayer)),
+);

apps/server/src/provider/Layers/PrimeAgentAdapter.ts

@@ -44,7 +44,9 @@ import {
 import { makeAcpNativeLoggerFactory } from "../acp/AcpNativeLogging.ts";
 import {
   makePrimeAgentAcpRuntime,
+  parsePrimeAgentAcpTerminalUpdate,
   primeAgentLaunchArgsIssue,
+  type PrimeAgentAcpTerminalUpdate,
 } from "../acp/PrimeAgentAcpSupport.ts";
 import type { PrimeAgentAdapterShape } from "../Services/PrimeAgentAdapter.ts";
 import { canonicalPrimeToolItemId } from "../prime/PrimeAgentDaemonRuntimeEvents.ts";
@@ -75,11 +77,19 @@ export interface PrimeAgentAdapterLiveOptions {
   readonly startupWarning?: string;
 }
 
+type PrimeAgentAcpTerminalSettlement =
+  | { readonly state: "settled"; readonly outcome: "result" | "error" }
+  | { readonly state: "invalid" };
+
 interface PrimeAgentActiveTurn {
   readonly id: TurnId;
   readonly cancellation: Deferred.Deferred<void>;
+  readonly terminalQuiescence: Deferred.Deferred<PrimeAgentAcpTerminalSettlement>;
   cancellationRequested: boolean;
   hasPublicAssistantTextAfterLatestToolBoundary: boolean;
+  nativePromptTurnId: number | undefined;
+  lastNativeEventSequence: number | undefined;
+  terminalQuiescenceExpected: boolean;
 }
 
 interface PrimeAgentSessionContext {
@@ -95,6 +105,55 @@ interface PrimeAgentSessionContext {
   stopped: boolean;
 }
 
+function observePrimeAgentAcpTerminalUpdate(
+  ctx: PrimeAgentSessionContext | undefined,
+  update: PrimeAgentAcpTerminalUpdate,
+): Effect.Effect<void> {
+  const activeTurn = ctx?.activeTurn;
+  if (activeTurn === undefined) return Effect.void;
+  if (update.phase === "invalid") {
+    activeTurn.terminalQuiescenceExpected = true;
+    return Deferred.succeed(activeTurn.terminalQuiescence, { state: "invalid" }).pipe(
+      Effect.asVoid,
+    );
+  }
+  if (
+    activeTurn.lastNativeEventSequence !== undefined &&
+    update.eventSequence <= activeTurn.lastNativeEventSequence
+  ) {
+    return Effect.void;
+  }
+  activeTurn.lastNativeEventSequence = update.eventSequence;
+
+  if (update.phase === "responseBoundary") {
+    if (activeTurn.nativePromptTurnId !== undefined) {
+      activeTurn.terminalQuiescenceExpected = true;
+      return Deferred.succeed(activeTurn.terminalQuiescence, { state: "invalid" }).pipe(
+        Effect.asVoid,
+      );
+    }
+    activeTurn.nativePromptTurnId = update.promptTurnId;
+    activeTurn.terminalQuiescenceExpected = update.terminalQuiescenceExpected;
+    return Effect.void;
+  }
+
+  if (
+    activeTurn.nativePromptTurnId === undefined ||
+    activeTurn.nativePromptTurnId !== update.promptTurnId ||
+    !activeTurn.terminalQuiescenceExpected
+  ) {
+    activeTurn.terminalQuiescenceExpected = true;
+    return Deferred.succeed(activeTurn.terminalQuiescence, { state: "invalid" }).pipe(
+      Effect.asVoid,
+    );
+  }
+  activeTurn.terminalQuiescenceExpected = true;
+  return Deferred.succeed(activeTurn.terminalQuiescence, {
+    state: "settled",
+    outcome: update.outcome,
+  }).pipe(Effect.asVoid);
+}
+
 export function parsePrimeAgentResumeMarker(raw: unknown): boolean {
   return isPrimeAgentCompatibleResumeCursor(raw);
 }
@@ -449,6 +508,7 @@ export function makePrimeAgentAdapter(
           );
 
           const sessionScope = yield* Scope.make("sequential");
+          let sessionContext: PrimeAgentSessionContext | undefined;
           let sessionScopeTransferred = false;
           yield* Effect.addFinalizer(() =>
             sessionScopeTransferred ? Effect.void : Scope.close(sessionScope, Exit.void),
@@ -467,6 +527,12 @@ export function makePrimeAgentAdapter(
             continueSession: parsePrimeAgentResumeMarker(input.resumeCursor),
             model,
             clientInfo: { name: "pylon", version: "0.0.0" },
+            observeSessionUpdate: (notification) => {
+              const update = parsePrimeAgentAcpTerminalUpdate(notification);
+              return update === undefined
+                ? Effect.void
+                : observePrimeAgentAcpTerminalUpdate(sessionContext, update);
+            },
             ...acpNativeLoggers,
           }).pipe(
             Effect.provideService(Crypto.Crypto, crypto),
@@ -524,6 +590,7 @@ export function makePrimeAgentAdapter(
             stopRequested: false,
             stopped: false,
           };
+          sessionContext = ctx;
 
           const notificationFiber = yield* Stream.runDrain(
             Stream.mapEffect(acp.getEvents(), (event) =>
@@ -777,8 +844,12 @@ export function makePrimeAgentAdapter(
               const activeTurn: PrimeAgentActiveTurn = {
                 id: turnId,
                 cancellation: yield* Deferred.make<void>(),
+                terminalQuiescence: yield* Deferred.make<PrimeAgentAcpTerminalSettlement>(),
                 cancellationRequested: false,
                 hasPublicAssistantTextAfterLatestToolBoundary: false,
+                nativePromptTurnId: undefined,
+                lastNativeEventSequence: undefined,
+                terminalQuiescenceExpected: false,
               };
               ctx.activeTurn = activeTurn;
               ctx.lastPlanFingerprint = undefined;
@@ -802,7 +873,7 @@ export function makePrimeAgentAdapter(
               turnId,
               payload: { model: ctx.session.model ?? "default" },
             });
-            const result = yield* Effect.raceFirst(
+            const promptExit = yield* Effect.raceFirst(
               ctx.acp.prompt({ prompt }),
               Deferred.await(activeTurn.cancellation).pipe(
                 Effect.as({ stopReason: "cancelled" as const }),
@@ -811,13 +882,52 @@ export function makePrimeAgentAdapter(
               Effect.mapError((error) =>
                 mapAcpToAdapterError(PROVIDER, input.threadId, "session/prompt", error),
               ),
+              Effect.exit,
             );
+            // Prime Agent 0.8 publishes a response boundary before the ACP response and
+            // an authoritative terminal-quiescence envelope after descendant work settles.
+            // Older releases publish neither, so their prompt response remains terminal.
+            // A stopped session has already shut down its notification consumer, so it
+            // must not enqueue a barrier that can no longer be acknowledged.
+            const promptCancelled =
+              Exit.isSuccess(promptExit) && promptExit.value.stopReason === "cancelled";
+            if (!promptCancelled && !activeTurn.cancellationRequested && !ctx.stopRequested) {
+              yield* ctx.acp.drainEvents;
+            }
+            const terminal =
+              activeTurn.terminalQuiescenceExpected && !promptCancelled
+                ? yield* Effect.raceFirst(
+                    Deferred.await(activeTurn.terminalQuiescence),
+                    Deferred.await(activeTurn.cancellation).pipe(
+                      Effect.as({ state: "cancelled" as const }),
+                    ),
+                  )
+                : undefined;
+            if (terminal?.state === "invalid") {
+              return yield* new ProviderAdapterRequestError({
+                provider: PROVIDER,
+                method: "session/prompt",
+                detail: "Prime Agent returned invalid terminal-quiescence metadata.",
+              });
+            }
+            if (Exit.isFailure(promptExit) && terminal?.state !== "cancelled") {
+              return yield* Effect.failCause(promptExit.cause);
+            }
+            const result = Exit.isSuccess(promptExit)
+              ? promptExit.value
+              : ({ stopReason: "cancelled" } as const);
             const settled = yield* settleActiveTurn(
               ctx,
               turnId,
-              result.stopReason === "cancelled"
-                ? { state: "cancelled" }
-                : { state: "completed", stopReason: result.stopReason ?? null },
+              terminal?.state === "settled" && terminal.outcome === "error"
+                ? {
+                    state: "failed",
+                    errorMessage: PRIME_AGENT_TURN_FAILED,
+                    terminalFailure: true,
+                  }
+                : terminal?.state === "cancelled" || result.stopReason === "cancelled"
+                  ? { state: "cancelled" }
+                  : { state: "completed", stopReason: result.stopReason ?? null },
               true,
             );
             if (!settled && !activeTurn.cancellationRequested && !ctx.stopRequested) {

apps/server/src/provider/acp/AcpSessionRuntime.ts

@@ -55,6 +55,8 @@ export interface AcpSpawnInput {
   readonly args: ReadonlyArray<string>;
   readonly cwd?: string;
   readonly env?: NodeJS.ProcessEnv;
+  /** Whether to merge the current process environment into `env`. Defaults to true. */
+  readonly extendEnv?: boolean;
 }
 
 export interface AcpSessionRuntimeOptions {
@@ -80,6 +82,10 @@ export interface AcpSessionRuntimeOptions {
   readonly shouldDiscardSessionUpdate?: (
     notification: EffectAcpSchema.SessionNotification,
   ) => boolean;
+  /** Observes accepted root-session metadata without copying it into the generic event stream. */
+  readonly observeSessionUpdate?: (
+    notification: EffectAcpSchema.SessionNotification,
+  ) => Effect.Effect<void, never>;
 }
 
 export interface AcpSessionRequestLogEvent {
@@ -336,13 +342,17 @@ export const make = (
     const spawnCommand = yield* resolveSpawnCommand(
       options.spawn.command,
       options.spawn.args,
-      options.spawn.env ? { env: options.spawn.env, extendEnv: true } : {},
+      options.spawn.env
+        ? { env: options.spawn.env, extendEnv: options.spawn.extendEnv ?? true }
+        : {},
     );
     const child = yield* spawner
       .spawn(
         ChildProcess.make(spawnCommand.command, spawnCommand.args, {
           ...(options.spawn.cwd ? { cwd: options.spawn.cwd } : {}),
-          ...(options.spawn.env ? { env: options.spawn.env, extendEnv: true } : {}),
+          ...(options.spawn.env
+            ? { env: options.spawn.env, extendEnv: options.spawn.extendEnv ?? true }
+            : {}),
           shell: spawnCommand.shell,
         }),
       )
@@ -400,6 +410,9 @@ export const make = (
         ) {
           return;
         }
+        if (options.observeSessionUpdate !== undefined) {
+          yield* options.observeSessionUpdate(notification);
+        }
         yield* handleSessionUpdate({
           queue: eventQueue,
           modeStateRef,