Merge pull request #2635 from langbot-app/test/core-contract-coverage-gaps

test(core): verify runtime behavior and close PostgreSQL coverage gaps

.github/workflows/run-tests.yml

@@ -84,6 +84,93 @@ jobs:
           echo "" >> $GITHUB_STEP_SUMMARY
           echo "Test Status: ${{ job.status }}" >> $GITHUB_STEP_SUMMARY
 
+  postgres-contracts:
+    name: PostgreSQL Contract Tests
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    # Only this disposable, loopback-bound CI service uses trust authentication.
+    # Admission fixtures create a separate unprivileged role with no password;
+    # the literal postgres@ URL lets that fixture switch to the runtime role.
+    env:
+      TEST_POSTGRES_URL: postgresql+asyncpg://postgres@localhost:5432/langbot_test
+      LANGBOT_ADMISSION_TEST_URL: postgresql+asyncpg://postgres@localhost:5432/langbot_admission_test
+    services:
+      postgres:
+        image: pgvector/pgvector:pg16
+        env:
+          POSTGRES_USER: postgres
+          POSTGRES_DB: langbot_test
+          POSTGRES_HOST_AUTH_METHOD: trust
+        ports:
+          - '127.0.0.1:5432:5432'
+        options: >-
+          --health-cmd="pg_isready -U postgres -d langbot_test"
+          --health-interval=5s
+          --health-timeout=5s
+          --health-retries=5
+
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v4
+
+      - name: Set up Python
+        uses: actions/setup-python@v5
+        with:
+          python-version: '3.12'
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v4
+
+      - name: Install dependencies
+        run: uv sync --dev --frozen
+
+      - name: Create isolated admission test database
+        run: >-
+          docker exec "${{ job.services.postgres.id }}"
+          psql -U postgres -d postgres -v ON_ERROR_STOP=1
+          -c 'CREATE DATABASE langbot_admission_test;'
+
+      - name: Run required PostgreSQL contracts
+        run: |
+          uv run pytest \
+            tests/integration/persistence/test_pipeline_admission_postgres.py \
+            tests/integration/persistence/test_rag_document_identity.py \
+            tests/integration/persistence/test_rag_interrupted_ingestion.py \
+            tests/integration/persistence/test_migration_branch_convergence.py \
+            tests/integration/persistence/test_runner_timestamps_postgres.py \
+            -k postgres -q --tb=short -rs --durations=10 \
+            --junitxml=postgres-contract-results.xml
+          python - <<'PY'
+          from collections import Counter
+          import xml.etree.ElementTree as ET
+
+          cases = list(ET.parse('postgres-contract-results.xml').getroot().iter('testcase'))
+          expected = {
+              'test_pipeline_admission_postgres': 27,
+              'test_rag_document_identity': 34,
+              'test_rag_interrupted_ingestion': 22,
+              'test_migration_branch_convergence': 7,
+              'test_runner_timestamps_postgres': 32,
+          }
+          counts = Counter(case.get('classname', '').rsplit('.', 1)[-1] for case in cases)
+          for module, minimum in expected.items():
+              assert counts[module] >= minimum, f'{module}: expected at least {minimum} cases, got {counts[module]}'
+          unsuccessful = [
+              case.get('name') for case in cases
+              if any(case.find(tag) is not None for tag in ('skipped', 'failure', 'error'))
+          ]
+          assert not unsuccessful, f'PostgreSQL contracts must pass without skips: {unsuccessful}'
+          print(f'PostgreSQL contracts: {len(cases)} passed, 0 skipped; {dict(counts)}')
+          PY
+
+      - name: Upload PostgreSQL contract results
+        if: always()
+        uses: actions/upload-artifact@v4
+        with:
+          name: postgres-contract-results
+          path: postgres-contract-results.xml
+          if-no-files-found: ignore
+
   e2e:
     name: E2E Startup Tests
     runs-on: ubuntu-latest
@@ -148,9 +235,16 @@ jobs:
   coverage:
     name: Coverage Gate
     runs-on: ubuntu-latest
-    needs: [test, integration]
+    needs: [test, integration, postgres-contracts]
+    if: always()
 
     steps:
+      - name: Verify required test jobs
+        run: |
+          test "${{ needs.test.result }}" = success
+          test "${{ needs.integration.result }}" = success
+          test "${{ needs.postgres-contracts.result }}" = success
+
       - name: Checkout code
         uses: actions/checkout@v4
 

docs/POSTGRES_CONTRACT_TESTS.md

