diff --git a/.github/workflows/optional-contracts.yml b/.github/workflows/optional-contracts.yml
index f7e18f890..2a2809b6a 100644
--- a/.github/workflows/optional-contracts.yml
+++ b/.github/workflows/optional-contracts.yml
@@ -9,7 +9,6 @@ on:
       - 'pyproject.toml'
       - 'pytest.ini'
       - 'uv.lock'
-      - 'scripts/assert-required-tests.py'
       - '.github/workflows/optional-contracts.yml'
   push:
     branches: [master, develop]
@@ -19,7 +18,6 @@ on:
       - 'pyproject.toml'
       - 'pytest.ini'
       - 'uv.lock'
-      - 'scripts/assert-required-tests.py'
       - '.github/workflows/optional-contracts.yml'
   workflow_dispatch:
 
@@ -87,12 +85,8 @@ jobs:
         env:
           LANGBOT_CLI_BIN: ${{ runner.temp }}/langbot-cli/lbctl
         run: |
-          uv run --no-sync pytest \
-            tests/unit_tests/api/service/test_cli_diagnostics.py::test_compiled_cli_against_core_http \
+          uv run --no-sync pytest --required-suite compiled_cli \
             -q --tb=short -rs --junitxml=compiled-cli-results.xml
-          python scripts/assert-required-tests.py compiled-cli-results.xml \
-            --module tests.unit_tests.api.service.test_cli_diagnostics \
-            --name-prefix test_compiled_cli_against_core_http --expected-tests 1
 
       - name: Upload compiled CLI contract result
         if: always()
@@ -151,12 +145,8 @@ jobs:
         env:
           LANGBOT_LEGACY_PLUGIN_ROOT: ${{ github.workspace }}/.contract-sources/langbot-plugins
         run: |
-          uv run --no-sync pytest \
-            tests/unit_tests/agent/test_legacy_identity_boundary.py::test_real_plugin_identity_helper_with_host_payload \
+          uv run --no-sync pytest --required-suite plugin_identity \
             -q --tb=short -rs --junitxml=plugin-identity-results.xml
-          python scripts/assert-required-tests.py plugin-identity-results.xml \
-            --module tests.unit_tests.agent.test_legacy_identity_boundary \
-            --name-prefix test_real_plugin_identity_helper_with_host_payload --expected-tests 16
 
       - name: Upload plugin identity results
         if: always()
@@ -198,11 +188,9 @@ jobs:
 
           print(f"pyseekdb={version('pyseekdb')}, pylibseekdb={version('pylibseekdb')}")
           PY
-          uv run --no-sync pytest tests/integration/vector/test_seekdb.py \
-            -m slow -q --tb=short -rs --durations=5 \
+          uv run --no-sync pytest --required-suite seekdb \
+            -q --tb=short -rs --durations=5 \
             --basetemp="$SEEKDB_TEST_TMP" --junitxml=seekdb-contract-results.xml
-          python scripts/assert-required-tests.py seekdb-contract-results.xml \
-            --module tests.integration.vector.test_seekdb --name-prefix test_ --expected-tests 2
 
       - name: Upload SeekDB results and startup logs
         if: always()
diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml
index 583dfd5a5..bd54e498f 100644
--- a/.github/workflows/run-tests.yml
+++ b/.github/workflows/run-tests.yml
@@ -11,7 +11,7 @@ on:
       - 'uv.lock'
       - 'run_tests.sh'
       - 'scripts/test-*.sh'
-      - 'scripts/assert-required-tests.py'
+      - 'pytest.ini'
   push:
     branches:
       - master
@@ -50,11 +50,8 @@ jobs:
 
       - name: Run required security regressions
         run: |
-          uv run pytest tests/test_cwe94_debug_exec.py \
+          uv run pytest --required-suite security \
             -q --tb=short -rs --junitxml=cwe94-results.xml
-          python scripts/assert-required-tests.py cwe94-results.xml \
-            --module tests.test_cwe94_debug_exec \
-            --name-prefix test_no_ --expected-tests 2
 
       - name: Upload security regression results
         if: always()
@@ -149,45 +146,8 @@ jobs:
 
       - 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 \
-            tests/unit_tests/api/service/test_monitoring_sessions.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': 6,
-              'test_runner_timestamps_postgres': 32,
-              'test_monitoring_sessions': 1,
-          }
-          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]}'
-          convergence_cases = {
-              case.get('name') for case in cases
-              if case.get('classname', '').rsplit('.', 1)[-1] == 'test_migration_branch_convergence'
-          }
-          assert 'test_empty_database_startup_schema_then_real_migrations[postgresql]' in convergence_cases, (
-              'The fresh-schema migration and idempotency contract must execute'
-          )
-          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
+          uv run pytest --required-suite postgres \
+            -q --tb=short -rs --durations=10 --junitxml=postgres-contract-results.xml
 
       - name: Upload PostgreSQL contract results
         if: always()
@@ -241,7 +201,7 @@ jobs:
         run: uv sync --dev
 
       - name: Run E2E startup tests
