Python: Give a hosted Foundry agent a single source of conversation history

[Atharva Vichare (atty57)]

Motivation & Context

A hosted /responses agent feeds the model the conversation transcript more than once per request, and the duplication compounds every turn. With a real model the agent visibly repeats its replies, getting worse as the conversation grows, plus the matching token overspend. In the reporter's repro, by turn 3 the model sees turn 1 three times (23 messages instead of 9).

_handle_inner_agent gives agent.run(...) two sources of history at once: run_kwargs["messages"], the full platform transcript from context.get_history() plus the new input, and run_kwargs["session"], the session loaded from the session store. The existing guard pops the transient history buffer (_HOSTED_RESPONSES_HISTORY_SOURCE_ID) out of session.state, but the session's top-level service_session_id survives and core resumes it. The run therefore continues a service-side thread that already holds the whole transcript while messages carries that transcript again. The service appends the duplicated request to the thread, so the next turn's thread contains it twice, hence the superlinear growth.

This is rarely seen because the default FoundryAgentSessionStore's reads currently fail and return None, so the service thread is never resumed. Any working session store exposes it.

.NET fixed this class of bug in #7525 and refined it in #7572; this is the Python counterpart.

Description & Review Guide

• What are the major changes?

  Server-side storage is turned off for the run, so the platform record is the only history in play:

  • chat_options["store"] = False when hosting manages history. The chat client then keeps nothing of its own and reports no conversation id, so no second thread exists to resume. This removes the cause rather than cleaning up after it, and avoids creating a downstream thread per turn.
  • If a service session id lands on the session anyway, the client stored the turn despite that setting and a second unreconciled record now exists. The request fails and the session is left unsaved, so later turns cannot resume onto the duplicated thread. A container configured this way is a server fault, not a bad request.
  • New allow_stored_output_enabled keyword on ResponsesHostServer, defaulting to False. Setting it to True leaves the chat client exactly as the container configured it; nothing is overridden or checked, and reconciling the two records is the container's responsibility.

  Three regression tests in TestAgentSessionPersistence cover the duplicated transcript, the client that stores despite store=False, and the opt-in. Each fails without the source change.

• What is the impact of these changes?

  Hosted agents whose chat client stores conversation state service-side (the default for FoundryChatClient and OpenAI Responses when store is not False) now send each message to the model exactly once.

  Behaviour changes for those agents: the downstream service is asked not to store, and an agent that stores anyway now fails loudly instead of silently duplicating. allow_stored_output_enabled=True restores the previous hands-off behaviour. The constructor keyword is additive and defaulted, so no signature breaks.

  Note that Python: [Bug]: Foundry Hosted agent - delay before final response when streaming #7487, the ~5s streaming delay seen with the Foundry project endpoint and store: False, was closed as not caused by the SDK and not reproduced as consistent. This change makes store: False the default path for hosted agents, so if that delay is real it would become more visible; allow_stored_output_enabled=True is the escape hatch.

• What do you want reviewers to focus on?

  Whether disabling downstream storage by default is the direction you want for Python, matching .NET: Add Options for Hosted Agent to Allow Backend Storage #7572, versus neutralizing service_session_id on the hosted session. Also whether the misconfigured-client check belongs in the finally block as written, and whether a readiness-time probe like the .NET one is wanted here as a follow-up.

Related Issue

Fixes #7955

Contribution Checklist

• The code builds clean without any errors or warnings
• All unit tests pass, and I have added new tests where possible
• The PR follows the Contribution Guidelines
• This PR is linked to an issue and there is no other open PR for this issue (see Related Issue above).
• This is not a breaking change. If it is a breaking change, add the breaking change label (or add "[BREAKING]" to the title prefix, before or after any language prefix) — a workflow keeps the label and title prefix in sync automatically.

[Copilot]

Pull request overview

Makes hosted Foundry agents rely exclusively on platform-managed conversation history.

Changes:

• Disables downstream response storage by default.
• Detects clients that store despite configuration.
• Adds an opt-in and regression tests.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.

File	Description
_responses.py	Adds storage policy and violation handling.
test_responses.py	Tests storage, failure, and opt-in behavior.

---

💡 Add a code-review agent skill for context-aware, tailored reviews. Learn more in the docs.

[Sachin Kahawala (sachinkahawala)]

Reporter of #7955 here — thanks for picking this up so quickly, and the direction question you pose at the end is exactly the right one. I ran my repro bench from the issue against this branch (plus an upgrade simulation), so here are verification results rather than opinions.

✅ Verified: the fix works for fresh conversations. With a fake service that honours store=False (keeps nothing, returns no conversation id), three chained turns on this branch are clean — store=False on every service call, no thread ever created, each message reaches the model exactly once, where the released 1.0.0b260827 shows turn 1 reaching the model three times by turn 3.

⚠️ Finding 1: conversations that existed before this fix get wedged permanently. I simulated an upgrade: turns 1–2 on released 1.0.0b260827 (session persisted with service_session_id="svc-thread-1" via a FileSessionStore), then turn 3 on this branch continuing the same chain. Result:

• The loaded session's stale service_session_id is still forwarded as conversation_id (core gives it precedence), so the model input for turn 3 is still duplicated (my repro's duplicate detector fires: turn 1 seen ×3), and then
• the finally check sees session.service_session_id is not None and fails the request with the new server_error — even though the client honoured store=False perfectly. The check can't distinguish "client stored anyway" from "session carried a pre-fix thread id".
• Because the session is deliberately left unsaved, the same stale session loads on every later turn, so every subsequent turn of that conversation fails the same way until the conversation is abandoned.

(If the real service instead rejects store=false + an existing conversation reference outright, the outcome is the same — a hard failure on every turn of a pre-existing conversation.)

Suggested remedy, which I think also answers your own review question: clear service_session_id when loading the session in hosted-history mode (i.e. neutralize on load), and treat only an id that newly appears after a store=False run as the misconfiguration violation. Neutralize-on-load composes with your store=False default rather than competing with it — legacy sessions then heal gracefully on their first post-upgrade turn instead of bricking.

⚠️ Finding 2: this makes store: False the default path for all hosted agents, and #7487 is real in production. The PR notes the ~5 s delay "was not reproduced as consistent". For what it's worth, in my production deployment (Foundry Hosted Agent, FoundryChatClient via the Foundry project endpoint, streaming, gpt-5-mini, measured Aug 2026) the delay was consistent: roughly +5 s per model call whenever store: False was set, on every call — it's the reason my agent runs with store left on, which is how #7955 surfaced in the first place. If store=False becomes the hosted default, I'd expect that regression to become visible to every hosted Python agent on the project endpoint. Worth re-validating against a live Foundry endpoint before this merges (happy to share my measurement setup, and I can re-measure on my deployment if useful).

My repro/upgrade-sim scripts are self-contained (no Azure needed) — glad to share them here or test iterations of this branch. If it's easier, I'm also happy to push a follow-up implementing neutralize-on-load: I have it working on top of this branch against my bench (fresh conversations stay clean, and the pre-fix-conversation case above completes cleanly instead of failing).

[Atharva Vichare (atty57)]

Thanks — this is a genuinely useful report, and Finding 1 is real. Fixed in 7dcf9cb.

Finding 1. Confirmed exactly as you describe. _prepare_run_options takes conversation_id from _get_chat_conversation_id(active_session) whenever a session exists (_agents.py:1490), so a session loaded with a pre-fix service_session_id still forwards it and store=False never clears it. The finally check then can't distinguish "client stored anyway" from "session carried a stale id", and because a failed turn is deliberately not saved, every later turn reloads the same stale session and fails identically.

Your neutralize-on-load remedy is also what makes that check exact instead of heuristic, which is why I took it as written: session.service_session_id = None alongside the existing pop of the transient history buffer means any id present in finally can only have been set by this run. Gated on not allow_stored_output_enabled, so the opt-in still leaves the session untouched.