@@ -0,0 +1,81 @@
+# Disposable PostgreSQL contract tests
+
+The fast integration suite deliberately runs without an external database. At
+the current baseline, 83 cases in three persistence modules require PostgreSQL:
+
+| Module | PostgreSQL cases | Contract |
+| --- | ---: | --- |
+| `test_pipeline_admission_postgres.py` | 27 | Real row-lock admission, tenant isolation, rollback, cancellation and commit reconciliation |
+| `test_rag_document_identity.py` | 34 | Persisted engine document identity, scoped deletion and migration convergence |
+| `test_rag_interrupted_ingestion.py` | 22 | Interrupted/cancelled ingestion, delayed SDK writes, recovery and retained uploads |
+
+Another 39 PostgreSQL cases are marked slow and deselected by Fast Integration
+Tests rather than reported as skipped:
+
+| Module | PostgreSQL cases | Contract |
+| --- | ---: | --- |
+| `test_migration_branch_convergence.py` | 7 | Real historical/populated upgrade branches, certification backfill and merge-only downgrade |
+| `test_runner_timestamps_postgres.py` | 32 | Non-superuser timestamp/lease/retention contracts against both metadata and published-migration schemas |
+
+The PostgreSQL Contract Tests job in `.github/workflows/run-tests.yml` provisions
+a dedicated `pgvector/pgvector:pg16` service, sets both required URL variables,
+and runs all 122 PostgreSQL cases. Admission uses a separate database from RAG,
+convergence and timestamp tests: its populated `public` schema must not affect
+other fixtures' schema discovery through their `schema,public` search path.
+Its JUnit check requires each module's baseline
+case count and rejects skips, failures and errors. Results are uploaded even when
+the job fails. SQLite variants continue running in Fast Integration Tests.
+
+## Run locally
+
+Use a new disposable database. The admission fixture drops and recreates the
+entire `public` schema and creates a test runtime role. Never use a development,
+staging or production database, or a shared database containing useful data.
+
+For example, start an isolated container bound only to the local loopback address:
+
+```bash
+docker run --rm -d --name langbot-contract-pg \
+  -e POSTGRES_USER=postgres -e POSTGRES_DB=langbot_test \
+  -e POSTGRES_HOST_AUTH_METHOD=trust \
+  -p 127.0.0.1:55432:5432 pgvector/pgvector:pg16
+docker exec langbot-contract-pg pg_isready -U postgres -d langbot_test
+```
+
+Wait until `pg_isready` reports that the server is accepting connections, then:
+
+```bash
+docker exec langbot-contract-pg psql -U postgres -d postgres \
+  -v ON_ERROR_STOP=1 -c 'CREATE DATABASE langbot_admission_test;'
+uv sync --dev --frozen
+export TEST_POSTGRES_URL=postgresql+asyncpg://postgres@localhost:55432/langbot_test
+export LANGBOT_ADMISSION_TEST_URL=postgresql+asyncpg://postgres@localhost:55432/langbot_admission_test
+uv run pytest \
+  tests/integration/persistence/test_pipeline_admission_postgres.py \
+  tests/integration/persistence/test_rag_document_identity.py \
+  tests/integration/persistence/test_rag_interrupted_ingestion.py \
+  tests/integration/persistence/test_migration_branch_convergence.py \
+  tests/integration/persistence/test_runner_timestamps_postgres.py \
+  -k postgres -q --tb=short -rs --durations=10 \
+  --junitxml=postgres-contract-results.xml
+docker stop langbot-contract-pg
+```
+
+Trust authentication is limited to this throwaway, loopback-bound service. The
+admission fixture needs the literal `postgres@` URL to derive a separate
+unprivileged runtime-role connection. RAG fixtures create and clean up their own
+unique schemas, as do convergence fixtures. Timestamp fixtures create a unique
+non-superuser, non-RLS-bypass role and owned schema for each case and remove both
+afterward. The admin connection therefore needs role/schema creation privileges.
+These settings do not belong in an application configuration.
+
+## Other skip categories
+
+- The monitoring PostgreSQL unit regression is separately selected by the
+  Migrations (PostgreSQL) job in `.github/workflows/test-migrations.yml`
+- For plugin identity, compiled CLI and embedded SeekDB prerequisites, see
+  [the dependency contract setup in PR #2636](https://github.com/langbot-app/LangBot/pull/2636).
+
+These remaining prerequisites should be reported explicitly rather than treated
+as passing tests. The four chat-handler logging tests now initialize the normal
+application import graph before importing the handler and no longer skip.

tests/README.md

@@ -291,6 +291,11 @@ Note: PostgreSQL tests are **not** included in fast integration gate because the
 - Are marked with `@pytest.mark.slow`
 - Need `TEST_POSTGRES_URL` environment variable
 
+A separate **PostgreSQL Contract Tests** job provisions a disposable database for
+pipeline admission, RAG recovery, migration convergence, and timestamp/lease contracts.
+See [the PostgreSQL contract guide](../docs/POSTGRES_CONTRACT_TESTS.md) for scope,
+local commands, and remaining optional prerequisites.
+
 CI workflow `.github/workflows/test-migrations.yml` runs:
 - SQLite tests in `test-migrations-sqlite` job (fast, no external services)
 - PostgreSQL tests in `test-migrations-postgres` job (uses PostgreSQL service container)

tests/integration/persistence/test_runner_timestamps_postgres.py

@@ -34,7 +34,11 @@ TABLES = [AgentRun.__table__, AgentRunEvent.__table__, AgentRuntime.__table__, E
 
 def _published_schema(connection):
     with Operations.context(MigrationContext.configure(connection)):
-        for revision in ('58846a8d7a81_add_event_log_and_transcript_tables', '8d3a1f2c4b6e_add_agent_run_ledger'):
+        for revision in (
+            '58846a8d7a81_add_event_log_and_transcript_tables',
+            '8d3a1f2c4b6e_add_agent_run_ledger',
+            '0036_execution_host_leases',
+        ):
             importlib.import_module(f'langbot.pkg.persistence.alembic.versions.{revision}').upgrade()
 
 
@@ -119,6 +123,7 @@ async def test_run_lifecycle_deadline_events_and_stats(pg_engine):
             bot_id=None,
             event_type='message.received',
             source='debug',
+            data={},
             delivery=SimpleNamespace(model_dump=lambda **kwargs: {}),
         ),
         binding=SimpleNamespace(binding_id='binding', agent_id='agent', processor_id='agent', processor_type='agent'),

tests/integration/pipeline/test_full_flow.py

@@ -2,7 +2,8 @@
 Pipeline full-flow integration tests.
 
 Tests real pipeline stages with fake runner/provider.
-Validates message processing through PreProcessor, Processor, and SendResponseBackStage.
+Validates RuntimePipeline dispatch through PreProcessor, Processor, ResponseWrapper,
+and SendResponseBackStage, with external services replaced by deterministic doubles.
 
 Uses RuntimePipeline directly (not PipelineManager) to avoid DB dependency.
 
@@ -13,6 +14,7 @@ from __future__ import annotations
 
 import pytest
 import asyncio
+from types import SimpleNamespace
 from unittest.mock import AsyncMock, Mock
 
 from tests.factories import FakeApp, text_query, mock_platform_adapter
@@ -23,76 +25,20 @@ from tests.factories.platform import FakePlatform
 pytestmark = pytest.mark.integration
 
 
-# ============== FIXTURE FOR SYS.MODULES ISOLATION ==============
+# ============== NORMAL APPLICATION IMPORT BOOTSTRAP ==============
 
 
 @pytest.fixture(scope='module')
-def mock_circular_import_chain():
+def load_pipeline_modules():
+    """Initialize the real import graph before importing individual stages.
+
+    This matches normal startup and the pipeline unit-test bootstrap. No
+    sys.modules/package attributes are replaced, so later tests keep real
+    module identities and the complete stage registry.
     """
-    Break circular import chain for pipeline modules using isolated_sys_modules.
+    from importlib import import_module
 
-    Chain: pipeline → core.app → http_controller → groups/plugins
-
-    We mock minimal modules to allow importing RuntimePipeline, StageInstContainer,
-    and stage classes without triggering full application initialization.
-
-    After mocking, we import the stage modules so decorators register them.
-    """
-    from tests.utils.import_isolation import isolated_sys_modules, MockLifecycleControlScope
-
-    # Mock core.entities with LifecycleControlScope enum
-    mock_core_entities = Mock()
-    mock_core_entities.LifecycleControlScope = MockLifecycleControlScope
-
-    # Mock core.app - Application class is referenced but not instantiated
-    mock_core_app = Mock()
-
-    # Mock utils.importutil to avoid unrelated import-time registrations.
-    mock_importutil = Mock()
-    mock_importutil.import_modules_in_pkg = lambda pkg: None
-    mock_importutil.import_modules_in_pkgs = lambda pkgs: None
-
-    # Modules to clear (force re-import after mocking)
-    clear = [
-        'langbot.pkg.pipeline.stage',
-        'langbot.pkg.pipeline.entities',
-        'langbot.pkg.pipeline.pipelinemgr',
-        'langbot.pkg.pipeline.preproc.preproc',
-        'langbot.pkg.pipeline.process.process',
-        'langbot.pkg.pipeline.process.handler',
-        'langbot.pkg.pipeline.process.handlers.chat',
-        'langbot.pkg.pipeline.process.handlers.command',
-        'langbot.pkg.pipeline.respback.respback',
-    ]
-
-    with isolated_sys_modules(
-        mocks={
-            'langbot.pkg.core.entities': mock_core_entities,
-            'langbot.pkg.core.app': mock_core_app,
-            'langbot.pkg.utils.importutil': mock_importutil,
-            'langbot.pkg.pipeline.controller': Mock(),
-            'langbot.pkg.pipeline.pipelinemgr': Mock(),
-        },
-        clear=clear,
-    ):
-        # Import stage modules AFTER clearing so decorators register them
-        from importlib import import_module
-
-        # Import stage base first
-        import_module('langbot.pkg.pipeline.stage')
-
-        # Import entities
-        import_module('langbot.pkg.pipeline.entities')
-
-        # Import specific stages to register them
-        import_module('langbot.pkg.pipeline.preproc.preproc')
-        import_module('langbot.pkg.pipeline.process.process')
-        import_module('langbot.pkg.pipeline.respback.respback')
-
-        # Import pipelinemgr for RuntimePipeline
-        import_module('langbot.pkg.pipeline.pipelinemgr')
-
-        yield
+    import_module('langbot.pkg.pipeline.pipelinemgr')
 
 
 # ============== FAKE RUNNER ==============
@@ -172,35 +118,18 @@ def pipeline_app():
     """
     app = FakeApp()
 
-    # Session/conversation mocks for PreProcessor
-    mock_session = Mock()
-    mock_session.launcher_type = Mock()
-    mock_session.launcher_type.value = 'person'
-    mock_session.launcher_id = 12345
-    mock_session.sender_id = 12345
-    mock_session.use_prompt_name = 'default'
-    mock_session.using_conversation = None
+    # Real SDK entities keep plugin event validation meaningful.
+    from langbot_plugin.api.entities.builtin.provider.session import Conversation, Session, LauncherTypes
+    from langbot_plugin.api.entities.builtin.provider.prompt import Prompt
 
-    # Create a simple class to mimic Prompt behavior
-    class MockPrompt:
-        def __init__(self, name, messages):
-            self.name = name
-            self.messages = messages
-
-        def copy(self):
-            return MockPrompt(self.name, list(self.messages))
-
-    # Create real lists for messages
-    prompt_messages_list = []
-    messages_list = []
-
-    mock_prompt = MockPrompt('default', prompt_messages_list)
-    mock_conversation = Mock()
-    mock_conversation.prompt = mock_prompt
-    mock_conversation.messages = messages_list
-    mock_conversation.uuid = 'test-conversation-uuid'
-    mock_conversation.update_time = None
-    mock_conversation.create_time = None
+    mock_session = Session(launcher_type=LauncherTypes.PERSON, launcher_id=12345, sender_id=12345)
+    mock_conversation = Conversation(
+        prompt=Prompt(name='default', messages=[]),
+        messages=[],
+        pipeline_uuid='test-pipeline-uuid',
+        bot_uuid='test-bot-uuid',
+        uuid='test-conversation-uuid',
+    )
 
     async def get_scoped_session(query):
         context = query._execution_context
@@ -223,6 +152,18 @@ def pipeline_app():
 
     # Tool manager mock
     app.tool_mgr.get_all_tools = AsyncMock(return_value=[])
+    app.tool_mgr.get_resolved_tool_catalog = AsyncMock(return_value=[])
+    app.workspace_service = AsyncMock()
+    app.workspace_service.get_execution_binding.return_value = SimpleNamespace(
+        instance_uuid='test-instance',
+        workspace_uuid='test-workspace',
+        placement_generation=1,
+    )
+    app.query_pool.remove_query = AsyncMock(return_value=True)
+    app.bot_service = AsyncMock()
+    app.bot_service.get_bot.return_value = {'name': 'Test bot'}
+    app.monitoring_service = AsyncMock()
+    app.monitoring_service.record_message.return_value = 'test-monitoring-message'
 
     # Telemetry mock (required by ChatMessageHandler)
     app.telemetry = Mock()
@@ -251,7 +192,10 @@ def set_fake_runner(pipeline_app):
         orchestrator = Mock()
         orchestrator.try_claim_steering_from_query = AsyncMock(return_value=False)
 
+        orchestrator.observed_queries = []
+
         async def run_from_query(query):
+            orchestrator.observed_queries.append(query)
             async for result in runner.run(query):
                 yield result
 
@@ -283,6 +227,7 @@ def create_minimal_pipeline_config():
             'misc': {
                 'at-sender': False,
                 'quote-origin': False,
+                'track-function-calls': False,
                 'exception-handling': 'show-hint',
                 'failure-hint': 'Request failed.',
             },
@@ -320,7 +265,7 @@ async def collect_processor_results(processor, query, stage_name):
 # ============== TESTS ==============
 
 
-@pytest.mark.usefixtures('mock_circular_import_chain')
+@pytest.mark.usefixtures('load_pipeline_modules')
 class TestPipelineStageChainReal:
     """Tests for real pipeline stage chain."""
 
@@ -348,7 +293,7 @@ class TestPipelineStageChainReal:
         assert 'SendResponseBackStage' in stage.preregistered_stages
 
 
-@pytest.mark.usefixtures('mock_circular_import_chain')
+@pytest.mark.usefixtures('load_pipeline_modules')
 class TestPreProcessorStage:
     """Tests for PreProcessor stage alone."""
 
@@ -410,7 +355,7 @@ class TestPreProcessorStage:
         assert result.new_query.user_message.role == 'user'
 
 
-@pytest.mark.usefixtures('mock_circular_import_chain')
+@pytest.mark.usefixtures('load_pipeline_modules')
 class TestProcessorStage:
     """Tests for MessageProcessor stage."""
 
@@ -517,7 +462,7 @@ class TestProcessorStage:
         assert query.resp_messages[0] == reply_chain
 
 
-@pytest.mark.usefixtures('mock_circular_import_chain')
+@pytest.mark.usefixtures('load_pipeline_modules')
 class TestRunnerExceptionFlow:
     """Tests for runner exception handling."""
 
@@ -638,7 +583,7 @@ class TestRunnerExceptionFlow:
         assert results[0].user_notice is None
 
 
-@pytest.mark.usefixtures('mock_circular_import_chain')
+@pytest.mark.usefixtures('load_pipeline_modules')
 class TestSendResponseBackStage:
     """Tests for SendResponseBackStage."""
 
@@ -768,136 +713,220 @@ class TestSendResponseBackStage:
         pipeline_app.plugin_connector.notify_plugin_diagnostic.assert_not_called()
 
 
-@pytest.mark.usefixtures('mock_circular_import_chain')
+@pytest.fixture
+async def runtime_pipeline_factory(pipeline_app, load_pipeline_modules):
+    """Use real dispatch and stages; fake only the surrounding services."""
+    from langbot.pkg.pipeline import pipelinemgr
+    from langbot.pkg.pipeline.preproc.preproc import PreProcessor
+    from langbot.pkg.pipeline.process.process import Processor
+    from langbot.pkg.pipeline.wrapper.wrapper import ResponseWrapper
+    from langbot.pkg.pipeline.respback.respback import SendResponseBackStage
+
+    async def create(query):
+        context = query._execution_context
+        query.instance_uuid = context.instance_uuid
+        query.workspace_uuid = context.workspace_uuid
+        query.placement_generation = context.placement_generation
+        query.resp_message_chain = []
+        query.adapter.on_monitoring_message_created = AsyncMock()
+        stage_specs = [
+            ('PreProcessor', PreProcessor),
+            ('MessageProcessor', Processor),
+            ('ResponseWrapper', ResponseWrapper),
+            ('SendResponseBackStage', SendResponseBackStage),
+        ]
+        containers = []
+        for name, cls in stage_specs:
+            instance = cls(pipeline_app)
+            await instance.initialize(query.pipeline_config)
+            containers.append(pipelinemgr.StageInstContainer(name, instance))
+        entity = SimpleNamespace(
+            uuid=query.pipeline_uuid,
+            workspace_uuid=context.workspace_uuid,
+            name='Contract pipeline',
+            config=query.pipeline_config,
+            extensions_preferences={'enable_all_plugins': False, 'plugins': [{'author': 'test', 'name': 'observer'}]},
+        )
+        return pipelinemgr.RuntimePipeline(pipeline_app, entity, containers, context)
+
+    return create
+
+
+@pytest.fixture
+def passthrough_plugin(pipeline_app):
+    """Return real SDK contexts, like a plugin which does not intercept a call."""
+    from langbot_plugin.api.entities.context import EventContext
+
+    def context_for(event, _bound_plugins):
+        return EventContext.from_event(event)
+
+    pipeline_app.plugin_connector.emit_event.side_effect = context_for
+    return context_for
+
+
+@pytest.mark.usefixtures('load_pipeline_modules')
 class TestStageChainIntegration:
-    """Tests for full stage chain (PreProcessor -> Processor -> SendResponseBackStage)."""
+    """Runtime dispatch must preserve input, output, plugin boundaries, and cleanup."""
 
     @pytest.mark.asyncio
-    async def test_full_chain_text_message_flow(self, pipeline_app, fake_platform_adapter, set_fake_runner):
-        """
-        Full chain: text message -> PreProcessor -> Processor -> SendResponseBackStage.
-
-        Validates:
-        - PreProcessor sets up session, user_message
-        - Processor calls runner and populates resp_messages
-        - SendResponseBackStage calls adapter.reply_message
-        """
-        from langbot.pkg.pipeline import entities
-        from langbot.pkg.pipeline.preproc import preproc
-        from langbot.pkg.pipeline.process import process
-        from langbot.pkg.pipeline.respback import respback
-
+    async def test_full_chain_text_message_flow(
+        self, pipeline_app, fake_platform_adapter, set_fake_runner, runtime_pipeline_factory, passthrough_plugin
+    ):
         adapter, platform = fake_platform_adapter
-
-        # Set fake runner
-        fake_runner = FakeRunner().returns('LANGBOT_FAKE_PONG')
-        set_fake_runner(fake_runner)
-
-        # Create query
-        config = create_minimal_pipeline_config()
-        query = text_query('ping')
+        set_fake_runner(FakeRunner.returns('LANGBOT_FAKE_PONG'))
+        query = text_query('ping', pipeline_config=create_minimal_pipeline_config())
+        query.pipeline_config['output']['misc']['quote-origin'] = True
         query.adapter = adapter
-        query.pipeline_config = config
-        query.resp_messages = []
-        query.resp_message_chain = []
+        runtime = await runtime_pipeline_factory(query)
 
-        # Mock plugin_connector for PreProcessor and Processor events
-        mock_event_ctx_preproc = Mock()
-        mock_event_ctx_preproc.event = Mock()
-        mock_event_ctx_preproc.event.default_prompt = []
-        mock_event_ctx_preproc.event.prompt = []
+        await runtime.run(query)
 
-        mock_event_ctx_processor = Mock()
-        mock_event_ctx_processor.is_prevented_default = Mock(return_value=False)
-        mock_event_ctx_processor.event = Mock()
-        mock_event_ctx_processor.event.user_message_alter = None
-
-        pipeline_app.plugin_connector.emit_event = AsyncMock()
-        pipeline_app.plugin_connector.emit_event.side_effect = [
-            mock_event_ctx_preproc,  # PreProcessor PromptPreProcessing
-            mock_event_ctx_processor,  # Processor NormalMessageReceived
+        assert pipeline_app.agent_run_orchestrator.observed_queries == [query]
+        assert str(query.user_message.get_content_platform_message_chain()) == 'ping'
+        assert [message.content for message in query.resp_messages] == ['LANGBOT_FAKE_PONG']
+        assert [str(chain) for chain in query.resp_message_chain] == ['LANGBOT_FAKE_PONG']
+        outbound = platform.get_outbound_messages()
+        assert len(outbound) == 1, pipeline_app.logger.error.call_args_list
+        assert outbound[0]['source'] is query.message_event
+        assert str(outbound[0]['message']) == 'LANGBOT_FAKE_PONG'
+        assert outbound[0]['quote_origin'] is True
+        assert platform.get_outbound_chunks() == []
+        assert not query.variables.get('_monitoring_has_error'), pipeline_app.logger.error.call_args_list
+        pipeline_app.logger.error.assert_not_called()
+        emitted = pipeline_app.plugin_connector.emit_event.await_args_list
+        assert [call.args[0].event_name for call in emitted] == [
+            'PersonMessageReceived',
+            'PromptPreProcessing',
+            'PersonNormalMessageReceived',
+            'NormalMessageResponded',
         ]
+        assert [call.args[1] for call in emitted] == [['test/observer']] * 4
+        pipeline_app.query_pool.remove_query.assert_awaited_once_with(query)
 
-        # Create stages
-        preproc_stage = preproc.PreProcessor(pipeline_app)
-        processor_stage = process.Processor(pipeline_app)
-        await processor_stage.initialize(config)
-        respback_stage = respback.SendResponseBackStage(pipeline_app)
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        'blocked_event', ['PersonMessageReceived', 'PersonNormalMessageReceived', 'NormalMessageResponded']
+    )
+    async def test_chain_stops_on_interrupt(
+        self,
+        pipeline_app,
+        fake_platform_adapter,
+        set_fake_runner,
+        runtime_pipeline_factory,
+        passthrough_plugin,
+        blocked_event,
+    ):
+        adapter, platform = fake_platform_adapter
+        set_fake_runner(FakeRunner.returns('MUST_NOT_BE_DELIVERED'))
+        query = text_query('hello', pipeline_config=create_minimal_pipeline_config())
+        query.adapter = adapter
+        runtime = await runtime_pipeline_factory(query)
 
-        # Run PreProcessor
-        result1 = await preproc_stage.process(query, 'PreProcessor')
-        assert result1.result_type.value == entities.ResultType.CONTINUE.value
-        query = result1.new_query
+        def intercept(event, bound_plugins):
+            context = passthrough_plugin(event, bound_plugins)
+            if event.event_name == blocked_event:
+                context.prevent_default()
+            return context
 
-        # Run Processor
-        results = await collect_processor_results(processor_stage, query, 'MessageProcessor')
-        assert len(results) >= 1
+        pipeline_app.plugin_connector.emit_event.side_effect = intercept
+        await runtime.run(query)
 
-        # Build resp_message_chain from resp_messages
+        event_names = [call.args[0].event_name for call in pipeline_app.plugin_connector.emit_event.await_args_list]
+        assert event_names[-1] == blocked_event
+        expected_runner_queries = [query] if blocked_event == 'NormalMessageResponded' else []
+        assert pipeline_app.agent_run_orchestrator.observed_queries == expected_runner_queries
+        assert query.resp_message_chain == []
+        assert platform.get_outbound_messages() == []
+        assert platform.get_outbound_chunks() == []
+        assert not query.variables.get('_monitoring_has_error'), pipeline_app.logger.error.call_args_list
+        pipeline_app.logger.error.assert_not_called()
+        pipeline_app.query_pool.remove_query.assert_awaited_once_with(query)
+
+    @pytest.mark.asyncio
+    async def test_wrapper_plugin_reply_reaches_adapter(
+        self, pipeline_app, fake_platform_adapter, set_fake_runner, runtime_pipeline_factory, passthrough_plugin
+    ):
         from tests.factories.message import text_chain
 
-        for resp_msg in query.resp_messages:
-            if resp_msg.content:
-                query.resp_message_chain.append(text_chain(resp_msg.content))
+        adapter, platform = fake_platform_adapter
+        set_fake_runner(FakeRunner.returns('original runner answer'))
+        query = text_query('hello', pipeline_config=create_minimal_pipeline_config())
+        query.adapter = adapter
+        runtime = await runtime_pipeline_factory(query)
 
-        # Run SendResponseBackStage
-        result3 = await respback_stage.process(query, 'SendResponseBackStage')
-        assert result3.result_type.value == entities.ResultType.CONTINUE.value
+        def alter_response(event, bound_plugins):
+            context = passthrough_plugin(event, bound_plugins)
+            if event.event_name == 'NormalMessageResponded':
+                context.event.reply_message_chain = text_chain('plugin replacement')
+            return context
+
+        pipeline_app.plugin_connector.emit_event.side_effect = alter_response
+        await runtime.run(query)
 
-        # Verify adapter was called
         outbound = platform.get_outbound_messages()
-        assert len(outbound) >= 1
+        assert len(outbound) == 1, pipeline_app.logger.error.call_args_list
+        assert str(outbound[0]['message']) == 'plugin replacement'
+        assert [str(chain) for chain in query.resp_message_chain] == ['plugin replacement']
+        pipeline_app.query_pool.remove_query.assert_awaited_once_with(query)
 
     @pytest.mark.asyncio
-    async def test_chain_stops_on_interrupt(self, pipeline_app, fake_platform_adapter):
-        """
-        Chain should stop when a stage returns INTERRUPT.
-
-        PreProcessor returns CONTINUE, Processor returns INTERRUPT (prevent_default).
-        """
-        from langbot.pkg.pipeline import entities
-        from langbot.pkg.pipeline.preproc import preproc
-        from langbot.pkg.pipeline.process import process
-
+    async def test_runner_failure_delivers_hint_and_cleans_up(
+        self, pipeline_app, fake_platform_adapter, set_fake_runner, runtime_pipeline_factory, passthrough_plugin
+    ):
         adapter, platform = fake_platform_adapter
-
-        # Create query
-        query = text_query('hello')
+        set_fake_runner(FakeRunner.raises(RuntimeError('provider unavailable')))
+        query = text_query('hello', pipeline_config=create_minimal_pipeline_config())
         query.adapter = adapter
-        query.pipeline_config = create_minimal_pipeline_config()
+        runtime = await runtime_pipeline_factory(query)
 
-        # Mock plugin_connector - PreProcessor continues, Processor interrupts
-        mock_event_ctx_preproc = Mock()
-        mock_event_ctx_preproc.event = Mock()
-        mock_event_ctx_preproc.event.default_prompt = []
-        mock_event_ctx_preproc.event.prompt = []
+        await runtime.run(query)
 
-        mock_event_ctx_processor = Mock()
-        mock_event_ctx_processor.is_prevented_default = Mock(return_value=True)
-        mock_event_ctx_processor.event = Mock()
-        mock_event_ctx_processor.event.reply_message_chain = None
-
-        pipeline_app.plugin_connector.emit_event = AsyncMock()
-        pipeline_app.plugin_connector.emit_event.side_effect = [
-            mock_event_ctx_preproc,  # PreProcessor PromptPreProcessing
-            mock_event_ctx_processor,  # Processor NormalMessageReceived
+        outbound = platform.get_outbound_messages()
+        assert len(outbound) == 1, pipeline_app.logger.error.call_args_list
+        assert str(outbound[0]['message']) == 'Request failed.'
+        assert query.variables['_monitoring_has_error'] is True
+        assert query.resp_messages == []
+        assert query.resp_message_chain == []
+        assert [call.args[0].event_name for call in pipeline_app.plugin_connector.emit_event.await_args_list] == [
+            'PersonMessageReceived',
+            'PromptPreProcessing',
+            'PersonNormalMessageReceived',
         ]
+        pipeline_app.query_pool.remove_query.assert_awaited_once_with(query)
 
-        # Create stages
-        preproc_stage = preproc.PreProcessor(pipeline_app)
-        processor_stage = process.Processor(pipeline_app)
-        await processor_stage.initialize(query.pipeline_config)
+    @pytest.mark.asyncio
+    async def test_streaming_chain_delivers_each_chunk_with_one_response_id(
+        self, pipeline_app, set_fake_runner, runtime_pipeline_factory, passthrough_plugin
+    ):
+        from langbot_plugin.api.entities.builtin.provider.message import MessageChunk
 
-        # Run PreProcessor
-        result1 = await preproc_stage.process(query, 'PreProcessor')
-        assert result1.result_type.value == entities.ResultType.CONTINUE.value
-        query = result1.new_query
+        class StreamingRunner(FakeRunner):
+            async def run(self, query):
+                yield MessageChunk(role='assistant', content='first', all_content='first', is_final=False)
+                yield MessageChunk(role='assistant', content=' second', all_content='first second', is_final=True)
 
-        # Run Processor - should INTERRUPT
-        results = await collect_processor_results(processor_stage, query, 'MessageProcessor')
+        platform = FakePlatform(stream_output_supported=True)
+        adapter = mock_platform_adapter(platform)
+        adapter.create_message_card = AsyncMock(side_effect=platform.create_message_card)
+        set_fake_runner(StreamingRunner)
+        query = text_query('stream please', pipeline_config=create_minimal_pipeline_config())
+        query.adapter = adapter
+        runtime = await runtime_pipeline_factory(query)
 
-        assert len(results) == 1
-        assert results[0].result_type.value == entities.ResultType.INTERRUPT.value
+        await runtime.run(query)
 
-        # Chain stops here - no resp_messages
-        assert len(query.resp_messages) == 0
+        chunks = platform.get_outbound_chunks()
+        assert [(str(chunk['message']), chunk['is_final']) for chunk in chunks] == [
+            ('first', False),
+            ('first second', True),
+        ], pipeline_app.logger.error.call_args_list
+        assert all(chunk['source'] is query.message_event for chunk in chunks)
+        response_ids = [chunk['bot_message'].resp_message_id for chunk in chunks]
+        assert response_ids[0] and response_ids == [response_ids[0]] * 2
+        adapter.create_message_card.assert_awaited_once_with(response_ids[0], query.message_event)
+        assert [str(chain) for chain in query.resp_message_chain] == ['first second']
+        assert platform.get_outbound_messages() == []
+        assert pipeline_app.agent_run_orchestrator.observed_queries == [query]
+        assert not query.variables.get('_monitoring_has_error'), pipeline_app.logger.error.call_args_list
+        pipeline_app.logger.error.assert_not_called()
+        pipeline_app.query_pool.remove_query.assert_awaited_once_with(query)

tests/unit_tests/api/service/test_bot_service.py

@@ -867,7 +867,8 @@ class TestBotServiceSendMessage:
         with pytest.raises(Exception, match='Invalid message_chain format'):
             await service.send_message(WORKSPACE_UUID, 'bot-uuid', 'group', '123', {'invalid': 'format'})
 
-    async def test_send_message_valid_call(self):
+    @pytest.mark.parametrize(('target_type', 'target_id'), [('group', 123), ('person', 'user-456')])
+    async def test_send_message_valid_call(self, target_type, target_id):
         """Sends message through adapter when all valid."""
         # Setup
         ap = SimpleNamespace()
@@ -881,14 +882,18 @@ class TestBotServiceSendMessage:
         service = BotService(ap)
         service.get_bot = AsyncMock(return_value={'uuid': 'bot-uuid'})
 
-        # Execute with valid message chain format
-        message_chain_data = {'messages': [{'type': 'text', 'data': {'text': 'Hello'}}]}
+        from langbot_plugin.api.entities.builtin.platform.message import MessageChain, Plain
 
-        # Patch the import location - the module imports inside the function
-        with patch('langbot_plugin.api.entities.builtin.platform.message.MessageChain') as MockMessageChain:
-            mock_chain = Mock()
-            MockMessageChain.model_validate = Mock(return_value=mock_chain)
-            await service.send_message(WORKSPACE_UUID, 'bot-uuid', 'group', '123', message_chain_data)
+        expected_chain = MessageChain([Plain(text='Hello')])
+        message_chain_data = expected_chain.model_dump()
+        await service.send_message(WORKSPACE_UUID, 'bot-uuid', target_type, target_id, message_chain_data)
 
-        # Verify adapter.send_message was called
-        runtime_bot.adapter.send_message.assert_called_once_with('group', '123', mock_chain)
+        service.get_bot.assert_awaited_once_with(WORKSPACE_UUID, 'bot-uuid', include_secret=False)
+        ap.platform_mgr.get_bot_by_uuid.assert_awaited_once_with(WORKSPACE_UUID, 'bot-uuid')
+        runtime_bot.adapter.send_message.assert_awaited_once()
+        sent_target_type, sent_target_id, sent_chain = runtime_bot.adapter.send_message.await_args.args
+        assert sent_target_type == target_type
+        assert sent_target_id == str(target_id)
+        assert isinstance(sent_chain, MessageChain)
+        assert sent_chain.model_dump() == expected_chain.model_dump()
+        assert str(sent_chain) == 'Hello'

tests/unit_tests/api/service/test_user_service.py

@@ -16,6 +16,10 @@ from __future__ import annotations
 import pytest
 import jwt
 import datetime
+
+import sqlalchemy
+from argon2 import PasswordHasher
+from argon2.exceptions import VerifyMismatchError
 from unittest.mock import AsyncMock, Mock
 from types import SimpleNamespace
 
@@ -510,10 +514,17 @@ class TestUserServiceResetPassword:
         service = UserService(ap)
 
         # Execute
-        await service.reset_password('test@example.com', 'new_password')
+        await service.reset_password(' Test@Example.COM ', 'new_password')
 
-        # Verify - execute_async was called with update
-        ap.persistence_mgr.execute_async.assert_called_once()
+        ap.persistence_mgr.execute_async.assert_awaited_once()
+        statement = ap.persistence_mgr.execute_async.await_args.args[0]
+        assert isinstance(statement, sqlalchemy.sql.dml.Update)
+        assert statement.table.name == User.__tablename__
+        params = statement.compile().params
+        assert params['normalized_email_1'] == '[email protected]'
+        assert PasswordHasher().verify(params['password'], 'new_password')
+        with pytest.raises(VerifyMismatchError):
+            PasswordHasher().verify(params['password'], 'old-password')
 
 
 class TestUserServiceChangePassword:

tests/unit_tests/api/service/test_webhook_service.py

@@ -377,104 +377,59 @@ class TestWebhookServiceGetWebhook:
 class TestWebhookServiceUpdateWebhook:
     """Tests for update_webhook method."""
 
-    async def test_update_webhook_name_only(self):
-        """Updates only the name field."""
-        # Setup
-        ap = SimpleNamespace()
-        ap.persistence_mgr = SimpleNamespace()
-        ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_write_result())
+    @pytest.mark.parametrize(
+        'changes',
+        [
+            {'name': 'Updated Name'},
+            {'url': 'https://updated.example.com'},
+            {'description': 'Updated description'},
+            {'enabled': False},
+            {
+                'name': 'All Updated',
+                'url': 'https://all.updated.example.com',
+                'description': 'All updated description',
+                'enabled': False,
+            },
+        ],
+        ids=['name-only', 'url-only', 'description-only', 'enabled-only', 'all-fields'],
+    )
+    async def test_update_webhook_persists_requested_fields(self, tenant_webhook_service, changes):
+        """Persist requested fields without changing omitted fields or another Workspace."""
+        original = {
+            'name': 'Original Name',
+            'url': 'https://original.example.com',
+            'description': 'Original description',
+            'enabled': True,
+        }
+        created = await tenant_webhook_service.create_webhook(ISOLATION_WORKSPACE_A, **original)
+        other = await tenant_webhook_service.create_webhook(ISOLATION_WORKSPACE_B, **original)
 
-        service = WebhookService(ap)
+        assert await tenant_webhook_service.update_webhook(ISOLATION_WORKSPACE_A, created['id'], **changes) is True
 
-        # Execute
-        await service.update_webhook(WORKSPACE_UUID, 1, name='Updated Name')
+        updated = await tenant_webhook_service.get_webhook(ISOLATION_WORKSPACE_A, created['id'], include_secret=True)
+        assert {field: updated[field] for field in original} == original | changes
+        assert updated['id'] == created['id']
+        assert updated['workspace_uuid'] == ISOLATION_WORKSPACE_A
+        assert (
+            await tenant_webhook_service.get_webhook(ISOLATION_WORKSPACE_B, other['id'], include_secret=True) == other
+        )
 
-        # Verify
-        ap.persistence_mgr.execute_async.assert_called_once()
-
-    async def test_update_webhook_url_only(self):
-        """Updates only the url field."""
-        # Setup
-        ap = SimpleNamespace()
-        ap.persistence_mgr = SimpleNamespace()
-        ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_write_result())
-
-        service = WebhookService(ap)
-
-        # Execute
-        await service.update_webhook(WORKSPACE_UUID, 1, url='http://updated.example.com')
-
-        # Verify
-        ap.persistence_mgr.execute_async.assert_called_once()
-
-    async def test_update_webhook_description_only(self):
-        """Updates only the description field."""
-        # Setup
-        ap = SimpleNamespace()
-        ap.persistence_mgr = SimpleNamespace()
-        ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_write_result())
-
-        service = WebhookService(ap)
-
-        # Execute
-        await service.update_webhook(WORKSPACE_UUID, 1, description='Updated description')
-
-        # Verify
-        ap.persistence_mgr.execute_async.assert_called_once()
-
-    async def test_update_webhook_enabled_only(self):
-        """Updates only the enabled field."""
-        # Setup
-        ap = SimpleNamespace()
-        ap.persistence_mgr = SimpleNamespace()
-        ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_write_result())
-
-        service = WebhookService(ap)
-
-        # Execute
-        await service.update_webhook(WORKSPACE_UUID, 1, enabled=False)
-
-        # Verify
-        ap.persistence_mgr.execute_async.assert_called_once()
-
-    async def test_update_webhook_all_fields(self):
-        """Updates all fields at once."""
-        # Setup
-        ap = SimpleNamespace()
-        ap.persistence_mgr = SimpleNamespace()
-        ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_write_result())
-
-        service = WebhookService(ap)
-
-        # Execute
-        await service.update_webhook(
-            WORKSPACE_UUID,
-            1,
-            name='All Updated',
-            url='http://all.updated.com',
-            description='All updated description',
+    async def test_update_webhook_no_fields(self, tenant_webhook_service):
+        """An empty update reports scoped existence and leaves persisted values unchanged."""
+        created = await tenant_webhook_service.create_webhook(
+            ISOLATION_WORKSPACE_A,
+            'Unchanged',
+            'https://unchanged.example.com',
+            description='Keep this description',
             enabled=False,
         )
 
-        # Verify
-        ap.persistence_mgr.execute_async.assert_called_once()
-
-    async def test_update_webhook_no_fields(self):
-        """Does nothing when no fields provided."""
-        # Setup
-        ap = SimpleNamespace()
-        ap.persistence_mgr = SimpleNamespace()
-        existing = _create_mock_webhook(webhook_id=1)
-        ap.persistence_mgr.execute_async = AsyncMock(return_value=_create_mock_result(first_item=existing))
-        ap.persistence_mgr.serialize_model = Mock(return_value={'id': 1})
-
-        service = WebhookService(ap)
-
-        # Execute - no update parameters
-        await service.update_webhook(WORKSPACE_UUID, 1)
-
-        # No write is issued; one scoped existence lookup is performed.
-        ap.persistence_mgr.execute_async.assert_called_once()
+        assert await tenant_webhook_service.update_webhook(ISOLATION_WORKSPACE_A, created['id']) is True
+        assert (
+            await tenant_webhook_service.get_webhook(ISOLATION_WORKSPACE_A, created['id'], include_secret=True)
+            == created
+        )
+        assert await tenant_webhook_service.update_webhook(ISOLATION_WORKSPACE_B, created['id']) is False
 
 
 class TestWebhookServiceDeleteWebhook:
@@ -695,6 +650,10 @@ async def test_update_and_delete_are_scoped(tenant_webhook_service):
         name='new',
         enabled=False,
     )
