[Bug]: Queueing a follow-up repeats a pending subagent approval notification #34979

#!/bin/sh
set -eu
repo=${1:-$PWD}
case "$(uname -s)" in
 Darwin) : "${PLAYWRIGHT_BROWSERS_PATH:=$HOME/Library/Caches/ms-playwright}" ;;
 *) : "${PLAYWRIGHT_BROWSERS_PATH:=$HOME/.cache/ms-playwright}" ;;
esac
export PLAYWRIGHT_BROWSERS_PATH
stage=$(mktemp -d "${VERIFY_SCRATCH:?set VERIFY_SCRATCH to a private scratch directory}/repro.XXXXXX")
export HOME="$stage/userhome" TMPDIR="$stage/tmp" T3CODE_HOME="$stage/t3home"
mkdir -p "$HOME" "$TMPDIR" "$T3CODE_HOME"
export PATH="$repo/node_modules/.bin:$PATH"
server="$repo/apps/server/src/orchestration-v2/zz-verify-approval-renotify.test.ts"
browser="$repo/apps/web/src/components/zz-verify-approval-renotify.browser.test.ts"
export VERIFY_SNAPSHOTS="$repo/apps/web/src/components/zz-approval-snapshots.json"
cleanup() { rm -f "$server" "$browser" "$VERIFY_SNAPSHOTS"; rm -rf "$stage"; }
trap cleanup EXIT HUP INT TERM
cat > "$stage/generate-fixtures.py" <<'PYTHON_FIXTURE'
from pathlib import Path
out=Path(__file__).parent
root=Path(__import__('sys').argv[1])
server='''import { assert, it } from "@effect/vitest";
import { ProviderSessionId, ProviderDriverKind, CommandId, MessageId, EventId, ProjectId, ProviderInstanceId, RuntimeRequestId, ThreadId, OrchestrationV2ThreadShell } from "@t3tools/contracts";
import * as DateTime from "effect/DateTime";
import * as Effect from "effect/Effect";
import * as Layer from "effect/Layer";
import * as Schema from "effect/Schema";
import { writeFileSync } from "node:fs";
import * as SqlitePersistence from "../persistence/Sqlite.ts";
import * as ProjectionStore from "./ProjectionStore.ts";
import * as Orchestrator from "./Orchestrator.ts";
import * as EventSink from "./EventSink.ts";
import * as ProviderReplayHarness from "./testkit/ProviderReplayHarness.ts";
import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
const instanceId = ProviderInstanceId.make("codex");
const adapter = {instanceId, driver: ProviderDriverKind.make("codex"), getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), planSelectionTransition: () => Effect.succeed({type: "apply_on_next_turn" as const}), openSession: () => Effect.die("No provider process used")} as never;
const database = SqlitePersistence.layerMemory;
const layer = Layer.mergeAll(database, ProjectionStore.layer.pipe(Layer.provide(database)), ProviderReplayHarness.layerWithRegistry({name: "approval-renotify-verification"}, ProviderAdapterRegistry.layerFromAdapters([adapter]), {databaseLayer: database, runEffectWorker: false}));
it.effect("production queue and approval snapshots", () => Effect.gen(function* () {
 const orchestrator = yield* Orchestrator.OrchestratorV2;
 const projections = yield* ProjectionStore.ProjectionStoreV2;
 const sink = yield* EventSink.EventSinkV2;
 const instanceId = ProviderInstanceId.make("codex");
 const modelSelection = { instanceId, model: "gpt-5.4" };
 const parentId = ThreadId.make("parent");
 let childId = ThreadId.make("child");
 for (const threadId of [parentId]) yield* orchestrator.dispatch({ type: "thread.create", commandId: CommandId.make(`create-${threadId}`), threadId, projectId: ProjectId.make("project"), title: threadId, modelSelection, runtimeMode: "approval-required", interactionMode: "default", branch: null, worktreePath: null, createdBy: "user", creationSource: "web", ...(threadId === childId ? {lineage: { rootThreadId: parentId, parentThreadId: parentId, relationshipToParent: "subagent" }} : {}) });
 for (const threadId of [parentId]) yield* orchestrator.dispatch({type: "message.dispatch", commandId: CommandId.make(`start-${threadId}`), threadId, messageId: MessageId.make(`message-${threadId}`), text: "Work", attachments: [], dispatchMode: {type: "start_immediately"}, createdBy: "user", creationSource: "web"});
 const parentRun = (yield* projections.getThreadProjection(parentId)).runs[0]!;
 yield* orchestrator.dispatch({type: "delegated_task.request", commandId: CommandId.make("delegate"), parentThreadId: parentId, parentRunId: parentRun.id, parentNodeId: parentRun.rootNodeId!, task: "Research", completionWake: "settled_only", modelSelection, runtimeMode: "approval-required", interactionMode: "default", createdBy: "agent", creationSource: "mcp"});
 childId = (yield* projections.getThreadProjection(parentId)).subagents[0]!.childThreadId!;
 const now = yield* DateTime.now;
 for (const threadId of [parentId, childId]) {
  const projection = yield* projections.getThreadProjection(threadId);
  const run = projection.runs[0]!;
  yield* sink.write({events: [{id: EventId.make(`running-${threadId}`), type: "run.updated", threadId, runId: run.id, occurredAt: now, payload: {...run, status: "running", startedAt: now}}]});
 }
 const shell = (threadId: typeof parentId) => projections.getThreadShell(threadId).pipe(Effect.map(value => { assert.isNotNull(value); return value!; }));
 const idle = [yield* shell(parentId), yield* shell(childId)];
 const childProjection = yield* projections.getThreadProjection(childId);
 const childRun = childProjection.runs[0]!;
 const request = {id: RuntimeRequestId.make("child-approval"), nodeId: childRun.rootNodeId!, providerTurnId: null, nativeRequestRef: null, kind: "command" as const, status: "pending" as const, responseCapability: {type: "live" as const, providerSessionId: ProviderSessionId.make("provider-session")}, createdAt: now, resolvedAt: null};
 yield* sink.write({events: [{id: EventId.make("approval"), type: "runtime-request.updated", threadId: childId, runId: childRun.id, occurredAt: now, payload: request}]});
 const pending = [yield* shell(parentId), yield* shell(childId)];
 yield* orchestrator.dispatch({type: "message.dispatch", commandId: CommandId.make("queue-followup"), threadId: parentId, messageId: MessageId.make("followup"), text: "Additional context", attachments: [], dispatchMode: {type: "queue_after_active"}, createdBy: "user", creationSource: "web"});
 const queued = [yield* shell(parentId), yield* shell(childId)];
 assert.notEqual(pending[0].latestRunId, queued[0].latestRunId);
 assert.equal(queued[0].status, "queued");
 assert.equal(queued[0].pendingRuntimeRequest, null);
 assert.deepEqual(pending[1].pendingRuntimeRequest, queued[1].pendingRuntimeRequest);
 assert.equal(pending[1].lineage.relationshipToParent, "subagent");
 assert.equal(pending[1].creationSource, "mcp");
 yield* sink.write({events: [{id: EventId.make("resolve-child-approval"), type: "runtime-request.updated", threadId: childId, runId: childRun.id, occurredAt: now, payload: {...request, status: "resolved", resolvedAt: now}}]});
 const controlIdle = [yield* shell(parentId), yield* shell(childId)];
 const ownRequest = {...request, id: RuntimeRequestId.make("parent-approval"), nodeId: parentRun.rootNodeId!};
 yield* sink.write({events: [{id: EventId.make("parent-approval"), type: "runtime-request.updated", threadId: parentId, runId: parentRun.id, occurredAt: now, payload: ownRequest}]});
 const controlPending = [yield* shell(parentId), yield* shell(childId)];
 yield* orchestrator.dispatch({type: "message.dispatch", commandId: CommandId.make("control-queue"), threadId: parentId, messageId: MessageId.make("control-followup"), text: "Additional context", attachments: [], dispatchMode: {type: "queue_after_active"}, createdBy: "user", creationSource: "web"});
 const controlQueued = [yield* shell(parentId), yield* shell(childId)];
 assert.notEqual(controlPending[0].latestRunId, controlQueued[0].latestRunId);
 assert.deepEqual(controlPending[0].pendingRuntimeRequest, controlQueued[0].pendingRuntimeRequest);
 const encode = Schema.encodeSync(Schema.toCodecJson(OrchestrationV2ThreadShell));
 writeFileSync(process.env.VERIFY_SNAPSHOTS!, JSON.stringify({controlIdle: controlIdle.map(encode), controlPending: controlPending.map(encode), controlQueued: controlQueued.map(encode), idle: idle.map(encode), pending: pending.map(encode), queued: queued.map(encode)}, null, 2));
 console.log("PRODUCTION_QUEUE", JSON.stringify({before: pending[0].latestRunId, after: queued[0].latestRunId, status: queued[0].status, childRequestId: queued[1].pendingRuntimeRequest?.id, parentRequest: queued[0].pendingRuntimeRequest}));
}).pipe(Effect.provide(layer)));
'''
(root/'apps/server/src/orchestration-v2/zz-verify-approval-renotify.test.ts').write_text(server)
PYTHON_FIXTURE
mkdir -p "$stage/fixtures/apps/server/src/orchestration-v2"
python3 "$stage/generate-fixtures.py" "$stage/fixtures"
cp "$stage/fixtures/apps/server/src/orchestration-v2/zz-verify-approval-renotify.test.ts" "$server"
cat > "$stage/browser.fixture.ts" <<'BROWSER_FIXTURE'
import type { AnyRouter } from "@tanstack/react-router";
import { OrchestrationV2ThreadShell, ThreadId, RuntimeRequestId, EnvironmentId } from "@t3tools/contracts";
import * as Schema from "effect/Schema";
import { PrimaryConnectionTarget } from "@t3tools/client-runtime/connection";
import { createEnvironmentThreadShellAtoms } from "@t3tools/client-runtime/state/threads";
import { createEnvironmentSnapshotAtom, EMPTY_SHELL_STATE, applyShellStreamEvent } from "@t3tools/client-runtime/state/shell";
import snapshots from "./zz-approval-snapshots.json";
import { resolveBackgroundThreadRequest } from "./backgroundThreadAnnouncement";
import * as Option from "effect/Option";
import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
import { expect, it, vi } from "vite-plus/test";

