feat: publish custom findings and persist dedupe groups

Dockerfile

@@ -4,7 +4,7 @@ FROM node:22-bookworm-slim@sha256:6c74791e557ce11fc957704f6d4fe134a7bc8d6f5ca440
 
 WORKDIR /build/sdk/typescript
 
-COPY sdk/typescript/package.json sdk/typescript/pnpm-lock.yaml ./
+COPY sdk/typescript/package.json sdk/typescript/pnpm-lock.yaml sdk/typescript/pnpm-workspace.yaml ./
 COPY plugins/codex-security/mcp-app/package.json plugins/codex-security/mcp-app/package-lock.json /build/plugins/codex-security/mcp-app/
 
 RUN corepack enable \

README.md

@@ -50,10 +50,15 @@ Use the included Docker Compose configuration for scans of many repositories. Se
 
 The [findings service](sdk/typescript/README.md#findings-service-preview) runs
 from the SDK in Docker, stores findings and embeddings in SQLite, and lists
-findings with pagination. It also returns potential duplicates by embedding
-similarity within a repository or an explicit all-repository scope. The SDK and
-`codex-security dedupe` CLI command retrieve those candidates and run independent
-Codex reviews locally; `--all-repositories` opts into the broader scope.
+findings with pagination. Its read-only dashboard at `/dashboard` refreshes every
+five seconds and shows stored findings and duplicate groups from the service's
+database. It also returns potential duplicates by embedding similarity within a
+repository or an explicit all-repository scope. The
+`codex-security publish scan --to custom --findings-url http://localhost:3000`
+command uploads completed findings and their repository ID. The SDK and
+`codex-security dedupe` command retrieve candidates, run independent Codex
+reviews locally, and persist accepted duplicate groups; `--all-repositories`
+opts into the broader scope.
 
 ## Other providers
 

docker/fixtures/mock-reviews.mjs

@@ -66,7 +66,7 @@ const server = createServer(async (request, response) => {
       const execute = functions.find((tool) => tool.name === "exec");
       assert.match(execute.description, /### `exec_command`/);
       const same = findings.every(
-        (finding) => finding.extensions.smokeGroup === "duplicate",
+        (finding) => finding.extensions?.smokeGroup !== "distinct",
       );
       const result =
         stage === "screen"

plugins/codex-security/scripts/workbench_cli.py

@@ -339,7 +339,12 @@ def parse_args(description: str) -> argparse.Namespace:
         publication.add_argument("--input-file", required=True)
 
     subparsers.add_parser("database-info")
+    subparsers.add_parser("dashboard")
+    subparsers.add_parser("finding-workflow")
     subparsers.add_parser("store-findings")
+    subparsers.add_parser("store-dedupe-groups")
+    dedupe_groups = subparsers.add_parser("list-dedupe-groups")
+    dedupe_groups.add_argument("--finding-id", required=True)
     potential_duplicates = subparsers.add_parser("find-potential-duplicates")
     potential_duplicates.add_argument("--finding-id", required=True)
     scope = potential_duplicates.add_mutually_exclusive_group(required=True)

plugins/codex-security/scripts/workbench_dashboard.py

@@ -0,0 +1,115 @@
+"""Read-only dashboard projections for stored findings and duplicate groups."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sqlite3
+import sys
+from pathlib import Path
+from typing import Any
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+from workbench_findings import list_dedupe_groups
+
+
+FINDING_RECORDS = """
+SELECT findings.id, json_extract(details_json, '$.title') AS title,
+    COALESCE(repositories.ids, '[]') AS repositoryIds,
+    json_extract(details_json, '$.severity.level') AS severity,
+    findings.created_at AS createdAt, findings.updated_at AS updatedAt
+FROM findings LEFT JOIN (
+    SELECT finding_id, json_group_array(repository_id) AS ids
+    FROM finding_repositories GROUP BY finding_id
+) AS repositories ON repositories.finding_id = findings.id
+WHERE details_json IS NOT NULL
+"""
+
+GROUP_RECORDS = """
+SELECT groups.id, groups.id AS title,
+    (SELECT json_group_array(DISTINCT repository_id)
+     FROM finding_dedupe_group_members AS members
+     JOIN finding_repositories ON finding_repositories.finding_id = members.finding_id
+     WHERE members.group_id = groups.id) AS repositoryIds,
+    groups.created_at AS createdAt, groups.created_at AS updatedAt,
+    (SELECT COUNT(*) FROM finding_dedupe_group_members WHERE group_id = groups.id) AS memberCount
+FROM finding_dedupe_groups AS groups
+"""
+
+RECORDS = {
+    "findings": FINDING_RECORDS,
+    "groups": GROUP_RECORDS,
+}
+
+
+def item(row: sqlite3.Row) -> dict[str, Any]:
+    result = dict(row)
+    result["repositoryIds"] = json.loads(result["repositoryIds"])
+    return result
+
+
+def detail(connection: sqlite3.Connection, view: str, selected: dict[str, Any]) -> dict[str, Any]:
+    result: dict[str, Any] = {"item": selected}
+    selected_id = selected["id"]
+    if view == "findings":
+        result["finding"] = json.loads(connection.execute(
+            "SELECT details_json FROM findings WHERE id = ?", (selected_id,),
+        ).fetchone()[0])
+        result["groups"] = list_dedupe_groups(connection, selected_id)["groups"]
+    else:
+        result["group"] = {
+            "groupId": selected_id, "createdAt": selected["createdAt"],
+            "findingIds": [r[0] for r in connection.execute(
+                "SELECT finding_id FROM finding_dedupe_group_members WHERE group_id = ? ORDER BY finding_id",
+                (selected_id,),
+            )],
+        }
+    return result
+
+
+def dashboard(connection: sqlite3.Connection, query: dict[str, Any]) -> dict[str, Any]:
+    """One snapshot, no artifact reads, model calls, or writes."""
+    view = query["view"]
+    records = RECORDS[view]
+    clauses: list[str] = []
+    values: list[Any] = []
+    if query.get("query"):
+        connection.create_function("casefold", 1, str.casefold, deterministic=True)
+        columns = ["id", "title", "repositoryIds"]
+        clauses.append("(" + " OR ".join(f"instr(casefold(COALESCE({c}, '')), casefold(?)) > 0" for c in columns) + ")")
+        values.extend([query["query"]] * len(columns))
+    if query.get("repository"):
+        clauses.append("EXISTS (SELECT 1 FROM json_each(repositoryIds) WHERE value = ?)")
+        values.append(query["repository"])
+    where = " WHERE " + " AND ".join(clauses) if clauses else ""
+    order = "createdAt DESC, id" if query["sort"] == "newest" else "updatedAt DESC, id"
+    connection.execute("BEGIN")
+    with connection:
+        repositories = connection.execute("""
+            SELECT DISTINCT repository_id AS id, repository_id AS label
+            FROM finding_repositories ORDER BY repository_id
+        """).fetchall()
+        total = connection.execute(f"SELECT COUNT(*) FROM ({records}) {where}", values).fetchone()[0]
+        rows = connection.execute(
+            f"SELECT * FROM ({records}) {where} ORDER BY {order} LIMIT ? OFFSET ?",
+            (*values, query["limit"], query["offset"]),
+        ).fetchall()
+        selected = connection.execute(
+            f"SELECT * FROM ({records}) WHERE id = ?", (query["id"],),
+        ).fetchone() if query.get("id") else None
+        next_offset = query["offset"] + len(rows)
+        return {
+            "overview": {
+                "findings": connection.execute("SELECT COUNT(*) FROM findings WHERE details_json IS NOT NULL").fetchone()[0],
+                "groups": connection.execute("SELECT COUNT(*) FROM finding_dedupe_groups").fetchone()[0],
+            },
+            "repositories": [dict(row) for row in repositories],
+            "items": [item(row) for row in rows], "total": total,
+            "limit": query["limit"], "offset": query["offset"],
+            "nextOffset": next_offset if next_offset < total else None,
+            "detail": detail(connection, view, item(selected)) if selected is not None else None,
+        }
+
+
+if __name__ == "__main__":
+    argparse.ArgumentParser(description=__doc__).parse_args()

plugins/codex-security/scripts/workbench_db.py

@@ -81,8 +81,16 @@
     SQLITE_RETRY_ATTEMPTS,
 )
 from workbench_feedback import get_scan_feedback
+from workbench_dashboard import dashboard
 from workbench_finding_index import index_findings
-from workbench_findings import find_potential_duplicates, list_stored_findings, store_findings
+from workbench_finding_workflows import finding_workflow, register_workflow_scan
+from workbench_findings import (
+    find_potential_duplicates,
+    list_dedupe_groups,
+    list_stored_findings,
+    store_dedupe_groups,
+    store_findings,
+)
 from workbench_remediation import remediation_claim_is_active
 from workbench_scan_start import (
     archive_scan,
@@ -1652,10 +1660,12 @@ def register_cli_scan(connection: sqlite3.Connection, args: argparse.Namespace)
         raise SystemExit("The scan artifact directory must be empty before the scan starts.")
 
     user_context = None
+    workflow_id = None
     if args.registration_json_stdin:
         registration = json.load(sys.stdin)
         recipe_json = json.dumps(registration["recipe"], ensure_ascii=False, separators=(",", ":"))
         user_context = registration.get("userContext")
+        workflow_id = registration.get("workflowId")
     else:
         recipe_json = sys.stdin.read() if args.recipe_json_stdin else args.recipe_json
     recipe = parse_scan_recipe(recipe_json, repository)
@@ -1751,6 +1761,8 @@ def register_cli_scan(connection: sqlite3.Connection, args: argparse.Namespace)
                 scan_id,
             ),
         )
+        if workflow_id is not None:
+            register_workflow_scan(connection, workflow_id, scan_id, str(scan_dir), timestamp)
         connection.commit()
     except BaseException:
         connection.rollback()
@@ -4025,11 +4037,19 @@ def main() -> None:
             result = export_findings(connection, args)
         elif args.command == "database-info":
             result = {"databasePath": str(database_path())}
+        elif args.command == "finding-workflow":
+            result = finding_workflow(connection, json.load(sys.stdin), now())
+        elif args.command == "dashboard":
+            result = dashboard(connection, json.load(sys.stdin))
         elif args.command == "store-findings":
             payload = json.load(sys.stdin)
             result = store_findings(connection, payload["entries"], now(), payload.get("repositoryId"))
         elif args.command == "find-potential-duplicates":
             result = find_potential_duplicates(connection, args.finding_id, args.repository_id)
+        elif args.command == "store-dedupe-groups":
+            result = store_dedupe_groups(connection, json.load(sys.stdin)["groups"], now())
+        elif args.command == "list-dedupe-groups":
+            result = list_dedupe_groups(connection, args.finding_id)
         elif args.command == "list-stored-findings":
             result = list_stored_findings(connection, limit=args.limit, offset=args.offset)
         else: