fix(#3200): forward context-discipline flags into spawned agent pods

orchestrator/kubernetes_spawner.py

@@ -16,6 +16,7 @@
 import sys
 import threading
 import time
+from collections.abc import Mapping
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from pathlib import Path
@@ -205,6 +206,52 @@ def _resolve_wait_producer_allowlist(phase: str | None, role: str, repo: str | N
 )
 
 
+# Context-discipline flags (#3200) that the orchestrator forwards from its OWN
+# env into every spawned agent pod when set. These switches are read *in-pod* —
+# by the agent (``egg_agent.context_discipline`` / ``egg_agent.session``) and by
+# the wrapper's in-pod prompt composer (``event_prompt._context_discipline_enabled``)
+# — but the orchestrator deployment is the one process an operator can flip them
+# on. Nothing else wires them into the Job env (no ``envFrom``, and ``sandbox_env``
+# is built from pipeline fields), so without this forward ``kubectl set env`` on
+# the orchestrator is a no-op for agents. Forwarding them here makes that the
+# single operator knob. Default-OFF semantics are preserved: an unset/blank flag
+# is simply not forwarded, so the pod stays on the legacy path byte-for-byte. The
+# forward runs BEFORE the ``extra_env`` merge, so a per-spawn override still
+# wins. ``EGG_SESSION_RESUME`` rides along as the narrower staged-rollout knob;
+# the per-spawn substrate (``EGG_SESSION_STATE_FILE`` / ``EGG_RESEED_THRESHOLD``)
+# needs real per-pod wiring and is deliberately NOT a blind forward.
+#
+# NOTE: the #3249 emit-only measurement knob (``EGG_CONTEXT_MEASUREMENT``) is
+# deliberately NOT forwarded yet — it has no in-pod consumer (no
+# ``egg_agent.measurement`` module exists), and the flag-name auto-discovery
+# convention (``sandbox/tests/test_context_discipline_flag.py``) means the
+# eventual consumer's flag name isn't pinned. Forwarding a guessed name now would
+# be a silent no-op. Add it here once #3249's emit consumer lands under a fixed
+# name.
+_FORWARDED_DISCIPLINE_ENV_KEYS: tuple[str, ...] = (
+    "EGG_CONTEXT_DISCIPLINE",
+    "EGG_SESSION_RESUME",
+)
+
+
+def _forwarded_discipline_env(source: Mapping[str, str]) -> dict[str, str]:
+    """Return the context-discipline flags set and non-blank in *source*.
+
+    Pure: selects the :data:`_FORWARDED_DISCIPLINE_ENV_KEYS` subset of *source*
+    (typically ``os.environ``) whose value is set and non-blank. An unset or
+    blank flag is omitted, never forwarded as an empty string — so the pod's
+    default-OFF parse is identical to the flag simply being absent. Note this is
+    a non-blank filter, not a truthiness filter: a value like ``"false"`` is
+    forwarded as-is and parsed OFF in-pod.
+    """
+    out: dict[str, str] = {}
+    for key in _FORWARDED_DISCIPLINE_ENV_KEYS:
+        value = source.get(key)
+        if value and value.strip():
+            out[key] = value
+    return out
+
+
 # --- #3064 slice-2: orchestrator-owned one-shot event spawns ------------
 #
 # When the orchestrator (not the in-pod wait-loop) owns the BRC event loop
@@ -1747,6 +1794,13 @@ def spawn_agent_job(
             if wait_allowlist:
                 environment["EGG_WAIT_PRODUCER_ALLOWLIST"] = wait_allowlist
 
+            # Forward operator-set context-discipline flags (#3200) from
+            # the orchestrator's own env into the pod. Read in-pod but flippable
+            # only on the orchestrator deployment, so this is the single
+            # operator knob (kubectl set env). Placed before the extra_env merge
+            # so a per-spawn override still wins. See _forwarded_discipline_env.
+            environment.update(_forwarded_discipline_env(os.environ))
+
             # Caller's extra_env overrides defaults, except protected keys
             if extra_env:
                 for key, value in extra_env.items():

orchestrator/tests/test_kubernetes_spawner.py

@@ -927,6 +927,97 @@ def test_spawn_custom_image(self, spawner, mock_k8s_client):
         assert call_kwargs["image"] == "custom-image:v2"
 
 
+# ---------------------------------------------------------------------------
+# TestContextDisciplineFlagForward (#3200)
+# ---------------------------------------------------------------------------
+
+
+class TestForwardedDisciplineEnvHelper:
+    """The pure ``_forwarded_discipline_env`` selector."""
+
+    def test_selects_only_set_non_blank_flags(self):
+        from kubernetes_spawner import _forwarded_discipline_env
+
+        # Non-blank filter, not truthiness: a "false" value is forwarded as-is
+        # (and parsed OFF in-pod), while unrelated keys are dropped.
+        source = {
+            "EGG_CONTEXT_DISCIPLINE": "true",
+            "EGG_SESSION_RESUME": "false",
+            "UNRELATED": "x",
+        }
+        assert _forwarded_discipline_env(source) == {
+            "EGG_CONTEXT_DISCIPLINE": "true",
+            "EGG_SESSION_RESUME": "false",
+        }
+
+    def test_omits_unset_and_blank(self):
+        from kubernetes_spawner import _forwarded_discipline_env
+
+        # Blank / whitespace-only never forwards as an empty string — the pod's
+        # default-OFF parse must be identical to the flag being absent.
+        assert _forwarded_discipline_env({"EGG_CONTEXT_DISCIPLINE": "  "}) == {}
+        assert _forwarded_discipline_env({}) == {}
+
+    def test_covers_every_declared_key(self):
+        from kubernetes_spawner import (
+            _FORWARDED_DISCIPLINE_ENV_KEYS,
+            _forwarded_discipline_env,
+        )
+
+        source = dict.fromkeys(_FORWARDED_DISCIPLINE_ENV_KEYS, "on")
+        assert set(_forwarded_discipline_env(source)) == set(_FORWARDED_DISCIPLINE_ENV_KEYS)
+
+
+class TestContextDisciplineFlagForward:
+    """spawn_agent_job forwards the orchestrator's context-discipline flags."""
+
+    def test_flags_forwarded_into_pod_env_when_set(
+        self, spawner, mock_k8s_client, mock_gateway, monkeypatch
+    ):
+        monkeypatch.setenv("EGG_CONTEXT_DISCIPLINE", "true")
+        monkeypatch.setenv("EGG_SESSION_RESUME", "1")
+        result = spawner.spawn_agent_job(
+            pipeline_id="pipe-cd",
+            agent_role=AgentRole.CODER,
+            phase="implement",
+            branch="egg/issue-cd",
+            repos=["owner/repo"],
+        )
+        env = result.environment
+        assert env["EGG_CONTEXT_DISCIPLINE"] == "true"
+        assert env["EGG_SESSION_RESUME"] == "1"
+
+    def test_flags_absent_when_unset(self, spawner, mock_k8s_client, mock_gateway, monkeypatch):
+        monkeypatch.delenv("EGG_CONTEXT_DISCIPLINE", raising=False)
+        monkeypatch.delenv("EGG_SESSION_RESUME", raising=False)
+        result = spawner.spawn_agent_job(
+            pipeline_id="pipe-cd2",
+            agent_role=AgentRole.CODER,
+            phase="implement",
+            branch="egg/issue-cd2",
+            repos=["owner/repo"],
+        )
+        env = result.environment
+        assert "EGG_CONTEXT_DISCIPLINE" not in env
+        assert "EGG_SESSION_RESUME" not in env
+
+    def test_extra_env_overrides_forwarded_flag(
+        self, spawner, mock_k8s_client, mock_gateway, monkeypatch
+    ):
+        # A per-spawn extra_env value wins over the orchestrator-global forward
+        # (the forward runs before the extra_env merge).
+        monkeypatch.setenv("EGG_CONTEXT_DISCIPLINE", "true")
+        result = spawner.spawn_agent_job(
+            pipeline_id="pipe-cd3",
+            agent_role=AgentRole.CODER,
+            phase="implement",
+            branch="egg/issue-cd3",
+            repos=["owner/repo"],
+            extra_env={"EGG_CONTEXT_DISCIPLINE": "false"},
+        )
+        assert result.environment["EGG_CONTEXT_DISCIPLINE"] == "false"
+
+
 # ---------------------------------------------------------------------------
 # TestSpawnRetry (#1839)
 # ---------------------------------------------------------------------------