Failed headerless inline /new leaves an unreachable binding and forks retries

From the repository root, save this as repro_api_new_inline_failure.mts and run:

pnpm install --frozen-lockfile
pnpm exec tsx repro_api_new_inline_failure.mts

import { Readable } from "node:stream";
import { ApiServerChannel } from "./src/adapters/channel/api-server/ApiServerChannel.js";
import { ApiServerSessionMapper } from "./src/adapters/channel/api-server/ApiServerSessionMapper.js";
import type { Gateway, GatewayEvent, GatewaySubmitTurnInput } from "./src/gateway/index.js";

type FailureMode = "throw-before-yield" | "partial-then-throw";
type Call = Pick<GatewaySubmitTurnInput, "sessionKey" | "channelKey" | "message">;
type ResponseSummary = {
  status: number;
  responseSessionHeaderPresent: boolean;
  hasGatewayUnavailable: boolean;
  hasChannelSubmitFailed: boolean;
  containsPartialReply: boolean;
  containsDone: boolean;
};

const FIXTURE_SESSION_UUID = "00000000-0000-0000-0000-000000000042";

class FakeResponse {
  statusCode = 200;
  readonly headers = new Map<string, string>();
  private readonly chunks: string[] = [];

  setHeader(name: string, value: string): void {
    this.headers.set(name.toLowerCase(), String(value));
  }

  write(chunk: string): boolean {
    this.chunks.push(String(chunk));
    return true;
  }

  flushHeaders(): void {}

  end(chunk?: string): void {
    if (chunk !== undefined) this.chunks.push(String(chunk));
  }

  get body(): string {
    return this.chunks.join("");
  }
}

function makeMapper(): ApiServerSessionMapper {
  return new ApiServerSessionMapper({ activeByChatId: {} }, () => FIXTURE_SESSION_UUID);
}

function makeRequest(content: string, stream: boolean, sessionId?: string): Readable & {
  method?: string;
  url?: string;
  headers: Record<string, string>;
} {
  const headers: Record<string, string> = {
    host: "fixture.invalid",
    "content-type": "application/json",
  };
  if (sessionId !== undefined) headers["x-hermes-session-id"] = sessionId;
  const request = Readable.from([Buffer.from(JSON.stringify({
    model: "fixture-model",
    messages: [{ role: "user", content }],
    stream,
  }))]) as Readable & { method?: string; url?: string; headers: Record<string, string> };
  request.method = "POST";
  request.url = "/v1/chat/completions";
  request.headers = headers;
  return request;
}

async function post(
  channel: ApiServerChannel,
  content: string,
  stream: boolean,
  sessionId?: string,
): Promise<{ response: ResponseSummary; sessionHeader: string | null }> {
  const response = new FakeResponse();
  await (channel as unknown as {
    handleRequest(req: unknown, res: FakeResponse): Promise<void>;
  }).handleRequest(makeRequest(content, stream, sessionId), response);
  return {
    response: {
      status: response.statusCode,
      responseSessionHeaderPresent: response.headers.has("x-hermes-session-id"),
      hasGatewayUnavailable: response.body.includes("gateway_unavailable"),
      hasChannelSubmitFailed: response.body.includes("channel_submit_failed"),
      containsPartialReply: response.body.includes("fixture reply"),
      containsDone: response.body.includes("data: [DONE]"),
    },
    sessionHeader: response.headers.get("x-hermes-session-id") ?? null,
  };
}

function makeGateway(mode: "success" | FailureMode, calls: Call[]): Gateway {
  return {
    submitTurn(input: GatewaySubmitTurnInput): AsyncIterable<GatewayEvent> {
      calls.push({
        sessionKey: input.sessionKey,
        channelKey: input.channelKey,
        message: input.message,
      });
      return (async function* (): AsyncGenerator<GatewayEvent> {
        if (mode === "throw-before-yield") {
          throw new Error("synthetic submit failure");
        }
        yield { type: "assistant_text_delta", text: "fixture reply" };
        if (mode === "partial-then-throw") {
          throw new Error("synthetic partial failure");
        }
        yield {
          type: "turn_completed",
          usage: { inputTokens: 0, outputTokens: 0, totalTokens: 0 },
          finishReason: "completed",
        };
      })();
    },
  } as unknown as Gateway;
}

