API completions admits non-user final roles as user turns and mutates session state

[MicroMilo]

Summary

Buffered and streaming matrices admit assistant, system, tool, developer, unknown, missing, null, numeric, and object roles with HTTP 200 and a Gateway call. The roleless Gateway input is converted to AgentInput and persisted as canonical role=user. An assistant /new can rotate the mapper, and a blocked assistant /compact keeps activeChats occupied so a later valid user request receives session_busy.

Expected behavior

For this turn-oriented API, only a final role=user message should create a Gateway turn. Other roles and malformed role values must be rejected before mapper, activeChats, Gateway, and SSE admission.

Actual behavior

Buffered and streaming matrices admit assistant, system, tool, developer, unknown, missing, null, numeric, and object roles with HTTP 200 and a Gateway call. The roleless Gateway input is converted to AgentInput and persisted as canonical role=user. An assistant /new can rotate the mapper, and a blocked assistant /compact keeps activeChats occupied so a later valid user request receives session_busy.

Impact

Invalid callers can mutate session identity, create turns with lost role semantics, and block valid requests even though the API reports success.

Reproduction

With the API server running, POST /v1/chat/completions using a fixed session header and a non-empty message. Repeat with final roles assistant, system, tool, developer, an unknown string, a missing role, null, a number, and an object, in both stream=false and stream=true modes. Only role=user should create a turn. The observed result is HTTP 200 and Gateway admission for the non-user/malformed cases; some of those turns also rotate /new state or occupy the active-session lock.

Minimal reproduction script

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

pnpm install --frozen-lockfile
pnpm exec tsx repro_api_role_forwarding.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";

const CHAT_ID = "api-side-effect-session";
const OLD_SESSION = "api_server:chat=api-side-effect-session:s_old";

class CaptureResponse {
  statusCode = 0;
  headers = new Map();
  chunks = [];
  ended = false;

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

  flushHeaders() {}

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

