diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 31bb414..b8614bb 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -12,22 +12,7 @@ on:
     branches: [ main ]
 
 jobs:
-  lint-commits:
-    # Note: To re-run `lint-commits` after fixing the PR title, close-and-reopen the PR.
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd  # v6.0.2
-      - name: Use Node.js
-        uses: actions/setup-node@53b83947a5a98c8d113130e565377fae1a50d02f # v6.3.0
-        with:
-          node-version: 22.x
-      - name: Check PR title
-        run: |
-          node "$GITHUB_WORKSPACE/.github/workflows/lintcommit.js"
-
   build:
-    needs: lint-commits
-
     runs-on: ubuntu-latest
     strategy:
       fail-fast: false
diff --git a/.github/workflows/lintcommit.js b/.github/workflows/lintcommit.js
deleted file mode 100644
index 1f120cb..0000000
--- a/.github/workflows/lintcommit.js
+++ /dev/null
@@ -1,178 +0,0 @@
-// Checks that a PR title conforms to conventional commits
-// (https://www.conventionalcommits.org/).
-//
-// To run self-tests, run this script:
-//
-//     node lintcommit.js test
-
-import { readFileSync, appendFileSync } from "fs";
-
-const types = new Set([
-  "build",
-  "chore",
-  "ci",
-  "config",
-  "deps",
-  "docs",
-  "feat",
-  "fix",
-  "perf",
-  "refactor",
-  "revert",
-  "style",
-  "test",
-  "types",
-]);
-
-const scopes = new Set(["sdk", "examples"]);
-
-/**
- * Checks that a pull request title, or commit message subject, follows the expected format:
- *
- *      type(scope): message
- *
- * Returns undefined if `title` is valid, else an error message.
- */
-function validateTitle(title) {
-  const parts = title.split(":");
-  const subject = parts.slice(1).join(":").trim();
-
-  if (title.startsWith("Merge")) {
-    return undefined;
-  }
-
-  if (parts.length < 2) {
-    return "missing colon (:) char";
-  }
-
-  const typeScope = parts[0];
-
-  const [type, scope] = typeScope.split(/\(([^)]+)\)$/);
-
-  if (/\s+/.test(type)) {
-    return `type contains whitespace: "${type}"`;
-  } else if (!types.has(type)) {
-    return `invalid type "${type}"`;
-  } else if (!scope && typeScope.includes("(")) {
-    return `must be formatted like type(scope):`;
-  } else if (scope && scope.length > 30) {
-    return "invalid scope (must be <=30 chars)";
-  } else if (scope && /[^- a-z0-9]+/.test(scope)) {
-    return `invalid scope (must be lowercase, ascii only): "${scope}"`;
-  } else if (scope && !scopes.has(scope)) {
-    return `invalid scope "${scope}" (valid scopes are ${Array.from(scopes).join(", ")})`;
-  } else if (subject.length === 0) {
-    return "empty subject";
-  } else if (subject.length > 50) {
-    return "invalid subject (must be <=50 chars)";
-  }
-
-  return undefined;
-}
-
-function run() {
-  const eventData = JSON.parse(
-    readFileSync(process.env.GITHUB_EVENT_PATH, "utf8"),
-  );
-  const pullRequest = eventData.pull_request;
-
-  // console.log(eventData)
-
-  if (!pullRequest) {
-    console.info("No pull request found in the context");
-    return;
-  }
-
-  const title = pullRequest.title;
-
-  const failReason = validateTitle(title);
-  const msg = failReason
-    ? `
-Invalid pull request title: \`${title}\`
-
-* Problem: ${failReason}
-* Expected format: \`type(scope): subject...\`
-    * type: one of (${Array.from(types).join(", ")})
-    * scope: optional, lowercase, <30 chars
-    * subject: must be <50 chars
-* Hint: *close and re-open the PR* to re-trigger CI (after fixing the PR title).
-`
-    : `Pull request title matches the expected format`;
-
-  if (process.env.GITHUB_STEP_SUMMARY) {
-    appendFileSync(process.env.GITHUB_STEP_SUMMARY, msg);
-  }
-
-  if (failReason) {
-    console.error(msg);
-    process.exit(1);
-  } else {
-    console.info(msg);
-  }
-}
-
-function _test() {
-  const tests = {
-    " foo(scope): bar": 'type contains whitespace: " foo"',
-    "build: update build process": undefined,
-    "chore: update dependencies": undefined,
-    "ci: configure CI/CD": undefined,
-    "config: update configuration files": undefined,
-    "deps: bump aws-sdk group with 5 updates": undefined,
-    "docs: update documentation": undefined,
-    "feat(sdk): add new feature": undefined,
-    "feat(sdk):": "empty subject",
-    "feat foo):": 'type contains whitespace: "feat foo)"',
-    "feat(foo)): sujet": 'invalid type "feat(foo))"',
-    "feat(foo: sujet": 'invalid type "feat(foo"',
-    "feat(Q Foo Bar): bar":
-      'invalid scope (must be lowercase, ascii only): "Q Foo Bar"',
-    "feat(sdk): bar": undefined,
-    "feat(sdk): x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x x ":
-      "invalid subject (must be <=50 chars)",
-    "feat: foo": undefined,
-    "fix: foo": undefined,
-    "fix(sdk): resolve issue": undefined,
-    "foo (scope): bar": 'type contains whitespace: "foo "',
-    "invalid title": "missing colon (:) char",
-    "perf: optimize performance": undefined,
-    "refactor: improve code structure": undefined,
-    "revert: feat: add new feature": undefined,
-    "style: format code": undefined,
-    "test: add new tests": undefined,
-    "types: add type definitions": undefined,
-    "Merge staging into feature/lambda-get-started": undefined,
-    "feat(foo): fix the types":
-      'invalid scope "foo" (valid scopes are sdk, examples)',
-  };
-
-  let passed = 0;
-  let failed = 0;
-
-  for (const [title, expected] of Object.entries(tests)) {
-    const result = validateTitle(title);
-    if (result === expected) {
-      console.log(`✅ Test passed for "${title}"`);
-      passed++;
-    } else {
-      console.log(
-        `❌ Test failed for "${title}" (expected "${expected}", got "${result}")`,
-      );
-      failed++;
-    }
-  }
-
-  console.log(`\n${passed} tests passed, ${failed} tests failed`);
-}
-
-function main() {
-  const mode = process.argv[2];
-
-  if (mode === "test") {
-    _test();
-  } else {
-    run();
-  }
-}
-
-main();
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index f0404de..df5feff 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -30,6 +30,14 @@ There is a convenience script for the above that you can run from the root of th
 ops/ci-checks.sh
 ```
 
+This script also validates your commit messages against the [Conventional Commits](https://www.conventionalcommits.org/) format.
+If you have uncommitted changes, it will skip commit message validation with a warning - commit first, then re-run to validate.
+
+You can also run the commit message check independently:
+```
+python ops/lintcommit.py
+```
+
 ## Coding Standards
 Consistency is important for maintainability. Please adhere to the house-style of the repo, unless there's a really
 good reason to break pattern.
diff --git a/ops/__tests__/test_lintcommit.py b/ops/__tests__/test_lintcommit.py
new file mode 100644
index 0000000..cc6e3b1
--- /dev/null
+++ b/ops/__tests__/test_lintcommit.py
@@ -0,0 +1,120 @@
+#!/usr/bin/env python3
+
+import os
+import sys
+import unittest
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
+
+from lintcommit import validate_message, validate_subject
+
+
+class TestValidateSubject(unittest.TestCase):
+    # Valid subjects
+    def test_valid_feat(self) -> None:
+        self.assertIsNone(validate_subject("feat: add new feature"))
+
+    def test_valid_fix(self) -> None:
+        self.assertIsNone(validate_subject("fix: resolve issue"))
+
+    def test_valid_fix_with_scope(self) -> None:
+        self.assertIsNone(validate_subject("fix(sdk): resolve issue"))
+
+    def test_valid_build(self) -> None:
+        self.assertIsNone(validate_subject("build: update build process"))
+
+    def test_valid_chore(self) -> None:
+        self.assertIsNone(validate_subject("chore: update dependencies"))
+
+    def test_valid_ci(self) -> None:
+        self.assertIsNone(validate_subject("ci: configure CI/CD"))
+
+    def test_valid_deps(self) -> None:
+        self.assertIsNone(validate_subject("deps: bump aws-sdk group with 5 updates"))
+
+    def test_valid_docs(self) -> None:
+        self.assertIsNone(validate_subject("docs: update documentation"))
+
+    def test_valid_feat_with_scope(self) -> None:
+        self.assertIsNone(validate_subject("feat(sdk): add new feature"))
+
+    def test_valid_feat_scope_bar(self) -> None:
+        self.assertIsNone(validate_subject("feat(sdk): bar"))
+
+    def test_valid_feat_foo(self) -> None:
+        self.assertIsNone(validate_subject("feat: foo"))
+
+    def test_valid_fix_foo(self) -> None:
+        self.assertIsNone(validate_subject("fix: foo"))
+
+    # Invalid subjects
+    def test_invalid_type(self) -> None:
+        self.assertEqual(validate_subject("config: foo"), 'invalid type "config"')
+
+    def test_missing_colon(self) -> None:
+        self.assertEqual(validate_subject("invalid title"), "missing colon (:) char")
+
+    def test_period_at_end(self) -> None:
+        self.assertEqual(
+            validate_subject("feat: add thing."),
+            "subject must not end with a period",
+        )
+
+    def test_empty_subject(self) -> None:
+        self.assertEqual(validate_subject("feat: "), "empty subject")
+
+    def test_subject_too_long(self) -> None:
+        long_subject: str = "feat: " + "a" * 51
+        result: str | None = validate_subject(long_subject)
+        self.assertIsNotNone(result)
+        self.assertIn("invalid subject", result)  # type: ignore[arg-type]
+
+    def test_type_with_whitespace(self) -> None:
+        self.assertEqual(
+            validate_subject("fe at: foo"), 'type contains whitespace: "fe at"'
+        )
+
+    def test_scope_not_closed(self) -> None:
+        self.assertEqual(
+            validate_subject("feat(sdk: foo"), "must be formatted like type(scope):"
+        )
+
+    def test_scope_too_long(self) -> None:
+        long_scope: str = "a" * 31
+        result: str | None = validate_subject(f"feat({long_scope}): foo")
+        self.assertIsNotNone(result)
+        self.assertIn("invalid scope", result)  # type: ignore[arg-type]
+
+    def test_scope_uppercase(self) -> None:
+        result: str | None = validate_subject("feat(SDK): foo")
+        self.assertIsNotNone(result)
+        self.assertIn("invalid scope", result)  # type: ignore[arg-type]
+
+
+class TestValidateMessage(unittest.TestCase):
+    def test_valid_subject_only(self) -> None:
+        error, warnings = validate_message("feat: add thing")
+        self.assertIsNone(error)
+        self.assertEqual(warnings, [])
+
+    def test_valid_with_body(self) -> None:
+        error, warnings = validate_message("feat: add thing\n\nThis is the body.")
+        self.assertIsNone(error)
+        self.assertEqual(warnings, [])
+
+    def test_missing_blank_line(self) -> None:
+        _, warnings = validate_message("feat: add thing\nNo blank line.")
+        self.assertIn("missing blank line between subject and body", warnings)
+
+    def test_long_body_line(self) -> None:
+        _, warnings = validate_message("feat: add thing\n\n" + "x" * 80)
+        self.assertEqual(len(warnings), 1)
+        self.assertIn("exceeds 72 chars", warnings[0])
+
+    def test_empty_message(self) -> None:
+        error, _ = validate_message("")
+        self.assertEqual(error, "empty commit message")
+
+    def test_invalid_subject_in_message(self) -> None:
+        error, _ = validate_message("invalid title")
+        self.assertEqual(error, "missing colon (:) char")
diff --git a/ops/ci-checks.sh b/ops/ci-checks.sh
index c2ad664..838ebe5 100755
--- a/ops/ci-checks.sh
+++ b/ops/ci-checks.sh
@@ -13,3 +13,6 @@ echo SUCCESS: typings
 # static analysis
 hatch fmt
 echo SUCCESS: linting/fmt
+
+# commit message validation
+python ops/lintcommit.py
diff --git a/ops/lintcommit.py b/ops/lintcommit.py
new file mode 100644
index 0000000..db8dde0
--- /dev/null
+++ b/ops/lintcommit.py
@@ -0,0 +1,207 @@
+#!/usr/bin/env python3
+# Checks that commit messages conform to conventional commits
+# (https://www.conventionalcommits.org/).
+#
+# To run tests:
+#
+#     python -m pytest ops/__tests__/test_lintcommit.py
+
+from __future__ import annotations
+
+import os
+import re
+import sys
+
+TYPES: set[str] = {
+    "build",
+    "chore",
+    "ci",
+    "deps",
+    "docs",
+    "feat",
+    "fix",
+    "perf",
+    "refactor",
+    "style",
+    "test",
+}
+
+MAX_SUBJECT_LENGTH: int = 50
+MAX_SCOPE_LENGTH: int = 30
+MAX_BODY_LINE_LENGTH: int = 72
+
+
+def validate_subject(subject_line: str) -> str | None:
+    """Validate a commit message subject line.
+
+    Returns None if valid, else an error message string.
+    """
+    parts: list[str] = subject_line.split(":", maxsplit=1)
+
+    if len(parts) < 2:
+        return "missing colon (:) char"
+
+    type_scope: str = parts[0]
+    subject: str = parts[1].strip()
+
+    # Parse type and optional scope: type or type(scope)
+    scope: str | None = None
+    commit_type: str = type_scope
+
+    if "(" in type_scope:
+        paren_start: int = type_scope.index("(")
+        commit_type = type_scope[:paren_start]
+
+        if not type_scope.endswith(")"):
+            return "must be formatted like type(scope):"
+
+        scope = type_scope[paren_start + 1 : -1]
+
+    if " " in commit_type:
+        return f'type contains whitespace: "{commit_type}"'
+
+    if commit_type not in TYPES:
+        return f'invalid type "{commit_type}"'
+
+    if scope is not None:
+        if len(scope) > MAX_SCOPE_LENGTH:
+            return f"invalid scope (must be <={MAX_SCOPE_LENGTH} chars)"
+
+        if re.search(r"[^- a-z0-9]", scope):
+            return f'invalid scope (must be lowercase, ascii only): "{scope}"'
+
+    if len(subject) == 0:
+        return "empty subject"
+
+    if len(subject) > MAX_SUBJECT_LENGTH:
+        return f"invalid subject (must be <={MAX_SUBJECT_LENGTH} chars)"
+
+    if subject.endswith("."):
+        return "subject must not end with a period"
+
+    return None
+
+
+def validate_body(body: str) -> list[str]:
+    """Validate the body of a commit message.
+
+    Returns a list of warnings (not hard errors) for body issues.
+    """
+    warnings: list[str] = []
+    for i, line in enumerate(body.splitlines(), start=1):
+        if len(line) > MAX_BODY_LINE_LENGTH:
+            warnings.append(
+                f"body line {i} exceeds {MAX_BODY_LINE_LENGTH} chars ({len(line)} chars)"
+            )
+    return warnings
+
+
+def validate_message(message: str) -> tuple[str | None, list[str]]:
+    """Validate a full commit message (subject + optional body).
+
+    Returns (error, warnings) where error is None if the subject is valid.
+    """
+    lines: list[str] = message.strip().splitlines()
+    if not lines:
+        return ("empty commit message", [])
+
+    subject_line: str = lines[0]
+    error: str | None = validate_subject(subject_line)
+
+    warnings: list[str] = []
+    # Check for blank line between subject and body
+    if len(lines) > 1 and lines[1].strip() != "":
+        warnings.append("missing blank line between subject and body")
+
+    if len(lines) > 2:
+        body: str = "\n".join(lines[2:])
+        warnings.extend(validate_body(body))
+
+    return (error, warnings)
+
+
+def _format_error(title: str, reason: str) -> str:
+    valid_types: str = ", ".join(sorted(TYPES))
+    return f"""Invalid commit message: `{title}`
+
+* Problem: {reason}
+* Expected format: `type(scope): subject...`
+    * type: one of ({valid_types})
+    * scope: optional, lowercase, <{MAX_SCOPE_LENGTH} chars
+    * subject: must be <{MAX_SUBJECT_LENGTH} chars
+"""
+
+
+def run_local() -> None:
+    """Validate local commit messages ahead of origin/main.
+
+    If there are uncommitted changes, prints a warning and skips validation.
+    """
+    import subprocess
+
+    # Check for uncommitted changes
+    status: subprocess.CompletedProcess[str] = subprocess.run(
+        ["git", "status", "--porcelain"],
+        capture_output=True,
+        text=True,
+    )
+    if status.stdout.strip():
+        print(
+            "WARNING: uncommitted changes detected, skipping commit message validation.\n"
+            "Commit your changes and re-run to validate."
+        )
+        return
+
+    # Get all commit messages ahead of origin/main
+    result: subprocess.CompletedProcess[str] = subprocess.run(
+        ["git", "log", "origin/main..HEAD", "--format=%H%n%B%n---END---"],
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode != 0:
+        print(f"git log failed: {result.stderr}", file=sys.stderr)
+        sys.exit(1)
+
+    raw: str = result.stdout.strip()
+    if not raw:
+        print("No local commits ahead of origin/main")
+        return
+
+    blocks: list[str] = raw.split("---END---")
+    has_errors: bool = False
+
+    for block in blocks:
+        block = block.strip()
+        if not block:
+            continue
+
+        lines: list[str] = block.splitlines()
+        sha: str = lines[0][:7]
+        message: str = "\n".join(lines[1:]).strip()
+
+        if not message:
+            continue
+
+        error, warnings = validate_message(message)
+        subject: str = message.splitlines()[0]
+
+        if error:
+            print(f"FAIL {sha}: {subject}", file=sys.stderr)
+            print(f"  Error: {error}", file=sys.stderr)
+            has_errors = True
+        else:
+            print(f"PASS {sha}: {subject}")
+
+        for warning in warnings:
+            print(f"  Warning: {warning}")
+
+    if has_errors:
+        sys.exit(1)
+
+
+def main() -> None:
+    run_local()
+
+
+if __name__ == "__main__":
+    main()