feat: measure security scan tokens across workers

sdk/typescript/_bundled_plugin/references/final-report.md

@@ -20,6 +20,8 @@ Every scan mode uses the same final report pipeline. The model authors canonical
 
 When `complete_codex_security_scan` is available, use it to complete the scan. In Codex CLI or another terminal/chat host without that tool, run `python <plugin_dir>/scripts/finalize_scan_contract.py --scan-dir <scan_dir> --source-root <repo_root>` after writing the completed canonical JSON. Do not mark the scan goal complete until this command succeeds and the generated markdown report exists.
 
+After `complete_codex_security_scan` succeeds, include its returned `usage.totalTokens`, `usage.inputTokens`, and `usage.cachedInputTokens` in the final response when `usage.coverage` is `complete` or `partial`; explicitly label a partial measurement. If coverage is `unavailable`, say that token usage could not be measured instead of reporting zero or estimating a cost. Report only measured completion metadata in a terminal/chat host. Token usage is workbench metadata, not a reason to modify sealed scan artifacts or the deterministic report.
+
 Before completion, verify on disk that the workflow-owned `scan-manifest.json`, `findings.json`, and `coverage.json` exist and contain the completed canonical JSON. Completion is finalization only: it validates and seals already-authored canonical artifacts and generates `report.md`; it does not create missing artifacts or run skipped scan phases.
 
 If any required scan phase, canonical-artifact write, or on-disk existence check fails before completion, stop the current response and surface the exact workflow blocker. Do not call completion with missing artifacts, return a final report or no-findings result, satisfy a structured output schema, or emit benchmark JSON. Leave the durable scan available for a later continuation instead of canceling or failing it solely because canonical assembly is blocked.

sdk/typescript/_bundled_plugin/scripts/workbench_cli.py

@@ -142,7 +142,6 @@ def parse_args(description: str) -> argparse.Namespace:
 
     get_scan_feedback = subparsers.add_parser("get-scan-feedback")
     get_scan_feedback.add_argument("--scan-id", required=True)
-
     list_scans = subparsers.add_parser("list-scans")
     list_scans.add_argument("--query")
     list_scans.add_argument("--target-id")
@@ -224,6 +223,7 @@ def parse_args(description: str) -> argparse.Namespace:
     complete_scan.add_argument("--scan-id", required=True)
     complete_scan.add_argument("--claim-token")
     complete_scan.add_argument("--cost-json")
+    complete_scan.add_argument("--thread-id")
 
     cancel_scan = subparsers.add_parser("cancel-scan")
     cancel_scan.add_argument("--scan-id", required=True)

sdk/typescript/_bundled_plugin/scripts/workbench_db.py

@@ -39,6 +39,7 @@
 import workbench_progress as progress
 import workbench_remediation as remediation
 import workbench_scan_history as scan_history