function installGateway(channel: ApiServerChannel, gateway: Gateway | undefined): void {
  (channel as unknown as { gateway?: Gateway }).gateway = gateway;
}

function mapperEntries(channel: ApiServerChannel): Array<[string, string]> {
  return Object.entries((channel as unknown as {
    mapper: { snapshot(): { activeByChatId: Record<string, string> } };
  }).mapper.snapshot().activeByChatId);
}

function assertFixture(condition: boolean, message: string): void {
  if (!condition) throw new Error(`fixture assertion failed: ${message}`);
}

function summarizeFailure(
  failed: { response: ResponseSummary; sessionHeader: string | null },
  retry: { response: ResponseSummary; sessionHeader: string | null },
  afterFailure: Array<[string, string]>,
  afterRetry: Array<[string, string]>,
  failedCalls: Call[],
  retryCalls: Call[],
  expectedFailureStatus: number,
): Record<string, unknown> {
  const initialBinding = afterFailure[0];
  const retryCall = retryCalls[0];
  const failedCall = failedCalls[0];
  const initialChatId = initialBinding?.[0] ?? null;
  const initialSessionKey = initialBinding?.[1] ?? null;

  assertFixture(initialBinding !== undefined, "inline /new must create a mapper entry before failure");
  assertFixture(initialSessionKey.includes(":s_"), "mapper entry must be the reset session");
  assertFixture(failed.response.status === expectedFailureStatus, "failure status must match the selected mode");
  assertFixture(failed.sessionHeader === null, "buffered failure must not expose a session header");
  assertFixture(retry.response.status === 200, "headerless retry must reach the normal completion path");
  assertFixture(retryCall !== undefined, "retry must call the Gateway");
  assertFixture(retryCall.sessionKey.endsWith(":general"), "headerless retry must use the general fallback");
  assertFixture(retryCall.sessionKey !== initialSessionKey, "retry must not reuse the orphaned reset binding");
  assertFixture(retry.sessionHeader !== null, "successful retry should expose its generated chat id");
  assertFixture(retry.sessionHeader !== initialChatId, "retry chat id must differ from the failed request chat id");

  return {
    failed: {
      ...failed.response,
      responseSessionHeaderReturned: failed.sessionHeader !== null,
    },
    mapperAfterFailure: {
      entryCount: afterFailure.length,
      resetBindingRetained: initialSessionKey.includes(":s_"),
      bindingReachableByReturnedFailureToken: failed.sessionHeader !== null,
    },
    failedGatewayCall: {
      occurred: failedCall !== undefined,
      usedCreatedResetBinding: failedCall?.sessionKey === initialSessionKey,
      message: failedCall?.message ?? null,
    },
    retry: {
      requestSessionHeader: null,
      ...retry.response,
      responseSessionHeaderReturned: retry.sessionHeader !== null,
      responseTokenMatchesFailedChatId: retry.sessionHeader === initialChatId,
      usesInitialMappedSession: retryCall?.sessionKey === initialSessionKey,
      usesGeneralFallback: retryCall?.sessionKey.endsWith(":general") ?? false,
      message: retryCall?.message ?? null,
    },
    mapperAfterRetry: {
      entryCount: afterRetry.length,
      originalResetBindingStillRetained: afterRetry.some(([chatId, sessionKey]) =>
        chatId === initialChatId && sessionKey === initialSessionKey),
    },
  };
}

async function runNoGateway(): Promise<Record<string, unknown>> {
  const channel = new ApiServerChannel({
    host: "127.0.0.1",
    port: 0,
    modelName: "fixture-model",
    mapper: makeMapper(),
  });
  const failed = await post(channel, "/new first prompt", false);
  const afterFailure = mapperEntries(channel);
  const retryCalls: Call[] = [];
  installGateway(channel, makeGateway("success", retryCalls));
  const retry = await post(channel, "retry without returned session header", false);
  return summarizeFailure(
    failed,
    retry,
    afterFailure,
    mapperEntries(channel),
    [],
    retryCalls,
    503,
  );
}

