feat: classify finding severity with custom rubrics

README.md

@@ -70,6 +70,12 @@ command uploads completed findings and their repository ID. The SDK and
 reviews locally, and persist accepted duplicate groups; `--all-repositories`
 opts into the broader scope.
 
+Use `codex-security classify-severity --scan SCAN_ID --rubric /path/to/policy.md`
+to assess selected findings under your own policy before publishing tickets.
+Scan classification checkpoints each finding in SQLite and reuses matching
+assessments on reruns; `--reprocess` forces reassessment. The SDK exposes the same
+classification operation; original scan severity stays unchanged. See [severity classification](sdk/typescript/README.md#classify-finding-severity).
+
 ## Other providers
 
 To use another inference provider, set its API key and select a model:

plugins/codex-security/plugin-files.json

@@ -77,6 +77,7 @@
     "scripts/workbench_scan_start.py",
     "scripts/workbench_scan_usage.py",
     "scripts/workbench_schema.py",
+    "scripts/workbench_severity.py",
     "scripts/workbench_source_excerpt.py",
     "scripts/workbench_target.py",
     "scripts/workbench_target_state.py",

plugins/codex-security/scripts/workbench_cli.py

@@ -354,6 +354,9 @@ def parse_args(description: str) -> argparse.Namespace:
     subparsers.add_parser("database-info")
     subparsers.add_parser("dashboard")
     subparsers.add_parser("finding-workflow")
+    subparsers.add_parser("severity-classification")
+    severity = subparsers.add_parser("read-severity-classification")
+    severity.add_argument("--scan-id", required=True)
     subparsers.add_parser("store-findings")
     subparsers.add_parser("store-dedupe-groups")
     dedupe_groups = subparsers.add_parser("list-dedupe-groups")

plugins/codex-security/scripts/workbench_db.py

@@ -40,6 +40,7 @@
 import workbench_saved_results as saved_results
 import workbench_scan_history as scan_history
 import workbench_scan_usage as scan_usage
+import workbench_severity as severity
 from filesystem_identity import (
     serialize_filesystem_identity as serialize_filesystem_identity,
 )
@@ -3451,6 +3452,10 @@ def main() -> None:
         result = inspect_setup(args)
         print(json.dumps(result, allow_nan=False, sort_keys=True))
         return
+    if args.command == "read-severity-classification":
+        result = severity.read_classification(database_path(), args.scan_id)
+        print(json.dumps(result, allow_nan=False, sort_keys=True))
+        return
     if args.command == "inspect-linear-publication":
         result = inspect_linear_publication(args)
         print(json.dumps(result, allow_nan=False, sort_keys=True))
@@ -3621,6 +3626,8 @@ def main() -> None:
             result = export_findings(connection, args)
         elif args.command == "database-info":
             result = {"databasePath": str(database_path())}
+        elif args.command == "severity-classification":
+            result = severity.checkpoint(connection, json.load(sys.stdin), now())
         elif args.command == "finding-workflow":
             result = finding_workflow(connection, json.load(sys.stdin), now())
         elif args.command == "dashboard":

plugins/codex-security/scripts/workbench_schema.py

@@ -836,6 +836,37 @@
         ON scan_comparisons(after_scan_id, before_scan_id);
         """,
     ),
+    (
+        41,
+        "checkpoint finding severity assessments",
+        """
+        CREATE TABLE finding_severity_assessments (
+            finding_id TEXT PRIMARY KEY REFERENCES findings(id) ON DELETE CASCADE,
+            occurrence_id TEXT,
+            input_sha256 TEXT NOT NULL,
+            rubric_sha256 TEXT,
+            knowledge_base_sha256 TEXT,
+            assessed_at TEXT NOT NULL,
+            source TEXT NOT NULL CHECK (source IN ('existing-severity', 'rubric')),
+            decision TEXT NOT NULL CHECK (decision IN ('assessed', 'excluded')),
+            level TEXT CHECK (level IN ('critical', 'high', 'medium', 'low', 'informational')),
+            rubric_label TEXT,
+            rationale TEXT NOT NULL,
+            confidence TEXT CHECK (confidence IN ('high', 'medium', 'low')),
+            review_trigger TEXT,
+            CHECK ((decision = 'assessed' AND level IS NOT NULL)
+                OR (decision = 'excluded' AND level IS NULL AND rubric_label IS NULL))
+        );
+
+        CREATE TABLE scan_severity_classifications (
+            scan_id TEXT PRIMARY KEY,
+            finding_ids_json TEXT NOT NULL,
+            assessed_at TEXT NOT NULL,
+            rubric_sha256 TEXT,
+            knowledge_base_sha256 TEXT
+        );
+        """,
+    ),
 )
 
 

plugins/codex-security/scripts/workbench_severity.py

@@ -0,0 +1,119 @@
+"""Per-finding severity checkpoints and the selection requested for each scan."""
+
+import argparse
+import json
+import sqlite3
+from contextlib import closing
+from pathlib import Path
+from typing import Any
+from urllib.parse import quote
+
+from workbench_finding_index import upsert_finding
+
+FIELDS = {
+    "findingId": "finding_id",
+    "occurrenceId": "occurrence_id",
+    "inputSha256": "input_sha256",
+    "rubricSha256": "rubric_sha256",
+    "knowledgeBaseSha256": "knowledge_base_sha256",
+    "assessedAt": "assessed_at",
+    "source": "source",
+    "decision": "decision",
+    "level": "level",
+    "rubricLabel": "rubric_label",
+    "rationale": "rationale",
+    "confidence": "confidence",
+    "reviewTrigger": "review_trigger",
+}
+
+
+def assessments(connection: sqlite3.Connection, finding_ids: list[str]) -> list[dict[str, Any]]:
+    rows = connection.execute(
+        """SELECT assessment.* FROM json_each(?) AS selected
+        JOIN finding_severity_assessments AS assessment ON assessment.finding_id = selected.value
+        ORDER BY selected.key""",
+        (json.dumps(finding_ids),),
+    )
+    return [{key: row[column] for key, column in FIELDS.items()} for row in rows]
+
+
+def checkpoint(
+    connection: sqlite3.Connection, payload: dict[str, Any], timestamp: str
+) -> dict[str, Any]:
+    if payload["action"] == "begin":
+        with connection:
+            connection.execute(
+                """INSERT INTO scan_severity_classifications (
+                    scan_id, finding_ids_json, assessed_at, rubric_sha256, knowledge_base_sha256
+                ) VALUES (?, ?, ?, ?, ?)
+                ON CONFLICT(scan_id) DO UPDATE SET
+                    finding_ids_json = excluded.finding_ids_json,
+                    assessed_at = excluded.assessed_at,
+                    rubric_sha256 = excluded.rubric_sha256,
+                    knowledge_base_sha256 = excluded.knowledge_base_sha256""",
+                (
+                    payload["scanId"],
+                    json.dumps(payload["findingIds"]),
+                    payload["assessedAt"],
+                    payload["rubricSha256"],
+                    payload["knowledgeBaseSha256"],
+                ),
+            )
+            return {"assessments": assessments(connection, payload["findingIds"])}
+    if payload["action"] != "save":
+        raise SystemExit("Unknown severity checkpoint action.")
+    finding = payload["finding"]
+    assessment = {**payload["assessment"], "assessedAt": timestamp}
+    with connection:
+        # External scan directories may not have been indexed on this machine.
+        if (
+            connection.execute(
+                "SELECT 1 FROM findings WHERE id = ?", (finding["findingId"],)
+            ).fetchone()
+            is None
+        ):
+            upsert_finding(connection, finding, timestamp)
+        columns = ", ".join(FIELDS.values())
+        parameters = ", ".join("?" for _ in FIELDS)
+        updates = ", ".join(f"{column} = excluded.{column}" for column in FIELDS.values())
+        connection.execute(
+            f"""INSERT INTO finding_severity_assessments ({columns})
+            VALUES ({parameters})
+            ON CONFLICT(finding_id) DO UPDATE SET
+            {updates}""",
+            tuple(assessment[key] for key in FIELDS),
+        )
+    return {}
+
+
+def read_classification(database: Path, scan_id: str) -> dict[str, Any]:
+    uri = f"file:{quote(str(database), safe='')}?mode=ro"
+    with closing(sqlite3.connect(uri, uri=True, timeout=5)) as connection:
+        connection.row_factory = sqlite3.Row
+        connection.execute("BEGIN")
+        if (
+            connection.execute(
+                "SELECT 1 FROM sqlite_master WHERE type = 'table' "
+                "AND name = 'scan_severity_classifications'"
+            ).fetchone()
+            is None
+        ):
+            return {}
+        row = connection.execute(
+            "SELECT * FROM scan_severity_classifications WHERE scan_id = ?", (scan_id,)
+        ).fetchone()
+        if row is None:
+            return {}
+        finding_ids = json.loads(row["finding_ids_json"])
+        return {
+            "scanId": scan_id,
+            "findingIds": finding_ids,
+            "assessedAt": row["assessed_at"],
+            "rubricSha256": row["rubric_sha256"],
+            "knowledgeBaseSha256": row["knowledge_base_sha256"],
+            "assessments": assessments(connection, finding_ids),
+        }
+
+
+if __name__ == "__main__":
+    argparse.ArgumentParser(description=__doc__).parse_args()

plugins/codex-security/tests/test_workbench_db.py

@@ -69,6 +69,8 @@
     "finding_repositories",
     "finding_triage",
     "finding_workflow_reviews",
+    "finding_severity_assessments",
+    "scan_severity_classifications",
     "finding_workflows",
     "findings",
     "scan_artifacts",
@@ -1040,7 +1042,7 @@ def test_workbench_persists_progress_and_indexes_completed_findings(tmp_path: Pa
             )
         }
         assert tables == EXPECTED_TABLES
-        assert connection.execute("SELECT COUNT(*) FROM schema_migrations").fetchone() == (40,)
+        assert connection.execute("SELECT COUNT(*) FROM schema_migrations").fetchone() == (41,)
         assert connection.execute("SELECT COUNT(*) FROM findings").fetchone() == (1,)
         assert connection.execute("SELECT COUNT(*) FROM finding_locations").fetchone() == (1,)
 

plugins/codex-security/tests/test_workbench_deep_scan.py

@@ -279,7 +279,7 @@ def claim() -> dict[str, object]:
         return claim_deep_scan_coordinator(state_dir, codex_home, scan_id)
 
     with sqlite3.connect(state_dir / "workbench.sqlite3") as connection:
-        assert connection.execute("SELECT MAX(version) FROM schema_migrations").fetchone() == (40,)
+        assert connection.execute("SELECT MAX(version) FROM schema_migrations").fetchone() == (41,)
     assert claim()["deepScan"]["coordinatorGeneration"] == 2
     assert claim()["coordinatorDisposition"] == "observing"
     expire_deep_scan_coordinator(state_dir, scan_id)

plugins/codex-security/tests/test_workbench_setup_and_migrations.py

@@ -405,7 +405,7 @@ def test_workbench_serializes_concurrent_first_run_migrations(tmp_path: Path) ->
         {"databasePath": str(state_dir / "workbench.sqlite3")},
     ]
     with sqlite3.connect(state_dir / "workbench.sqlite3") as connection:
-        assert connection.execute("SELECT COUNT(*) FROM schema_migrations").fetchone() == (40,)
+        assert connection.execute("SELECT COUNT(*) FROM schema_migrations").fetchone() == (41,)
 
 
 @pytest.mark.parametrize("previous_history", ["main", "comparison-preview"])
@@ -866,6 +866,7 @@ def test_workbench_creates_single_final_schema(tmp_path: Path) -> None:
             (38, "store findings workflow metadata in columns"),
             (39, "store dedupe checkpoint bindings in columns"),
             (40, "index finding identity and comparison history"),
+            (41, "checkpoint finding severity assessments"),
         ]
         assert {row[1] for row in connection.execute("PRAGMA table_info(workspaces)")} >= {
             "diff_target_kind",
@@ -968,7 +969,7 @@ def test_workbench_upgrades_preexisting_database(tmp_path: Path) -> None:
         connection.execute("ALTER TABLE scans DROP COLUMN handoff_claim_token")
     run_workbench(state_dir, "database-info")
     with sqlite3.connect(database) as connection:
-        assert connection.execute("SELECT MAX(version) FROM schema_migrations").fetchone() == (40,)
+        assert connection.execute("SELECT MAX(version) FROM schema_migrations").fetchone() == (41,)
         assert {row[1] for row in connection.execute("PRAGMA table_info(scans)")} >= {
             "handoff_claimed_at",
             "handoff_claim_token",
@@ -1994,6 +1995,7 @@ def test_workbench_upgrades_released_database_schema(tmp_path: Path) -> None:
             (38, "store findings workflow metadata in columns"),
             (39, "store dedupe checkpoint bindings in columns"),
             (40, "index finding identity and comparison history"),
+            (41, "checkpoint finding severity assessments"),
         ]
         assert "capability_preflight_json" in {
             row[1] for row in connection.execute("PRAGMA table_info(workspaces)")
@@ -2076,6 +2078,7 @@ def test_workbench_upgrades_pre_release_phase_progress_migration(tmp_path: Path)
             (38, "store findings workflow metadata in columns"),
             (39, "store dedupe checkpoint bindings in columns"),
             (40, "index finding identity and comparison history"),
+            (41, "checkpoint finding severity assessments"),
         ]
         assert "continuation_thread_id" in {
             row[1] for row in connection.execute("PRAGMA table_info(scans)")
@@ -2166,6 +2169,7 @@ def test_workbench_upgrades_pre_release_preflight_progress_migration(tmp_path: P
             (38, "store findings workflow metadata in columns"),
             (39, "store dedupe checkpoint bindings in columns"),
             (40, "index finding identity and comparison history"),
+            (41, "checkpoint finding severity assessments"),
         ]
         assert "continuation_thread_id" in {
             row[1] for row in connection.execute("PRAGMA table_info(scans)")