import { stubGlobal } from "../../test/render";

const state = vi.hoisted(() => ({
  mode: "notifications",
  inApp: false,
  toast: vi.fn(),
  closeToast: vi.fn(),
  shells: null as unknown as { set: (id: string, value: unknown) => void },
  subagentRequests: null as unknown as { set: (value: unknown) => void },
  navigate: vi.fn(),
  settingsListeners: new Set<() => void>(),
}));
vi.mock(import("./ui/toast"), async (original) => {
  const actual = await original();
  return {
    ...actual,
    toastManager: {
      ...actual.toastManager,
      add: state.toast as never,
      close: state.closeToast as never,
    },
  };
});
vi.mock(import("../state/shell"), async (original) => {
  const actual = await original();
  return {
    ...actual,
    environmentShell: {
      ...actual.environmentShell,
      stateValueAtom: ((id: string) => shellAtom(id)) as never,
    },
  };
});
vi.mock(import("../state/presentation"), async (original) => {
  const actual = await original();
  return {
    ...actual,
    environmentSummaries: {
      ...actual.environmentSummaries,
      get environmentIdsAtom() {
        return environmentIdsAtom as never;
      },
    },
  };
});
vi.mock(import("../state/threads"), async (original) => {
  const actual = await original();
  return {
    ...actual,
    environmentThreadShells: {...actual.environmentThreadShells, pendingSubagentRequestsAtom: ((id: EnvironmentId) => realThreads.pendingSubagentRequestsAtom(id)) as never},
  };
});
vi.mock(import("../hooks/useSettings"), async (original) => ({
  ...(await original()),
  subscribeClientSettings: ((listener: () => void) => {
    state.settingsListeners.add(listener);
    return () => state.settingsListeners.delete(listener);
  }) as never,
  getClientSettings: (() => ({
    notificationMode: state.mode,
    inAppNotificationsEnabled: state.inApp,
  })) as never,
}));
vi.mock(import("../threadNotifications"), async (original) => ({
  ...(await original()),
  playNotificationSound: vi.fn(),
  unlockNotificationAudio: vi.fn(),
  setNotificationBadge: vi.fn(),
}));

