feat: persist resumable local findings workflows

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,8 +50,10 @@ 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
+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

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,6 +339,8 @@ 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")

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,7 +81,9 @@
     SQLITE_RETRY_ATTEMPTS,
 )
 from workbench_feedback import get_scan_feedback
+from workbench_dashboard import dashboard
 from workbench_finding_index import index_findings
+from workbench_finding_workflows import finding_workflow, register_workflow_scan
 from workbench_findings import (
     find_potential_duplicates,
     list_dedupe_groups,
@@ -1658,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)
@@ -1757,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()
@@ -4031,6 +4037,10 @@ 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"))

plugins/codex-security/scripts/workbench_finding_workflows.py

@@ -0,0 +1,190 @@
+"""Durable state for the opt-in local scan, publication, and dedupe workflow."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import hashlib
+import sqlite3
+import sys
+from pathlib import Path
+from typing import Any
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+from workbench_target import directory_content_digest, git_output, git_revision
+
+WORKFLOW_BINDINGS = {
+    "repositoryPath": "repository_path",
+    "scanRequestDigest": "scan_request_digest",
+    "scanId": "scan_id",
+    "scanDir": "scan_dir",
+    "artifactDigest": "artifact_digest",
+    "destination": "destination",
+}
+WORKFLOW_STAGES = ("scan", "publish", "dedupe")
+
+
+def read_workflow(connection: sqlite3.Connection, workflow_id: str) -> dict[str, Any] | None:
+    row = connection.execute(
+        "SELECT * FROM finding_workflows WHERE id = ?", (workflow_id,)
+    ).fetchone()
+    if row is None:
+        return None
+    state = {"id": row["id"], "stages": {}}
+    for field, column in WORKFLOW_BINDINGS.items():
+        if row[column] is not None:
+            state[field] = row[column]
+    if row["scope_repository_id"] is not None:
+        state["scope"] = {"repositoryId": row["scope_repository_id"]}
+    elif row["scope_all_repositories"] is not None:
+        state["scope"] = {"allRepositories": bool(row["scope_all_repositories"])}
+    results = json.loads(row["results_json"])
+    for stage in WORKFLOW_STAGES:
+        current = {"status": row[f"{stage}_status"]}
+        if row[f"{stage}_error"] is not None:
+            current["error"] = row[f"{stage}_error"]
+        if stage in results:
+            current["result"] = results[stage]
+        state["stages"][stage] = current
+    if "dedupePendingWrite" in results:
+        state["stages"]["dedupe"]["pendingWrite"] = results["dedupePendingWrite"]
+    return state
+
+
+def save_workflow(connection: sqlite3.Connection, state: dict[str, Any], timestamp: str) -> None:
+    values = {"id": state["id"]}
+    values.update({column: state.get(field) for field, column in WORKFLOW_BINDINGS.items()})
+    scope = state.get("scope", {})
+    values.update(
+        scope_repository_id=scope.get("repositoryId"),
+        scope_all_repositories=scope.get("allRepositories"),
+    )
+    results = {}
+    for stage in WORKFLOW_STAGES:
+        current = state["stages"][stage]
+        values[f"{stage}_status"] = current["status"]
+        values[f"{stage}_error"] = current.get("error")
+        if "result" in current:
+            results[stage] = current["result"]
+    if "pendingWrite" in state["stages"]["dedupe"]:
+        results["dedupePendingWrite"] = state["stages"]["dedupe"]["pendingWrite"]
+    values.update(
+        results_json=json.dumps(results, allow_nan=False), created_at=timestamp, updated_at=timestamp
+    )
+    updates = ", ".join(
+        f"{column} = excluded.{column}" for column in values if column not in {"id", "created_at"}
+    )
+    connection.execute(
+        f"INSERT INTO finding_workflows ({', '.join(values)}) "
+        f"VALUES ({', '.join('?' for _ in values)}) ON CONFLICT(id) DO UPDATE SET {updates}",
+        tuple(values.values()),
+    )
+
+
+def bind_workflow(state: dict[str, Any], binding: dict[str, Any]) -> None:
+    for field, value in binding.items():
+        if field not in WORKFLOW_BINDINGS and field != "scope":
+            raise SystemExit("Unknown workflow binding.")
+        if field in state and state[field] != value:
+            raise SystemExit(
+                f"Workflow {state['id']} is already bound to a different {field}. "
+                "Use another --workflow-id."
+            )
+        state[field] = value
+
+
+def register_workflow_scan(
+    connection: sqlite3.Connection, workflow_id: str, scan_id: str, scan_dir: str, timestamp: str
+) -> None:
+    """Called inside the scan-registration transaction, before model execution."""
+    state = read_workflow(connection, workflow_id)
+    if state is None:
+        raise SystemExit("The workflow must be started before registering its scan.")
+    if state["stages"]["scan"]["status"] == "completed":
+        raise SystemExit("The workflow scan is already complete.")
+    previous = state.get("scanId")
+    if previous is not None:
+        row = connection.execute("SELECT status FROM scans WHERE id = ?", (previous,)).fetchone()
+        if row is not None and row["status"] == "complete":
+            raise SystemExit("Reuse the workflow's completed scan instead of registering another.")
+    state["scanId"] = scan_id
+    state["scanDir"] = scan_dir
+    save_workflow(connection, state, timestamp)
+
+
+def finding_workflow(
+    connection: sqlite3.Connection, payload: dict[str, Any], timestamp: str
+) -> dict[str, Any]:
+    workflow_id = payload["id"]
+    if not isinstance(workflow_id, str) or not workflow_id.strip():
+        raise SystemExit("workflowId must be a nonempty string.")
+    if payload["action"] == "get":
+        return {"workflow": read_workflow(connection, workflow_id)}
+    if payload["action"] == "source":
+        target = Path(payload["repository"]).resolve(strict=True)
+        return {"source": {
+            "repository": str(target),
+            "revision": git_revision(target),
+            "refsDigest": hashlib.sha256((git_output(target, "show-ref") or "").encode()).hexdigest(),
+            "content": directory_content_digest(target, include_ignored=True),
+        }}
+    if payload["action"] == "get-review":
+        row = connection.execute(
+            "SELECT result_json FROM finding_workflow_reviews WHERE workflow_id = ? AND review_key = ?",
+            (workflow_id, payload["key"]),
+        ).fetchone()
+        return {"review": json.loads(row["result_json"]) if row is not None else None}
+    if payload["action"] == "save-review":
+        binding = payload["binding"]
+        source = binding["source"]
+        scope = binding["scope"]
+        with connection:
+            connection.execute(
+                """INSERT INTO finding_workflow_reviews
+                (workflow_id, review_key, review_contract_version, codex_version,
+                 source_repository_path, source_revision, source_refs_digest, source_content_digest,
+                 scope_repository_id, scope_all_repositories, model, effort, settings_digest,
+                 prompt_digest, contract_digest, result_json, created_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                ON CONFLICT(workflow_id, review_key) DO NOTHING""",
+                (
+                    workflow_id, payload["key"], binding["version"], binding["codexVersion"],
+                    source["repository"], source["revision"], source["refsDigest"], source["content"],
+                    scope.get("repositoryId"), scope.get("allRepositories"), binding["model"],
+                    binding["effort"], binding.get("settingsDigest"), binding["promptDigest"],
+                    binding["contractDigest"], json.dumps(payload["result"], allow_nan=False), timestamp,
+                ),
+            )
+        return {}
+    connection.execute("BEGIN IMMEDIATE")
+    with connection:
+        state = read_workflow(connection, workflow_id)
+        if state is None:
+            state = {
+                "id": workflow_id,
+                "stages": {stage: {"status": "pending"} for stage in WORKFLOW_STAGES},
+            }
+        bind_workflow(state, payload.get("binding", {}))
+        action = payload["action"]
+        if action != "bind":
+            stage = payload["stage"]
+            if stage not in state["stages"]:
+                raise SystemExit("Unknown workflow stage.")
+            current = state["stages"][stage]
+            if current["status"] != "completed":
+                if action == "begin":
+                    current["status"] = "running"
+                elif action == "complete":
+                    state["stages"][stage] = {"status": "completed", "result": payload["result"]}
+                elif action == "fail":
+                    current.update(status="failed", error=payload["error"])
+                elif action == "prepare-dedupe":
+                    current.update(result=payload["result"], pendingWrite=payload["pendingWrite"])
+                else:
+                    raise SystemExit("Unknown workflow action.")
+        save_workflow(connection, state, timestamp)
+        return {"workflow": state}
+
+
+if __name__ == "__main__":
+    argparse.ArgumentParser(description=__doc__).parse_args()