Python: [Bug]: Foundry hosting feeds the model duplicated conversation history when a working session store is combined with service-side conversation storage #7955

"""Repro: hosted /responses feeds the model the conversation transcript twice
(superlinearly) when a working agent-session store is combined with a chat
client that stores conversation state service-side.

    pip install --pre agent-framework-foundry-hosting
    python repro.py            # buggy: duplicated history, grows every turn
    python repro.py --null-store  # workaround: session store returns None -> clean

No Azure resources needed. The script starts a local ResponsesHostServer on
:8188, drives three chained turns, and prints exactly what the model receives
on each service call. The fake chat client emulates service-side conversation
storage (what FoundryChatClient / OpenAI Responses do when ``store`` is not
False): requests carrying ``conversation_id`` are evaluated against the stored
thread plus the request messages, and responses return the thread id, which
the framework records on the session as ``service_session_id``.
"""

import json
import os
import subprocess
import sys
import time
import urllib.request

LOG = os.path.join(os.path.dirname(os.path.abspath(__file__)), "repro_model_calls.jsonl")
PORT = int(os.environ.get("REPRO_PORT", "8188"))
NULL_STORE = "--null-store" in sys.argv


# --------------------------------------------------------------------------
# server half (run with: python repro.py --serve [--null-store])
# --------------------------------------------------------------------------
def serve() -> None:
    from agent_framework import (
        Agent,
        BaseChatClient,
        ChatResponse,
        ChatResponseUpdate,
        FunctionInvocationLayer,
        ResponseStream,
        SessionStore,
    )
    from agent_framework_foundry_hosting import AgentSessionStoreProvider, ResponsesHostServer

    def log(event, payload):
        with open(LOG, "a") as f:
            f.write(json.dumps({"event": event, "payload": payload}) + "\n")

    def describe(messages):
        out = []
        for m in messages:
            role = getattr(getattr(m, "role", None), "value", getattr(m, "role", None))
            out.append({"role": str(role), "text": getattr(m, "text", None)})
        return out

    class ServiceStorageChatClient(FunctionInvocationLayer, BaseChatClient):
        """Fake model with service-side conversation storage (store=on)."""

        STORES_BY_DEFAULT = True
        _calls = 0
        _threads: dict = {}

        def _inner_get_response(self, *, messages, stream, options, **kwargs):
            ServiceStorageChatClient._calls += 1
            n = ServiceStorageChatClient._calls

            conv = options.get("conversation_id")
            stored = list(self._threads.get(conv, [])) if conv else []
            request = describe(messages)
            # what the real service's model would see: stored thread + request
            log("model_call", {
                "call": n,
                "conversation_id": conv,
                "messages": ([dict(m, origin="service-thread") for m in stored]
                             + [dict(m, origin="request") for m in request]),
            })

            last = messages[-1] if messages else None
            answered = any(getattr(c, "type", None) == "function_result"
                           for c in getattr(last, "contents", []))
            contents = ([{"type": "text", "text": f"ack-{n}"}] if answered else
                        [{"type": "function_call", "call_id": f"call-{n}",
                          "name": "get_time", "arguments": "{}"}])

            tid = conv or "svc-thread-1"
            self._threads[tid] = stored + request + [
                {"role": "assistant", "text": f"ack-{n}" if answered else "(tool call)"}
            ]

            if stream:
                async def _updates():
                    yield ChatResponseUpdate(role="assistant", contents=contents, conversation_id=tid)
                return ResponseStream(_updates(), finalizer=ChatResponse.from_updates)

            async def _response():
                from agent_framework import Message
                return ChatResponse(messages=[Message(role="assistant", contents=contents)],
                                    conversation_id=tid)
            return _response()

    class NullSessionStore(SessionStore):
        async def get(self, session_id):
            return None

        async def set(self, session_id, session):
            return None

    class Provider(AgentSessionStoreProvider):
        def __init__(self):
            # SessionStore() is the framework's own stock in-memory store.
            self._store = NullSessionStore() if NULL_STORE else SessionStore()

        def get_store(self, *, config, platform_context):
            return self._store

    def get_time() -> str:
        """Return the current time."""
        return "noon"

    open(LOG, "w").close()
    agent = Agent(client=ServiceStorageChatClient(), instructions="You are a helpful assistant.",
                  tools=[get_time])
    ResponsesHostServer(agent=agent, agent_session_store_provider=Provider()).run()


