feat(typescript): deduplicate scans through SDK and CLI

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,7 +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.
+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

@@ -0,0 +1,148 @@
+import assert from "node:assert/strict";
+import { once } from "node:events";
+import { appendFile, mkdtemp, writeFile } from "node:fs/promises";
+import { createServer } from "node:http";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+
+let sequence = 0;
+function sameDecision(findings) {
+  return {
+    decision: "SAME",
+    rationale: "Synthetic shared correction for the complete original reports.",
+    canonicalFindingId: findings[0].findingId,
+    mergedFinding: {
+      ...findings[0],
+      title: findings.map((finding) => finding.title).join("; "),
+      extensions: { ...findings[0].extensions, mergedOriginals: findings },
+    },
+  };
+}
+const server = createServer(async (request, response) => {
+  try {
+    assert.equal(request.method, "POST");
+    assert.equal(request.url, "/v1/responses");
+    const chunks = [];
+    for await (const chunk of request) chunks.push(chunk);
+    const body = JSON.parse(Buffer.concat(chunks).toString("utf8"));
+    const responseId = `response_${++sequence}`;
+    let item;
+    if (body.input.some((entry) => entry.type === "function_call_output")) {
+      item = {
+        type: "message",
+        id: `message_${sequence}`,
+        role: "assistant",
+        status: "completed",
+        phase: "final_answer",
+        content: [{ type: "output_text", text: "Submitted.", annotations: [] }],
+      };
+    } else {
+      const prompt = body.input
+        .flatMap((entry) => entry.content ?? [])
+        .filter((content) => content.type === "input_text")
+        .map((content) => content.text)
+        .find((text) => text.includes('{"findings":['));
+      assert.ok(prompt);
+      const { findings } = JSON.parse(
+        prompt.slice(prompt.lastIndexOf("\n\n") + 2),
+      );
+      const stage = prompt.startsWith("Review the complete assigned")
+        ? "screen"
+        : "pair";
+      if (stage === "pair") assert.equal(findings.length, 2);
+      assert.equal(
+        body.model,
+        stage === "screen" ? "gpt-5.6-luna" : "gpt-5.6-sol",
+      );
+      assert.equal(body.reasoning.effort, "xhigh");
+      const tools = body.input
+        .filter((entry) => entry.type === "additional_tools")
+        .flatMap((entry) => entry.tools);
+      const validator = tools.find((tool) => tool.name === "review_validator");
+      assert.equal(validator?.type, "namespace");
+      assert.equal(validator.tools[0].name, "submit_decisions");
+      assert.equal(validator.tools[0].type, "function");
+      const functions = tools.find((tool) => tool.name === "functions").tools;
+      const execute = functions.find((tool) => tool.name === "exec");
+      assert.match(execute.description, /### `exec_command`/);
+      const same = findings.every(
+        (finding) => finding.extensions?.smokeGroup !== "distinct",
+      );
+      const result =
+        stage === "screen"
+          ? {
+              decisions: findings.slice(1).map((finding) => ({
+                findingIds: [findings[0].findingId, finding.findingId],
+                ...sameDecision([findings[0], finding]),
+              })),
+            }
+          : same
+            ? sameDecision(findings)
+            : {
+                decision: "DISTINCT",
+                rationale: "Synthetic review of the original reports.",
+              };
+      await appendFile(
+        join(process.env.CODEX_SECURITY_STATE_DIR, "review-calls.jsonl"),
+        JSON.stringify({
+          stage,
+          findingIds: findings.map((finding) => finding.findingId),
+        }) + "\n",
+      );
+      item = {
+        type: "function_call",
+        id: `item_${sequence}`,
+        call_id: `call_${sequence}`,
+        name: "submit_decisions",
+        namespace: "review_validator",
+        arguments: JSON.stringify(result),
+        status: "completed",
+      };
+    }
+    response.writeHead(200, {
+      "Content-Type": "text/event-stream",
+      Connection: "close",
+    });
+    for (const event of [
+      {
+        type: "response.created",
+        response: { id: responseId, status: "in_progress", output: [] },
+      },
+      { type: "response.output_item.added", output_index: 0, item },
+      { type: "response.output_item.done", output_index: 0, item },
+      {
+        type: "response.completed",
+        response: {
+          id: responseId,
+          status: "completed",
+          output: [item],
+          usage: { input_tokens: 1, output_tokens: 1, total_tokens: 2 },
+        },
+      },
+    ])
+      response.write(
+        `event: ${event.type}\ndata: ${JSON.stringify(event)}\n\n`,
+      );
+    response.end();
+  } catch (error) {
+    console.error(error);
+    response.writeHead(400).end("Synthetic model request failed validation.");
+  }
+});
+server.listen(0, "127.0.0.1");
+await once(server, "listening");
+server.unref();
+const modelHome = await mkdtemp(join(tmpdir(), "findings-models-"));
+await writeFile(
+  join(modelHome, "config.toml"),
+  `model_provider = "smoke"
+[model_providers.smoke]
+name = "Local smoke model"
+base_url = "http://127.0.0.1:${server.address().port}/v1"
+wire_api = "responses"
+env_key = "OPENAI_API_KEY"
+supports_websockets = false
+`,
+  { mode: 0o600 },
+);
+process.env.CODEX_HOME = modelHome;

plugins/codex-security/scripts/workbench_cli.py

@@ -339,7 +339,17 @@ 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)
+    scope.add_argument("--repository-id")
+    scope.add_argument("--all-repositories", action="store_true")
     stored_findings = subparsers.add_parser("list-stored-findings")
     stored_findings.add_argument("--limit", type=positive_int, required=True)
     stored_findings.add_argument("--offset", type=non_negative_int, 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 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,8 +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":
-            result = store_findings(connection, json.load(sys.stdin), now())
+            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: