#3820 · Provider bridge validation diagnostics and response contracts

import { experimental_defineProviderBridge, PROVIDER_BRIDGE_PROTOCOL_VERSION, threadStartParamsSchema, turnStartParamsSchema } from "@get-bb/plugin-sdk/provider-bridge";
import { z } from "zod";
const inbound = z.object({ id: z.union([z.string(), z.number()]), method: z.string(), params: z.unknown() });
let pending: { threadId: string; deltas: unknown[] } | null = null;
function write(message: Record<string, unknown>) { process.stdout.write(JSON.stringify({ jsonrpc: "2.0", ...message }) + "\n"); }
function emit(threadId: string, deltas: unknown[]) { write({ method: "thread/delta", params: { threadId, deltas } }); }
export function handleLine(line: string) {
  const request = inbound.parse(JSON.parse(line));
  const { id, method, params } = request;
  if (method === "initialize") {
    write({ id, result: { protocolVersion: PROVIDER_BRIDGE_PROTOCOL_VERSION, capabilities: { grammarVersions: [3, 3] } } });
  } else if (method === "thread/start") {
    const { threadId } = threadStartParamsSchema.parse(params);
    write({ method: "thread/identity", params: { threadId, providerThreadId: "synthetic-session" } });
    emit(threadId, [{ kind: "session.reset" }]);
    write({ id, result: { providerThreadId: "synthetic-session" } });
  } else if (method === "turn/start") {
    const { threadId, input } = turnStartParamsSchema.parse(params);
    const first = input[0];
    const mode = first?.type === "text" ? first.text : "valid-only";
    const providerTurnId = "synthetic-turn";
    const key = { providerItemId: "synthetic-message" };
    const valid = [
      { kind: "item.open", key, item: { type: "agentMessage", text: "" }, providerTurnId },
      { kind: "item.close", key, status: "completed", item: { type: "agentMessage", text: "synthetic response" }, providerTurnId },
      { kind: "turn.boundary", status: "completed", providerTurnId },
    ];
    const invalid = { kind: "contextWindow", used: "invalid-number", estimated: false, attach: "open", providerTurnId };
    emit(threadId, [{ kind: "turn.open", providerTurnId }]);
    if (mode === "mixed-first" || mode === "mixed-last") {
      pending = { threadId, deltas: valid };
      emit(threadId, mode === "mixed-first" ? [invalid, ...valid] : [...valid, invalid]);
    } else if (mode === "split") {
      emit(threadId, [invalid]);
      emit(threadId, valid);
    } else {
      emit(threadId, valid);
    }
    write({ id, result: {} });
  } else if (method === "provider/health") {
    if (pending !== null) { emit(pending.threadId, pending.deltas); pending = null; }
    write({ id, result: { supported: false } });
  } else {
    write({ id, result: {} });
  }
}
export const experimental_providerBridge = experimental_defineProviderBridge({ handleLine });

import { writeFileSync, rmSync } from "node:fs";
import { fileURLToPath } from "node:url";
import { afterAll, describe, expect, it } from "vitest";
import { getThread, listEvents } from "@bb/db";
import type { ThreadEvent } from "@bb/domain";
import { groupHostDaemonEvents } from "@bb/host-daemon-contract";
import { createAgentRuntime } from "../../../../packages/agent-runtime/src/runtime.js";
import { createScriptedEchoLaunch, fullRuntimeOptions, wait } from "../../../../packages/agent-runtime/src/test/runtime-test-harness.js";
import { threadDeltaNotificationParamsSchema } from "../../../../packages/provider-bridge-protocol/src/thread-delta.js";
import { createTestAppHarness } from "../helpers/test-app.js";
import { registerFakeProviders } from "../helpers/provider-registry.js";
import { seedEnvironment, seedHostSession, seedProjectWithSource, seedThread, seedThreadRuntimeState } from "../helpers/seed.js";
import { internalAuthHeaders } from "../helpers/commands.js";
import { textInput } from "../helpers/prompt-input.js";
import { getActiveTurnId } from "../../src/services/threads/thread-events.js";
const evidence: object[] = [];
afterAll(() => writeFileSync("issue-3820-results.json", JSON.stringify(evidence, null, 2) + "\n"));
async function until(predicate: () => boolean) {
  const end = Date.now() + 4000;
  while (!predicate()) { if (Date.now() >= end) throw new Error("bounded wait expired"); await wait(10); }
}
describe("mixed delta notifications through runtime and persistence", () => {
  it.each(["mixed-first", "mixed-last", "split", "valid-only"])("%s", async mode => {
    const harness = await createTestAppHarness({ seedFirstPartyProviders: false });
    await registerFakeProviders(harness.deps.providerRegistry, harness.deps.pluginHostArtifacts);
    const { host, session } = seedHostSession(harness.deps);
    const { project } = seedProjectWithSource(harness.deps, { hostId: host.id });
    const environment = seedEnvironment(harness.deps, { hostId: host.id, projectId: project.id, status: "ready" });
    const thread = seedThread(harness.deps, { projectId: project.id, environmentId: environment.id, providerId: "fake", status: "active" });
    seedThreadRuntimeState(harness.deps, { threadId: thread.id, environmentId: environment.id, providerThreadId: "synthetic-session", model: "gpt-5-mini", reasoningLevel: "medium", permissionMode: "full" });
    const events: ThreadEvent[] = [];
    const stderr: string[] = [];
    const bridgeLaunch = createScriptedEchoLaunch({ pluginId: "provider-synthetic-deltas", digest: "synthetic-deltas", modulePath: fileURLToPath(new URL("../../../../packages/agent-runtime/src/test/bridges/issue-3820-bridge.ts", import.meta.url)) });
    const runtime = createAgentRuntime({ workspacePath: harness.config.dataDir, onEvent: e => events.push(e), onStderr: line => stderr.push(line), onToolCall: async () => { throw new Error("unexpected synthetic tool call"); } });
    const options = { ...fullRuntimeOptions, model: "gpt-5-mini" };
    let forwarded = 0;
    async function persist() {
      for (; forwarded < events.length; forwarded++) {
        const event = events[forwarded]!;
        const response = await harness.app.request("/internal/session/events", { method: "POST", headers: internalAuthHeaders(harness), body: JSON.stringify({ sessionId: session.id, eventGroups: groupHostDaemonEvents([{ threadId: thread.id, event }]) }) });
        expect(response.status).toBe(200);
      }
    }
    function snapshot() {
      const types = listEvents(harness.db, { threadId: thread.id }).map(row => row.type);
      return { runtimeHasActiveTurn: runtime.getActiveTurnId(thread.id) !== null, serverStatus: getThread(harness.db, thread.id)?.status, serverHasActiveTurn: getActiveTurnId(harness.deps, thread.id) !== null, persistedCompletions: types.filter(t => t === "turn/completed").length, persistedItemCompletions: types.filter(t => t === "item/completed").length, diagnosticEvents: events.filter(e => e.type === "system/error" || e.type === "provider/error" || e.type === "provider/warning").length };
    }
    try {
      const bad = threadDeltaNotificationParamsSchema.safeParse({ threadId: thread.id, deltas: [{ kind: "contextWindow", used: "invalid-number", estimated: false, attach: "open", providerTurnId: "synthetic-turn" }] });
      expect(bad.success).toBe(false);
      if (!bad.success) expect(bad.error.issues.map(i => i.path.join("."))).toEqual(["deltas.0.used"]);
      await runtime.startThread({ environmentId: environment.id, threadId: thread.id, projectId: project.id, providerId: "fake", bridgeLaunch, options });
      await runtime.runTurn({ clientRequestId: "creq_222222222w", threadId: thread.id, input: textInput(mode), options });
      await until(() => events.some(e => e.type === "turn/started"));
      await wait(150);
      await persist();
      const beforeResend = snapshot();
      const mixed = mode.startsWith("mixed");
      expect(beforeResend).toEqual({ runtimeHasActiveTurn: mixed, serverStatus: mixed ? "active" : "idle", serverHasActiveTurn: mixed, persistedCompletions: mixed ? 0 : 1, persistedItemCompletions: mixed ? 0 : 1, diagnosticEvents: 0 });
      if (mixed) {
        await runtime.providerHealth({ providerId: "fake", bridgeLaunch });
        await until(() => events.some(e => e.type === "turn/completed"));
        await persist();
      }
      const final = snapshot();
      expect(final).toEqual({ runtimeHasActiveTurn: false, serverStatus: "idle", serverHasActiveTurn: false, persistedCompletions: 1, persistedItemCompletions: 1, diagnosticEvents: 0 });
      expect(events.filter(e => e.type === "item/completed")).toEqual([expect.objectContaining({ item: expect.objectContaining({ type: "agentMessage", text: "synthetic response" }) })]);
      expect(stderr).toEqual([]);
      evidence.push({ case: mode, invalidField: "deltas.0.used", beforeResend, final, runtimeStderrLines: stderr.length, runtimeEventTypes: events.map(e => e.type), persistedEventTypes: listEvents(harness.db, { threadId: thread.id }).map(row => row.type) });
    } finally { await runtime.shutdown(); rmSync(bridgeLaunch.dataDir, { recursive: true, force: true }); await harness.cleanup(); }
  }, 15000);
});