import { playNotificationSound } from "../threadNotifications";
import { startThreadNotifications } from "./ThreadNotificationCoordinator";

let registry: AtomRegistry.AtomRegistry;
const shellAtom = Atom.family((_id: string) => Atom.make<any>(EMPTY_SHELL_STATE));
const environmentIdsAtom = Atom.make<ReadonlyArray<string>>([]);
const environmentId = EnvironmentId.make("one");
const snapshotAtom = createEnvironmentSnapshotAtom(Atom.family((id: EnvironmentId) => Atom.map(shellAtom(id), AsyncResult.success)));
const catalogValueAtom = Atom.make({isReady: true, entries: new Map([[environmentId, {target: new PrimaryConnectionTarget({environmentId, label: "test", httpBaseUrl: "https://example.test", wsBaseUrl: "wss://example.test"}), profile: Option.none(), enabled: true}]])});
const realThreads = createEnvironmentThreadShellAtoms({catalogValueAtom, snapshotAtom: snapshotAtom as never, hiddenThreadTagsAtom: () => Atom.make(null)});
state.shells = { set: (id, value) => registry.set(shellAtom(id), value) };


class TestNotification extends EventTarget {
  static permission = "granted";
  static sent: TestNotification[] = [];
  close = vi.fn();
  get tag() {
    return this.options.tag ?? "";
  }
  constructor(
    readonly title: string,
    readonly options: NotificationOptions,
  ) {
    super();
    TestNotification.sent.push(this);
  }
}

