fix(box): harden sandbox session isolation

src/langbot/pkg/box/service.py

@@ -172,6 +172,17 @@ class BoxService:
             .get('box-session-id-template', '{launcher_type}_{launcher_id}')
         )
         variables = dict(query.variables or {})
+        launcher_type = getattr(query, 'launcher_type', None)
+        if hasattr(launcher_type, 'value'):
+            launcher_type = launcher_type.value
+        launcher_id = getattr(query, 'launcher_id', None)
+        sender_id = getattr(query, 'sender_id', None)
+        query_id = getattr(query, 'query_id', None)
+
+        variables.setdefault('query_id', str(query_id or 'unknown'))
+        variables.setdefault('launcher_type', str(launcher_type or 'query'))
+        variables.setdefault('launcher_id', str(launcher_id or query_id or 'unknown'))
+        variables.setdefault('sender_id', str(sender_id or launcher_id or query_id or 'unknown'))
         variables.setdefault('global', 'global')
         return template.format_map(collections.defaultdict(lambda: 'unknown', variables))
 

tests/integration_tests/box/test_box_integration.py

@@ -322,7 +322,7 @@ async def test_full_service_to_remote_runtime(tmp_path):
         assert result['ok'] is True
         assert result['status'] == 'completed'
         assert 'service-path' in result['stdout']
-        assert result['session_id'] == '42'
+        assert result['session_id'] == 'query_42'
     finally:
         server_task.cancel()
         client_task.cancel()

tests/unit_tests/box/test_box_service.py

@@ -264,6 +264,38 @@ async def test_box_service_defaults_session_id_from_query():
     assert backend.start_calls == ['person_test_user']
 
 
+@pytest.mark.asyncio
+async def test_box_service_session_id_uses_query_attributes_without_variables():
+    logger = Mock()
+    backend = FakeBackend(logger)
+    runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
+    service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
+    await service.initialize()
+
+    query = pipeline_query.Query.model_construct(query_id=7, launcher_type='group', launcher_id='room-1')
+    result = await service.execute_tool({'command': 'pwd'}, query)
+
+    assert result['session_id'] == 'group_room-1'
+    assert result['ok'] is True
+    assert backend.start_calls == ['group_room-1']
+
+
+@pytest.mark.asyncio
+async def test_box_service_session_id_falls_back_to_query_id_for_synthetic_queries():
+    logger = Mock()
+    backend = FakeBackend(logger)
+    runtime = BoxRuntime(logger=logger, backends=[backend], session_ttl_sec=300)
+    service = BoxService(make_app(logger), client=_InProcessBoxRuntimeClient(logger, runtime))
+    await service.initialize()
+
+    query = pipeline_query.Query.model_construct(query_id=7)
+    result = await service.execute_tool({'command': 'pwd'}, query)
+
+    assert result['session_id'] == 'query_7'
+    assert result['ok'] is True
+    assert backend.start_calls == ['query_7']
+
+
 @pytest.mark.asyncio
 async def test_box_service_fails_closed_when_backend_unavailable():
     logger = Mock()

tests/unit_tests/provider/test_skill_tools.py

@@ -522,7 +522,7 @@ class TestNativeToolLoaderSkillPaths:
             ap.box_service = SimpleNamespace(
                 available=True,
                 default_workspace=tmpdir,
-                execute_spec_payload=AsyncMock(return_value={'ok': True}),
+                execute_tool=AsyncMock(return_value={'ok': True}),
             )
             ap.skill_mgr = SimpleNamespace(refresh_skill_from_disk=Mock())
             loader = NativeToolLoader(ap)
@@ -540,11 +540,9 @@ class TestNativeToolLoaderSkillPaths:
             )
 
             assert result == {'ok': True}
-            spec_payload = ap.box_service.execute_spec_payload.await_args.args[0]
-            assert spec_payload['cmd'] == 'python /workspace/scripts/run.py'
-            assert spec_payload['workdir'] == '/workspace'
-            assert spec_payload['host_path'] == tmpdir
-            assert spec_payload['session_id'] == 'skill-person_123-demo'
+            tool_parameters = ap.box_service.execute_tool.await_args.args[0]
+            assert tool_parameters['command'] == 'python /workspace/.skills/demo/scripts/run.py'
+            assert tool_parameters['workdir'] == '/workspace/.skills/demo'
             ap.skill_mgr.refresh_skill_from_disk.assert_called_once_with('demo')
 
     @pytest.mark.asyncio