async function runBufferedFailure(mode: FailureMode): Promise<Record<string, unknown>> {
  const channel = new ApiServerChannel({
    host: "127.0.0.1",
    port: 0,
    modelName: "fixture-model",
    mapper: makeMapper(),
  });
  const failedCalls: Call[] = [];
  installGateway(channel, makeGateway(mode, failedCalls));
  const failed = await post(channel, "/new first prompt", false);
  const afterFailure = mapperEntries(channel);
  const retryCalls: Call[] = [];
  installGateway(channel, makeGateway("success", retryCalls));
  const retry = await post(channel, "retry without returned session header", false);
  return {
    mode,
    ...summarizeFailure(
      failed,
      retry,
      afterFailure,
      mapperEntries(channel),
      failedCalls,
      retryCalls,
      500,
    ),
  };
}

async function runSuppliedHeaderControl(): Promise<Record<string, unknown>> {
  const channel = new ApiServerChannel({
    host: "127.0.0.1",
    port: 0,
    modelName: "fixture-model",
    mapper: makeMapper(),
  });
  const failedCalls: Call[] = [];
  installGateway(channel, makeGateway("throw-before-yield", failedCalls));
  const failed = await post(channel, "/new first prompt", false, "known-inline-client");
  const afterFailure = mapperEntries(channel);
  const initialBinding = afterFailure[0];
  const retryCalls: Call[] = [];
  installGateway(channel, makeGateway("success", retryCalls));
  const retry = await post(channel, "retry with caller-supplied session header", false, "known-inline-client");

  assertFixture(initialBinding?.[0] === "known-inline-client", "control must use the supplied chat id");
  assertFixture(failed.sessionHeader === null, "buffered error may omit the supplied header");
  assertFixture(retryCalls[0]?.sessionKey === initialBinding?.[1], "supplied header must address the reset binding");
  assertFixture(retry.response.status === 200, "supplied-header retry must succeed");

  return {
    failed: {
      ...failed.response,
      responseSessionHeaderReturned: failed.sessionHeader !== null,
    },
    retry: {
      ...retry.response,
      requestSessionHeader: "known-inline-client",
      responseSessionHeaderPresent: retry.sessionHeader !== null,
      reusedSuppliedChatId: retry.sessionHeader === "known-inline-client",
      usesInitialMappedSession: retryCalls[0]?.sessionKey === initialBinding?.[1],
    },
    mapperAfterFailure: {
      entryCount: afterFailure.length,
      resetBindingRetained: initialBinding?.[1].includes(":s_") ?? false,
    },
  };
}

async function runStreamingControl(): Promise<Record<string, unknown>> {
  const channel = new ApiServerChannel({
    host: "127.0.0.1",
    port: 0,
    modelName: "fixture-model",
    mapper: makeMapper(),
  });
  const calls: Call[] = [];
  installGateway(channel, makeGateway("partial-then-throw", calls));
  const failed = await post(channel, "/new first prompt", true);
  assertFixture(failed.response.status === 200, "streaming fixture should keep the established SSE status");
  assertFixture(failed.sessionHeader !== null, "streaming path sets a header before submitTurn");
  assertFixture(failed.response.hasChannelSubmitFailed, "streaming fixture must expose the synthetic failure");
  assertFixture(!failed.response.containsDone, "partial stream failure must omit [DONE]");
  return {
    failed: {
      ...failed.response,
      responseSessionHeader: failed.sessionHeader !== null,
    },
    gatewayCallOccurred: calls.length === 1,
    bindingRetained: mapperEntries(channel).length === 1,
    scopeNote: "Streaming failure exposes the token before iteration; terminal SSE behavior is a separate finding.",
  };
}

const results = {
  noGatewayBuffered: await runNoGateway(),
  throwBeforeYieldBuffered: await runBufferedFailure("throw-before-yield"),
  partialThenThrowBuffered: await runBufferedFailure("partial-then-throw"),
  callerSuppliedHeaderControl: await runSuppliedHeaderControl(),
  streamingControl: await runStreamingControl(),
};

console.log(JSON.stringify({
  fixture: "api-server-inline-new-failure-binding-witness",
  transport: "in-memory IncomingMessage/ServerResponse equivalent",
  endpoint: "POST /v1/chat/completions",
  input: {
    firstMessage: "/new first prompt",
    retryMessage: "retry without returned session header",
    firstRequestSessionHeader: null,
    retryRequestSessionHeader: null,
  },
  sanitization: "No UUIDs or raw session keys are emitted; comparisons are represented as booleans and stable classes.",
  results,
}));