+    updated = await tenant_webhook_service.get_webhook(ISOLATION_WORKSPACE_A, created['id'], include_secret=True)
+    assert updated['name'] == 'new'
+    assert updated['enabled'] is False
+    assert updated['url'] == 'https://a.invalid/old'
     assert await tenant_webhook_service.get_enabled_webhooks(ISOLATION_WORKSPACE_A) == []
     assert await tenant_webhook_service.delete_webhook(ISOLATION_WORKSPACE_A, created['id'])
     assert await tenant_webhook_service.get_webhook(ISOLATION_WORKSPACE_A, created['id']) is None

tests/unit_tests/pipeline/test_chat_handler_logging.py

@@ -2,14 +2,11 @@ from __future__ import annotations
 
 from unittest.mock import Mock
 
-import pytest
 import langbot_plugin.api.entities.builtin.provider.message as provider_message
 
-# TODO: unskip once the handler ↔ app circular import is resolved
-pytest.skip(
-    'circular import in handler ↔ app; will be unblocked once resolved',
-    allow_module_level=True,
-)
+# Initialize the application import graph before importing a leaf handler. This
+# matches normal startup and avoids handler -> app -> handler partial imports.
+from langbot.pkg.core import app  # noqa: F401
 
 from langbot.pkg.pipeline.process.handler import MessageHandler  # noqa: E402