[Bug]: ACP v2 plan replacements and removals lose their target between turns #35827

#!/bin/sh
set -u
result=2
checkout=$(cd "${1:?Usage: sh repro.sh /path/to/t3code}" && pwd) || exit 2
scratch=$(mktemp -d) || exit 2
cleanup() {
  git -C "$checkout" worktree remove --force "$scratch/wt" >/dev/null 2>&1 || :
  rm -rf "$scratch"
}
trap cleanup EXIT HUP INT TERM
finish() {
  case "$1" in
    0) echo 'RESULT: expected behavior';;
    1) echo 'RESULT: defect reproduced';;
    *) echo 'RESULT: inconclusive';;
  esac
  exit "$1"
}
node --version || finish 2
python3 --version || finish 2
git --version || finish 2
"$checkout/node_modules/.bin/vp" --version || finish 2
git -C "$checkout" rev-parse HEAD || finish 2
mkdir -p "$scratch/home" "$scratch/tmp" "$scratch/state" || finish 2
export HOME="$scratch/home" TMPDIR="$scratch/tmp" T3CODE_HOME="$scratch/state"
export XDG_CACHE_HOME="$scratch/cache" XDG_CONFIG_HOME="$scratch/config"
export T3CODE_DEV_SEED=0
if ! git -C "$checkout" worktree add --detach "$scratch/wt" HEAD >"$scratch/setup.log" 2>&1; then finish 2; fi
cat > "$scratch/case.ts" <<'CASE'
  it.live("g08 plan replacement and removal survive turn boundaries", () =>
    Effect.gen(function* () {
      const instanceId = ProviderInstanceId.make("g08-plans");
      const threadId = ThreadId.make("g08-plans");
      let notify: Parameters<AcpSessionRuntime.AcpSessionRuntime["Service"]["handleSessionUpdate"]>[0];
      let ordinal = 0;
      const adapter = yield* makeAcpAdapterV2({
        instanceId,
        selfInvocation: yield* resolveSelfInvocation(),
        flavor: {
          driver: ACP_TEST_DRIVER,
          capabilities: AcpProviderCapabilitiesV2,
          makeRuntime: makeMockRuntime({
            childProcessSpawner: yield* ChildProcessSpawner.ChildProcessSpawner,
            mockAgentPath: yield* bundledMockAgentPath,
            wrapRuntime: (runtime) => ({
              ...runtime,
              handleSessionUpdate: (handler) => Effect.sync(() => { notify = handler; }).pipe(Effect.andThen(runtime.handleSessionUpdate(handler))),
              prompt: () => Effect.gen(function* () {
                ordinal++;
                const send = (update: EffectAcpSchema.SessionUpdate) => notify({ sessionId: "mock-session-1", update });
                const plan = (id: string, text: string) => send({ sessionUpdate: "plan_update", plan: { type: "items", planId: id, entries: [{ content: text, priority: "medium", status: "pending" }] } });
                yield* send({ sessionUpdate: "state_update", state: "running" });
                if (ordinal === 1) {
                  yield* plan("control", "First");
                  yield* plan("control", "Replacement");
                  yield* send({ sessionUpdate: "plan_removed", planId: "control" });
                  yield* plan("later-remove", "Remove on next turn");
                  yield* plan("later-replace", "Old contents");
                } else {
                  yield* send({ sessionUpdate: "plan_removed", planId: "later-remove" });
                  yield* plan("later-replace", "New contents");
                }
                yield* send({ sessionUpdate: "state_update", state: "idle", stopReason: "end_turn" });
                return { stopReason: "end_turn" as const };
              }),
            }),
          }),
        },
      });
      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("g08-plans"), modelSelection, runtimePolicy });
      const providerThread = yield* runtime.ensureThread({ threadId, modelSelection, runtimePolicy });
      const now = yield* DateTime.now;
      const run = (ordinal: number) => Effect.gen(function* () {
        yield* runtime.startTurn(makeTurnInput({ threadId, providerThread, instanceId, runtimePolicy, now, ordinal }));
        return Array.from(yield* runtime.events.pipe(Stream.takeUntil((event) => event.type === "turn.terminal"), Stream.runCollect)).flatMap((event) => event.type === "plan.updated" ? [event.plan] : []);
      });
      const first = yield* run(1);
      assert.equal(first.length, 5);
      assert.equal(first[0]!.id, first[1]!.id);
      assert.equal(first[1]!.id, first[2]!.id);
      assert.equal(first[1]!.kind, "todo_list");
      assert.deepEqual(first[1]!.kind === "todo_list" ? first[1]!.steps.map((s) => s.text) : [], ["Replacement"]);
      assert.equal(first[2]!.status, "superseded");
      console.log("PLAN_CONTROL_OK same-turn replacement and removal");
      const second = yield* run(2);
      const removed = second.find((p) => p.id === first[3]!.id && p.status === "superseded");
      const replacement = second.find((p) => p.kind === "todo_list" && p.steps.some((s) => s.text === "New contents"));
      console.log("PLAN_OBSERVED", JSON.stringify({ firstTurnPlanEvents: first.length, secondTurnPlanEvents: second.length, removedPriorPlan: !!removed, reusedReplacedPlanId: replacement?.id === first[4]!.id }));
      assert.isDefined(removed, "plan_removed on a later turn must supersede the existing session plan");
      assert.equal(replacement?.id, first[4]!.id, "the same session plan must be replaced, not duplicated");
    }).pipe(Effect.provide(layerTest), Effect.scoped),
  );

CASE
python3 - "$checkout" "$scratch" <<'PYTHON'
from pathlib import Path
import os,sys
root=Path(sys.argv[1]); scratch=Path(sys.argv[2]); wt=scratch/'wt'
# Link dependencies individually so new runner caches stay in the disposable worktree.
for directory, dirs, files in os.walk(root):
    dirs[:]=[d for d in dirs if d not in ('.git','.repos','.t3')]
    if 'node_modules' not in dirs:
        continue
    dirs.remove('node_modules')
    nm=Path(directory)/'node_modules'; target=wt/nm.relative_to(root)
    target.mkdir(parents=True,exist_ok=True)
    for entry in nm.iterdir():
        (target/entry.name).symlink_to(entry,target_is_directory=entry.is_dir())
p=wt/'apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts'
s=p.read_text(); needle='  beforeAll(() => bundleAgents);'
if s.count(needle)!=1:
    raise RuntimeError('test fixture insertion point changed')
p.write_text(s.replace(needle,needle+'\n'+(scratch/'case.ts').read_text(),1))
PYTHON
if [ "$?" -ne 0 ]; then finish 2; fi
cd "$scratch/wt" || finish 2
node_modules/.bin/vp test run apps/server/src/orchestration-v2/Adapters/AcpAdapterV2.test.ts -t 'g08 plan' >"$scratch/test.log" 2>&1
status=$?
# The control must pass before a failing assertion can count as this defect.
python3 - "$scratch/test.log" <<'PYTHON'
import sys
for line in open(sys.argv[1]):
    if line.startswith(('PLAN_CONTROL_OK','PLAN_OBSERVED')):
        print(line.rstrip())
PYTHON
if ! grep -q 'PLAN_CONTROL_OK' "$scratch/test.log"; then finish 2; fi
if [ "$status" -eq 0 ]; then finish 0; fi
if grep -q 'plan_removed on a later turn must supersede' "$scratch/test.log"; then finish 1; fi
finish 2