diff --git a/.github/workflows/change-control.yml b/.github/workflows/change-control.yml
new file mode 100644
index 00000000..f4c74998
--- /dev/null
+++ b/.github/workflows/change-control.yml
@@ -0,0 +1,36 @@
+---
+name: Change control
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened, labeled, unlabeled, edited]
+  pull_request_review:
+    types: [submitted, dismissed]
+
+concurrency:
+  group: change-control-${{ github.event.pull_request.number }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+  issues: write
+  pull-requests: write
+  checks: read
+  actions: read
+
+jobs:
+  change-control:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v7
+      - name: Run tests
+        run: python3 tools/change_control/self_test.py
+      - name: Check modification records
+        env:
+          GH_TOKEN: ${{ github.token }}
+          PR_NUMBER: ${{ github.event.pull_request.number }}
+          CAN_COMMENT: ${{ github.event.pull_request.head.repo.full_name == github.repository }}
+        run: |-
+          args=()
+          [ "$CAN_COMMENT" = 'true' ] || args+=(--no-comment)
+          python3 -m tools.change_control --repository '${{ github.repository }}' --pr "$PR_NUMBER" "${args[@]}"
diff --git a/docs/process/enforcement-mode b/docs/process/enforcement-mode
new file mode 100644
index 00000000..1ef71804
--- /dev/null
+++ b/docs/process/enforcement-mode
@@ -0,0 +1 @@
+warn
diff --git a/tools/change_control/__main__.py b/tools/change_control/__main__.py
new file mode 100644
index 00000000..e85f201c
--- /dev/null
+++ b/tools/change_control/__main__.py
@@ -0,0 +1,134 @@
+# SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc.
+# SPDX-License-Identifier: Apache-2.0
+"""Run change-control checks against pull-request state obtained through gh api."""
+
+import argparse
+import json
+import shlex
+import subprocess
+import sys
+from pathlib import Path
+
+from tools.safety_lint.model import LintError
+
+from .checks import cr_number, evaluate, load_mode, render_report, upsert_comment
+
+
+class GhApi:
+    """Small fail-closed subprocess adapter around gh api."""
+
+    def __init__(self, command, repository):
+        self.command = shlex.split(command)
+        self.repository = repository
+
+    def __call__(self, method, path, body=None, paginate=False, collection_key=None):
+        endpoint = path.replace('{repo}', self.repository)
+        command = [*self.command, 'api', endpoint]
+        if paginate:
+            command.extend(['--paginate', '--slurp'])
+        if method != 'GET':
+            command.extend(['--method', method])
+        if body:
+            for key, value in body.items():
+                command.extend(['--field', f'{key}={value}'])
+        result = subprocess.run(command, check=False, capture_output=True, text=True)
+        if result.returncode:
+            detail = result.stderr.strip() or 'no error detail'
+            raise RuntimeError(f'gh api failed for {endpoint}: {detail}')
+        try:
+            response = json.loads(result.stdout or '{}')
+        except json.JSONDecodeError as error:
+            raise RuntimeError(f'gh api returned invalid JSON for {endpoint}') from error
+        if not paginate:
+            return response
+        if not isinstance(response, list):
+            raise RuntimeError(f'paginated gh api response is not a page list for {endpoint}')
+        if collection_key:
+            merged = []
+            for page in response:
+                if not isinstance(page, dict) or not isinstance(page.get(collection_key), list):
+                    raise RuntimeError(f'paginated gh api response lacks {collection_key} for {endpoint}')
+                merged.extend(page[collection_key])
+            return {collection_key: merged}
+        if not all(isinstance(page, list) for page in response):
+            raise RuntimeError(f'paginated gh api response contains a non-list page for {endpoint}')
+        return [item for page in response for item in page]
+
+
+def _require(mapping, path):
+    value = mapping
+    for key in path:
+        if not isinstance(value, dict) or key not in value:
+            raise RuntimeError(f'partial GitHub response missing {".".join(path)}')
+        value = value[key]
+    return value
+
+
+def collect(api, repository, pr_number):
+    """Collect the complete GitHub snapshot used by pure policy evaluation."""
+    prefix = f'repos/{repository}'
+    pr = api('GET', f'{prefix}/pulls/{pr_number}')
+    _require(pr, ('user', 'login'))
+    head = _require(pr, ('head', 'sha'))
+    if 'body' not in pr or 'labels' not in pr:
+        raise RuntimeError('partial GitHub response missing PR body or labels')
+    cr = cr_number(pr['body'], repository)
+    issue = api('GET', f'{prefix}/issues/{cr}') if cr else {'labels': [], 'body': ''}
+    comments = api('GET', f'{prefix}/issues/{cr}/comments', paginate=True) if cr else []
+    reviews = api('GET', f'{prefix}/pulls/{pr_number}/reviews', paginate=True)
+    files = api('GET', f'{prefix}/pulls/{pr_number}/files', paginate=True)
+    checks = api('GET', f'{prefix}/commits/{head}/check-runs', paginate=True, collection_key='check_runs')
+    workflows = api(
+        'GET',
+        f'{prefix}/actions/runs?head_sha={head}',
+        paginate=True,
+        collection_key='workflow_runs',
+    )
+    pr_comments = api('GET', f'{prefix}/issues/{pr_number}/comments', paginate=True)
+    for name, value in (('files', files), ('comments', comments), ('reviews', reviews), ('PR comments', pr_comments)):
+        if not isinstance(value, list):
+            raise RuntimeError(f'partial GitHub response: {name} is not a list')
+    return {
+        'repository': repository,
+        'pr': pr,
+        'issue': issue,
+        'issue_comments': comments,
+        'reviews': reviews,
+        'files': files,
+        'check_runs': _require(checks, ('check_runs',)),
+        'workflow_runs': _require(workflows, ('workflow_runs',)),
+        'pr_comments': pr_comments,
+    }
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser()
+    parser.add_argument('--root', default='.')
+    parser.add_argument('--repository', required=True)
+    parser.add_argument('--pr', required=True, type=int)
+    parser.add_argument('--gh', default='gh')
+    parser.add_argument('--no-comment', action='store_true')
+    args = parser.parse_args(argv)
+    try:
+        mode = load_mode(args.root)
+        api = GhApi(args.gh, args.repository)
+        data = collect(api, args.repository, args.pr)
+        results = evaluate(Path(args.root), data)
+        report = render_report(mode, results)
+        print(report)
+        findings = any(item.status == 'fail' for item in results)
+        if not args.no_comment:
+            try:
+                upsert_comment(api, args.pr, report, data['pr_comments'])
+            except RuntimeError as error:
+                print(f'change-control: comment publication warning: {error}', file=sys.stderr)
+                if mode == 'enforce':
+                    return 2
+        return 1 if findings and mode == 'enforce' else 0
+    except (OSError, RuntimeError, ValueError, KeyError, LintError) as error:
+        print(f'change-control: cannot run: {error}', file=sys.stderr)
+        return 2
+
+
+if __name__ == '__main__':
+    sys.exit(main())
diff --git a/tools/change_control/checks.py b/tools/change_control/checks.py
new file mode 100644
index 00000000..82cc033e
--- /dev/null
+++ b/tools/change_control/checks.py
@@ -0,0 +1,385 @@
+# SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc.
+# SPDX-License-Identifier: Apache-2.0
+"""Evaluate existence and ordering of modification records without judging adequacy."""
+
+import re
+from dataclasses import dataclass
+from pathlib import Path
+
+from tools.safety_lint.parse_srs import parse_srs
+
+from .issue_form import parse_issue_form
+
+AUTHORIZERS = ('iliabaranov', 'rajasimman-madhivanan', 'davidt315')
+CLASS_RANK = {'A': 1, 'B': 2, 'C': 3}
+COMMENT_MARKER = '<!-- change-control -->'
+
+
+@dataclass(frozen=True)
+class CheckResult:
+    """One caller-visible policy result."""
+
+    check_id: str
+    status: str
+    message: str
+
+
+def load_mode(root):
+    """Read the auditable mode switch and reject every value except warn or enforce."""
+    path = Path(root) / 'docs/process/enforcement-mode'
+    try:
+        value = path.read_text(encoding='utf-8')
+    except OSError as error:
+        raise RuntimeError(f'cannot read enforcement mode: {error}') from error
+    if value not in ('warn\n', 'enforce\n'):
+        raise RuntimeError('enforcement-mode must contain exactly warn or enforce followed by a newline')
+    return value.strip()
+
+
+def _headings(path):
+    return [
+        line.strip() for line in Path(path).read_text(encoding='utf-8').splitlines() if re.match(r'^#(?:\s|\d)', line)
+    ]
+
+
+def impact_analysis_complete(text, headings):
+    """Check that headings occur in order and each has non-whitespace content beneath it."""
+    positions = []
+    cursor = 0
+    for heading in headings:
+        match = re.search(rf'(?m)^{re.escape(heading)}\s*$', text[cursor:])
+        if not match:
+            return False, [heading]
+        start = cursor + match.start()
+        end = cursor + match.end()
+        positions.append((heading, start, end))
+        cursor = end
+    missing = []
+    for index, (heading, _, end) in enumerate(positions):
+        next_start = positions[index + 1][1] if index + 1 < len(positions) else len(text)
+        if not text[end:next_start].strip():
+            missing.append(heading)
+    return not missing, missing
+
+
+def minimum_class(paths, wire_changed=False):
+    """Return the mechanical classification floor for changed repository paths."""
+    floor = 'C' if wire_changed else 'A'
+    for path in paths:
+        if (
+            path.startswith('pstop_c/')
+            or path in ('firmware/main/main.c', 'machn/main/main.c')
+            or path.startswith('docs/safety/')
+        ):
+            candidate = 'C'
+        elif path.startswith(('components/', 'common/')) or path in (
+            'firmware/sdkconfig.defaults',
+            'machn/sdkconfig.defaults',
+        ):
+            candidate = 'B'
+        elif path.startswith('.github/workflows/') or path.startswith('scripts/'):
+            candidate = 'B'
+        else:
+            candidate = 'A'
+        if CLASS_RANK[candidate] > CLASS_RANK[floor]:
+            floor = candidate
+    return floor
+
+
+def _labels(entity):
+    return {label['name'] if isinstance(label, dict) else label for label in entity.get('labels', [])}
+
+
+def cr_number(body, repository):
+    """Return one local CR number, rejecting foreign or ambiguous candidates."""
+    if repository.count('/') != 1:
+        return None
+    current_owner, current_name = repository.casefold().split('/')
+    numbers = set()
+    foreign = False
+    for line in (body or '').splitlines():
+        local = re.fullmatch(r'\s*(?:closes|refs)\s+#(\d+)\s*', line, re.IGNORECASE)
+        if local:
+            numbers.add(int(local.group(1)))
+            continue
+        url = re.fullmatch(
+            r'\s*(?:(?:closes|refs)\s+)?https://github\.com/([^/\s]+)/([^/\s]+)/issues/(\d+)/?\s*',
+            line,
+            re.IGNORECASE,
+        )
+        if url:
+            owner, name, number = url.groups()
+            if (owner.casefold(), name.casefold()) == (current_owner, current_name):
+                numbers.add(int(number))
+            else:
+                foreign = True
+    return next(iter(numbers)) if len(numbers) == 1 and not foreign else None
+
+
+def _issue_fields_complete(root, body):
+    form = parse_issue_form(Path(root) / '.github/ISSUE_TEMPLATE/change-request.yml')
+    missing = []
+    for field in form['body']:
+        if not field['required']:
+            continue
+        label = field['label']
+        match = re.search(rf'(?ms)^###\s+{re.escape(label)}\s*$\n(.*?)(?=^###\s|\Z)', body or '')
+        if not match or not match.group(1).strip() or match.group(1).strip() == '_No response_':
+            missing.append(field['id'])
+    return missing
+
+
+def _find_ia(comments, headings):
+    for comment in comments:
+        body = comment.get('body', '')
+        if headings and headings[0] in body:
+            return body
+    return ''
+
+
+def _find_short_ia(comments):
+    for comment in comments:
+        body = comment.get('body', '')
+        if all(re.search(phrase, body, re.IGNORECASE) for phrase in ('what changed', 'what it could affect', 'tests')):
+            return body
+    return ''
+
+
+def _cited_sr_tokens(text):
+    return set(re.findall(r'\bSR-[A-Z]+-[0-9A-Za-z]+(?:-[0-9A-Za-z]+)*\b', text))
+
+
+def _is_authorization(comment, author=''):
+    login = comment.get('user', {}).get('login', '').lower()
+    return (
+        login in AUTHORIZERS
+        and login != author.lower()
+        and bool(re.search(r'(?im)^\s*(?:decision:\s*)?authori[sz]ed\b', comment.get('body', '')))
+    )
+
+
+def _named_tests(text):
+    sections = re.findall(r'(?ms)^# 7\. Verification plan for this change\s*$\n(.*?)(?=^# 8\.|\Z)', text)
+    names = []
+    for section in sections:
+        names.extend(re.findall(r'`([^`]+)`', section))
+        for line in section.splitlines():
+            if line.lstrip().startswith(('-', '+')) and ':' in line:
+                value = line.split(':', 1)[1].strip()
+                if value and value.lower() not in ('none', 'n/a'):
+                    names.append(value)
+            if line.strip().startswith('|'):
+                cells = [cell.strip() for cell in line.strip().strip('|').split('|')]
+                if len(cells) >= 2 and cells[0] not in ('Purpose', '---'):
+                    value = cells[1]
+                    if value and value not in ('---', 'None', 'N/A'):
+                        names.extend(part.strip() for part in re.split(r'<br\s*/?>|,', value) if part.strip())
+    if not sections:
+        match = re.search(r'(?im)^\s*(?:which\s+)?tests(?:\s+will\s+be\s+run)?\s*:\s*(.+)$', text)
+        if match:
+            names.extend(part.strip(' `') for part in match.group(1).split(',') if part.strip(' `'))
+    return list(dict.fromkeys(name.strip() for name in names if name.strip()))
+
+
+def _evidence_matches(name, evidence_name):
+    """Match an IA entry to one complete check or workflow name, never a substring."""
+    planned = ' '.join(name.casefold().split())
+    observed = ' '.join(evidence_name.casefold().split())
+    return bool(planned and observed and planned == observed)
+
+
+def _approvers(data, require_head=True):
+    head = data['pr']['head']['sha']
+    author = data['pr']['user']['login'].lower()
+    latest = {}
+    for review in data.get('reviews', []):
+        login = review.get('user', {}).get('login', '').lower()
+        if login:
+            latest[login] = review
+    return {
+        login
+        for login, review in latest.items()
+        if login in AUTHORIZERS
+        and login != author
+        and review.get('state') == 'APPROVED'
+        and (not require_head or review.get('commit_id') == head)
+    }
+
+
+def evaluate(root, data):
+    """Evaluate E1-E7 against a complete, synthetic-or-live GitHub state snapshot."""
+    root = Path(root)
+    pr = data['pr']
+    labels = _labels(pr)
+    linked_cr = cr_number(pr.get('body', ''), data['repository'])
+    issue = data.get('issue', {})
+    issue_labels = _labels(issue)
+    comments = data.get('issue_comments', [])
+    headings = _headings(root / 'docs/process/templates/IMPACT_ANALYSIS.md')
+    ia = _find_ia(comments, headings)
+    short_ia = _find_short_ia(comments)
+    emergency = 'emergency' in labels
+    if emergency and not ia:
+        ia = short_ia
+    results = []
+
+    if 'needs-change-request' in labels:
+        results.append(CheckResult('E1', 'pending', 'maintainer Change Request required before review begins'))
+    else:
+        author = pr['user']['login']
+        required_authorizers = 2 if 'class-c' in labels or emergency else 1
+        authorization_comments = {}
+        for comment in comments:
+            if _is_authorization(comment, author):
+                authorization_comments[comment['user']['login'].lower()] = comment
+        authorized_comment = len(authorization_comments) >= required_authorizers
+        authorization_times = [
+            comment.get('created_at') for comment in authorization_comments.values() if comment.get('created_at')
+        ]
+        ia_times = [
+            comment.get('created_at')
+            for comment in comments
+            if comment.get('body', '') == ia and comment.get('created_at')
+        ]
+        before_implementation = not pr.get('created_at') or (
+            len(authorization_times) >= required_authorizers and max(authorization_times) <= pr['created_at']
+        )
+        after_analysis = not ia_times or (
+            len(authorization_times) >= required_authorizers and max(ia_times) <= min(authorization_times)
+        )
+        ordered = before_implementation and after_analysis
+        missing_fields = _issue_fields_complete(root, issue.get('body', '')) if linked_cr else ['change-request-link']
+        okay = (
+            linked_cr is not None
+            and 'change-request' in issue_labels
+            and 'status:authorized' in issue_labels
+            and authorized_comment
+            and ordered
+            and not missing_fields
+        )
+        detail = (
+            f'authorized Change Request has {required_authorizers} distinct pre-implementation authorizer(s)'
+            if okay
+            else f'Change Request missing, ambiguous, incomplete, or lacks {required_authorizers} distinct pre-implementation authorizer(s) ({", ".join(missing_fields)})'
+        )
+        results.append(CheckResult('E1', 'pass' if okay else 'fail', detail))
+
+    if emergency and ia == short_ia and short_ia:
+        complete, empty = True, []
+    else:
+        complete, empty = impact_analysis_complete(ia, headings) if ia else (False, ['Impact Analysis'])
+    results.append(
+        CheckResult(
+            'E2',
+            'pass' if complete else 'fail',
+            'all IA sections exist and are nonblank; content truth and adequacy are not assessed'
+            if complete
+            else f'IA sections missing or blank: {", ".join(empty)}; content truth and adequacy are not assessed',
+        )
+    )
+
+    canonical = {requirement.sr_id for requirement in parse_srs(root / 'docs/safety/SAFETY_REQUIREMENTS.md')}
+    cited = _cited_sr_tokens(ia)
+    invalid = sorted(cited - canonical)
+    results.append(
+        CheckResult(
+            'E3',
+            'fail' if invalid else 'pass',
+            f'invalid requirement IDs: {", ".join(invalid)}' if invalid else 'all cited requirement IDs exist',
+        )
+    )
+
+    class_labels = sorted(label for label in labels if re.fullmatch(r'class-[abc]', label))
+    paths = [item['filename'] for item in data.get('files', [])]
+    wire_changed = any(path.startswith('pstop_c/pstop/include/pstop/') for path in paths)
+    floor = minimum_class(paths, wire_changed)
+    if len(class_labels) != 1:
+        results.append(CheckResult('E4', 'fail', 'exactly one class-a, class-b, or class-c label is required'))
+        selected = None
+    else:
+        selected = class_labels[0][-1].upper()
+        under = CLASS_RANK[selected] < CLASS_RANK[floor]
+        results.append(
+            CheckResult(
+                'E4',
+                'fail' if under else 'pass',
+                f'Class {selected}; mechanical floor Class {floor}; checks existence/order, not classification adequacy',
+            )
+        )
+
+    approvers = _approvers(data)
+    if selected == 'C':
+        status = 'pass' if len(approvers) >= 2 else 'fail'
+        results.append(
+            CheckResult('E5', status, f'current distinct non-author approving authorizers: {len(approvers)}/2')
+        )
+    else:
+        results.append(CheckResult('E5', 'not-applicable', 'two-review requirement applies to Class C'))
+
+    names = _named_tests(ia)
+    head = pr['head']['sha']
+    evidence = {
+        item.get('name', '')
+        for item in data.get('check_runs', [])
+        if item.get('head_sha') == head and item.get('conclusion') == 'success'
+    }
+    evidence.update(
+        item.get('name', item.get('path', ''))
+        for item in data.get('workflow_runs', [])
+        if item.get('head_sha') == head and item.get('conclusion') == 'success'
+    )
+    missing_tests = [name for name in names if not any(_evidence_matches(name, item) for item in evidence)]
+    explanation = 'check-run/workflow evidence cannot prove commands or tests inside a job executed'
+    if not names:
+        e6_status = 'fail'
+        e6_message = f'IA verification plan names no specific tests; {explanation}'
+    elif missing_tests:
+        e6_status = 'fail'
+        e6_message = f'missing head-SHA evidence: {", ".join(missing_tests)}; {explanation}'
+    else:
+        e6_status = 'pass'
+        e6_message = f'all named evidence matched; {explanation}'
+    results.append(
+        CheckResult(
+            'E6',
+            e6_status,
+            e6_message,
+        )
+    )
+
+    if emergency:
+        short_form = bool(ia and names and short_ia)
+        status = 'pass' if len(approvers) >= 2 and short_form else 'fail'
+        results.append(
+            CheckResult(
+                'E7',
+                status,
+                'emergency path requires two approvals and short-form IA; retrospective due within five working days of release',
+            )
+        )
+    else:
+        results.append(CheckResult('E7', 'not-applicable', 'PR is not labelled emergency'))
+    return results
+
+
+def render_report(mode, results):
+    """Render one deterministic PR comment with the active mode visible."""
+    lines = [f'mode: {mode}', '', '| Check | Result | Explanation |', '|---|---|---|']
+    for item in results:
+        message = item.message.replace('|', '\\|')
+        lines.append(f'| {item.check_id} | {item.status} | {message} |')
+    lines.extend([
+        '',
+        'These checks verify artifact existence and ordering only, not truth, adequacy, or safety sufficiency.',
+    ])
+    return '\n'.join(lines)
+
+
+def upsert_comment(api, pr_number, report, comments):
+    """Create or update at most one marker-owned report comment."""
+    body = f'{COMMENT_MARKER}\n{report}'
+    existing = next((comment for comment in comments if COMMENT_MARKER in comment.get('body', '')), None)
+    if existing:
+        api('PATCH', f'repos/{{repo}}/issues/comments/{existing["id"]}', {'body': body})
+    else:
+        api('POST', f'repos/{{repo}}/issues/{pr_number}/comments', {'body': body})
diff --git a/tools/change_control/fixtures/fake_gh.py b/tools/change_control/fixtures/fake_gh.py
new file mode 100755
index 00000000..990c599d
--- /dev/null
+++ b/tools/change_control/fixtures/fake_gh.py
@@ -0,0 +1,46 @@
+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc.
+# SPDX-License-Identifier: Apache-2.0
+"""Deterministic gh replacement used only by change-control tests."""
+
+import json
+import os
+import sys
+from pathlib import Path
+
+
+def main():
+    """Serve API responses from the JSON file named by FAKE_GH_DATA."""
+    data = json.loads(Path(os.environ['FAKE_GH_DATA']).read_text(encoding='utf-8'))
+    if data.get('exit_code'):
+        print(data.get('stderr', 'fake gh failure'), file=sys.stderr)
+        return int(data['exit_code'])
+    args = sys.argv[1:]
+    if not args or args[0] != 'api':
+        return 2
+    endpoint = args[1] if len(args) > 1 else ''
+    method = 'GET'
+    if '--method' in args:
+        method = args[args.index('--method') + 1]
+    key = f'{method} {endpoint}'
+    calls = os.environ.get('FAKE_GH_CALLS')
+    if calls:
+        with Path(calls).open('a', encoding='utf-8') as stream:
+            stream.write(key + '\n')
+    response = data.get('responses', {}).get(key)
+    if response is None:
+        print(f'unconfigured fake gh request: {key}', file=sys.stderr)
+        return 1
+    if isinstance(response, dict) and '__pages__' in response:
+        response = response['__pages__']
+    elif '--slurp' in args:
+        response = [response]
+    if isinstance(response, str):
+        print(response)
+    else:
+        print(json.dumps(response))
+    return 0
+
+
+if __name__ == '__main__':
+    sys.exit(main())
diff --git a/tools/change_control/issue_form.py b/tools/change_control/issue_form.py
new file mode 100644
index 00000000..362a5261
--- /dev/null
+++ b/tools/change_control/issue_form.py
@@ -0,0 +1,140 @@
+# SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc.
+# SPDX-License-Identifier: Apache-2.0
+"""Parse and validate the constrained YAML used by the Change Request issue form."""
+
+import re
+from pathlib import Path
+
+
+def _scalar(value):
+    value = value.strip()
+    if not value:
+        return ''
+    if value in ('true', 'false'):
+        return value == 'true'
+    if value.isdigit():
+        return int(value)
+    if value.startswith('[') and value.endswith(']'):
+        return [part.strip().strip('\'"') for part in value[1:-1].split(',') if part.strip()]
+    return value.strip('\'"')
+
+
+def parse_issue_form(path):
+    """Return issue-form controls from the repository's deliberately limited YAML subset."""
+    lines = Path(path).read_text(encoding='utf-8').splitlines()
+    if any('\t' in line for line in lines):
+        raise ValueError('tabs are not valid indentation')
+    document = {'body': []}
+    item = None
+    section = None
+    options = False
+    active_attribute_key = None
+    active_attribute_indent = None
+    in_body = False
+    for number, raw in enumerate(lines, 1):
+        if not raw.strip() or raw.lstrip().startswith('#') or raw.strip() == '---':
+            continue
+        indent = len(raw) - len(raw.lstrip(' '))
+        text = raw.strip()
+        if indent == 0:
+            active_attribute_key = None
+            active_attribute_indent = None
+            match = re.fullmatch(r'([a-z_]+):(?:\s*(.*))?', text)
+            if not match:
+                raise ValueError(f'{path}:{number}: unsupported top-level YAML')
+            key, value = match.groups()
+            if key == 'body':
+                if value:
+                    raise ValueError(f'{path}:{number}: body must be a sequence')
+                in_body = True
+            else:
+                if in_body:
+                    raise ValueError(f'{path}:{number}: top-level key after body')
+                document[key] = _scalar(value or '')
+            continue
+        if not in_body:
+            raise ValueError(f'{path}:{number}: unexpected indentation')
+        if indent == 2 and text.startswith('- type: '):
+            item = {'type': _scalar(text[8:]), 'required': False, 'options': []}
+            document['body'].append(item)
+            section = None
+            options = False
+            active_attribute_key = None
+            active_attribute_indent = None
+            continue
+        if item is None:
+            raise ValueError(f'{path}:{number}: body entry must begin with type')
+        if indent == 4 and re.fullmatch(r'(id|attributes|validations):.*', text):
+            key, value = text.split(':', 1)
+            if key == 'id':
+                item['id'] = _scalar(value)
+                section = None
+            else:
+                if value.strip():
+                    raise ValueError(f'{path}:{number}: {key} must be a mapping')
+                section = key
+            options = False
+            active_attribute_key = None
+            active_attribute_indent = None
+            continue
+        if indent == 6 and section in ('attributes', 'validations'):
+            if ':' not in text:
+                raise ValueError(f'{path}:{number}: expected mapping value')
+            key, value = text.split(':', 1)
+            if section == 'validations' and key == 'required':
+                item['required'] = _scalar(value)
+                active_attribute_key = None
+                active_attribute_indent = None
+            elif section == 'attributes' and key == 'options':
+                if value.strip():
+                    raise ValueError(f'{path}:{number}: options must be a sequence')
+                options = True
+                active_attribute_key = None
+                active_attribute_indent = None
+            elif section == 'attributes':
+                item[key] = _scalar(value)
+                options = False
+                active_attribute_key = key
+                active_attribute_indent = indent
+            else:
+                raise ValueError(f'{path}:{number}: unsupported validation')
+            continue
+        if indent == 8 and options and text.startswith('- '):
+            item['options'].append(_scalar(text[2:]))
+            continue
+        if (
+            section == 'attributes'
+            and not options
+            and active_attribute_key is not None
+            and indent > active_attribute_indent
+        ):
+            item[active_attribute_key] = f'{item[active_attribute_key]} {_scalar(text)}'.strip()
+            continue
+        raise ValueError(f'{path}:{number}: unsupported indentation or YAML construct')
+    validate_issue_form(document)
+    return document
+
+
+def validate_issue_form(document):
+    """Reject forms that GitHub could silently replace with a blank issue page."""
+    for key in ('name', 'description', 'title', 'labels'):
+        if not document.get(key):
+            raise ValueError(f'issue form missing top-level {key}')
+    body = document.get('body')
+    if not isinstance(body, list) or not body:
+        raise ValueError('issue form body must contain controls')
+    ids = []
+    for field in body:
+        if field.get('type') not in ('input', 'textarea', 'dropdown'):
+            raise ValueError(f'unsupported issue form field type: {field.get("type")}')
+        if not field.get('id') or not field.get('label'):
+            raise ValueError('every issue form field needs id and label')
+        ids.append(field['id'])
+        if field['type'] == 'dropdown':
+            if not field.get('options') or not isinstance(field.get('default'), int):
+                raise ValueError('dropdown needs options and integer default')
+            if not 0 <= field['default'] < len(field['options']):
+                raise ValueError('dropdown default is outside options')
+    if len(ids) != len(set(ids)):
+        raise ValueError('duplicate issue form field id')
+    return document
diff --git a/tools/change_control/test_change_control.py b/tools/change_control/test_change_control.py
new file mode 100644
index 00000000..68dd376a
--- /dev/null
+++ b/tools/change_control/test_change_control.py
@@ -0,0 +1,800 @@
+# SPDX-FileCopyrightText: 2026 Polymath Robotics, Inc.
+# SPDX-License-Identifier: Apache-2.0
+"""Spec-driven tests for the focused warn-mode change-control checker."""
+
+import json
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+from tools.change_control.__main__ import GhApi  # noqa: E402
+from tools.change_control.checks import (  # noqa: E402
+    AUTHORIZERS,
+    evaluate,
+    impact_analysis_complete,
+    minimum_class,
+    upsert_comment,
+)
+from tools.change_control.issue_form import parse_issue_form, validate_issue_form  # noqa: E402
+
+FORM = ROOT / '.github/ISSUE_TEMPLATE/change-request.yml'
+
+
+def snapshot(**overrides):
+    """Return a complete synthetic GitHub state shaped like the real API snapshot."""
+    ia = (ROOT / 'docs/process/templates/IMPACT_ANALYSIS.md').read_text(encoding='utf-8')
+    filled = ia.replace('Modules changed:', 'Modules changed: tools/change_control').replace(
+        '|  |  |', '| None | None |'
+    )
+    data = {
+        'repository': 'polymathrobotics/protective-stop',
+        'pr': {
+            'user': {'login': 'contributor'},
+            'body': 'Closes #17',
+            'labels': [{'name': 'class-b'}],
+            'head': {'sha': 'abc123'},
+        },
+        'files': [{'filename': 'tools/change_control/checks.py'}],
+        'issue': {
+            'labels': [{'name': 'change-request'}, {'name': 'status:authorized'}],
+            'body': '### Reason for the change\nNeeded\n### Hazards that may be affected\nNone identified, because tooling only\n### Description of the proposed change\nTooling\n### Baseline affected\nmain\n### Proposed class\nB',
+        },
+        'issue_comments': [
+            {'user': {'login': AUTHORIZERS[0]}, 'body': 'Authorized: proceed.'},
+            {'user': {'login': 'analyst'}, 'body': filled},
+        ],
+        'reviews': [],
+        'check_runs': [{'name': 'change-control', 'conclusion': 'success', 'head_sha': 'abc123'}],
+        'workflow_runs': [],
+    }
+    data.update(overrides)
+    return data
+
+
+class IssueFormTests(unittest.TestCase):
+    def _parse_field(self, attributes):
+        form = (
+            'name: Test\n'
+            'description: Test form\n'
+            "title: '[Test] '\n"
+            'labels: [test]\n'
+            'body:\n'
+            '  - type: textarea\n'
+            '    id: field\n'
+            '    attributes:\n'
+            f'{attributes}'
+            '    validations:\n'
+            '      required: true\n'
+        )
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'form.yml'
+            path.write_text(form, encoding='utf-8')
+            return parse_issue_form(path)['body'][0]
+
+    def test_issue_form_is_valid_yaml_and_parses(self):
+        """The checked-in issue form must parse as the deliberately supported YAML subset."""
+        parsed = parse_issue_form(FORM)
+        self.assertEqual(parsed['name'], 'Change Request')
+        self.assertGreater(len(parsed['body']), 5)
+
+    def test_required_fields_present(self):
+        """All five creation-time fields mandated by the procedure must be required."""
+        fields = {field['id']: field for field in parse_issue_form(FORM)['body']}
+        self.assertTrue(
+            all(fields[name]['required'] for name in ('reason', 'hazards', 'description', 'baseline', 'class'))
+        )
+
+    def test_class_dropdown_defaults_to_c(self):
+        """An unclassified request must conservatively default to Class C."""
+        fields = {field['id']: field for field in parse_issue_form(FORM)['body']}
+        self.assertEqual(fields['class']['options'][fields['class']['default']], 'C')
+
+    def test_every_source_field_is_represented(self):
+        """The issue form must carry every Change Request source section without a duplicate field list."""
+        ids = {field['id'] for field in parse_issue_form(FORM)['body']}
+        expected = {
+            'reason',
+            'hazards',
+            'description',
+            'baseline',
+            'requester',
+            'impact-analysis',
+            'class',
+            'authorization',
+            'implementation',
+            'gate-0',
+            'gate-1',
+            'review',
+            'deviations',
+            'release',
+            'status',
+        }
+        self.assertEqual(ids, expected)
+
+    def test_source_field_details_survive_issue_form_conversion(self):
+        """YAML conversion must retain source details needed to complete implementation and Gate 1 records."""
+        fields = {field['id']: field for field in parse_issue_form(FORM)['body']}
+        self.assertIn('Yes / No, with link', fields['implementation']['description'])
+        self.assertIn('Run by', fields['gate-1']['description'])
+        self.assertIn('Forward -', fields['gate-1']['description'])
+        self.assertIn('Backward -', fields['gate-1']['description'])
+
+    def test_plain_label_continuation_is_folded(self):
+        """A continued attributes label must be joined to its first line with one space."""
+        field = self._parse_field('      label: First label line\n        second label line\n')
+        self.assertEqual(field['label'], 'First label line second label line')
+
+    def test_plain_description_continuation_is_folded(self):
+        """A continued attributes description must be joined to its first line with one space."""
+        field = self._parse_field(
+            '      label: Field\n      description: First description line\n        second description line\n'
+        )
+        self.assertEqual(field['description'], 'First description line second description line')
+
+    def test_plain_placeholder_continuation_is_folded(self):
+        """A continued attributes placeholder must be joined to its first line with one space."""
+        field = self._parse_field(
+            '      label: Field\n      placeholder: First placeholder line\n        second placeholder line\n'
+        )
+        self.assertEqual(field['placeholder'], 'First placeholder line second placeholder line')
+
+    def test_real_gate_one_description_is_complete(self):
+        """The checked-in folded Gate 1 description must retain its final continuation text."""
+        fields = {field['id']: field for field in parse_issue_form(FORM)['body']}
+        self.assertTrue(fields['gate-1']['description'].endswith('requirements covered.'))
+
+    def test_orphan_attribute_continuation_is_rejected(self):
+        """Indented text without an active scalar key must fail rather than disappear."""
+        with self.assertRaises(ValueError):
+            self._parse_field('        orphan continuation\n      label: Field\n')
+
+    def test_option_continuation_is_rejected(self):
+        """Deeper text below an option must never be folded into an attributes scalar."""
+        attributes = '      label: Field\n      options:\n        - First\n          unsupported option continuation\n'
+        with self.assertRaises(ValueError):
+            self._parse_field(attributes)
+
+    def test_validation_continuation_is_rejected(self):
+        """Deeper validation text must never be appended to the preceding attributes scalar."""
+        attributes = (
+            '      label: Field\n    validations:\n      required: true\n        unsupported validation continuation\n'
+        )
+        with self.assertRaises(ValueError):
+            self._parse_field(attributes)
+
+    def test_malformed_issue_form_is_rejected(self):
+        """Malformed indentation must fail instead of silently degrading to a blank issue."""
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / 'form.yml'
+            path.write_text('name: Bad\nbody:\n - type: input\n    id: broken\n', encoding='utf-8')
+            with self.assertRaises(ValueError):
+                parse_issue_form(path)
+
+    def test_blank_issue_fallback_risk_is_rejected(self):
+        """A form lacking required top-level metadata or body controls must be invalid."""
+        with self.assertRaises(ValueError):
+            validate_issue_form({'name': 'Change Request', 'body': []})
+
+
+class RepositoryPolicyTests(unittest.TestCase):
+    def test_issue_chooser_keeps_blank_issues_and_links_security_policy(self):
+        """Public reports must remain available while safety defects are directed to the private policy."""
+        config = (ROOT / '.github/ISSUE_TEMPLATE/config.yml').read_text(encoding='utf-8')
+        self.assertIn('blank_issues_enabled: true', config)
+        self.assertIn('https://github.com/polymathrobotics/protective-stop/security/policy', config)
+
+    def test_labels_json_has_no_duplicate_names(self):
+        """The reproducible label definition must contain unique names."""
+        labels = json.loads((ROOT / 'tools/change_control/labels.json').read_text(encoding='utf-8'))
+        names = [label['name'] for label in labels]
+        self.assertEqual(len(names), len(set(names)))
+
+    def test_labels_json_contains_the_complete_settled_label_set(self):
+        """The reproducible data must contain every label settled by the modification procedure plan."""
+        labels = json.loads((ROOT / 'tools/change_control/labels.json').read_text(encoding='utf-8'))
+        names = {label['name'] for label in labels}
+        self.assertEqual(
+            names,
+            {
+                'change-request',
+                'class-a',
+                'class-b',
+                'class-c',
+                'emergency',
+                'safety-defect',
+                'wire-break',
+                'needs-change-request',
+                'status:proposed',
+                'status:under-analysis',
+                'status:authorized',
+                'status:rejected',
+                'status:in-implementation',
+                'status:in-verification',
+                'status:merged',
+                'status:released',
+            },
+        )
+
+    def test_label_sync_dry_run_is_deterministic_and_network_free(self):
+        """Two dry runs must produce identical plans without invoking GitHub."""
+        environment = os.environ.copy()
+        environment['PATH'] = '/usr/bin:/bin'
+        first = subprocess.run(
+            ['scripts/sync_labels.sh', '--dry-run'],
+            cwd=ROOT,
+            env=environment,
+            check=False,
+            capture_output=True,
+            text=True,
+        )
+        second = subprocess.run(
+            ['scripts/sync_labels.sh', '--dry-run'],
+            cwd=ROOT,
+            env=environment,
+            check=False,
+            capture_output=True,
+            text=True,
+        )
+        self.assertEqual((first.returncode, second.returncode), (0, 0))
+        self.assertEqual(first.stdout, second.stdout)
+        self.assertEqual(first.stdout.count('would sync label:'), 16)
+
+    def test_label_sync_targets_the_intended_repository_explicitly(self):
+        """Label writes must not depend on ambiguous git-remote repository inference."""
+        script = (ROOT / 'scripts/sync_labels.sh').read_text(encoding='utf-8')
+        self.assertIn('--repo "$REPOSITORY"', script)
+
+    def test_every_label_referenced_in_the_procedure_exists_in_labels_json(self):
+        """Every backticked process label must exist in the reproducible label set."""
+        import re
+
+        procedure = (ROOT / 'docs/process/MODIFICATION_PROCEDURE.md').read_text(encoding='utf-8')
+        referenced = set(
+            re.findall(
+                r'`((?:class-[abc]|change-request|emergency|safety-defect|wire-break|needs-change-request|status:[a-z-]+))`',
+                procedure,
+            )
+        )
+        labels = {
+            item['name'] for item in json.loads((ROOT / 'tools/change_control/labels.json').read_text(encoding='utf-8'))
+        }
+        self.assertTrue(referenced)
+        self.assertEqual(referenced - labels, set())
+
+    def test_codeowners_parses_and_covers_root(self):
+        """One CODEOWNERS rule must cover the repository root with all verified authorizers."""
+        lines = [
+            line.split()
+            for line in (ROOT / '.github/CODEOWNERS').read_text(encoding='utf-8').splitlines()
+            if line and not line.startswith('#')
+        ]
+        self.assertEqual(lines, [['*', *('@' + name for name in AUTHORIZERS)]])
+
+    def test_procedure_authorizers_match_enforcement_and_codeowners(self):
+        """The procedure, enforcement code, and CODEOWNERS must name one identical authorizer set."""
+        import re
+
+        procedure = (ROOT / 'docs/process/MODIFICATION_PROCEDURE.md').read_text(encoding='utf-8')
+        named = set(re.findall(r'@(iliabaranov|rajasimman-madhivanan|davidt315)', procedure))
+        self.assertEqual(named, set(AUTHORIZERS))
+
+    @unittest.skipUnless(
+        os.environ.get('GH_TOKEN'), 'GH_TOKEN absent; CODEOWNERS handle resolution test visibly skipped'
+    )
+    def test_codeowners_handles_resolve(self):
+        """Every CODEOWNERS account must resolve through authenticated GitHub API access."""
+        for handle in AUTHORIZERS:
+            result = subprocess.run(['gh', 'api', f'users/{handle}'], check=False, capture_output=True, text=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+
+
+class ImpactAndClassificationTests(unittest.TestCase):
+    def test_e2_rejects_heading_present_but_empty(self):
+        """An IA heading followed only by whitespace must fail completeness checking."""
+        headings = ['# One', '# Two']
+        complete, missing = impact_analysis_complete('# One\n \t\n# Two\nanswer\n', headings)
+        self.assertFalse(complete)
+        self.assertIn('# One', missing)
+
+    def test_e2_accepts_na_content_without_judging_truth(self):
+        """The checker verifies nonblank content but does not judge whether N/A is adequate."""
+        self.assertEqual(impact_analysis_complete('# One\nN/A\n# Two\nN/A\n', ['# One', '# Two']), (True, []))
+
+    def test_e3_rejects_invalid_sr_ids(self):
+        """A cited token shaped like an SR but outside the canonical grammar must be a finding."""
+        data = snapshot()
+        data['issue_comments'][1]['body'] += '\nSR-BOGUS-99\n'
+        results = evaluate(ROOT, data)
+        self.assertEqual(next(item for item in results if item.check_id == 'E3').status, 'fail')
+
+    def test_e4_under_classification_is_a_finding(self):
+        """A class label below the path-derived floor must fail E4."""
+        data = snapshot(files=[{'filename': 'docs/safety/HARA.md'}])
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E4').status, 'fail')
+
+    def test_e4_over_classification_is_not_a_finding(self):
+        """A class label above the path-derived floor must be accepted."""
+        data = snapshot(files=[{'filename': 'README.md'}])
+        data['pr']['labels'] = [{'name': 'class-c'}]
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E4').status, 'pass')
+
+    def test_missing_class_label_is_a_finding(self):
+        """A PR without a classification label must fail classification checking."""
+        data = snapshot()
+        data['pr']['labels'] = []
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E4').status, 'fail')
+
+    def test_duplicate_class_labels_are_a_finding(self):
+        """Multiple classification labels are ambiguous and must fail E4."""
+        data = snapshot()
+        data['pr']['labels'] = [{'name': 'class-a'}, {'name': 'class-b'}]
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E4').status, 'fail')
+
+    def test_minimum_class_rules(self):
+        """Every settled path rule must produce its documented minimum classification floor."""
+        self.assertEqual(minimum_class(['pstop_c/x.c']), 'C')
+        self.assertEqual(minimum_class(['firmware/main/main.c']), 'C')
+        self.assertEqual(minimum_class(['components/x.c']), 'B')
+        self.assertEqual(minimum_class(['firmware/sdkconfig.defaults']), 'B')
+        self.assertEqual(minimum_class(['.github/workflows/x.yml']), 'B')
+
+
+class ApprovalAndLinkTests(unittest.TestCase):
+    def _class_c(self):
+        data = snapshot(files=[{'filename': 'docs/safety/HARA.md'}])
+        data['pr']['labels'] = [{'name': 'class-c'}]
+        data['reviews'] = [
+            {'user': {'login': AUTHORIZERS[0]}, 'state': 'APPROVED', 'commit_id': 'abc123'},
+            {'user': {'login': AUTHORIZERS[1]}, 'state': 'APPROVED', 'commit_id': 'abc123'},
+        ]
+        return data
+
+    def test_class_c_two_distinct_approvals_pass(self):
+        """Class C requires two distinct current approving authorizers who are not the author."""
+        self.assertEqual(next(item for item in evaluate(ROOT, self._class_c()) if item.check_id == 'E5').status, 'pass')
+
+    def test_duplicate_reviews_count_once(self):
+        """Repeated approvals by one account must count as one approval."""
+        data = self._class_c()
+        data['reviews'][1]['user']['login'] = AUTHORIZERS[0]
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E5').status, 'fail')
+
+    def test_author_approval_is_excluded(self):
+        """The PR author's own approval must never satisfy Class C approval."""
+        data = self._class_c()
+        data['pr']['user']['login'] = AUTHORIZERS[0]
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E5').status, 'fail')
+
+    def test_stale_approval_is_excluded(self):
+        """An approval for a commit other than the PR head must not count."""
+        data = self._class_c()
+        data['reviews'][1]['commit_id'] = 'oldsha'
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E5').status, 'fail')
+
+    def test_approval_without_commit_identity_is_excluded(self):
+        """An approval with no commit identity cannot establish review of the current diff."""
+        data = self._class_c()
+        data['reviews'][1].pop('commit_id')
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E5').status, 'fail')
+
+    def test_authorization_after_pr_is_an_ordering_finding(self):
+        """An authorization timestamp after implementation began must fail E1 ordering."""
+        data = snapshot()
+        data['pr']['created_at'] = '2026-09-10T10:00:00Z'
+        data['issue_comments'][0]['created_at'] = '2026-09-10T11:00:00Z'
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail')
+
+    def test_authorization_before_impact_analysis_is_an_ordering_finding(self):
+        """Authorization must follow the completed Impact Analysis rather than merely precede implementation."""
+        data = snapshot()
+        data['pr']['created_at'] = '2026-09-10T12:00:00Z'
+        data['issue_comments'][0]['created_at'] = '2026-09-10T10:00:00Z'
+        data['issue_comments'][1]['created_at'] = '2026-09-10T11:00:00Z'
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail')
+
+    def test_pr_author_cannot_authorize_own_change(self):
+        """The implementer must not satisfy the Change Request authorization requirement."""
+        data = snapshot()
+        data['pr']['user']['login'] = AUTHORIZERS[0]
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail')
+
+    def test_class_c_requires_two_distinct_cr_authorizers_before_implementation(self):
+        """Class C implementation cannot start after only one Change Request authorization."""
+        data = snapshot()
+        data['pr']['labels'] = [{'name': 'class-c'}]
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail')
+        data['issue_comments'].append({'user': {'login': AUTHORIZERS[1]}, 'body': 'Authorized: proceed.'})
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'pass')
+
+    def test_rejection_text_does_not_count_as_authorization(self):
+        """An authorizer saying a change is not authorized must not satisfy E1."""
+        data = snapshot()
+        data['issue_comments'][0]['body'] = 'Rejected: this is not authorized.'
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail')
+
+    def test_emergency_requires_two_approvals_and_short_form_ia(self):
+        """Emergency E7 must require two current approvals and all three short-form IA subjects."""
+        data = snapshot()
+        data['pr']['labels'].append({'name': 'emergency'})
+        data['issue_comments'][1]['body'] = (
+            'What changed: tooling\nWhat it could affect: process\nTests: change-control'
+        )
+        data['reviews'] = [
+            {'user': {'login': AUTHORIZERS[0]}, 'state': 'APPROVED', 'commit_id': 'abc123'},
+            {'user': {'login': AUTHORIZERS[1]}, 'state': 'APPROVED', 'commit_id': 'abc123'},
+        ]
+        data['check_runs'] = [{'name': 'change-control', 'conclusion': 'success', 'head_sha': 'abc123'}]
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E7').status, 'pass')
+
+    def test_ambiguous_change_request_link_fails(self):
+        """A bare issue number without Closes or Refs syntax must not be guessed as the CR."""
+        data = snapshot()
+        data['pr']['body'] = 'Issue #17'
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail')
+
+    def test_multiple_change_request_links_fail(self):
+        """Multiple candidate CR links must fail rather than selecting one."""
+        data = snapshot()
+        data['pr']['body'] = 'Closes #17\nRefs #18'
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail')
+
+    def test_same_repository_issue_url_is_accepted(self):
+        """A full issue URL for the current repository must identify its Change Request."""
+        data = snapshot(repository='polymathrobotics/protective-stop')
+        data['pr']['body'] = 'https://github.com/polymathrobotics/protective-stop/issues/17'
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'pass')
+
+    def test_foreign_repository_same_issue_number_is_rejected(self):
+        """A foreign issue URL must not map an equal issue number into the current repository."""
+        data = snapshot(repository='polymathrobotics/protective-stop')
+        data['pr']['body'] = 'Refs https://github.com/other/protective-stop/issues/17'
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail')
+
+    def test_foreign_repository_different_issue_number_is_rejected(self):
+        """A foreign issue URL must never select that number from the current repository."""
+        data = snapshot(repository='polymathrobotics/protective-stop')
+        data['pr']['body'] = 'Closes https://github.com/other/project/issues/91'
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail')
+
+    def test_mixed_local_and_foreign_issue_links_are_rejected(self):
+        """A local CR candidate mixed with any foreign candidate must be treated as ambiguous."""
+        data = snapshot(repository='polymathrobotics/protective-stop')
+        data['pr']['body'] = 'Closes #17\nRefs https://github.com/other/project/issues/91'
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'fail')
+
+    def test_duplicate_same_local_issue_link_is_accepted(self):
+        """Repeated equivalent local links must identify one unambiguous Change Request."""
+        data = snapshot(repository='polymathrobotics/protective-stop')
+        data['pr']['body'] = 'Closes #17\nRefs #17'
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'pass')
+
+    def test_repository_identity_comparison_is_case_insensitive(self):
+        """GitHub owner and repository casing must not make a same-repository URL foreign."""
+        data = snapshot(repository='PolyMathRobotics/Protective-Stop')
+        data['pr']['body'] = 'Refs https://github.com/POLYMATHROBOTICS/protective-stop/issues/17'
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'pass')
+
+    def test_needs_change_request_label_exempts_e1(self):
+        """External PRs awaiting a maintainer CR must produce a pending E1 result."""
+        data = snapshot()
+        data['pr']['labels'].append({'name': 'needs-change-request'})
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E1').status, 'pending')
+
+
+class EvidenceAndCommentTests(unittest.TestCase):
+    def test_e6_requires_at_least_one_named_test(self):
+        """An IA that names no specific test must fail the plan-versus-execution check."""
+        result = next(item for item in evaluate(ROOT, snapshot()) if item.check_id == 'E6')
+        self.assertEqual(result.status, 'fail')
+
+    def test_e6_missing_workflow_evidence_names_test(self):
+        """A test named by the IA without head-SHA check evidence must fail and be named."""
+        data = snapshot()
+        data['issue_comments'][1]['body'] += '\n# 7. Verification plan for this change\n`missing-check`\n'
+        result = next(item for item in evaluate(ROOT, data) if item.check_id == 'E6')
+        self.assertEqual(result.status, 'fail')
+        self.assertIn('missing-check', result.message)
+
+    def test_e6_ignores_evidence_for_other_sha(self):
+        """Evidence attached to an older commit must not satisfy the IA plan."""
+        data = snapshot()
+        data['issue_comments'][1]['body'] += '\n# 7. Verification plan for this change\n`host-check`\n'
+        data['check_runs'] = [{'name': 'host-check', 'conclusion': 'success', 'head_sha': 'oldsha'}]
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E6').status, 'fail')
+
+    def test_e6_ignores_check_evidence_without_head_sha(self):
+        """A successful check without explicit commit identity cannot satisfy the verification plan."""
+        data = snapshot()
+        data['issue_comments'][1]['body'] += '\n# 7. Verification plan for this change\n`host-check`\n'
+        data['check_runs'] = [{'name': 'host-check', 'conclusion': 'success'}]
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E6').status, 'fail')
+
+    def test_e6_ignores_workflow_evidence_with_null_head_sha(self):
+        """A successful workflow with null commit identity cannot satisfy the verification plan."""
+        data = snapshot()
+        data['issue_comments'][1]['body'] += '\n# 7. Verification plan for this change\n`host-check`\n'
+        data['check_runs'] = []
+        data['workflow_runs'] = [{'name': 'host-check', 'conclusion': 'success', 'head_sha': None}]
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E6').status, 'fail')
+
+    def test_e6_ignores_check_evidence_with_empty_head_sha(self):
+        """A successful check with empty commit identity cannot satisfy the verification plan."""
+        data = snapshot()
+        data['issue_comments'][1]['body'] += '\n# 7. Verification plan for this change\n`host-check`\n'
+        data['check_runs'] = [{'name': 'host-check', 'conclusion': 'success', 'head_sha': ''}]
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E6').status, 'fail')
+
+    def test_e6_accepts_workflow_evidence_with_exact_head_sha(self):
+        """A successful exactly named workflow explicitly attached to PR head must satisfy the plan."""
+        data = snapshot()
+        data['issue_comments'][1]['body'] += '\n# 7. Verification plan for this change\n`host-check`\n'
+        data['check_runs'] = []
+        data['workflow_runs'] = [{'name': 'host-check', 'conclusion': 'success', 'head_sha': 'abc123'}]
+        self.assertEqual(next(item for item in evaluate(ROOT, data) if item.check_id == 'E6').status, 'pass')
+
+    def test_e6_does_not_accept_a_substring_check_name(self):
+        """A short IA test token must not match an unrelated longer check-run name."""
+        data = snapshot()
+        data['issue_comments'][1]['body'] += '\n# 7. Verification plan for this change\n`host`\n'
+        data['check_runs'] = [{'name': 'host-check', 'conclusion': 'success', 'head_sha': 'abc123'}]
+        result = next(item for item in evaluate(ROOT, data) if item.check_id == 'E6')
+        self.assertEqual(result.status, 'fail')
+        self.assertIn('host', result.message)
+
+    def test_e6_reads_plain_table_cells(self):
+        """Plain test names in the IA verification table must be checked, not only backticked names."""
+        data = snapshot()
+        data['issue_comments'][1]['body'] = data['issue_comments'][1]['body'].replace(
+            '| Tests that validate the change itself |  |',
+            '| Tests that validate the change itself | missing-table-check |',
+        )
+        result = next(item for item in evaluate(ROOT, data) if item.check_id == 'E6')
+        self.assertEqual(result.status, 'fail')
+        self.assertIn('missing-table-check', result.message)
+
+    def test_upsert_comment_updates_existing_comment(self):
+        """An existing bot report must be updated instead of appending another comment."""
+        calls = []
+        upsert_comment(
+            lambda method, path, body=None: calls.append((method, path, body)),
+            9,
+            'report',
+            [{'id': 44, 'body': '<!-- change-control -->old'}],
+        )
+        self.assertEqual(calls[0][0:2], ('PATCH', 'repos/{repo}/issues/comments/44'))
+
+    def test_upsert_comment_creates_when_absent(self):
+        """A bot report must be created exactly once when no marker exists."""
+        calls = []
+        upsert_comment(lambda method, path, body=None: calls.append((method, path, body)), 9, 'report', [])
+        self.assertEqual(
+            calls, [('POST', 'repos/{repo}/issues/9/comments', {'body': '<!-- change-control -->\nreport'})]
+        )
+
+
+class CliTests(unittest.TestCase):
+    def test_gh_api_flattens_all_paginated_list_pages(self):
+        """Policy evaluation must see every item returned across GitHub list pages."""
+        with tempfile.TemporaryDirectory() as directory:
+            data = Path(directory) / 'gh.json'
+            data.write_text(
+                json.dumps({'responses': {'GET items': {'__pages__': [[{'id': 1}], [{'id': 2}]]}}}),
+                encoding='utf-8',
+            )
+            with mock.patch.dict(os.environ, {'FAKE_GH_DATA': str(data)}):
+                api = GhApi(f'{sys.executable} {ROOT / "tools/change_control/fixtures/fake_gh.py"}', 'acme/project')
+                self.assertEqual(api('GET', 'items', paginate=True), [{'id': 1}, {'id': 2}])
+
+    def test_gh_api_merges_all_paginated_collection_pages(self):
+        """Check and workflow evidence must include every GitHub response page."""
+        with tempfile.TemporaryDirectory() as directory:
+            data = Path(directory) / 'gh.json'
+            pages = [{'check_runs': [{'id': 1}]}, {'check_runs': [{'id': 2}]}]
+            data.write_text(
+                json.dumps({'responses': {'GET checks': {'__pages__': pages}}}),
+                encoding='utf-8',
+            )
+            with mock.patch.dict(os.environ, {'FAKE_GH_DATA': str(data)}):
+                api = GhApi(f'{sys.executable} {ROOT / "tools/change_control/fixtures/fake_gh.py"}', 'acme/project')
+                self.assertEqual(
+                    api('GET', 'checks', paginate=True, collection_key='check_runs'),
+                    {'check_runs': [{'id': 1}, {'id': 2}]},
+                )
+
+    def test_gh_api_rejects_non_list_paginated_page(self):
+        """A malformed list page must fail closed rather than hide omitted GitHub records."""
+        with self.assertRaisesRegex(RuntimeError, 'non-list page'):
+            self._fake_api({'GET items': {'__pages__': [[{'id': 1}], {'id': 2}]}})('GET', 'items', paginate=True)
+
+    def test_gh_api_rejects_missing_collection_in_paginated_page(self):
+        """A malformed collection page must fail closed rather than produce partial evidence."""
+        with self.assertRaisesRegex(RuntimeError, 'lacks check_runs'):
+            self._fake_api({'GET checks': {'__pages__': [{'check_runs': []}, {}]}})(
+                'GET', 'checks', paginate=True, collection_key='check_runs'
+            )
+
+    def test_gh_api_rejects_invalid_json(self):
+        """A successful API process with malformed JSON must still be an unable-to-run error."""
+        with self.assertRaisesRegex(RuntimeError, 'invalid JSON'):
+            self._fake_api({'GET items': 'not-json'})('GET', 'items')
+
+    def test_workflows_disable_writes_for_fork_pull_requests(self):
+        """Fork pull requests must still run checks without attempting unavailable comment writes."""
+        change = (ROOT / '.github/workflows/change-control.yml').read_text(encoding='utf-8')
+        self.assertIn('CAN_COMMENT', change)
+        self.assertIn('--no-comment', change)
+
+    def test_marker_comment_workflows_cancel_superseded_pr_runs(self):
+        """The marker-comment writer must cancel stale runs in its per-PR concurrency group."""
+        change = (ROOT / '.github/workflows/change-control.yml').read_text(encoding='utf-8')
+        self.assertIn('group: change-control-${{ github.event.pull_request.number }}', change)
+        self.assertEqual(change.count('cancel-in-progress: true'), 1)
+
+    def test_warn_mode_exits_zero_with_findings(self):
+        """Warn mode must report findings while returning success to the caller."""
+        result = self._run_cli('warn', {'responses': {}})
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn('mode: warn', result.stdout)
+
+    def test_enforce_mode_exits_one_with_findings(self):
+        """Enforce mode must return one for the same findings that warn mode tolerates."""
+        result = self._run_cli('enforce', {'responses': {}})
+        self.assertEqual(result.returncode, 1, result.stderr)
+
+    def test_mode_file_rejects_unknown_value(self):
+        """An invalid mode must return two rather than silently defaulting to warn."""
+        result = self._run_cli('maybe', {'responses': {}})
+        self.assertEqual(result.returncode, 2)
+
+    def test_missing_mode_file_returns_two(self):
+        """An unreadable mode source must return two rather than silently defaulting to warn."""
+        with tempfile.TemporaryDirectory() as directory:
+            result = subprocess.run(
+                [
+                    sys.executable,
+                    '-m',
+                    'tools.change_control',
+                    '--root',
+                    directory,
+                    '--repository',
+                    'acme/project',
+                    '--pr',
+                    '7',
+                    '--gh',
+                    f'{sys.executable} {ROOT / "tools/change_control/fixtures/fake_gh.py"}',
+                ],
+                cwd=ROOT,
+                check=False,
+                capture_output=True,
+                text=True,
+            )
+        self.assertEqual(result.returncode, 2)
+        self.assertIn('cannot read enforcement mode', result.stderr)
+
+    def test_gh_api_failure_returns_two(self):
+        """A failed gh subprocess must make the checker unable to run, not create policy findings."""
+        result = self._run_cli('warn', {'exit_code': 1, 'stderr': 'API unavailable'})
+        self.assertEqual(result.returncode, 2)
+        self.assertIn('API unavailable', result.stderr)
+
+    def test_partial_json_returns_two(self):
+        """A partial GitHub response must fail closed as an execution error."""
+        responses = self._responses()
+        responses['GET repos/acme/project/pulls/7'] = {'body': 'Closes #17'}
+        result = self._run_cli('warn', {'responses': responses})
+        self.assertEqual(result.returncode, 2)
+
+    def test_no_comment_avoids_write_api(self):
+        """Fork-safe change-control checks must not call the comment API when writes are unavailable."""
+        result, calls = self._run_cli('warn', {'responses': self._responses()}, no_comment=True, record_calls=True)
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertNotIn('POST repos/acme/project/issues/7/comments', calls)
+
+    def test_warn_mode_survives_comment_permission_failure(self):
+        """Warn-mode findings remain visible in logs when GitHub denies advisory comment writes."""
+        responses = self._responses()
+        responses.pop('POST repos/acme/project/issues/7/comments')
+        result = self._run_cli('warn', {'responses': responses})
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn('comment publication warning', result.stderr)
+
+    def test_enforce_mode_fails_closed_on_comment_permission_failure(self):
+        """Enforce mode must return unable-to-run when its required report comment cannot be published."""
+        responses = self._responses()
+        responses.pop('POST repos/acme/project/issues/7/comments')
+        result = self._run_cli('enforce', {'responses': responses})
+        self.assertEqual(result.returncode, 2)
+        self.assertIn('comment publication warning', result.stderr)
+
+    def _fake_api(self, responses):
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        data = Path(temporary.name) / 'gh.json'
+        data.write_text(json.dumps({'responses': responses}), encoding='utf-8')
+        patcher = mock.patch.dict(os.environ, {'FAKE_GH_DATA': str(data)})
+        patcher.start()
+        self.addCleanup(patcher.stop)
+        return GhApi(f'{sys.executable} {ROOT / "tools/change_control/fixtures/fake_gh.py"}', 'acme/project')
+
+    def _responses(self):
+        data = snapshot()
+        data['pr']['labels'] = []
+        return {
+            'GET repos/acme/project/pulls/7': data['pr'],
+            'GET repos/acme/project/pulls/7/files': data['files'],
+            'GET repos/acme/project/issues/17': data['issue'],
+            'GET repos/acme/project/issues/17/comments': data['issue_comments'],
+            'GET repos/acme/project/pulls/7/reviews': data['reviews'],
+            'GET repos/acme/project/commits/abc123/check-runs': {'check_runs': data['check_runs']},
+            'GET repos/acme/project/actions/runs?head_sha=abc123': {'workflow_runs': data['workflow_runs']},
+            'GET repos/acme/project/issues/7/comments': [],
+            'POST repos/acme/project/issues/7/comments': {},
+        }
+
+    def _run_cli(self, mode, fake_data, no_comment=False, record_calls=False):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            (root / 'docs/process').mkdir(parents=True)
+            (root / 'docs/process/enforcement-mode').write_text(mode + '\n', encoding='utf-8')
+            (root / 'docs/process/templates').mkdir()
+            shutil.copy2(
+                ROOT / 'docs/process/templates/IMPACT_ANALYSIS.md', root / 'docs/process/templates/IMPACT_ANALYSIS.md'
+            )
+            (root / 'docs/safety').mkdir(parents=True)
+            shutil.copy2(ROOT / 'docs/safety/SAFETY_REQUIREMENTS.md', root / 'docs/safety/SAFETY_REQUIREMENTS.md')
+            (root / '.github/ISSUE_TEMPLATE').mkdir(parents=True)
+            shutil.copy2(FORM, root / '.github/ISSUE_TEMPLATE/change-request.yml')
+            data_path = root / 'gh.json'
+            data_path.write_text(
+                json.dumps(
+                    fake_data
+                    if fake_data.get('exit_code')
+                    else {'responses': fake_data.get('responses') or self._responses()}
+                ),
+                encoding='utf-8',
+            )
+            environment = os.environ.copy()
+            environment['FAKE_GH_DATA'] = str(data_path)
+            environment['PYTHONPATH'] = str(ROOT)
+            calls_path = root / 'gh-calls.txt'
+            if record_calls:
+                environment['FAKE_GH_CALLS'] = str(calls_path)
+            command = [
+                sys.executable,
+                '-m',
+                'tools.change_control',
+                '--root',
+                str(root),
+                '--repository',
+                'acme/project',
+                '--pr',
+                '7',
+                '--gh',
+                f'{sys.executable} {ROOT / "tools/change_control/fixtures/fake_gh.py"}',
+            ]
+            if no_comment:
+                command.append('--no-comment')
+            result = subprocess.run(
+                command,
+                cwd=ROOT,
+                env=environment,
+                check=False,
+                capture_output=True,
+                text=True,
+            )
+            if record_calls:
+                calls = calls_path.read_text(encoding='utf-8') if calls_path.exists() else ''
+                return result, calls
+            return result
+
+
+if __name__ == '__main__':
+    unittest.main(verbosity=2)