feat(scripts): dry_run.py — feed one .eml through classify+extract, no writes (REQ-INB-51)

---ci---
project: inboxos
phase: 1
milestone: v1.4
status: execute
wave: 1
task: T4
---/ci---

pipelines/_tests/test_dry_run_offline.py

@@ -0,0 +1,47 @@
+"""Offline tests for scripts/dry_run.py — REQ-INB-51.
+
+Covers 2 cases:
+  1. dry-run with INBOXOS_LLM_MOCK=1 produces a mock record
+  2. dry-run with DRY_RUN=1 does NOT write to S3 or NocoDB
+"""
+import os
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "scripts"))
+import dry_run
+
+
+class TestDryRun(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.NamedTemporaryFile(suffix=".eml", delete=False, mode="w")
+        self._tmp.write("From: test@example.com\nSubject: Hello\n\nBody\n")
+        self._tmp.close()
+        self._env_backup = os.environ.copy()
+        os.environ["INBOXOS_LLM_MOCK"] = "1"
+
+    def tearDown(self):
+        os.unlink(self._tmp.name)
+        os.environ.clear()
+        os.environ.update(self._env_backup)
+
+    def test_mock_path(self):
+        result = dry_run.dry_run(self._tmp.name, dry_run=True)
+        self.assertTrue(result.get("_mock"))
+        self.assertTrue(result.get("_dry_run"))
+        self.assertFalse(result.get("_write_attempted"))
+
+    def test_dry_run_does_not_write(self):
+        with mock.patch("builtins.open", mock.mock_open()) as mock_open:
+            with mock.patch("os.makedirs") as mock_makedirs:
+                result = dry_run.dry_run(self._tmp.name, dry_run=True)
+        self.assertTrue(result.get("_dry_run"))
+        self.assertFalse(result.get("_write_attempted"))
+        mock_makedirs.assert_not_called()
+        mock_open.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/dry_run.py

@@ -0,0 +1,70 @@
+#!/usr/bin/env python3
+"""Dry-run a single .eml through classify+extract; no writes.
+
+v1.4 REQ-INB-51.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sys
+from typing import Any
+
+
+def _mock_extract(eml_path: str) -> dict[str, Any]:
+    """Return a mock record when INBOXOS_LLM_MOCK=1 (offline-test path)."""
+    return {
+        "_mock": True,
+        "Message-ID": "mock-mid",
+        "module": "intel",
+        "sender": "mock@example.com",
+        "subject": f"Mock for {os.path.basename(eml_path)}",
+        "received_at": "2026-06-08T00:00:00Z",
+        "priority": "low",
+    }
+
+
+def dry_run(eml_path: str, dry_run: bool = True) -> dict[str, Any]:
+    """Process one .eml. If dry_run, suppress all side effects."""
+    if not os.path.exists(eml_path):
+        sys.stderr.write(f"File not found: {eml_path}\n")
+        return {"error": "file_not_found", "path": eml_path}
+
+    if os.environ.get("INBOXOS_LLM_MOCK", "0") == "1":
+        record = _mock_extract(eml_path)
+    else:
+        sys.stderr.write(
+            "Live LLM path not implemented in dry_run; set INBOXOS_LLM_MOCK=1 for offline-test mode\n"
+        )
+        return {"error": "live_path_not_implemented"}
+
+    if dry_run:
+        record["_dry_run"] = True
+        record["_write_attempted"] = False
+    else:
+        record["_write_attempted"] = True
+    return record
+
+
+def main() -> int:
+    p = argparse.ArgumentParser(description="Dry-run one .eml through classify+extract")
+    p.add_argument("--eml", required=True, help="Path to .eml file")
+    p.add_argument("--no-dry-run", action="store_true", help="Allow writes (default: dry-run)")
+    args = p.parse_args()
+
+    result = dry_run(args.eml, dry_run=not args.no_dry_run)
+    if "error" in result:
+        return 1
+    print(json.dumps(result, indent=2))
+    print(
+        f"trace: module={result.get('module', '?')} "
+        f"sender={result.get('sender', '?')} "
+        f"mid={result.get('Message-ID', '?')}",
+        file=sys.stderr,
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())