# --------------------------------------------------------------------------
# driver half (default entry point)
# --------------------------------------------------------------------------
def post(text, previous_response_id=None):
    body = {"input": text, "stream": False, "store": True, "model": "repro-model"}
    if previous_response_id:
        body["previous_response_id"] = previous_response_id
    req = urllib.request.Request(f"http://127.0.0.1:{PORT}/responses",
                                 data=json.dumps(body).encode(),
                                 headers={"content-type": "application/json"}, method="POST")
    with urllib.request.urlopen(req, timeout=60) as resp:
        return json.loads(resp.read())


def drive() -> None:
    env = dict(os.environ, PORT=str(PORT),
               AGENTSERVER_STATE_ROOT=os.path.join(os.path.dirname(LOG), "repro_state"))
    args = [sys.executable, os.path.abspath(__file__), "--serve"]
    if NULL_STORE:
        args.append("--null-store")
    server = subprocess.Popen(args, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
    try:
        for _ in range(60):
            try:
                urllib.request.urlopen(f"http://127.0.0.1:{PORT}/readiness", timeout=1)
                break
            except Exception:
                time.sleep(0.5)
        r1 = post("Turn one: my favourite colour is teal.")
        r2 = post("Turn two: I live in Melbourne.", r1["id"])
        post("Turn three: what do you know about me?", r2["id"])
    finally:
        server.terminate()

    calls = [json.loads(l)["payload"] for l in open(LOG) if json.loads(l)["event"] == "model_call"]
    for c in calls:
        print(f"\n--- model call {c['call']} "
              f"(conversation_id={c['conversation_id']}, {len(c['messages'])} messages) ---")
        for m in c["messages"]:
            print(f"  [{m.get('origin', '?'):<14}] {m['role']:<10} {(m['text'] or '')[:70]}")

    last = calls[-1]["messages"]
    counts: dict = {}
    for m in last:
        if m["text"]:
            counts[(m["role"], m["text"])] = counts.get((m["role"], m["text"]), 0) + 1
    dupes = {k: v for k, v in counts.items() if v > 1}
    if dupes:
        print("\nDUPLICATED MESSAGES IN FINAL MODEL INPUT:")
        for (role, text), v in dupes.items():
            print(f"  x{v}  {role}: {text[:60]}")
        sys.exit(1)
    print("\nOK: no duplicates in final model input.")


if __name__ == "__main__":
    if "--serve" in sys.argv:
        PORT = int(os.environ.get("PORT", PORT))
        serve()
    else:
        drive()

No exception is raised — the failure is silent transcript duplication (plus token overspend). Output of `python repro.py`, third turn (the fake service's model input = stored service thread + request messages; origin labels added by the repro):

    --- model call 5 (conversation_id=svc-thread-1, 21 messages) ---
      [service-thread] user       Turn one: my favourite colour is teal.
      [service-thread] assistant  (tool call)
      [service-thread] tool
      [service-thread] assistant  ack-2
      [service-thread] user       Turn one: my favourite colour is teal.      <-- turn 1 again
      [service-thread] assistant
      [service-thread] tool
      [service-thread] assistant  ack-2
      [service-thread] user       Turn two: I live in Melbourne.
      [service-thread] assistant  (tool call)
      [service-thread] tool
      [service-thread] assistant  ack-4
      [request       ] user       Turn one: my favourite colour is teal.      <-- turn 1 a third time
      [request       ] assistant
      [request       ] tool
      [request       ] assistant  ack-2
      [request       ] user       Turn two: I live in Melbourne.              <-- turn 2 again
      [request       ] assistant
      [request       ] tool
      [request       ] assistant  ack-4
      [request       ] user       Turn three: what do you know about me?

    DUPLICATED MESSAGES IN FINAL MODEL INPUT:
      x3  user: Turn one: my favourite colour is teal.
      x3  assistant: (tool call)
      x3  assistant: ack-2
      x2  user: Turn two: I live in Melbourne.
      x2  assistant: ack-4

With `--null-store` (workaround), the same three turns produce each message exactly once and the script prints `OK: no duplicates in final model input.`