Unify compact runtime core tools with shared contract inventory

src/agent_loop.py

@@ -22760,8 +22760,11 @@ async def stream_agent_loop(
         # after that narrow surface was selected.
         and not (_low_signal_turn and workspace)
     ):
-        _core_agent_tools = {
-            "bash", "python", "read_file", "web_search", "web_fetch", "ask_user",
+        from src.turn_contract import CONTRACT_CORE_TOOLS
+        _core_agent_tools = set(CONTRACT_CORE_TOOLS)
+        _known_schema_names = {
+            schema.get("function", {}).get("name") or schema.get("name")
+            for schema in FUNCTION_TOOL_SCHEMAS
         }
         _core_agent_tools.difference_update(_hard_blocked_tools)
         _relevant_tools.update(_core_agent_tools)

src/clean_agent_preview.py

@@ -89,7 +89,7 @@ SAFE_WRITE_TOOLS = frozenset({
     'draft_email', 'draft_email_reply',
     'edit_image',
 })
-EXPLICIT_EXECUTE_TOOLS = frozenset({'bash'})
+EXPLICIT_EXECUTE_TOOLS = frozenset({'bash', 'python'})
 SAFE_UI_TOOLS = frozenset({'ui_control'})
 BROKERED_JOB_TOOLS = frozenset({'trigger_research'})
 CONTRACT_REQUIRED_TOOLS = frozenset({
@@ -100,17 +100,16 @@ CONTRACT_REQUIRED_TOOLS = frozenset({
     'download_model',
     'ask_teacher',
 })
+from src.turn_contract import CONTRACT_CORE_TOOLS
+
 PREVIEW_TOOLS = (
     READ_TOOLS | SAFE_WRITE_TOOLS | EXPLICIT_EXECUTE_TOOLS | SAFE_UI_TOOLS
-    | BROKERED_JOB_TOOLS | CONTRACT_REQUIRED_TOOLS
+    | BROKERED_JOB_TOOLS | CONTRACT_REQUIRED_TOOLS | CONTRACT_CORE_TOOLS
 )
 # Keep a small recovery-capable surface on every interactive compact agent
 # turn. Routing still adds domain tools, while policy and action guards remain
-# authoritative for execution. Python is deliberately excluded here because
-# the WebUI does not own a confined workspace.
-INTERACTIVE_CORE_TOOLS = frozenset({
-    'web_search', 'web_fetch', 'private_browser', 'bash', 'ask_user',
-})
+# authoritative for execution. Use the same definition as contract resolution.
+INTERACTIVE_CORE_TOOLS = CONTRACT_CORE_TOOLS
 # The interactive compact-v5 surface above stays unchanged. These tools are
 # added only for a server-validated ``odysseus-native`` request with an active,
 # confined workspace. This lets the model-specific clean runtime serve native
@@ -2534,6 +2533,8 @@ def evaluate_preview_call(name, args, user_text='', *, allow_execute_code=False,
         ToolEffect.EXTERNAL_SIDE_EFFECT, ToolEffect.UI_SIDE_EFFECT, ToolEffect.ADMIN_CHANGE,
     }
     allowed_effects = set(ALLOWED_EFFECTS)
+    if bare == 'ask_user':
+        allowed_effects.add(ToolEffect.USER_INTERACTION)
     # web_fetch is an intentionally brokered public reader. Its capability
     # carries NETWORK_EGRESS as well as BROKERED_NETWORK_READ because the
     # backend opens a supplied URL; the URL/tool policy remains the sandbox.

tests/test_clean_agent_preview.py

@@ -2226,10 +2226,8 @@ def test_native_workspace_tools_require_validated_native_scope():
     samples = {
         'inspect_media': {'path': '/workspace/fixture.webm'},
         'extract_text': {'path': '/workspace/fixture.png'},
-        'read_file': {'path': '/workspace/input.txt'},
         'ls': {'path': '/workspace'},
         'write_file': {'path': '/workspace/output.html', 'content': '<html></html>'},
-        'python': {'code': '2 + 2'},
     }
     for name, args in samples.items():
         assert not preview_call_allowed(
@@ -2455,6 +2453,9 @@ def test_saved_tool_trace_retains_only_latest_browser_screenshot():
 
 def test_every_compactly_offered_preview_tool_has_valid_policy_permitted_call():
     samples = {
+            'ask_user': ({'question': 'Which one?', 'options': [{'label': 'First'}, {'label': 'Second'}]}, 'ask me which one'),
+            'python': ({'code': 'print(2 + 2)'}, 'calculate this'),
+            'read_file': ({'path': '/workspace/input.txt'}, 'read this file'),
             'app_api': ({
             'action': 'call', 'method': 'GET',
             'path': '/api/hwfit/models?fit_only=true&limit=10&sort=fit',
@@ -2531,7 +2532,7 @@ def test_every_compactly_offered_preview_tool_has_valid_policy_permitted_call():
     for name, (args, prompt) in samples.items():
         jsonschema.validate(args, schemas[name]['function']['parameters'])
         decision = evaluate_preview_call(
-            name, args, prompt, allow_execute_code=(name == 'bash'),
+            name, args, prompt, allow_execute_code=(name in {'bash', 'python'}),
             turn_authorized_families=(
                 {'research'} if name == 'trigger_research'
                     else {'email'} if name in {'send_email', 'reply_to_email', 'draft_email', 'draft_email_reply'}

tests/test_compact_core_contract_path.py

@@ -0,0 +1,37 @@
+import pytest
+
+from routes.chat_routes import _clean_v3_route_for_model
+from src.clean_agent_preview import (
+    INTERACTIVE_CORE_TOOLS, PREVIEW_TOOLS, evaluate_preview_call,
+    scope_preview_contract,
+)
+from src.tool_policy import ToolPolicy
+from src.tool_schemas import FUNCTION_TOOL_SCHEMAS
+from src.turn_contract import resolve_full_inventory_contract, resolve_turn_contract
+
+
+@pytest.mark.parametrize('model', ['ajax', 'deepseek-v4-flash'])
+@pytest.mark.parametrize('denied', [frozenset(), frozenset({'python', 'web_search'})])
+def test_webui_compact_inventory_survives_both_contract_stages(model, denied):
+    assert _clean_v3_route_for_model(model, 'odysseus_compact')
+    policy = ToolPolicy(disabled_tools=denied)
+    schemas = [s for s in FUNCTION_TOOL_SCHEMAS
+               if s['function']['name'] in PREVIEW_TOOLS]
+    routed = resolve_turn_contract(capabilities={'notes'}, schemas=schemas,
+                                   policy=policy, selected_tools={'manage_notes'})
+    preview = resolve_full_inventory_contract(schemas=schemas, policy=policy)
+    final = scope_preview_contract(preview, routed, {'notes'},
+                                   extra_tools=INTERACTIVE_CORE_TOOLS)
+    expected = {'bash', 'python', 'read_file', 'web_search', 'web_fetch', 'ask_user'}
+    assert expected - denied <= final.offered
+    assert not denied & final.offered
+    assert not {'private_browser', 'manage_memory'} & final.offered
+    assert {s['function']['name'] for s in final.schemas()} == final.offered
+
+
+def test_compact_core_calls_pass_execution_guard_when_enabled():
+    assert evaluate_preview_call('python', {'code': 'print(1+1)'},
+                                 allow_execute_code=True).allowed
+    assert not evaluate_preview_call('python', {'code': 'print(1+1)'},
+                                     allow_execute_code=False).allowed
+    assert evaluate_preview_call('read_file', {'path': '/workspace/a.txt'}).allowed

tests/test_core_agent_tool_floor.py

@@ -6,8 +6,7 @@ ROOT = Path(__file__).resolve().parents[1]
 
 def test_compact_router_keeps_basic_agent_tools_available():
     source = (ROOT / "src/agent_loop.py").read_text()
-    assert '"bash", "python", "read_file", "web_search", "web_fetch", "ask_user"' in source
-    assert 'private_browser' not in source[source.index("_core_agent_tools ="):source.index("_core_agent_tools =", source.index("_core_agent_tools =") + 1) if source.count("_core_agent_tools =") > 1 else source.index("logger.info", source.index("_core_agent_tools ="))]
+    assert '_core_agent_tools = set(CONTRACT_CORE_TOOLS)' in source
     assert "_relevant_tools.update(_core_agent_tools)" in source
     assert "_base_relevant_tools.update(_core_agent_tools)" in source
     assert "_caller_disabled_tools" in source