Added a regression test for the upgrade path: turn 1, then the stored session mutated to carry service_session_id="svc-thread-1" as a pre-fix version would have left it, then turn 2 on this branch. Without the change it fails with your symptom (status == "failed"); with it, the stale id is not forwarded (conversation_ids == [None, None]), the input is not duplicated, the turn completes, and the session is saved clean. 240 tests pass in packages/foundry_hosting.

I kept it in this PR rather than a follow-up since the repo asks for one open PR per issue — but I'd still like to run your upgrade-sim against it, since your bench covers the real chain and mine simulates the pre-fix state. Please do paste the scripts or link a gist.

Finding 2 is a maintainer call, and your measurement is a stronger signal than #7487's "not reproduced as consistent" — it's the exact population this change affects: hosted Python agents on the Foundry project endpoint. Worth being explicit that the two halves of this PR are separable:

• Neutralize-on-load alone fixes Python: [Bug]: Foundry hosting feeds the model duplicated conversation history when a working session store is combined with service-side conversation storage #7955. With the id cleared each turn no service thread is ever resumed, so the model sees each message once even with store left on. The cost is an orphan thread per turn — wasteful, not incorrect.
• store=False on top is the cleanup: no orphan threads, no reliance on the service ignoring an unreferenced one. But it makes Python: [Bug]: Foundry Hosted agent - delay before final response when streaming #7487 the default path for every hosted agent.

So if the +5 s reproduces on a live endpoint, the safe shape is neutralize-on-load as the default with store=False behind the opt-in — the inverse of the current defaults, and a one-line flip from where the branch now stands. I'd rather not flip it on my own read; that's the same direction question in the description, now with a concrete alternative attached.

@microsoft/agent-framework-python — the ask is: (a) keep allow_stored_output_enabled=False as the default and treat the +5 s as #7487 to fix separately, or (b) default to leaving store alone and ship the neutralize-on-load half only. Happy to push (b) if that's the call. And Sachin Kahawala (@sachinkahawala), if you can re-measure the delay on your deployment against this branch, that would settle which one this should be.

[Sachin Kahawala (sachinkahawala)]

Ran the bench against 7dcf9cb — Finding 1 is fixed, verified on both paths:

• Fresh chain (turns 1–3, session store = the framework's stock SessionStore, client honours store=False): every service call gets store=False, conversation_id=None, no thread created, each message reaches the model once. Exit clean.
• Upgrade sim (turns 1–2 on released 1.0.0b260827 with a FileSessionStore — session persisted carrying service_session_id="svc-thread-1" — then turn 3 on this branch continuing the same chain): the stale id is dropped on load (conversation_id=None forwarded), the model input for turn 3 contains each message exactly once, the turn completes, and the session is saved clean. The wedge from my earlier comment is gone.

Your regression test's simulated pre-fix state matches what the real chain produces, for what it's worth — the mutation you describe is exactly the session shape my released-version run persists.

Bench script below as requested — it's the upgrade-sim variant of the repro already in #7955 (--fresh --turns 1,2 under the released package, then --continue --turns 3 under this branch, sharing a FileSessionStore and a JSON-persisted fake service thread). No Azure needed.

On the measurement: I'll re-run the store: False timing on my live deployment (Foundry project endpoint, streaming, released stack — the delay is a property of the endpoint, not of this branch) and report numbers here, so the (a)/(b) call can be made on data.

repro_pr_check.py — upgrade-sim bench

"""Verification bench for PR #7957 against issue #7955.

Like repro.py, but:
- the fake service HONOURS ``store``: when store is False it keeps nothing and
  returns no conversation_id (matching a well-behaved Responses service);
- sessions use FileSessionStore and the fake service persists its threads to
  disk, so a conversation can be continued by a later process (upgrade sim);
- ``--turns 1,2`` / ``--turns 3`` with ``--continue`` split the chain.

Usage:
  python repro_pr_check.py --fresh --turns 1,2,3     # full chain, fresh state
  python repro_pr_check.py --fresh --turns 1,2       # first two turns only
  python repro_pr_check.py --continue --turns 3      # continue prior chain
"""

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

HERE = os.path.dirname(os.path.abspath(__file__))
WORK = os.path.join(HERE, "prcheck_state")
LOG = os.path.join(WORK, "model_calls.jsonl")
THREADS = os.path.join(WORK, "service_threads.json")
CHAIN = os.path.join(WORK, "chain.json")
SESSIONS = os.path.join(WORK, "sessions")
PORT = int(os.environ.get("REPRO_PORT", "8288"))

TURN_TEXT = {
    1: "Turn one: my favourite colour is teal.",
    2: "Turn two: I live in Melbourne.",
    3: "Turn three: what do you know about me?",
}


def serve() -> None:
    from agent_framework import (
        Agent,
        BaseChatClient,
        ChatResponse,
        ChatResponseUpdate,
        FileSessionStore,
        FunctionInvocationLayer,
        ResponseStream,
    )
    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

    def load_threads():
        try:
            return json.load(open(THREADS))
        except Exception:
            return {}

    class HonestServiceClient(FunctionInvocationLayer, BaseChatClient):
        """Stores server-side by default, but honours store=False."""

        STORES_BY_DEFAULT = True
        _calls = 0

        def _inner_get_response(self, *, messages, stream, options, **kwargs):
            HonestServiceClient._calls += 1
            n = HonestServiceClient._calls
            store = options.get("store")
            conv = options.get("conversation_id")
            threads = load_threads()
            stored = list(threads.get(conv, [])) if conv else []
            request = describe(messages)
            log("model_call", {
                "call": n, "store_option": store, "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 = None
            if store is not False:  # honour store=False: keep nothing, no thread id
                tid = conv or "svc-thread-1"
                threads[tid] = stored + request + [
                    {"role": "assistant", "text": f"ack-{n}" if answered else "(tool call)"}
                ]
                json.dump(threads, open(THREADS, "w"))
            elif conv:
                # store=False but a conversation id was still passed: the service
                # would still evaluate against that thread (it just stores nothing).
                tid = None

            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 Provider(AgentSessionStoreProvider):
        def __init__(self):
            self._store = FileSessionStore(SESSIONS)

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

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

    agent = Agent(client=HonestServiceClient(), instructions="You are a helpful assistant.",
                  tools=[get_time])
    ResponsesHostServer(agent=agent, agent_session_store_provider=Provider()).run()


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(turns, cont) -> None:
    os.makedirs(WORK, exist_ok=True)
    if not cont:
        for p in (LOG, THREADS, CHAIN):
            if os.path.exists(p):
                os.remove(p)
        import shutil
        shutil.rmtree(SESSIONS, ignore_errors=True)
        open(LOG, "a").close()

    chain = json.load(open(CHAIN)) if os.path.exists(CHAIN) else {}
    env = dict(os.environ, PORT=str(PORT), AGENTSERVER_STATE_ROOT=os.path.join(WORK, "agentserver"))
    server = subprocess.Popen([sys.executable, os.path.abspath(__file__), "--serve"],
                              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)
        for t in turns:
            prev = chain.get(str(t - 1))
            r = post(TURN_TEXT[t], prev)
            chain[str(t)] = r["id"]
            print(f"turn{t}: status={r.get('status')}"
                  + (f"  error={json.dumps(r.get('error'))[:180]}" if r.get("status") == "failed" else ""))
        json.dump(chain, open(CHAIN, "w"))
    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']} (store={c['store_option']}, "
              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]}")

    if calls:
        last = calls[-1]["messages"]
        counts = {}
        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:
        turns = [1, 2, 3]
        for i, a in enumerate(sys.argv):
            if a == "--turns":
                turns = [int(x) for x in sys.argv[i + 1].split(",")]
        drive(turns, "--continue" in sys.argv)

[Eduard van Valkenburg (eavanvalkenburg)]

I'm going to close this PR, we are having some design discussions on how we want to approach this, and I will create the PR to address the issue once that has played out, thanks Atharva Vichare (@atty57)