BUG: Reloading changed AGENTS.md can overflow the context window and lock an existing session

#!/usr/bin/env python3
"""Cold-resume AGENTS.md duplication probe. Linux + bubblewrap, no real API.

python3 reproduce.py [--codex /absolute/path/to/codex] [--freeze] [--keep]
--freeze checks the proposed agents_md_reload_policy="freeze" configuration.
An older CLI may silently ignore that key; assertions deliberately expose this.
"""

import argparse
import http.server
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
import threading

MARKER = "UNCHANGED_AGENTS_PAYLOAD_7c309f"
OLD = "2026-01-01T00:00:00Z"
NEW = "2026-01-01T00:00:01Z"
REPLY = "SYNTHETIC_PREVIOUS_REPLY_a8f194"


def arguments():
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument(
        "--codex", default="codex", help="Executable on PATH or absolute path"
    )
    p.add_argument(
        "--freeze", action="store_true", help="Assert proposed freeze behavior"
    )
    p.add_argument(
        "--keep", action="store_true", help="Keep synthetic requests and CLI logs"
    )
    p.add_argument("--inside", action="store_true", help=argparse.SUPPRESS)
    return p.parse_args()


def sandbox(args):
    bwrap = shutil.which("bwrap")
    executable = shutil.which(args.codex)
    if not bwrap or not executable:
        raise RuntimeError(
            "Requires Linux bubblewrap and a Codex executable; refusing unisolated execution"
        )
    executable = str(Path(executable).resolve())
    scratch = Path(tempfile.mkdtemp(prefix="codex-agents-repro-"))
    try:
        cmd = [
            bwrap,
            "--unshare-all",
            "--die-with-parent",
            "--new-session",
            "--clearenv",
            "--ro-bind",
            "/usr",
            "/usr",
        ]
        for name in ("/bin", "/lib", "/lib64"):
            path = Path(name)
            if path.is_symlink():
                cmd += ["--symlink", os.readlink(path), name]
            elif path.exists():
                cmd += ["--ro-bind", name, name]
        cmd += [
            "--proc",
            "/proc",
            "--dev",
            "/dev",
            "--tmpfs",
            "/tmp",
            "--tmpfs",
            "/run",
            "--bind",
            str(scratch),
            "/work",
            "--ro-bind",
            executable,
            "/opt/codex",
            "--ro-bind",
            str(Path(__file__).resolve()),
            "/app/reproduce.py",
            "--setenv",
            "PATH",
            "/usr/bin:/bin:/opt",
            "--setenv",
            "HOME",
            "/work/home",
            "--setenv",
            "CODEX_HOME",
            "/work/codex-home",
            "--setenv",
            "LANG",
            "C.UTF-8",
            "--setenv",
            "PYTHONDONTWRITEBYTECODE",
            "1",
            "--chdir",
            "/work",
            "--",
            "/usr/bin/python3",
            "/app/reproduce.py",
            "--inside",
            "--codex",
            "/opt/codex",
        ]
        if args.freeze:
            cmd.append("--freeze")
        # Both the local HTTP fixture and Codex run inside the SAME isolated netns.
        # Real HOME/config/credentials are not mounted; only scratch is writable.
        return subprocess.run(cmd, timeout=180).returncode
    finally:
        if args.keep:
            print("Synthetic evidence:", scratch, flush=True)
        else:
            shutil.rmtree(scratch)