-        run: uv run pytest tests/e2e --ignore=tests/e2e/test_local_runner_fake_provider.py -q --tb=short
+        run: uv run pytest tests/e2e -m "not local_agent" -q --tb=short
 
       - name: Check Docker for the required LocalAgent Box contract
         run: docker info
@@ -250,11 +210,8 @@ jobs:
         env:
           LANGBOT_LEGACY_PLUGIN_ROOT: ${{ github.workspace }}/.contract-sources/langbot-plugins
         run: |
-          uv run --no-sync pytest tests/e2e/test_local_runner_fake_provider.py \
+          uv run --no-sync pytest --required-suite local_agent \
             -q --tb=short -rs --junitxml=local-agent-results.xml
-          python scripts/assert-required-tests.py local-agent-results.xml \
-            --module tests.e2e.test_local_runner_fake_provider \
-            --name-prefix test_local_runner_ --expected-tests 6
 
       - name: Upload LocalAgent contract results
         if: always()
diff --git a/docs/DEPENDENCY_CONTRACT_TESTS.md b/docs/DEPENDENCY_CONTRACT_TESTS.md
index 48168d529..6c038c99f 100644
--- a/docs/DEPENDENCY_CONTRACT_TESTS.md
+++ b/docs/DEPENDENCY_CONTRACT_TESTS.md
@@ -47,12 +47,12 @@ To run with an official checkout at the locked revision:
 
 ```bash
 export LANGBOT_LEGACY_PLUGIN_ROOT=/absolute/path/to/langbot-plugins
-uv run --frozen pytest tests/unit_tests/agent/test_legacy_identity_boundary.py -q -rs
+uv run --frozen pytest --required-suite plugin_identity -q -rs
 ```
 
 An unset variable allows an explicit local prerequisite skip. An empty variable,
 a missing file, a path escaping the fixture root, or a hash mismatch fails.
-CI selects the 16 consumer cases and requires all 16 to pass without skips.
+CI selects the `plugin_identity` suite and requires every selected case to pass without skips.
 
 ## Embedded SeekDB
 
@@ -64,10 +64,8 @@ The tests supply their own small embedding vectors; no model download is needed.
 
 ```bash
 uv sync --dev --extra seekdb --frozen
-uv run --no-sync pytest tests/integration/vector/test_seekdb.py \
-  -m slow -q -rs --durations=5 --junitxml=seekdb-contract-results.xml
-python scripts/assert-required-tests.py seekdb-contract-results.xml \
-  --module tests.integration.vector.test_seekdb --expected-tests 2
+uv run --no-sync pytest --required-suite seekdb \
+  -q -rs --durations=5 --junitxml=seekdb-contract-results.xml
 ```
 
 This isolated Linux job keeps native dependency startup failures visible instead