[
  {
    "case": "mixed-first",
    "invalidField": "deltas.0.used",
    "beforeResend": {
      "runtimeHasActiveTurn": true,
      "serverStatus": "active",
      "serverHasActiveTurn": true,
      "persistedCompletions": 0,
      "persistedItemCompletions": 0,
      "diagnosticEvents": 0
    },
    "final": {
      "runtimeHasActiveTurn": false,
      "serverStatus": "idle",
      "serverHasActiveTurn": false,
      "persistedCompletions": 1,
      "persistedItemCompletions": 1,
      "diagnosticEvents": 0
    },
    "runtimeStderrLines": 0,
    "runtimeEventTypes": [
      "thread/identity",
      "provider.env-resolved",
      "turn/started",
      "item/started",
      "item/completed",
      "turn/completed"
    ],
    "persistedEventTypes": [
      "thread/identity",
      "client/turn/requested",
      "thread/identity",
      "provider.env-resolved",
      "turn/started",
      "item/started",
      "item/completed",
      "turn/completed"
    ]
  },
  {
    "case": "mixed-last",
    "invalidField": "deltas.0.used",
    "beforeResend": {
      "runtimeHasActiveTurn": true,
      "serverStatus": "active",
      "serverHasActiveTurn": true,
      "persistedCompletions": 0,
      "persistedItemCompletions": 0,
      "diagnosticEvents": 0
    },
    "final": {
      "runtimeHasActiveTurn": false,
      "serverStatus": "idle",
      "serverHasActiveTurn": false,
      "persistedCompletions": 1,
      "persistedItemCompletions": 1,
      "diagnosticEvents": 0
    },
    "runtimeStderrLines": 0,
    "runtimeEventTypes": [
      "thread/identity",
      "provider.env-resolved",
      "turn/started",
      "item/started",
      "item/completed",
      "turn/completed"
    ],
    "persistedEventTypes": [
      "thread/identity",
      "client/turn/requested",
      "thread/identity",
      "provider.env-resolved",
      "turn/started",
      "item/started",
      "item/completed",
      "turn/completed"
    ]
  },
  {
    "case": "split",
    "invalidField": "deltas.0.used",
    "beforeResend": {
      "runtimeHasActiveTurn": false,
      "serverStatus": "idle",
      "serverHasActiveTurn": false,
      "persistedCompletions": 1,
      "persistedItemCompletions": 1,
      "diagnosticEvents": 0
    },
    "final": {
      "runtimeHasActiveTurn": false,
      "serverStatus": "idle",
      "serverHasActiveTurn": false,
      "persistedCompletions": 1,
      "persistedItemCompletions": 1,
      "diagnosticEvents": 0
    },
    "runtimeStderrLines": 0,
    "runtimeEventTypes": [
      "thread/identity",
      "provider.env-resolved",
      "turn/started",
      "item/started",
      "item/completed",
      "turn/completed"
    ],
    "persistedEventTypes": [
      "thread/identity",
      "client/turn/requested",
      "thread/identity",
      "provider.env-resolved",
      "turn/started",
      "item/started",
      "item/completed",
      "turn/completed"
    ]
  },
  {
    "case": "valid-only",
    "invalidField": "deltas.0.used",
    "beforeResend": {
      "runtimeHasActiveTurn": false,
      "serverStatus": "idle",
      "serverHasActiveTurn": false,
      "persistedCompletions": 1,
      "persistedItemCompletions": 1,
      "diagnosticEvents": 0
    },
    "final": {
      "runtimeHasActiveTurn": false,
      "serverStatus": "idle",
      "serverHasActiveTurn": false,
      "persistedCompletions": 1,
      "persistedItemCompletions": 1,
      "diagnosticEvents": 0
    },
    "runtimeStderrLines": 0,
    "runtimeEventTypes": [
      "thread/identity",
      "provider.env-resolved",
      "turn/started",
      "item/started",
      "item/completed",
      "turn/completed"
    ],
    "persistedEventTypes": [
      "thread/identity",
      "client/turn/requested",
      "thread/identity",
      "provider.env-resolved",
      "turn/started",
      "item/started",
      "item/completed",
      "turn/completed"
    ]
  }
]