fix(soul): resolve merge duplicates from blackbox port integration

Drop duplicate budget-nudge, ToolUseSkipped emit paths, and conflicting
tests introduced when merging feat/blackbox-agent-loop-port into main.

Author: elkaix

src/pythinker_code/prompts/__init__.py

@@ -9,12 +9,6 @@
 GOAL_SET = (Path(__file__).parent / "goal_set.md").read_text(encoding="utf-8")
 GOAL_CONTINUATION = (Path(__file__).parent / "goal_continuation.md").read_text(encoding="utf-8")
 GOAL_WRAP_UP = (Path(__file__).parent / "goal_wrap_up.md").read_text(encoding="utf-8")
-BUDGET_CONTINUATION_NUDGE = (
-    "You have used ~{pct:.0f}% of this session's configured spend ceiling "
-    "(${spent:.2f} of ${ceiling:.2f}). Pause new work and finish the current request. "
-    "If you need more, raise `loop_control.max_session_cost_usd` in config."
-)
-
 BUDGET_CONTINUATION_NUDGE = (
     "The session has crossed {ratio:.0%} of its configured spend ceiling "
     "(estimated ${spent:.2f} of ${ceiling:.2f}). Continue only if the remaining work is "

src/pythinker_code/soul/pythinkersoul.py

@@ -309,7 +309,7 @@ def _budget_nudge_message(*, session_cost_usd: float, ceiling: float, ratio: flo
         spent=session_cost_usd,
         ceiling=ceiling,
     )
-    return Message(role="user", content=[system_reminder(text[:450])])
+    return Message(role="user", content=[system_reminder(text[:500])])
 
 
 def _user_message_with_hook_context(

tests/core/test_config.py

@@ -61,7 +61,6 @@ def test_default_config_dump():
                 "prune_protect_last": 20,
                 "prune_min_chars": 2000,
                 "prune_tool_result_max_chars": 0,
-                "budget_nudge_ratio": 0.75,
             },
             "background": {
                 "max_running_tasks": 4,

tests/core/test_pythinkersoul_turn_balance.py

@@ -358,36 +358,3 @@ async def fake_checkpoint() -> None:
     await soul.turn(Message(role="user", content=[TextPart(text="hello")]))
     assert len(soul.context.history) == before + 1
     assert not any("spend ceiling" in message.extract_text(" ") for message in soul.context.history)
-
-
-@pytest.mark.asyncio
-async def test_token_budget_nudge_appends_user_message_when_over_threshold(
-    runtime: Runtime,
-    tmp_path: Path,
-    monkeypatch: pytest.MonkeyPatch,
-) -> None:
-    runtime.config.loop_control.max_session_cost_usd = 1.0
-    runtime.config.goal.auto_continue = False
-    soul = _make_soul(runtime, tmp_path)
-    soul._session_cost_usd = 2.0
-
-    async def fake_agent_loop() -> TurnOutcome:
-        return TurnOutcome(
-            stop_reason="no_tool_calls",
-            final_message=Message(role="assistant", content=[TextPart(text="done")]),
-            step_count=1,
-        )
-
-    async def fake_checkpoint() -> None:
-        return None
-
-    monkeypatch.setattr(soul, "_agent_loop", fake_agent_loop)
-    monkeypatch.setattr(soul, "_checkpoint", fake_checkpoint)
-
-    await soul.turn(Message(role="user", content=[TextPart(text="hello")]))
-    reminder_messages = [
-        message
-        for message in soul.context.history
-        if message.role == "user" and "spend ceiling" in message.extract_text(" ")
-    ]
-    assert len(reminder_messages) == 1

tests/core/test_toolset.py

@@ -751,15 +751,6 @@ def test_tool_defers_execution_started_reads_flag_only() -> None:
 # --- ToolUseSkipped wire event (opt-in per tool) ---
 
 
-class _RecordingWire:
-    def __init__(self, captured: list[object]) -> None:
-        self.soul_side = self
-        self._captured = captured
-
-    def send(self, msg: object) -> None:
-        self._captured.append(msg)
-
-
 class DummyToolEmitsSkipped(DummyToolA):
     emits_tool_use_skipped: ClassVar[bool] = True
 
@@ -808,39 +799,3 @@ async def call(self, arguments: object) -> ToolReturnValue:
         ("exit", "Shell"),
     ]
     assert not any(isinstance(e, type) and e.__name__ == "ToolUseSkipped" for e in captured)
-
-
-async def test_cross_step_dedup_emits_tool_use_skipped_when_opted_in(tmp_path: Path) -> None:
-    """Cross-step duplicate calls emit ToolUseSkipped only when the tool opts in."""
-    from unittest.mock import patch
-
-    from pythinker_code.hooks.engine import HookEngine
-    from pythinker_code.wire.types import ToolUseSkipped
-
-    ts = PythinkerToolset()
-    ts.add(DummyToolEmitsSkipped())
-    ts._hook_engine = HookEngine([], cwd=str(tmp_path))
-    captured: list[object] = []
-    args = '{"a":1}'
-
-    with patch("pythinker_code.soul.get_wire_or_none", return_value=_RecordingWire(captured)):
-        ts.begin_step([])
-        task1 = ts.handle(
-            ToolCall(id="tc1", function=ToolCall.FunctionBody(name="ToolA", arguments=args))
-        )
-        assert isinstance(task1, asyncio.Task)
-        await task1
-        prev = ts.end_step()
-        ts.begin_step(prev)
-        task2 = ts.handle(
-            ToolCall(id="tc2", function=ToolCall.FunctionBody(name="ToolA", arguments=args))
-        )
-        assert isinstance(task2, asyncio.Task)
-        await task2
-        ts.end_step()
-
-    skipped = [e for e in captured if isinstance(e, ToolUseSkipped)]
-    assert len(skipped) == 1
-    assert skipped[0].tool_call_id == "tc2"
-    assert skipped[0].tool_name == "ToolA"
-    assert skipped[0].reason == "dedup"

tests/tools/test_agent_tool.py

@@ -2723,28 +2723,3 @@ async def __call__(self, params):
     assert result.output.count("Shared cache key may collide.") >= 1
     assert "reviewer-0, reviewer-1" in result.output
     assert "batch_blockers:" not in result.output
-
-
-class _RecordingWire:
-    def __init__(self, captured: list[object]) -> None:
-        self.soul_side = self
-        self._captured = captured
-
-    def send(self, msg: object) -> None:
-        self._captured.append(msg)
-
-
-async def test_unknown_subagent_type_emits_fallback_wire_event(runtime: Runtime) -> None:
-    from unittest.mock import patch
-
-    from pythinker_code.tools.agent import AgentTool, Params
-
-    captured: list[object] = []
-    tool = AgentTool(runtime)
-    with (
-        tool_call_context("Agent"),
-        patch("pythinker_code.soul.get_wire_or_none", return_value=_RecordingWire(captured)),
-    ):
-        result = await tool(Params(description="x", prompt="y", subagent_type="does-not-exist"))
-    assert result.is_error
-    assert any(type(e).__name__ == "SubagentToolFallback" for e in captured)