@@ -87,7 +85,7 @@ CI builds the official [`langbot-app/langbot-cli` v0.2.0 source](https://github.
 pinned to commit `3bd20eecd4771b1fe5d0dca61d8b5523182e7bd9`, using Go 1.26.8.
 Module versions and checksums come from that checkout's `go.mod` and `go.sum`.
 The job verifies modules and builds with `CGO_ENABLED=0`, `-mod=readonly`, and
-`-trimpath`. It requires exactly one passing JUnit case and rejects skips.
+`-trimpath`. It requires the CLI HTTP behavior contract and rejects skips.
 
 To run locally with an official pinned CLI checkout and Go 1.26.8:
 
@@ -101,12 +99,35 @@ go build -mod=readonly -trimpath -o ./bin/lbctl ./cmd/lbctl
 export LANGBOT_CLI_BIN="$(pwd)/bin/lbctl"
 
 # In the LangBot Core checkout:
-uv run --frozen pytest \
-  tests/unit_tests/api/service/test_cli_diagnostics.py::test_compiled_cli_against_core_http \
-  -q --tb=short -rs
+uv run --frozen pytest --required-suite compiled_cli -q --tb=short -rs
 ```
 
 Without `LANGBOT_CLI_BIN`, this test remains an explicit local skip. When set, it
 must point to a real compiled executable. CI supplies an absolute path and treats
 a missing binary, command failure, or skipped case as a failure. Update the source
 pin deliberately and rerun the contract before accepting a newer CLI version.
+
+## Required suite policy
+
+CI provisions dependencies and invokes `pytest --required-suite SUITE`. The
+registered suites are `postgres`, `local_agent`, `security`, `compiled_cli`,
+`plugin_identity`, and `seekdb`. Tests carry explicit suite markers; mixed
+SQLite/PostgreSQL fixtures mark only their PostgreSQL parameter. New marked
+cases are discovered automatically, without editing workflow file lists or counts.
+
+`tests/utils/required_suites.py` lists the stable behavior obligations for each
+suite. `@pytest.mark.contract('name')` identifies the behavior a case exercises.
+Only a new or retired behavior obligation changes this policy; renaming, moving,
+or parameterizing an existing test does not. Unmarked sibling tests are not run
+by a required suite.
+
+Required execution fails if a behavior obligation is absent, no cases are
+selected, or any selected case fails, errors, skips, or xfails (including fixture
+setup/teardown). Partial `-k`, `-m`, and `--deselect` filters are rejected.
+`--collect-only` validates selection without claiming execution. Without
+`--required-suite`, optional local prerequisites may still skip as before.
+JUnit remains an artifact, not a second source of the expected test inventory.
+
+These checks prevent missing execution; they cannot prove that a behavior marker
+is truthful or that an assertion detects defects. Review and mutation checks
+still establish assertion quality.
diff --git a/pytest.ini b/pytest.ini
index d86cbba9f..8f543757a 100644
--- a/pytest.ini
+++ b/pytest.ini
@@ -28,6 +28,13 @@ addopts =
 
 # Markers
 markers =
+    contract(name): stable behavior obligation for required suites
+    postgres: required postgres execution suite
+    local_agent: required local_agent execution suite
+    security: required security execution suite
+    compiled_cli: required compiled_cli execution suite
+    plugin_identity: required plugin_identity execution suite
+    seekdb: required seekdb execution suite
     asyncio: mark test as async
     unit: mark test as unit test
     integration: mark test as integration test
diff --git a/scripts/assert-required-tests.py b/scripts/assert-required-tests.py
deleted file mode 100644
index 2ff1f6ced..000000000
--- a/scripts/assert-required-tests.py
+++ /dev/null
@@ -1,39 +0,0 @@
-"""Fail closed when a provisioned dependency-contract lane skips or loses tests."""
-
-from __future__ import annotations
-
-import argparse
-from pathlib import Path
-import xml.etree.ElementTree as ET
-
-
-def verify_report(path: Path, expected_module: str, expected_tests: int, name_prefix: str = '') -> int:
-    cases = list(ET.parse(path).getroot().iter('testcase'))
-    if len(cases) != expected_tests:
-        raise ValueError(f'Expected {expected_tests} tests, found {len(cases)} in {path}')
-    for case in cases:
-        if case.get('classname') != expected_module:
-            raise ValueError(f'Unexpected test module: {case.get("classname")!r}')
-        if not case.get('name', '').startswith(name_prefix):
-            raise ValueError(f'Unexpected test name: {case.get("name")!r}')
-        for outcome in ('skipped', 'failure', 'error'):
-            if case.find(outcome) is not None:
-                raise ValueError(f'Required test {case.get("name")!r} reported {outcome}')
-    return len(cases)
-
-
-def main() -> None:
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument('report', type=Path)
-    parser.add_argument('--module', required=True)
-    parser.add_argument('--expected-tests', type=int, required=True)
-    parser.add_argument('--name-prefix', default='')
-    args = parser.parse_args()
-    if args.expected_tests < 1:
-        parser.error('--expected-tests must be positive')
-    count = verify_report(args.report, args.module, args.expected_tests, args.name_prefix)
-    print(f'{args.module}: {count} required tests passed, 0 skipped')
-
-
-if __name__ == '__main__':
-    main()
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 000000000..8e63694e3
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,3 @@
+"""Repository-wide test execution policy."""
+
+pytest_plugins = ['tests.utils.required_suites']
diff --git a/tests/e2e/test_local_runner_fake_provider.py b/tests/e2e/test_local_runner_fake_provider.py
index 196b185e4..cbb953bad 100644
--- a/tests/e2e/test_local_runner_fake_provider.py
+++ b/tests/e2e/test_local_runner_fake_provider.py
@@ -26,7 +26,7 @@ import pytest
 from tests.e2e.utils.config_factory import create_minimal_config, create_test_directories
 from tests.utils.legacy_plugin_contract import verified_plugin_root
 
-pytestmark = pytest.mark.e2e
+pytestmark = [pytest.mark.e2e, pytest.mark.local_agent]
 
 
 LOCAL_RUNNER_ID = 'plugin:langbot-team/LocalAgent/default'
@@ -557,6 +557,7 @@ def _run_local_agent_probe(tmpdir: Path, probe):
     return asyncio.run(_run())
 
 
+@pytest.mark.contract('agent_ledger')
 def test_local_runner_uses_host_fake_provider_and_persists_ledger(
     local_agent_e2e_tmpdir,
     local_agent_e2e_config_path,
@@ -619,6 +620,7 @@ def test_local_runner_uses_host_fake_provider_and_persists_ledger(
         conn.close()
 
 
+@pytest.mark.contract('agent_tools')
 def test_local_runner_executes_authorized_tool_loop_through_host_action(
     local_agent_e2e_tmpdir,
     local_agent_e2e_config_path,
@@ -688,6 +690,7 @@ def test_local_runner_executes_authorized_tool_loop_through_host_action(
         conn.close()
 
 
+@pytest.mark.contract('agent_rag')
 def test_local_runner_retrieves_authorized_rag_context_through_host_action(
     local_agent_e2e_tmpdir,
     local_agent_e2e_config_path,
@@ -762,6 +765,7 @@ def test_local_runner_retrieves_authorized_rag_context_through_host_action(
         conn.close()
 
 
+@pytest.mark.contract('agent_compaction')
 def test_local_runner_compacts_history_and_persists_checkpoint(
     local_agent_e2e_tmpdir,
     local_agent_e2e_config_path,
@@ -855,6 +859,7 @@ def test_local_runner_compacts_history_and_persists_checkpoint(
         conn.close()
 
 
+@pytest.mark.contract('agent_multi_turn')
 def test_local_runner_combines_rag_compaction_and_multi_turn_tool_loop(
     local_agent_e2e_tmpdir,
     local_agent_e2e_config_path,
@@ -1017,6 +1022,7 @@ def test_local_runner_combines_rag_compaction_and_multi_turn_tool_loop(
         conn.close()
 
 
+@pytest.mark.contract('agent_box')
 def test_local_runner_owns_box_reuse_and_explicit_files(
     local_agent_e2e_tmpdir,
     local_agent_e2e_config_path,
diff --git a/tests/integration/persistence/test_migration_branch_convergence.py b/tests/integration/persistence/test_migration_branch_convergence.py
index 903b2794a..58e4a5796 100644
--- a/tests/integration/persistence/test_migration_branch_convergence.py
+++ b/tests/integration/persistence/test_migration_branch_convergence.py
@@ -39,7 +39,7 @@ ACCOUNT = '41100000-0000-4000-8000-000000000002'
 NOW = datetime.datetime(2026, 9, 1)
 
 
-@pytest.fixture(params=['sqlite', pytest.param('postgresql', marks=pytest.mark.slow)])
+@pytest.fixture(params=['sqlite', pytest.param('postgresql', marks=[pytest.mark.slow, pytest.mark.postgres])])
 async def convergence_engine(request, tmp_path):
     if request.param == 'sqlite':
         engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "convergence.db"}')
@@ -354,6 +354,7 @@ async def _assert_merged_schema_and_data(engine, source):
         ('beta', None),
     ],
 )
+@pytest.mark.contract('migration_history')
 async def test_historical_schema_and_populated_branch_upgrade(convergence_engine, source, checkpoint):
     engine = convergence_engine
     fixture = json.loads((FIXTURES / f'{source}_schema.json').read_text())
@@ -395,6 +396,7 @@ async def test_historical_schema_and_populated_branch_upgrade(convergence_engine
 
 
 @pytest.mark.asyncio
+@pytest.mark.contract('migration_fresh_schema')
 async def test_empty_database_startup_schema_then_real_migrations(convergence_engine):
     engine = convergence_engine
     async with engine.begin() as conn:
diff --git a/tests/integration/persistence/test_pipeline_admission_postgres.py b/tests/integration/persistence/test_pipeline_admission_postgres.py
index 6c95eae63..865a158eb 100644
--- a/tests/integration/persistence/test_pipeline_admission_postgres.py
+++ b/tests/integration/persistence/test_pipeline_admission_postgres.py
@@ -12,7 +12,11 @@ from langbot.pkg.persistence.pipeline_admission import lock_pipeline_admission
 from langbot.pkg.entity.persistence.agent_interaction import AgentInteraction
 
 URL = os.environ.get('LANGBOT_ADMISSION_TEST_URL')
-pytestmark = [pytest.mark.asyncio, pytest.mark.skipif(not URL, reason='disposable PostgreSQL URL required')]
+pytestmark = [
+    pytest.mark.postgres,
+    pytest.mark.asyncio,
+    pytest.mark.skipif(not URL, reason='disposable PostgreSQL URL required'),
+]
 
 
 @pytest.fixture
@@ -111,6 +115,7 @@ async def test_real_writer_waits_and_revalidates_after_transaction(env, outcome)
         assert (await writer)[0]['status'] == 'pending'
 
 
+@pytest.mark.contract('admission_isolation')
 async def test_real_writer_first_blocks_migration_and_other_tenant_isolated(env):
     await write(env)
     async with env.pm.tenant_uow(base.WS) as uow:
diff --git a/tests/integration/persistence/test_rag_document_identity.py b/tests/integration/persistence/test_rag_document_identity.py
index 5a729dd87..21cc6d7dc 100644
--- a/tests/integration/persistence/test_rag_document_identity.py
+++ b/tests/integration/persistence/test_rag_document_identity.py
@@ -54,7 +54,7 @@ DOCUMENT_IDENTITY_REVISION = '0025_rag_document_identity'
 CONTEXT = ExecutionContext(instance_uuid='instance-a', workspace_uuid='workspace-a', placement_generation=5)
 
 
-@pytest_asyncio.fixture(params=['sqlite', 'postgres'])
+@pytest_asyncio.fixture(params=['sqlite', pytest.param('postgres', marks=pytest.mark.postgres)])
 async def database(request, tmp_path):
     admin = None
     schema = 'ke_identity_' + uuid.uuid4().hex
@@ -183,6 +183,7 @@ async def read_file_row(runtime, file_id='host-id'):
 @pytest.mark.parametrize(
     'engine_id', ['dify-upstream', 'fastgpt-collection', 'ragflow-upstream', 'host-id', ' opaque ID ']
 )
+@pytest.mark.contract('rag_identity')
 async def test_ingest_persists_exact_engine_identity_and_restart_delete(runtime, engine_id):
     file = await seed(runtime)
     runtime.ap.plugin_connector.call_rag_ingest.return_value['document_id'] = engine_id
diff --git a/tests/integration/persistence/test_rag_interrupted_ingestion.py b/tests/integration/persistence/test_rag_interrupted_ingestion.py
index aa6b53dc6..8e6d77c26 100644
--- a/tests/integration/persistence/test_rag_interrupted_ingestion.py
+++ b/tests/integration/persistence/test_rag_interrupted_ingestion.py
@@ -25,6 +25,7 @@ def attach_task_manager(runtime):
 
 
 @pytest.mark.asyncio
+@pytest.mark.contract('rag_recovery')
 async def test_cancelled_ingestion_retains_recovery_resources(runtime):
     """Local cancellation is interrupted observation, not remote quiescence."""
     file = await seed(runtime)
diff --git a/tests/integration/persistence/test_runner_timestamps_postgres.py b/tests/integration/persistence/test_runner_timestamps_postgres.py
index 6cc988b7b..732b40490 100644
--- a/tests/integration/persistence/test_runner_timestamps_postgres.py
+++ b/tests/integration/persistence/test_runner_timestamps_postgres.py
@@ -25,7 +25,12 @@ from langbot.pkg.entity.persistence.base import Base
 from langbot.pkg.entity.persistence.event_log import EventLog
 from langbot.pkg.entity.persistence.transcript import Transcript
 
-pytestmark = [pytest.mark.integration, pytest.mark.slow, pytest.mark.asyncio]
+pytestmark = [
+    pytest.mark.postgres,
+    pytest.mark.integration,
+    pytest.mark.slow,
+    pytest.mark.asyncio,
+]
 
 UTC = dt.timezone.utc
 NOW = dt.datetime(2026, 9, 22, 4, 5, 6, 123000, tzinfo=UTC)
@@ -110,6 +115,7 @@ async def _legacy_run(engine, *, status='queued'):
         )
 
 
+@pytest.mark.contract('runner_timestamps')
 async def test_run_lifecycle_deadline_events_and_stats(pg_engine):
     store = run_ledger_store.RunLedgerStore(pg_engine)
     deadline = NOW + dt.timedelta(minutes=5)
diff --git a/tests/integration/vector/test_seekdb.py b/tests/integration/vector/test_seekdb.py
index 668072aba..6cf919bde 100644
--- a/tests/integration/vector/test_seekdb.py
+++ b/tests/integration/vector/test_seekdb.py
@@ -14,16 +14,14 @@ import uuid
 
 import pytest
 
-pytest.importorskip('pyseekdb')
-
-from langbot.pkg.vector.vdbs.seekdb import SeekDBVectorDatabase
-
-
-pytestmark = [pytest.mark.integration, pytest.mark.slow]
+pytestmark = [pytest.mark.integration, pytest.mark.slow, pytest.mark.seekdb]
 
 
 @pytest.fixture
 async def backend(tmp_path):
+    pytest.importorskip('pyseekdb')
+    from langbot.pkg.vector.vdbs.seekdb import SeekDBVectorDatabase
+
     app = SimpleNamespace(
         instance_config=SimpleNamespace(
             data={
@@ -51,6 +49,7 @@ async def backend(tmp_path):
 
 
 @pytest.mark.asyncio
+@pytest.mark.contract('vector_round_trip')
 async def test_upsert_and_text_round_trip(backend) -> None:
     database, collection = backend
     original = 'He said "hello".\nC:\\notes\\file.txt isn\'t empty. 中文'
@@ -80,6 +79,7 @@ async def test_upsert_and_text_round_trip(backend) -> None:
 
 
 @pytest.mark.asyncio
+@pytest.mark.contract('hybrid_relevance')
 async def test_full_text_and_hybrid_results_keep_relevance_order(backend) -> None:
     database, collection = backend
     documents = [
diff --git a/tests/test_cwe94_debug_exec.py b/tests/test_cwe94_debug_exec.py
index b31085981..ba91931a6 100644
--- a/tests/test_cwe94_debug_exec.py
+++ b/tests/test_cwe94_debug_exec.py
@@ -12,12 +12,17 @@ This test verifies that:
 import ast
 import pathlib
 
+import pytest
+
+pytestmark = pytest.mark.security
+
 # Resolve project root (one level up from tests/)
 _PROJECT_ROOT = pathlib.Path(__file__).resolve().parent.parent
 
 VULN_FILE = _PROJECT_ROOT / 'src' / 'langbot' / 'pkg' / 'api' / 'http' / 'controller' / 'groups' / 'system.py'
 
 
+@pytest.mark.contract('debug_exec_absent')
 def test_no_exec_call_in_system_controller():
     """Verify there is no exec() call in system.py that takes user input."""
     with open(VULN_FILE, 'r') as f:
@@ -38,6 +43,7 @@ def test_no_exec_call_in_system_controller():
     )
 
 
+@pytest.mark.contract('debug_route_absent')
 def test_no_debug_exec_route():
     """Verify the /debug/exec route is not registered."""
     with open(VULN_FILE, 'r') as f:
diff --git a/tests/unit_tests/agent/test_legacy_identity_boundary.py b/tests/unit_tests/agent/test_legacy_identity_boundary.py
index 216ade894..d5347acfb 100644
--- a/tests/unit_tests/agent/test_legacy_identity_boundary.py
+++ b/tests/unit_tests/agent/test_legacy_identity_boundary.py
@@ -205,14 +205,43 @@ def official_plugin_root():
 @pytest.mark.parametrize(
     'runner,folder,method,error_name,local_identity',
     [
-        ('DifyAgent', 'dify-agent', '_get_user_tag', 'DifyConfigError', 'person_Session_007'),
-        ('N8nAgent', 'n8n-agent', '_get_user_tag', 'N8nConfigError', 'person_Session_007'),
-        ('CozeAgent', 'coze-agent', '_get_user_id', 'CozeConfigError', 'person_launcher-123'),
-        ('TboxAgent', 'tbox-agent', '_get_user_id', 'TboxConfigError', 'bot-uuid-123'),
+        pytest.param(
+            'DifyAgent',
+            'dify-agent',
+            '_get_user_tag',
+            'DifyConfigError',
+            'person_Session_007',
+            marks=pytest.mark.contract('dify_identity'),
+        ),
+        pytest.param(
+            'N8nAgent',
+            'n8n-agent',
+            '_get_user_tag',
+            'N8nConfigError',
+            'person_Session_007',
+            marks=pytest.mark.contract('n8n_identity'),
+        ),
+        pytest.param(
+            'CozeAgent',
+            'coze-agent',
+            '_get_user_id',
+            'CozeConfigError',
+            'person_launcher-123',
+            marks=pytest.mark.contract('coze_identity'),
+        ),
+        pytest.param(
+            'TboxAgent',
+            'tbox-agent',
+            '_get_user_id',
+            'TboxConfigError',
+            'bot-uuid-123',
+            marks=pytest.mark.contract('tbox_identity'),
+        ),
     ],
 )
 @pytest.mark.parametrize('missing', [False, True], ids=['trusted-identity', 'missing-identity'])
 @pytest.mark.parametrize('profile', ['dedicated', 'shared'])
+@pytest.mark.plugin_identity
 async def test_real_plugin_identity_helper_with_host_payload(
     mock_query, monkeypatch, official_plugin_root, runner, folder, method, error_name, local_identity, missing, profile
 ):
diff --git a/tests/unit_tests/api/service/test_cli_diagnostics.py b/tests/unit_tests/api/service/test_cli_diagnostics.py
index 95b85e709..a78965f5b 100644
--- a/tests/unit_tests/api/service/test_cli_diagnostics.py
+++ b/tests/unit_tests/api/service/test_cli_diagnostics.py
@@ -236,6 +236,8 @@ async def test_extension_discovery_preserves_empty_bindings_and_redacts_plugins(
     ap.plugin_connector.require_workspace_context.assert_awaited_once_with(request_context())
 
 
+@pytest.mark.compiled_cli
+@pytest.mark.contract('cli_http')
 async def test_compiled_cli_against_core_http(service, tmp_path):  # noqa: F811
     """Optional cross-repo check: real HTTP/SQLite/Pipeline, synthetic runtime providers."""
     import json
diff --git a/tests/unit_tests/api/service/test_monitoring_sessions.py b/tests/unit_tests/api/service/test_monitoring_sessions.py
index 868f51589..891171ba3 100644
--- a/tests/unit_tests/api/service/test_monitoring_sessions.py
+++ b/tests/unit_tests/api/service/test_monitoring_sessions.py
@@ -21,6 +21,8 @@ pytestmark = pytest.mark.asyncio
 
 
 @pytest.mark.asyncio(loop_scope='module')
+@pytest.mark.postgres
+@pytest.mark.contract('monitoring_sessions')
 async def test_postgres_upgrade_rls_and_concurrent_bot_counts(cloud_database):  # noqa: F811
     import asyncio
     import importlib
diff --git a/tests/unit_tests/quality/test_required_contract_report.py b/tests/unit_tests/quality/test_required_contract_report.py
index 0c7c5cab5..79add8c29 100644
--- a/tests/unit_tests/quality/test_required_contract_report.py
+++ b/tests/unit_tests/quality/test_required_contract_report.py
@@ -1,72 +1,128 @@
-"""The optional-dependency CI guard must reject a false-green JUnit report."""
+"""Exercise the required-suite policy through real pytest collection and execution."""
 
-from __future__ import annotations
-
-import importlib.util
 from pathlib import Path
+import os
 import subprocess
 import sys
 
 import pytest
 
 
-SCRIPT = Path(__file__).resolve().parents[3] / 'scripts' / 'assert-required-tests.py'
-SPEC = importlib.util.spec_from_file_location('required_contract_report', SCRIPT)
-assert SPEC is not None and SPEC.loader is not None
-REPORT = importlib.util.module_from_spec(SPEC)
-SPEC.loader.exec_module(REPORT)
+ROOT = Path(__file__).resolve().parents[3]
+PASSING = """
+import pytest
+
+@pytest.mark.security
+@pytest.mark.contract('debug_exec_absent')
+@pytest.mark.parametrize('value', [1, 2, 3])
+def test_renamed_behavior(value):
+    assert value > 0
+
+@pytest.mark.security
+@pytest.mark.contract('debug_route_absent')
+def test_another_name():
+    pass
+
+def test_unrelated():
+    assert False
+"""
 
 
-def write_report(tmp_path, cases):
-    path = tmp_path / 'report.xml'
-    path.write_text(f'<testsuites><testsuite>{cases}</testsuite></testsuites>')
-    return path
-
-
-def test_accepts_exact_passing_cases(tmp_path):
-    path = write_report(
-        tmp_path, '<testcase classname="contracts" name="first"/><testcase classname="contracts" name="second"/>'
+def run_suite(tmp_path, source, *args):
+    (tmp_path / 'moved').mkdir(exist_ok=True)
+    (tmp_path / 'moved/test_any_name.py').write_text(source)
+    (tmp_path / 'pytest.ini').write_text('[pytest]\nmarkers =\n    security: suite\n    contract(name): behavior\n')
+    return subprocess.run(
+        [
+            sys.executable,
+            '-m',
+            'pytest',
+            '-p',
+            'tests.utils.required_suites',
+            '--required-suite',
+            'security',
+            '-q',
+            *args,
+        ],
+        cwd=tmp_path,
+        env={**os.environ, 'PYTHONPATH': str(ROOT)},
+        capture_output=True,
+        text=True,
+        timeout=30,
     )
-    assert REPORT.verify_report(path, 'contracts', 2) == 2
 
 
-@pytest.mark.parametrize('outcome', ['skipped', 'failure', 'error'])
-def test_rejects_every_nonpassing_outcome(tmp_path, outcome):
-    path = write_report(tmp_path, f'<testcase classname="contracts" name="required"><{outcome}/></testcase>')
-    with pytest.raises(ValueError, match=outcome):
-        REPORT.verify_report(path, 'contracts', 1)
+def test_renames_moves_and_parameter_growth_need_no_ci_edit(tmp_path):
+    result = run_suite(tmp_path, PASSING)
+    assert result.returncode == 0, result.stdout + result.stderr
+    assert '4 passed, 1 deselected' in result.stdout
+    result = run_suite(tmp_path, PASSING.replace('[1, 2, 3]', '[1, 2, 3, 4]'))
+    assert result.returncode == 0, result.stdout + result.stderr
+    assert '5 passed, 1 deselected' in result.stdout
 
 
-@pytest.mark.parametrize('count', [0, 2])
-def test_rejects_changed_case_count(tmp_path, count):
-    path = write_report(tmp_path, '<testcase classname="contracts" name="case"/>' * count)
-    with pytest.raises(ValueError, match='Expected 1 tests'):
-        REPORT.verify_report(path, 'contracts', 1)
+@pytest.mark.parametrize(
+    'statement',
+    [
+        "pytest.skip('dependency unavailable')",
+        "pytest.xfail('known bug')",
+        'assert False',
+        "raise RuntimeError('broken')",
+    ],
+)
+def test_required_cases_must_actually_pass(tmp_path, statement):
+    result = run_suite(tmp_path, PASSING.replace('    pass', f'    {statement}'))
+    assert result.returncode != 0
+    assert 'all selected cases must pass' in result.stdout
 
 
-def test_rejects_wrong_module(tmp_path):
-    path = write_report(tmp_path, '<testcase classname="unrelated" name="case"/>')
-    with pytest.raises(ValueError, match='Unexpected test module'):
-        REPORT.verify_report(path, 'contracts', 1)
+def test_unexpected_xpass_is_not_a_required_pass(tmp_path):
+    source = PASSING.replace('def test_another_name():', '@pytest.mark.xfail(strict=False)\ndef test_another_name():')
+    result = run_suite(tmp_path, source)
+    assert result.returncode != 0
+    assert 'all selected cases must pass' in result.stdout
 
 
-def test_rejects_missing_report(tmp_path):
-    with pytest.raises(FileNotFoundError):
-        REPORT.verify_report(tmp_path / 'missing.xml', 'contracts', 1)
+def test_missing_behavior_cannot_be_replaced_by_more_unrelated_cases(tmp_path):
+    source = PASSING.replace("@pytest.mark.contract('debug_route_absent')", '')
+    result = run_suite(tmp_path, source)
+    assert result.returncode != 0
+    assert "missing contracts: ['debug_route_absent']" in result.stderr
 
 
-def test_command_line_success_and_failure(tmp_path):
-    path = write_report(tmp_path, '<testcase classname="contracts" name="case"/>')
-    command = [sys.executable, str(SCRIPT), str(path), '--module', 'contracts', '--expected-tests']
-    good = subprocess.run([*command, '1'], capture_output=True, text=True)
-    assert good.returncode == 0
-    assert '1 required tests passed, 0 skipped' in good.stdout
-    bad = subprocess.run([*command, '0'], capture_output=True, text=True)
-    assert bad.returncode != 0
-    assert 'must be positive' in bad.stderr
+def test_empty_suite_fails(tmp_path):
+    result = run_suite(tmp_path, 'def test_unrelated(): pass')
+    assert result.returncode != 0
+    assert 'missing contracts' in result.stderr
 
 
-def test_rejects_wrong_function(tmp_path):
-    path = write_report(tmp_path, '<testcase classname="contracts" name="unrelated"/>')
-    with pytest.raises(ValueError, match='Unexpected test name'):
-        REPORT.verify_report(path, 'contracts', 1, 'test_required')
+@pytest.mark.parametrize('args', [('-k', 'renamed'), ('-m', 'security'), ('--deselect', 'moved/test_any_name.py')])
+def test_required_suite_rejects_partial_filters(tmp_path, args):
+    result = run_suite(tmp_path, PASSING, *args)
+    assert result.returncode != 0
+    assert 'cannot be combined' in result.stderr
+
+
+@pytest.mark.parametrize('phase', ['setup', 'teardown'])
+def test_fixture_skips_also_fail(tmp_path, phase):
+    fixture = '\n@pytest.fixture(autouse=True)\ndef dependency():\n'
+    if phase == 'teardown':
+        fixture += '    yield\n'
+    fixture += "    pytest.skip('dependency unavailable')\n"
+    result = run_suite(tmp_path, PASSING + fixture)
+    assert result.returncode != 0
+    assert 'all selected cases must pass' in result.stdout
+
+
+def test_optional_local_execution_still_allows_skips(tmp_path):
+    (tmp_path / 'test_optional.py').write_text("import pytest\ndef test_optional(): pytest.skip('no database')")
+    result = subprocess.run(
+        [sys.executable, '-m', 'pytest', '-p', 'tests.utils.required_suites', '-q'],
+        cwd=tmp_path,
+        env={**os.environ, 'PYTHONPATH': str(ROOT)},
+        capture_output=True,
+        text=True,
+        timeout=30,
+    )
+    assert result.returncode == 0, result.stdout + result.stderr
+    assert '1 skipped' in result.stdout
diff --git a/tests/utils/required_suites.py b/tests/utils/required_suites.py
new file mode 100644
index 000000000..d752e10c6
--- /dev/null
+++ b/tests/utils/required_suites.py
@@ -0,0 +1,71 @@
+"""Required suites select explicit markers and enforce execution, not case counts."""
+
+import pytest
+
+
+# These are behavior obligations, independent of file names and parametrization.
+REQUIRED_CONTRACTS = {
+    'postgres': {
+        'admission_isolation',
+        'rag_identity',
+        'rag_recovery',
+        'migration_history',
+        'migration_fresh_schema',
+        'runner_timestamps',
+        'monitoring_sessions',
+    },
+    'local_agent': {'agent_ledger', 'agent_tools', 'agent_rag', 'agent_compaction', 'agent_multi_turn', 'agent_box'},
+    'security': {'debug_exec_absent', 'debug_route_absent'},
+    'compiled_cli': {'cli_http'},
+    'plugin_identity': {'dify_identity', 'n8n_identity', 'coze_identity', 'tbox_identity'},
+    'seekdb': {'vector_round_trip', 'hybrid_relevance'},
+}
+
+
+def pytest_addoption(parser):
+    parser.addoption('--required-suite', choices=tuple(REQUIRED_CONTRACTS), help='Run a mandatory dependency suite')
+
+
+def pytest_configure(config):
+    suite = config.getoption('--required-suite')
+    if suite:
+        if config.option.markexpr or config.option.keyword or config.option.deselect:
+            raise pytest.UsageError('--required-suite cannot be combined with -m, -k or --deselect')
+        config.option.markexpr = suite
+        config.pluginmanager.register(RequiredSuite(suite), 'required-suite-policy')
+
+
+class RequiredSuite:
+    def __init__(self, suite):
+        self.suite = suite
+        self.items = {}
+        self.passed = set()
+        self.unsuccessful = set()
+
+    def pytest_collection_finish(self, session):
+        for item in session.items:
+            self.items[item.nodeid] = {mark.args[0] for mark in item.iter_markers('contract')}
+        covered = set().union(*self.items.values())
+        missing = REQUIRED_CONTRACTS[self.suite] - covered
+        if missing:
+            raise pytest.UsageError(f'Required suite {self.suite} is missing contracts: {sorted(missing)}')
+
+    def pytest_runtest_logreport(self, report):
+        if report.nodeid not in self.items:
+            return
+        if report.failed or report.skipped or hasattr(report, 'wasxfail'):
+            self.unsuccessful.add(report.nodeid)
+        if report.when == 'call' and report.passed:
+            self.passed.add(report.nodeid)
+
+    def pytest_sessionfinish(self, session, exitstatus):
+        if session.config.option.collectonly:
+            return
+        incomplete = set(self.items) - self.passed
+        if exitstatus == 0 and (incomplete or self.unsuccessful or not self.items):
+            session.exitstatus = pytest.ExitCode.TESTS_FAILED
+        reporter = session.config.pluginmanager.get_plugin('terminalreporter')
+        if reporter and (incomplete or self.unsuccessful):
+            reporter.write_sep('=', f'Required suite {self.suite}: all selected cases must pass without skips/xfails')
+            for nodeid in sorted(incomplete | self.unsuccessful):
+                reporter.write_line(nodeid)