  end(chunk) {
    if (chunk != null) this.chunks.push(String(chunk));
    this.ended = true;
  }

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

function makeRequest(role, content, stream = false) {
  const body = JSON.stringify({
    model: "side-effect-model",
    messages: [
      { role: "user", content: "sanitized-prior" },
      { role, content },
    ],
    stream,
  });
  const req = Readable.from([Buffer.from(body)]);
  req.method = "POST";
  req.url = "/v1/chat/completions";
  req.headers = {
    host: "fixture.invalid",
    "content-type": "application/json",
    "x-hermes-session-id": CHAT_ID,
  };
  return req;
}

function mapperWithOldSession() {
  return new ApiServerSessionMapper(
    { activeByChatId: { [CHAT_ID]: OLD_SESSION } },
    () => "00000000-0000-4000-8000-000000000099",
  );
}

function activeChatsSnapshot(channel) {
  return [...channel.activeChats];
}

function responseSummary(response) {
  let parsed = null;
  try { parsed = JSON.parse(response.body); } catch {}
  return {
    status: response.statusCode,
    contentType: response.headers.get("content-type") ?? null,
    sessionHeader: response.headers.get("x-hermes-session-id") ?? null,
    errorEvent: parsed?.error?.event ?? null,
    errorCode: parsed?.error?.code ?? null,
    object: parsed?.object ?? null,
    hasSseData: response.body.includes("data: "),
    hasDone: response.body.includes("data: [DONE]"),
    ended: response.ended,
  };
}

function request(channel, role, content, stream = false) {
  const response = new CaptureResponse();
  return channel.handleRequest(makeRequest(role, content, stream), response)
    .then(() => responseSummary(response));
}

function immediateGateway(calls) {
  return {
    async *submitTurn(input) {
      calls.push({
        sessionKey: input.sessionKey,
        channelKey: input.channelKey,
        message: input.message,
      });
      yield { type: "assistant_text_delta", text: "sanitized-reply" };
      yield {
        type: "turn_completed",
        usage: { inputTokens: 0, outputTokens: 0, totalTokens: 0 },
        finishReason: "completed",
      };
    },
  };
}

function blockingGateway(calls) {
  let release;
  const released = new Promise((resolve) => { release = resolve; });
  return {
    release,
    gateway: {
      async *submitTurn(input) {
        calls.push({
          sessionKey: input.sessionKey,
          channelKey: input.channelKey,
          message: input.message,
        });
        yield { type: "turn_started", runId: "sanitized-run-hold" };
        await released;
        yield { type: "assistant_text_delta", text: "sanitized-reply" };
        yield {
          type: "turn_completed",
          usage: { inputTokens: 0, outputTokens: 0, totalTokens: 0 },
          finishReason: "completed",
        };
      },
    },
  };
}

async function runIdleCase(name, content, stream) {
  const calls = [];
  const mapper = mapperWithOldSession();
  const channel = new ApiServerChannel({ mapper, modelName: "side-effect-model" });
  channel.gateway = immediateGateway(calls);
  const before = mapper.snapshot();
  const activeBefore = activeChatsSnapshot(channel);
  const response = await request(channel, "assistant", content, stream);
  return {
    name,
    state: "idle",
    role: "assistant",
    content,
    stream,
    before,
    after: mapper.snapshot(),
    activeBefore,
    activeAfter: activeChatsSnapshot(channel),
    calls,
    response,
  };
}

async function runActiveInvalidCase(name, content, stream) {
  const calls = [];
  const mapper = mapperWithOldSession();
  const channel = new ApiServerChannel({ mapper, modelName: "side-effect-model" });
  const blocked = blockingGateway(calls);
  channel.gateway = blocked.gateway;

  const firstResponse = new CaptureResponse();
  const firstPromise = channel.handleRequest(
    makeRequest("user", "sanitized-valid-hold"),
    firstResponse,
  );
  while (calls.length === 0) await new Promise((resolve) => setImmediate(resolve));

  const beforeInvalid = mapper.snapshot();
  const activeBeforeInvalid = activeChatsSnapshot(channel);
  const invalidResponse = await request(channel, "assistant", content, stream);
  const afterInvalid = mapper.snapshot();
  const activeAfterInvalid = activeChatsSnapshot(channel);
  blocked.release();
  await firstPromise;

  return {
    name,
    state: "valid-turn-already-active",
    role: "assistant",
    content,
    stream,
    beforeInvalid,
    afterInvalid,
    activeBeforeInvalid,
    activeAfterInvalid,
    calls,
    invalidResponse,
    firstResponse: responseSummary(firstResponse),
  };
}

async function runInvalidAdmissionCase(name, content, stream) {
  const calls = [];
  const mapper = mapperWithOldSession();
  const channel = new ApiServerChannel({ mapper, modelName: "side-effect-model" });
  const blocked = blockingGateway(calls);
  channel.gateway = blocked.gateway;

  const invalidResponse = new CaptureResponse();
  const invalidPromise = channel.handleRequest(
    makeRequest("assistant", content, stream),
    invalidResponse,
  );
  while (calls.length === 0) await new Promise((resolve) => setImmediate(resolve));

  const mapperDuring = mapper.snapshot();
  const activeDuring = activeChatsSnapshot(channel);
  const subsequentResponse = await request(channel, "user", "sanitized-user-after-invalid");
  blocked.release();
  await invalidPromise;

  return {
    name,
    state: "invalid-role-admitted-and-blocked",
    role: "assistant",
    content,
    stream,
    mapperDuring,
    activeDuring,
    mapperAfter: mapper.snapshot(),
    activeAfter: activeChatsSnapshot(channel),
    calls,
    invalidResponse: responseSummary(invalidResponse),
    subsequentResponse,
  };
}

const results = [];
for (const [name, content] of [
  ["idle-bare-new", "/new"],
  ["idle-new-suffix", "/new sanitized-follow-up"],
  ["idle-slash-command", "/compact"],
]) {
  results.push(await runIdleCase(name, content, false));
  results.push(await runIdleCase(`${name}-stream`, content, true));
}

results.push(await runActiveInvalidCase("active-invalid-bare-new", "/new", false));
results.push(await runActiveInvalidCase("active-invalid-slash-command", "/compact", true));
results.push(await runInvalidAdmissionCase("invalid-role-admits-slash-command", "/compact", false));
results.push(await runInvalidAdmissionCase("invalid-role-admits-slash-command-stream", "/compact", true));
results.push(await runInvalidAdmissionCase("invalid-role-admits-new-suffix", "/new sanitized-follow-up", false));

console.log(JSON.stringify({
  fixture: "api-server-invalid-role-state-side-effects-witness",
  transport: "in-memory IncomingMessage/ServerResponse equivalent",
  endpoint: "POST /v1/chat/completions",
  roleUnderTest: "assistant",
  sessionId: CHAT_ID,
  results,
}));

Relevant source locations

• src/adapters/channel/api-server/ApiServerChannel.ts:213-235
• src/adapters/channel/api-server/ApiServerChannel.ts:238-286
• src/adapters/channel/api-server/ApiServerChannel.ts:303-321
• src/adapters/channel/api-server/ApiServerChannel.ts:353-378
• src/adapters/channel/api-server/ApiServerSessionMapper.ts:13-28
• src/gateway/protocol/types.ts:86-120
• src/agent/turn/TurnInputProcessor.ts:9-31
• src/agent/turn/TurnRunner.ts:83-106

Suggested direction

Make the external-input path establish one durable, identity-bound state/receipt before returning success; propagate explicit terminal outcomes to every channel and client; and add a regression test for the reproduced boundary.

This report is about functional behavior, not security. The reproduction uses deterministic in-memory or isolated fixtures and contains no credentials or private data.