+import workbench_scan_usage as scan_usage
 from filesystem_identity import serialize_filesystem_identity as serialize_filesystem_identity
 from filesystem_identity import (
     stored_filesystem_identity_matches as stored_filesystem_identity_matches,
@@ -1357,7 +1358,12 @@ def complete_scan(
     cost_json = None if prepare_only else parse_scan_cost(args.cost_json)
     with scan_completion_lock(scan_id):
         return complete_scan_locked(
-            connection, scan_id, args.claim_token, cost_json, prepare_only=prepare_only
+            connection,
+            scan_id,
+            args.claim_token,
+            cost_json,
+            prepare_only=prepare_only,
+            thread_id=getattr(args, "thread_id", None),
         )
 
 
@@ -1368,6 +1374,7 @@ def complete_scan_locked(
     cost_json: str | None,
     *,
     prepare_only: bool = False,
+    thread_id: str | None = None,
 ) -> dict[str, Any]:
     scan = require_scan(connection, scan_id)
     if scan["status"] == "complete":
@@ -1460,6 +1467,15 @@ def complete_scan_locked(
             connection.rollback()
             raise
         return scan_context(connection, scan["id"])
+
+    if cost_json is None:
+        measured_usage = scan_usage.collect_scan_usage(
+            connection,
+            scan,
+            thread_id=thread_id,
+            completed_at=completion_timestamp,
+        )
+        cost_json = parse_scan_cost(scan_usage.measured_scan_cost_json(measured_usage))
     connection.execute("BEGIN IMMEDIATE")
     try:
         timestamp = manifest["scan"]["completedAt"]
@@ -1729,29 +1745,19 @@ def fail_scan(connection: sqlite3.Connection, args: argparse.Namespace) -> dict[
             args.claim_token,
             error_message="Scan failure is owned by another continuation.",
         )
+        message = optional_text(args.message, maximum=2400)
         updated = connection.execute(
             """
             UPDATE scans
             SET status = 'failed', failure_message = ?, completed_at = ?, updated_at = ?,
                 cost_json = ?
             WHERE id = ? AND status = 'running'
             """,
-            (
-                optional_text(args.message, maximum=2400),
-                timestamp,
-                timestamp,
-                cost_json,
-                scan["id"],
-            ),
+            (message, timestamp, timestamp, cost_json, scan["id"]),
         )
         if updated.rowcount != 1:
             raise SystemExit("Only a running scan can be marked failed.")
-        deep_scan.fail_from_parent_scan(
-            connection,
-            scan["id"],
-            optional_text(args.message, maximum=2400),
-            timestamp,
-        )
+        deep_scan.fail_from_parent_scan(connection, scan["id"], message, timestamp)
         progress_updated = connection.execute(
             "UPDATE scan_progress SET updated_at = ? WHERE scan_id = ?",
             (timestamp, scan["id"]),
@@ -2997,11 +3003,7 @@ def scan_result(
     return {
         "artifacts": artifacts,
         "canceledAt": scan["canceled_at"],
-        **(
-            {"cost": json.loads(scan["cost_json"], parse_constant=reject_non_finite_json)}
-            if scan["cost_json"] is not None
-            else {}
-        ),
+        **scan_usage.stored_scan_cost_fields(scan["cost_json"]),
         "contract": scan_contract(scan),
         "continuationThreadId": scan["continuation_thread_id"],
         "failureMessage": scan["failure_message"],

sdk/typescript/_bundled_plugin/scripts/workbench_feedback.py

@@ -8,7 +8,9 @@
 from pathlib import Path
 from typing import Any
 
+# Some plugin hosts launch Python with safe-path isolation enabled.
 sys.path.insert(0, str(Path(__file__).resolve().parent))
+
 from workbench_constants import (
     FINDING_LOCATION_PATH_BYTES,
     FINDING_SUMMARY_BYTES,

sdk/typescript/_bundled_plugin/scripts/workbench_scan_history.py

@@ -5,12 +5,16 @@
 import json
 import os
 import sqlite3
+import sys
 from pathlib import Path, PurePosixPath
 from typing import Any, Callable
 from urllib.parse import urlsplit
 
+# Some plugin hosts launch Python with safe-path isolation enabled.
+sys.path.insert(0, str(Path(__file__).resolve().parent))
 from report_projection import SEVERITY_ORDER
 from workbench_constants import FINDINGS_PAGE_MAX
+from workbench_scan_usage import stored_scan_cost_fields
 from workbench_target import git_output
 
 
@@ -221,7 +225,7 @@ def list_scans(
             {
                 "completedAt": row["completed_at"],
                 "continuationThreadId": row["continuation_thread_id"],
-                **({"cost": json.loads(row["cost_json"])} if row["cost_json"] else {}),
+                **stored_scan_cost_fields(row["cost_json"]),
                 "findingCount": row["finding_count"],
                 "handoffStatus": row["handoff_status"],
                 "mode": row["mode"],