diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index cd54a3c0..881b134a 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -85,3 +85,24 @@ jobs:
           else
             echo "::notice::'make api-check' not defined yet (XWING-2208) — skipping"
           fi
+
+  ratchet:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: astral-sh/setup-uv@v6
+        with:
+          enable-cache: true
+          python-version: "3.12"
+      - run: uv sync --frozen
+      # Drift ratchet (XWING-2210): fails on new hand-wiring; the baseline only
+      # shrinks - compared against the base branch's baseline, so a PR cannot
+      # raise its own to cover new hand-wiring.
+      - run: |
+          BASE="${{ github.base_ref || 'development' }}"
+          git fetch --depth=1 origin "$BASE"
+          REF=""
+          if git show "origin/$BASE:.ratchet.json" > /tmp/ratchet-reference.json 2>/dev/null; then
+            REF="--reference /tmp/ratchet-reference.json"
+          fi
+          uv run python -m traust_core.v1.checks.ratchet src --baseline .ratchet.json $REF
diff --git a/.ratchet.json b/.ratchet.json
new file mode 100644
index 00000000..2e5aa476
--- /dev/null
+++ b/.ratchet.json
@@ -0,0 +1,9 @@
+{
+  "src/traust_core/v1/providers/clock.py": {
+    "clock": 1
+  },
+  "src/traust_core/v1/providers/process.py": {
+    "env": 1,
+    "subprocess": 1
+  }
+}
diff --git a/AGENTS.md b/AGENTS.md
index d3c69f4c..f90b0d43 100644
--- a/AGENTS.md
+++ b/AGENTS.md
@@ -121,3 +121,17 @@ Python 3.11+ library. The platform API lives in `traust_core.v1`. Overview: [REA
 - Smallest change that does the job; no drive-by refactors
 - A new public module needs a README row and contract tests before merge
 - Lint: line length 100; ruff E/W/F/I/UP/B/SIM/PTH/RUF
+
+## Where we're headed
+
+- Inside traust-core, new code uses the platform pieces below. Outside traust-core, everything goes through the platform API (`traust-client` or the SDK); old code is deleted by its domain's cut-over plan
+- CI's drift ratchet (`python -m traust_core.v1.checks.ratchet`) fails a new hit (`path:line kind → use <API>`); baselines only shrink
+
+| Hand-wiring | Inside traust-core, use |
+| --- | --- |
+| `sqlite3.connect` / `psycopg.connect` | `SqlUnitOfWork` / `SqlRepository` |
+| `subprocess` | `ProcessRunner` or a tool provider |
+| `os.getenv` / `os.environ` | `Config` / `Context` |
+| `json.dump` / `write_text(json…)` | `ObjectStore` / `ArtifactPublisher` |
+| `datetime.now()` / `datetime.utcnow()` | `Clock` |
+| `requests` / `httpx` / `urllib.request` | a core provider (`SourceProvider`, `FeedProvider`, tool provider) |
diff --git a/src/traust_core/v1/checks/__init__.py b/src/traust_core/v1/checks/__init__.py
new file mode 100644
index 00000000..1cd4e1bf
--- /dev/null
+++ b/src/traust_core/v1/checks/__init__.py
@@ -0,0 +1,19 @@
+from typing import Any
+
+__all__ = [
+    "Hit",
+    "Result",
+    "Rule",
+    "compare",
+    "counts",
+    "scan",
+    "scan_source",
+]
+
+
+def __getattr__(name: str) -> Any:
+    if name in __all__:
+        from traust_core.v1.checks import ratchet
+
+        return getattr(ratchet, name)
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
diff --git a/src/traust_core/v1/checks/ratchet.py b/src/traust_core/v1/checks/ratchet.py
new file mode 100644
index 00000000..152bdabd
--- /dev/null
+++ b/src/traust_core/v1/checks/ratchet.py
@@ -0,0 +1,287 @@
+"""Drift ratchet over common hand-wiring forms (AST-based).
+
+A finite checker, not a prohibition: it catches the frequent shapes -
+direct calls, keyword arguments, simple concatenation - and the baseline
+only shrinks when CI passes --reference (the base branch's baseline).
+Novel forms are caught by review; add them here when they appear.
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import fnmatch
+import json
+import sys
+from collections.abc import Sequence
+from dataclasses import dataclass, field
+from pathlib import Path
+
+
+@dataclass(frozen=True, slots=True)
+class Rule:
+    kind: str
+    use_instead: str
+    calls: frozenset[str] = field(default_factory=frozenset)
+    import_modules: frozenset[str] = field(default_factory=frozenset)
+    allowed: tuple[str, ...] = ()
+
+
+# AST-based, so a comment or docstring that mentions one of these never counts.
+RULES: tuple[Rule, ...] = (
+    Rule(
+        "connect",
+        "a repository on SqlUnitOfWork",
+        calls=frozenset({"sqlite3.connect", "psycopg.connect"}),
+        allowed=("*/repositories/*", "*/repos/*"),
+    ),
+    Rule(
+        "subprocess",
+        "ProcessRunner or a tool provider",
+        import_modules=frozenset({"subprocess"}),
+    ),
+    Rule(
+        "env",
+        "a field on Config / Context",
+        calls=frozenset({"os.getenv", "os.environ.get", "os.environ.copy"}),
+        allowed=("*/context/*",),
+    ),
+    Rule(
+        "json-write",
+        "ObjectStore / ArtifactPublisher",
+        calls=frozenset({"json.dump"}),
+        # the ratchet writes its own baseline; that is the sanctioned writer
+        allowed=("*/repositories/*", "*/checks/ratchet.py"),
+    ),
+    Rule("clock", "an injected Clock", calls=frozenset({"datetime.now", "datetime.utcnow"})),
+    Rule(
+        "http",
+        "a core provider (SourceProvider, FeedProvider, tool provider)",
+        import_modules=frozenset({"requests", "httpx", "urllib.request"}),
+        # Outbound HTTP lives in providers; add the REST adapter path when it lands.
+        allowed=("*/providers/*",),
+    ),
+)
+
+_USE = {rule.kind: rule.use_instead for rule in RULES}
+
+
+@dataclass(frozen=True, slots=True)
+class Hit:
+    path: str
+    line: int
+    kind: str
+
+
+@dataclass(frozen=True, slots=True)
+class Result:
+    new: tuple[Hit, ...]
+    shrinkable: tuple[tuple[str, str, int, int], ...]
+
+    @property
+    def ok(self) -> bool:
+        return not self.new and not self.shrinkable
+
+
+Counts = dict[str, dict[str, int]]
+
+
+def _dotted(node: ast.expr) -> str | None:
+    parts: list[str] = []
+    cur: ast.expr = node
+    while isinstance(cur, ast.Attribute):
+        parts.append(cur.attr)
+        cur = cur.value
+    if isinstance(cur, ast.Name):
+        parts.append(cur.id)
+        return ".".join(reversed(parts))
+    return None
+
+
+def _aliases(tree: ast.AST) -> dict[str, str]:
+    out: dict[str, str] = {}
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            for a in node.names:
+                out[a.asname or a.name.split(".")[0]] = a.name if a.asname else a.name.split(".")[0]
+        elif isinstance(node, ast.ImportFrom) and node.module:
+            for a in node.names:
+                out[a.asname or a.name] = f"{node.module}.{a.name}"
+    return out
+
+
+def _resolve(dotted: str, aliases: dict[str, str]) -> str:
+    head, _, rest = dotted.partition(".")
+    if head not in aliases:
+        return dotted
+    return f"{aliases[head]}.{rest}" if rest else aliases[head]
+
+
+def _imports(tree: ast.AST) -> list[tuple[int, str]]:
+    found: list[tuple[int, str]] = []
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Import):
+            found += [(node.lineno, a.name) for a in node.names]
+        elif isinstance(node, ast.ImportFrom) and node.module:
+            found.append((node.lineno, node.module))
+            found += [(node.lineno, f"{node.module}.{a.name}") for a in node.names]
+    return found
+
+
+def _module_match(name: str, modules: frozenset[str]) -> bool:
+    return any(name == m or name.startswith(f"{m}.") for m in modules)
+
+
+def _dumps_inside(expr: ast.expr, aliases: dict[str, str]) -> bool:
+    """json.dumps as the written value, alone or concatenated (x + "\n")."""
+    if isinstance(expr, ast.Call):
+        inner = _dotted(expr.func)
+        return inner is not None and _resolve(inner, aliases) == "json.dumps"
+    if isinstance(expr, ast.BinOp):
+        return _dumps_inside(expr.left, aliases) or _dumps_inside(expr.right, aliases)
+    return False
+
+
+def scan_source(source: str) -> list[tuple[int, str]]:
+    """Return (line, kind) hits for one module's source, before path exemptions."""
+    tree = ast.parse(source)
+    aliases = _aliases(tree)
+    hits: list[tuple[int, str]] = []
+
+    for rule in RULES:
+        if rule.import_modules:
+            hits += [
+                (line, rule.kind)
+                for line, mod in _imports(tree)
+                if _module_match(mod, rule.import_modules)
+            ]
+
+    for node in ast.walk(tree):
+        if isinstance(node, ast.Subscript):
+            target = _dotted(node.value)
+            if target and _resolve(target, aliases) == "os.environ":
+                hits.append((node.lineno, "env"))
+        if not isinstance(node, ast.Call):
+            continue
+        func = node.func
+        if isinstance(func, ast.Attribute) and func.attr in {"now", "utcnow"}:
+            recv = _dotted(func.value)
+            if recv and _resolve(recv, aliases).endswith("datetime"):
+                hits.append((node.lineno, "clock"))
+        if isinstance(func, ast.Attribute) and func.attr == "write_text":
+            args = list(node.args[:1]) + [kw.value for kw in node.keywords]
+            if any(_dumps_inside(arg, aliases) for arg in args):
+                hits.append((node.lineno, "json-write"))
+        dotted = _dotted(func)
+        if dotted is None:
+            continue
+        qname = _resolve(dotted, aliases)
+        for rule in RULES:
+            if qname in rule.calls:
+                hits.append((node.lineno, rule.kind))
+    return hits
+
+
+def _exempt(path_posix: str, kind: str) -> bool:
+    patterns = next((r.allowed for r in RULES if r.kind == kind), ())
+    return any(fnmatch.fnmatch(path_posix, pat) for pat in patterns)
+
+
+def scan(roots: Sequence[Path]) -> list[Hit]:
+    hits: list[Hit] = []
+    for root in roots:
+        files = [root] if root.is_file() else sorted(root.rglob("*.py"))
+        for file in files:
+            if "__pycache__" in file.parts:
+                continue
+            posix = file.as_posix()
+            for line, kind in scan_source(file.read_text()):
+                if not _exempt(posix, kind):
+                    hits.append(Hit(posix, line, kind))
+    return hits
+
+
+def counts(hits: Sequence[Hit]) -> Counts:
+    out: Counts = {}
+    for hit in hits:
+        out.setdefault(hit.path, {}).setdefault(hit.kind, 0)
+        out[hit.path][hit.kind] += 1
+    return out
+
+
+def compare(hits: Sequence[Hit], baseline: Counts) -> Result:
+    actual = counts(hits)
+    new: list[Hit] = []
+    for hit in hits:
+        allowed = baseline.get(hit.path, {}).get(hit.kind, 0)
+        if actual[hit.path][hit.kind] > allowed:
+            new.append(hit)
+    shrinkable: list[tuple[str, str, int, int]] = []
+    for path, kinds in baseline.items():
+        for kind, base_n in kinds.items():
+            now_n = actual.get(path, {}).get(kind, 0)
+            if now_n < base_n:
+                shrinkable.append((path, kind, base_n, now_n))
+    return Result(tuple(new), tuple(sorted(shrinkable)))
+
+
+def baseline_growth(baseline: Counts, reference: Counts) -> list[tuple[str, str, int, int]]:
+    """Where a baseline rose above the reference (the base branch's)."""
+    growth: list[tuple[str, str, int, int]] = []
+    for path, kinds in baseline.items():
+        for kind, n in kinds.items():
+            ref_n = reference.get(path, {}).get(kind, 0)
+            if n > ref_n:
+                growth.append((path, kind, ref_n, n))
+    return sorted(growth)
+
+
+def load_baseline(path: Path) -> Counts:
+    if not path.is_file():
+        return {}
+    return json.loads(path.read_text())
+
+
+def write_baseline(path: Path, hits: Sequence[Hit]) -> None:
+    path.write_text(json.dumps(counts(hits), indent=2, sort_keys=True) + "\n")
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(prog="python -m traust_core.v1.checks.ratchet")
+    parser.add_argument("roots", nargs="+", type=Path)
+    parser.add_argument("--baseline", type=Path, default=Path(".ratchet.json"))
+    parser.add_argument("--write-baseline", action="store_true")
+    parser.add_argument("--summary", action="store_true")
+    parser.add_argument(
+        "--reference",
+        type=Path,
+        help="the base branch's baseline; the PR baseline may not grow past it",
+    )
+    args = parser.parse_args(argv)
+
+    hits = scan(args.roots)
+
+    if args.write_baseline:
+        write_baseline(args.baseline, hits)
+        print(f"wrote baseline {args.baseline} ({len(hits)} entries)")
+        return 0
+
+    baseline = load_baseline(args.baseline)
+    result = compare(hits, baseline)
+    for hit in result.new:
+        print(f"{hit.path}:{hit.line} {hit.kind} → use {_USE[hit.kind]}")
+    for path, kind, base_n, now_n in result.shrinkable:
+        print(f"baseline can shrink: {path} {kind} {base_n}→{now_n}")
+    failed = not result.ok
+    if args.reference is not None:
+        reference = load_baseline(args.reference)
+        for path, kind, ref_n, now_n in baseline_growth(baseline, reference):
+            print(f"baseline grew over reference: {path} {kind} {ref_n}→{now_n}")
+            failed = True
+    if args.summary:
+        print(f"{len(hits)} hits across {len(counts(hits))} files")
+    return 1 if failed else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tests/v1/test_ratchet.py b/tests/v1/test_ratchet.py
new file mode 100644
index 00000000..afeec6eb
--- /dev/null
+++ b/tests/v1/test_ratchet.py
@@ -0,0 +1,138 @@
+from pathlib import Path
+
+import pytest
+
+from traust_core.v1.checks import compare, scan, scan_source
+
+
+def _kinds(source: str) -> set[str]:
+    return {kind for _, kind in scan_source(source)}
+
+
+def test_connect_is_flagged() -> None:
+    assert "connect" in _kinds("import sqlite3\nsqlite3.connect('db')\n")
+    assert "connect" in _kinds("from sqlite3 import connect\nconnect('db')\n")
+
+
+def test_subprocess_import_is_flagged() -> None:
+    assert "subprocess" in _kinds("import subprocess\n")
+    assert "subprocess" in _kinds("from subprocess import run\nrun(['ls'])\n")
+
+
+def test_env_access_is_flagged() -> None:
+    assert "env" in _kinds("import os\nos.environ['X']\n")
+    assert "env" in _kinds("import os\nos.getenv('X')\n")
+    assert "env" in _kinds("import os\nos.environ.get('X')\n")
+
+
+def test_json_write_is_flagged() -> None:
+    assert "json-write" in _kinds("import json\njson.dump(d, f)\n")
+    src = "import json\nfrom pathlib import Path\nPath('x').write_text(json.dumps(d))\n"
+    assert "json-write" in _kinds(src)
+
+
+def test_clock_is_flagged() -> None:
+    assert "clock" in _kinds("from datetime import datetime\ndatetime.now()\n")
+    assert "clock" in _kinds("import datetime\ndatetime.datetime.utcnow()\n")
+
+
+def test_http_import_is_flagged() -> None:
+    assert "http" in _kinds("import requests\n")
+    assert "http" in _kinds("import httpx\n")
+    assert "http" in _kinds("import urllib.request\n")
+
+
+def test_docstring_mentioning_a_kind_does_not_count() -> None:
+    assert _kinds('"""This module avoids sqlite3.connect and os.getenv."""\nx = 1\n') == set()
+
+
+def test_repository_file_is_exempt_from_connect(tmp_path: Path) -> None:
+    repo = tmp_path / "repositories" / "sql.py"
+    repo.parent.mkdir(parents=True)
+    repo.write_text("import sqlite3\nsqlite3.connect('db')\n")
+    assert [h.kind for h in scan([tmp_path])] == []
+
+
+def test_provider_file_is_exempt_from_http(tmp_path: Path) -> None:
+    provider = tmp_path / "providers" / "feeds.py"
+    provider.parent.mkdir(parents=True)
+    provider.write_text("import httpx\n")
+    service = tmp_path / "services" / "vex.py"
+    service.parent.mkdir(parents=True)
+    service.write_text("import httpx\n")
+    assert [h.path for h in scan([tmp_path])] == [service.as_posix()]
+
+
+def test_new_hit_fails_against_empty_baseline(tmp_path: Path) -> None:
+    (tmp_path / "m.py").write_text("import subprocess\n")
+    result = compare(scan([tmp_path]), baseline={})
+    assert not result.ok
+    assert [h.kind for h in result.new] == ["subprocess"]
+
+
+def test_dropped_count_is_shrinkable(tmp_path: Path) -> None:
+    (tmp_path / "m.py").write_text("x = 1\n")
+    baseline = {(tmp_path / "m.py").as_posix(): {"subprocess": 2}}
+    result = compare(scan([tmp_path]), baseline)
+    assert not result.ok
+    assert result.shrinkable[0][1:] == ("subprocess", 2, 0)
+
+
+def test_matching_baseline_passes(tmp_path: Path) -> None:
+    (tmp_path / "m.py").write_text("import subprocess\n")
+    hits = scan([tmp_path])
+    baseline = {(tmp_path / "m.py").as_posix(): {"subprocess": 1}}
+    assert compare(hits, baseline).ok
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-q"]))
+
+
+def test_keyword_and_concatenated_json_writes_are_caught():
+    from traust_core.v1.checks import scan_source
+
+    hits = scan_source(
+        "import json\n"
+        "from pathlib import Path\n"
+        'Path("/tmp/x").write_text(json.dumps({"a": 1}) + "\\n")\n'
+        'Path("/tmp/x").write_text(data=json.dumps({"a": 1}))\n'
+    )
+    assert [kind for _, kind in hits] == ["json-write", "json-write"]
+
+
+def test_os_environ_copy_is_caught():
+    from traust_core.v1.checks import scan_source
+
+    hits = scan_source("import os\n\nsettings = os.environ.copy()\n")
+    assert [kind for _, kind in hits] == ["env"]
+
+
+def test_a_pr_cannot_raise_its_own_baseline(tmp_path):
+    import json
+
+    from traust_core.v1.checks.ratchet import main
+
+    root = tmp_path / "roots"
+    root.mkdir()
+    reference = tmp_path / "ref.json"
+    baseline = tmp_path / "baseline.json"
+    reference.write_text(json.dumps({"src/a.py": {"env": 1}}))
+    baseline.write_text(json.dumps({"src/a.py": {"env": 2}}))
+
+    assert main([str(root), "--baseline", str(baseline), "--reference", str(reference)]) == 1
+
+
+def test_reference_check_passes_when_the_baseline_only_shrinks(tmp_path):
+    import json
+
+    from traust_core.v1.checks.ratchet import main
+
+    root = tmp_path / "roots"
+    root.mkdir()
+    reference = tmp_path / "ref.json"
+    baseline = tmp_path / "baseline.json"
+    reference.write_text(json.dumps({"src/a.py": {"env": 1}}))
+    baseline.write_text(json.dumps({"src/a.py": {"env": 0}}))
+
+    assert main([str(root), "--baseline", str(baseline), "--reference", str(reference)]) == 0