def probe(args):
    root = Path("/work")
    home = root / "home"
    codex_home = root / "codex-home"
    work = root / "project"
    for directory in (home, codex_home, work):
        directory.mkdir()
    requests = []

    class Handler(http.server.BaseHTTPRequestHandler):
        def log_message(self, *unused):
            pass

        def do_POST(self):
            raw = self.rfile.read(int(self.headers["Content-Length"]))
            if self.headers.get("Content-Encoding"):
                self.send_error(400, "Compression disabled by this reproducer")
                return
            if self.path != "/v1/responses":
                self.send_error(404, "Only the fake Responses endpoint exists")
                return
            request = json.loads(raw)
            requests.append((request, len(raw)))
            (root / f"request-{len(requests)}.json").write_bytes(raw)
            msg = {
                "type": "message",
                "id": "msg_fixture",
                "status": "completed",
                "role": "assistant",
                "content": [{"type": "output_text", "text": REPLY, "annotations": []}],
            }
            events = [
                {
                    "type": "response.created",
                    "response": {
                        "id": "resp_fixture",
                        "object": "response",
                        "status": "in_progress",
                        "output": [],
                    },
                },
                {"type": "response.output_item.added", "output_index": 0, "item": msg},
                {
                    "type": "response.output_text.delta",
                    "item_id": "msg_fixture",
                    "output_index": 0,
                    "content_index": 0,
                    "delta": REPLY,
                },
                {"type": "response.output_item.done", "output_index": 0, "item": msg},
                {
                    "type": "response.completed",
                    "response": {
                        "id": "resp_fixture",
                        "object": "response",
                        "status": "completed",
                        "output": [msg],
                        "usage": {
                            "input_tokens": 100,
                            "output_tokens": 2,
                            "total_tokens": 102,
                        },
                    },
                },
            ]
            payload = "".join(
                "event: " + e["type"] + "\ndata: " + json.dumps(e) + "\n\n"
                for e in events
            ).encode()
            self.send_response(200)
            self.send_header("Content-Type", "text/event-stream")
            self.send_header("Content-Length", str(len(payload)))
            self.end_headers()
            self.wfile.write(payload)

    server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    agents = "# Synthetic context\nGenerated: " + OLD + "\n" + MARKER + "\n"
    agents += (
        "\n".join(f"Fixture fact {i}: ordinary test content." for i in range(150))
        + "\n"
    )
    path = work / "AGENTS.md"
    path.write_text(agents)
    command = [
        args.codex,
        "exec",
        "--ignore-user-config",
        "--ignore-rules",
        "--skip-git-repo-check",
        "--json",
        "-m",
        "gpt-5.6-sol",
    ]
    config = {
        "model_provider": "mock",
        "model_providers.mock.name": "Local fixture",
        "model_providers.mock.base_url": f"http://127.0.0.1:{server.server_port}/v1",
        "model_providers.mock.wire_api": "responses",
        "model_providers.mock.requires_openai_auth": False,
        "features.enable_request_compression": False,
        "skills.include_instructions": False,
        "features.shell_snapshot": False,
        "analytics.enabled": False,
        "feedback.enabled": False,
        "otel.exporter": "none",
        "otel.trace_exporter": "none",
    }
    if args.freeze:
        config["agents_md_reload_policy"] = "freeze"
    for key, value in config.items():
        command += ["-c", key + "=" + json.dumps(value)]

    def run(label, extra):
        before = len(requests)
        result = subprocess.run(
            command + extra, cwd=work, capture_output=True, text=True, timeout=45
        )
        (root / (label + ".stdout")).write_text(result.stdout)
        (root / (label + ".stderr")).write_text(result.stderr)
        if result.returncode:
            raise RuntimeError(
                f"{label}: CLI exited {result.returncode}: {result.stderr[-2000:]}"
            )
        assert len(requests) == before + 1, (
            f"{label}: expected exactly one fake API request"
        )
        request, byte_count = requests[-1]
        text = json.dumps(request)
        counts = {
            "payload_copies": text.count(MARKER),
            "old_timestamp": text.count(OLD),
            "new_timestamp": text.count(NEW),
            "prior_reply": text.count(REPLY),
        }
        row = {"turn": label, **counts, "request_bytes": byte_count}
        print(json.dumps(row), flush=True)
        return result, row

    try:
        print(
            subprocess.check_output([args.codex, "--version"], text=True).strip(),
            flush=True,
        )
        print(
            "Isolated network namespace; synthetic replies only; no real credentials.",
            flush=True,
        )
        first, a = run("initial", ["Reply briefly."])
        events = [
            json.loads(line)
            for line in first.stdout.splitlines()
            if line.startswith("{")
        ]
        sid = next(e["thread_id"] for e in events if e.get("type") == "thread.started")
        _, b = run("unchanged_resume", ["resume", sid, "Reply briefly."])
        edited = agents.replace(OLD, NEW)
        assert len(edited.encode()) == len(agents.encode())
        assert sum(x != y for x, y in zip(edited, agents)) == 1
        path.write_text(edited)
        _, c = run("edited_resume", ["resume", sid, "Reply briefly."])
        summary = {
            "mode": "freeze" if args.freeze else "default",
            "agents_bytes": len(agents.encode()),
            "changed_characters": 1,
            "request_byte_deltas": [
                b["request_bytes"] - a["request_bytes"],
                c["request_bytes"] - b["request_bytes"],
            ],
            "turns": [a, b, c],
            "token_counts": "synthetic, not a measurement",
        }
        (root / "summary.json").write_text(json.dumps(summary, indent=2) + "\n")
        print(json.dumps(summary), flush=True)
        assert a["payload_copies"] == b["payload_copies"] == 1
        assert a["old_timestamp"] == b["old_timestamp"] == 1
        assert a["new_timestamp"] == b["new_timestamp"] == 0
        assert b["prior_reply"] == 1 and c["prior_reply"] == 2, (
            "Conversation history was lost"
        )
        if args.freeze:
            assert (c["payload_copies"], c["old_timestamp"], c["new_timestamp"]) == (
                1,
                1,
                0,
            ), (
                "Freeze did not preserve the original AGENTS snapshot (old CLI may ignore the option)"
            )
        else:
            assert (c["payload_copies"], c["old_timestamp"], c["new_timestamp"]) == (
                2,
                1,
                1,
            ), "Default behavior differs: duplication no longer reproduced"
        print(
            "PASS: "
            + (
                "freeze holds the boot snapshot"
                if args.freeze
                else "duplication reproduced"
            ),
            flush=True,
        )
        return 0
    finally:
        server.shutdown()
        server.server_close()


if __name__ == "__main__":
    args = arguments()
    try:
        sys.exit(probe(args) if args.inside else sandbox(args))
    except (AssertionError, RuntimeError, subprocess.TimeoutExpired) as error:
        print("FAIL:", error, file=sys.stderr)
        sys.exit(1)