const decode = Schema.decodeUnknownSync(Schema.toCodecJson(OrchestrationV2ThreadShell));
const stages = Object.fromEntries(Object.entries(snapshots).map(([key, rows]) => [key, rows.map(row => decode(row))]));
function update(rows: typeof stages.idle) {
 let snapshot = registry.get(snapshotAtom("one")) ?? {schemaVersion: 1, snapshotSequence: 0, projects: [], threads: rows, archivedThreads: []};
 for (const thread of rows) snapshot = applyShellStreamEvent(snapshot, {kind: "thread.updated", location: "active", sequence: snapshot.snapshotSequence + 1, thread});
 state.shells.set("one", {status: "live", snapshot: Option.some(snapshot)});
}
function setup(initial = stages.idle) {
  vi.clearAllMocks();
  const stack = new AsyncDisposableStack();
  registry = AtomRegistry.make();
  const testRegistry = registry;
  stack.defer(() => testRegistry.dispose());
  state.mode = "notifications-and-sound";
  state.inApp = false;
  state.settingsListeners.clear();
  registry.set(environmentIdsAtom, ["one"]);
  update(initial);
  TestNotification.permission = "granted";
  TestNotification.sent = [];
  stack.use(stubGlobal("Notification", TestNotification));
  stack.use(vi.spyOn(window, "focus").mockImplementation(() => undefined));
  stack.use(vi.spyOn(document, "hasFocus").mockImplementation(() => false));
  stack.use(
    vi
      .spyOn(document, "visibilityState", "get")
      .mockImplementation(() => "visible" as DocumentVisibilityState),
  );
  const router = {
    state: { matches: [{ params: {} }] },
    navigate: state.navigate,
    subscribe: () => () => {},
  } as unknown as AnyRouter;
  return {
    start: () => {
      stack.defer(startThreadNotifications(testRegistry, router));
    },
    [Symbol.asyncDispose]: () => stack.disposeAsync(),
  };
}

const titles = () => TestNotification.sent.map(({ title }) => title);
it("production: ordinary approval control", async () => {
 await using app = setup(stages.controlIdle); app.start();
 update(stages.controlPending);
 update(stages.controlQueued);
 console.log("CONTROL", JSON.stringify({titles: titles(), soundCalls: vi.mocked(playNotificationSound).mock.calls.length}));
 expect(playNotificationSound).toHaveBeenCalledTimes(1);
 expect(titles()).toEqual(["Approval needed"]);
});
it("production: subagent approval repeats when follow-up is queued", async () => {
 await using app = setup(); app.start();
 update(stages.pending);
 const parent = registry.get(realThreads.threadShellAtom({environmentId, threadId: ThreadId.make("parent")}))!;
 const resolved = resolveBackgroundThreadRequest(parent);
 expect(parent.pendingSubagentRequest).toBe("approval");
 expect(resolved).toEqual({request: "approval", requestId: null});
 expect(titles()).toEqual(["Approval needed"]);
 update(stages.queued);
 console.log("SUBAGENT", JSON.stringify({resolved, titles: titles(), soundCalls: vi.mocked(playNotificationSound).mock.calls.length, latestRun: stages.queued[0].latestRunId, status: stages.queued[0].status}));
 expect(titles()).toEqual(["Approval needed", "Approval needed"]);
 expect(playNotificationSound).toHaveBeenCalledTimes(2);
});
BROWSER_FIXTURE
cp "$stage/browser.fixture.ts" "$browser"
cd "$repo"
printf '%s\n' '== Production orchestrator queue and shell projection =='
vp test run apps/server/src/orchestration-v2/zz-verify-approval-renotify.test.ts --silent=false
printf '%s\n' '== Production client rollup and notification coordinator, failure and control =='
cd apps/web
vp test run --project browser src/components/zz-verify-approval-renotify.browser.test.ts --silent=false
printf '%s\n' 'DEFECT PRESENT: unchanged subagent approval announced twice; ordinary approval once.'