feat(#3279): compute + inject EGG_RESEED_THRESHOLD per event spawn

orchestrator/concurrent_executor.py

@@ -37,6 +37,7 @@ def get_logger(name: str, **kwargs) -> logging.Logger:  # type: ignore[misc]
     UPSTREAM_ANTHROPIC,
     AgentModelDecision,
     classify_model,
+    reseed_threshold,
     resolve_agent_model,
 )
 from consensus_wrapper import build_consensus_wrapped_command
@@ -153,7 +154,15 @@ def spawn_event(
         # ``spawn_event_job`` via ``event_action`` below), not on an
         # ownership flag — the EGG_EVENT_LOOP_OWNER env was retired in #3164.
         env = self._ex.get_agent_env(agent_role)
-        command, upstream, upstream_model = self._ex._build_event_spawn_params(agent_role)
+        command, upstream, upstream_model, threshold = self._ex._build_event_spawn_params(
+            agent_role
+        )
+        # Export the per-model reseed threshold so the in-pod resume-vs-reseed
+        # gate (#3200 slice-8) and the #3249 measurement resolve a real-window
+        # boundary instead of None (the gate's ``no_threshold`` safe-reseed
+        # branch). Inert unless a discipline/resume/measurement flag is on in the
+        # pod — the only consumers read it (#3279); a plain default pod ignores it.
+        env["EGG_RESEED_THRESHOLD"] = str(threshold)
         return self._ex.spawn_fn(
             role=agent_role,
             branch=branch,
@@ -562,15 +571,31 @@ def stop_event_loop(self) -> None:
 
     def _build_event_spawn_params(
         self, role: AgentRole
-    ) -> tuple[list[str], str | None, str | None]:
-        """Return ``(command, upstream, upstream_model)`` for a role's event pod.
+    ) -> tuple[list[str], str | None, str | None, int]:
+        """Return ``(command, upstream, upstream_model, reseed_threshold)`` for a role's event pod.
 
         The event-pump template composes its own per-event prompt at runtime
         (``invoke_agent_for_event``), so the initial prompt is irrelevant —
         only model + effort matter, resolved identically to ``_spawn_agent``.
         ``upstream``/``upstream_model`` are returned only when they differ
         from the default Anthropic decision (mirroring ``_spawn_agent``'s
         conditional forwarding) so the default-Claude wire shape is unchanged.
+
+        ``reseed_threshold`` is the per-model token-occupancy boundary the
+        in-pod resume-vs-reseed gate (#3200 slice-8) compares against. The
+        orchestrator computes it here — it has the model decision, and the
+        agent pod can't (``orchestrator`` is off the pod's ``PYTHONPATH``, so
+        ``egg_agent.reseed.resolve_reseed_threshold`` resolves ``None`` without
+        the ``EGG_RESEED_THRESHOLD`` override, taking the gate's ``no_threshold``
+        safe-reseed branch every event). It is resolved against ``claude_code_alias``
+        — the same string passed to ``--model``. Both in-pod consumers (the reseed
+        gate, ``reseed.py:126-134``, and the #3249 measurement,
+        ``measurement.py:252-262``) read this ``EGG_RESEED_THRESHOLD`` override
+        first, so once it is always injected on the event path the injected
+        threshold and the emitted measurement agree because they read the *same env
+        var* — not because each independently re-resolves ``args.model``. And sub-1M
+        LiteLLM models (whose bare alias carries their real-backend identity) resolve
+        against their real window, not the ``[1m]``-implied 1M (#3279).
         """
         decision = self._resolve_model_decision(role)
         command = build_consensus_wrapped_command(
@@ -581,7 +606,8 @@ def _build_event_spawn_params(
         if decision.upstream != UPSTREAM_ANTHROPIC or decision.upstream_model is not None:
             upstream = decision.upstream
             upstream_model = decision.upstream_model
-        return command, upstream, upstream_model
+        threshold = reseed_threshold(decision.claude_code_alias)
+        return command, upstream, upstream_model, threshold
 
     def _orchestrator_side_confirm(self, tracker: Any, role: str) -> None:
         """Record a ``confirm``/``complete`` orchestrator-side — no pod (#3064).

orchestrator/tests/test_concurrent_executor.py

@@ -1608,3 +1608,83 @@ def test_default_recorded_on_execution(self):
         execution = executor._spawn_agent(AgentRole.CODER)
 
         assert execution.resolved_model == "opus"
+
+
+class TestEventSpawnReseedThreshold:
+    """#3279: the orchestrator computes the per-model reseed threshold at spawn
+    and exports it to the event pod as ``EGG_RESEED_THRESHOLD`` so the in-pod
+    resume-vs-reseed gate (#3200 slice-8) and the #3249 measurement resolve a
+    real-window boundary instead of ``None`` (the gate's ``no_threshold``
+    safe-reseed branch — ``orchestrator`` is off the pod's ``PYTHONPATH``).
+    """
+
+    def _event_spawner(self, pipeline, mock_spawn):
+        from concurrent_executor import ConcurrentPhaseExecutor, _ExecutorEventSpawner
+        from egg_orchestrator.types import AgentRole
+
+        executor = ConcurrentPhaseExecutor(pipeline, spawn_fn=mock_spawn)
+        return _ExecutorEventSpawner(
+            executor=executor,
+            roles=[AgentRole.CODER, AgentRole.REVIEWER_CODE],
+            slice_id=None,
+        )
+
+    def test_build_event_spawn_params_returns_default_threshold(self):
+        """Default (opus) → ``reseed_threshold`` is the 400k floor
+        (``min(400k, 0.80 * 1M)``), returned as the 4th tuple element.
+        """
+        from concurrent_executor import ConcurrentPhaseExecutor
+        from egg_orchestrator.types import AgentRole
+
+        pipeline = _make_pipeline()
+        executor = ConcurrentPhaseExecutor(pipeline, spawn_fn=MagicMock())
+
+        _command, _upstream, _upstream_model, threshold = executor._build_event_spawn_params(
+            AgentRole.CODER
+        )
+
+        assert threshold == 400_000
+
+    def test_event_spawn_injects_default_threshold_env(self):
+        """``spawn_event`` puts ``EGG_RESEED_THRESHOLD`` in the spawn_fn's
+        ``extra_env`` for a default-config (opus) pipeline.
+        """
+        pipeline = _make_pipeline()
+        mock_spawn = MagicMock(return_value=_kubernetes_spawn_result())
+        spawner = self._event_spawner(pipeline, mock_spawn)
+
+        spawner.spawn_event(role="coder", action="propose", dedupe_key="k1")
+
+        env = mock_spawn.call_args.kwargs["extra_env"]
+        assert env["EGG_RESEED_THRESHOLD"] == "400000"
+
+    def test_event_spawn_threshold_uses_real_window_for_sub_1m_model(self):
+        """A sub-1M LiteLLM model resolves against its REAL backend window, not
+        the ``[1m]``-implied 1M: ``kimi-k2.7-code`` → ``int(0.80 * 262_144)``.
+        Guards the mis-trigger bug #3200 task-2-1 calls out.
+        """
+        pipeline = _make_pipeline()
+        pipeline.config.agent_models = {"coder": "kimi-k2.7-code"}
+        mock_spawn = MagicMock(return_value=_kubernetes_spawn_result())
+        spawner = self._event_spawner(pipeline, mock_spawn)
+
+        spawner.spawn_event(role="coder", action="propose", dedupe_key="k1")
+
+        env = mock_spawn.call_args.kwargs["extra_env"]
+        assert env["EGG_RESEED_THRESHOLD"] == str(int(0.80 * 262_144))  # 209_715, NOT 400_000
+
+    def test_event_spawn_threshold_conservative_for_unregistered_litellm_model(self):
+        """An unregistered LiteLLM model (not in ``_SUB_1M_CONTEXT_MODELS``, not a
+        Claude alias) resolves against the conservative 200K window, so the
+        threshold is ``int(0.80 * 200_000)`` = ``160_000`` — the branch an operator
+        whose model isn't in the registry silently lands on.
+        """
+        pipeline = _make_pipeline()
+        pipeline.config.agent_models = {"coder": "qwen3-coder-30b"}
+        mock_spawn = MagicMock(return_value=_kubernetes_spawn_result())
+        spawner = self._event_spawner(pipeline, mock_spawn)
+
+        spawner.spawn_event(role="coder", action="propose", dedupe_key="k1")
+
+        env = mock_spawn.call_args.kwargs["extra_env"]
+        assert env["EGG_RESEED_THRESHOLD"] == "160000"