fix: repair incomplete deep-scan schema migrations

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/deep_scan_workbench.py

@@ -40,6 +40,8 @@ def register_subcommands(subparsers: Any, positive_int: Callable[[str], int]) ->
     begin_deep_scan.add_argument("--user-context")
     begin_deep_scan.add_argument("--scan-root")
     begin_deep_scan.add_argument("--claim-token")
+    begin_deep_scan.add_argument("--model")
+    begin_deep_scan.add_argument("--reasoning-effort")
     begin_deep_scan.add_argument("--available-parallelism", type=positive_int)
     begin_deep_scan.add_argument("--workflow-version", default=DEEP_SCAN_WORKFLOW_VERSION)
 
@@ -324,7 +326,7 @@ def independent_review_progress(
     connection: sqlite3.Connection, scan_id: str
 ) -> dict[str, int | str] | None:
     run = connection.execute(
-        "SELECT completion_sequence, updated_at FROM deep_scan_runs WHERE scan_id = ?",
+        "SELECT completion_sequence, phase, updated_at FROM deep_scan_runs WHERE scan_id = ?",
         (scan_id,),
     ).fetchone()
     if run is None:
@@ -342,6 +344,7 @@ def independent_review_progress(
     return {
         "active": int(active),
         "completed": int(run["completion_sequence"]),
+        "consolidating": run["phase"] == "reducing",
         "updatedAt": str(run["updated_at"]),
     }
 
@@ -569,6 +572,18 @@ def begin_deep_scan_for_scan(
     )
     if scan["mode"] != "deep":
         raise SystemExit("Deep Scan orchestration requires a scan in deep mode.")
+    model = optional_text(args.model, maximum=200)
+    reasoning_effort = optional_text(args.reasoning_effort, maximum=32)
+    if model is not None or reasoning_effort is not None:
+        connection.execute(
+            """
+            UPDATE scans
+            SET model = COALESCE(?, model), reasoning_effort = COALESCE(?, reasoning_effort)
+            WHERE id = ?
+            """,
+            (model, reasoning_effort, scan_id),
+        )
+        connection.commit()
     existing = connection.execute(
         "SELECT scan_id FROM deep_scan_runs WHERE scan_id = ?", (scan_id,)
     ).fetchone()
@@ -681,6 +696,8 @@ def begin_deep_scan_for_target(
             raise SystemExit("The scan artifact directory must be outside the selected target.")
         target_root.mkdir(parents=True, exist_ok=True)
         user_context = optional_text(args.user_context)
+        model = optional_text(args.model, maximum=200)
+        reasoning_effort = optional_text(args.reasoning_effort, maximum=32)
         workspace_id = str(uuid.uuid4())
         scan_id = str(uuid.uuid4())
         timestamp = now()
@@ -715,10 +732,10 @@ def begin_deep_scan_for_target(
             INSERT INTO scans (
                 id, workspace_id, target_id, target_path, target_revision, target_snapshot_digest,
                 target_device, target_inode, scope, mode, user_context,
-                deep_scan_owner_thread_id, scan_dir, status, phase, handoff_status,
-                started_at, created_at, updated_at
-            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'deep', ?, ?, ?, 'running', 'preflight',
-                'delivered', ?, ?, ?)
+                deep_scan_owner_thread_id, scan_dir, model, reasoning_effort, status, phase,
+                handoff_status, started_at, created_at, updated_at
+            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'deep', ?, ?, ?, ?, ?,
+                'running', 'preflight', 'delivered', ?, ?, ?)
             """,
             (
                 scan_id,
@@ -733,6 +750,8 @@ def begin_deep_scan_for_target(
                 user_context,
                 thread_id,
                 str(scan_dir),
+                model,
+                reasoning_effort,
                 timestamp,
                 timestamp,
                 timestamp,
@@ -1114,6 +1133,14 @@ def claim_deep_scan_dedup(
             "UPDATE deep_scan_runs SET phase = 'reducing', updated_at = ? WHERE scan_id = ?",
             (timestamp, scan_id),
         )
+        connection.execute(
+            """
+            UPDATE scan_progress
+            SET deep_review_pass = COALESCE(deep_review_pass, 0) + 1, updated_at = ?
+            WHERE scan_id = ?
+            """,
+            (timestamp, scan_id),
+        )
         connection.commit()
     except BaseException:
         connection.rollback()

sdk/typescript/_bundled_plugin/scripts/workbench_cli.py

@@ -113,6 +113,8 @@ def parse_args(description: str) -> argparse.Namespace:
     start_scan = subparsers.add_parser("start-scan")
     start_scan.add_argument("--workspace-id", required=True)
     start_scan.add_argument("--scan-root")
+    start_scan.add_argument("--model")
+    start_scan.add_argument("--reasoning-effort")
 
     disable_setup_ui = subparsers.add_parser("disable-setup-ui")
     disable_setup_ui.add_argument("--workspace-id", required=True)
@@ -129,6 +131,8 @@ def parse_args(description: str) -> argparse.Namespace:
     start_prompt_only_scan.add_argument("--diff-head-revision")
     start_prompt_only_scan.add_argument("--diff-content-digest")
     start_prompt_only_scan.add_argument("--scan-root")
+    start_prompt_only_scan.add_argument("--model")
+    start_prompt_only_scan.add_argument("--reasoning-effort")
 
     deep_scan.register_subcommands(subparsers, positive_int)
 
@@ -138,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")
@@ -209,6 +212,8 @@ def parse_args(description: str) -> argparse.Namespace:
     update_progress.add_argument("--reportable-findings-count", type=non_negative_int)
     update_progress.add_argument("--deep-review-pass", type=positive_int)
     update_progress.add_argument("--claim-token")
+    update_progress.add_argument("--model")
+    update_progress.add_argument("--reasoning-effort")
 
     prepare_scan_completion = subparsers.add_parser("prepare-scan-completion")
     prepare_scan_completion.add_argument("--scan-id", required=True)
@@ -218,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,
@@ -79,6 +80,7 @@
     SQLITE_RETRY_ATTEMPTS,
 )
 from workbench_feedback import get_scan_feedback
+from workbench_remediation import remediation_claim_is_active
 from workbench_scan_start import (
     archive_scan,
     compact_timestamp,
@@ -88,7 +90,12 @@
     scan_target_identity,
     stored_diff_target,
 )
-from workbench_schema import MIGRATIONS, normalize_pre_release_migrations, sql_statements
+from workbench_schema import (
+    MIGRATIONS,
+    normalize_pre_release_migrations,
+    repair_deep_scan_migration,
+    sql_statements,
+)
 from workbench_source_excerpt import finding_source_excerpt
 from workbench_target import (
     clean_worktree_content_digest,
@@ -137,23 +144,6 @@ def stale_claim_before(seconds: int = CLAIM_LEASE_SECONDS) -> str:
     )
 
 
-def remediation_claim_is_active(remediation: sqlite3.Row) -> bool:
-    if remediation["pending_action_claim_token"] is None:
-        return False
-    delivered_at = remediation["pending_action_delivered_at"]
-    claimed_at = delivered_at or remediation["pending_action_claimed_at"]
-    if not isinstance(claimed_at, str):
-        return True
-    try:
-        parsed = datetime.fromisoformat(claimed_at)
-        if parsed.tzinfo is None:
-            return True
-    except ValueError:
-        return True
-    lease_seconds = DELIVERED_ACTION_LEASE_SECONDS if delivered_at else CLAIM_LEASE_SECONDS
-    return parsed > datetime.now(timezone.utc) - timedelta(seconds=lease_seconds)
-
-
 def state_dir() -> Path:
     state_dir = os.environ.get("CODEX_SECURITY_STATE_DIR")
     if state_dir:
@@ -308,6 +298,8 @@ def apply_migrations(connection: sqlite3.Connection) -> None:
         }
         for version, name, sql in MIGRATIONS:
             if version in applied:
+                if version == 11:
+                    repair_deep_scan_migration(connection)
                 continue
             for statement in sql_statements(sql):
                 connection.execute(statement)
@@ -1172,6 +1164,8 @@ def start_scan(connection: sqlite3.Connection, args: argparse.Namespace) -> dict
             target_summary=target_summary,
             scope_file_count=scope_file_count,
             timestamp=timestamp,
+            model=args.model,
+            reasoning_effort=args.reasoning_effort,
         )
         if manages_transaction:
             connection.commit()
@@ -1309,6 +1303,8 @@ def start_prompt_only_scan(
             scope_file_count=scope_file_count,
             timestamp=timestamp,
             handoff_status="delivered",
+            model=args.model,
+            reasoning_effort=args.reasoning_effort,
         )
         connection.commit()
     except BaseException:
@@ -1369,7 +1365,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),
         )
 
 
@@ -1380,6 +1381,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":
@@ -1472,6 +1474,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"]
@@ -1741,29 +1752,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"]),
@@ -3004,15 +3005,12 @@ def scan_result(
         progress_result["independentReviews"] = {
             "active": independent_reviews["active"],
             "completed": independent_reviews["completed"],
+            "consolidating": independent_reviews["consolidating"],
         }
     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"],
@@ -3024,8 +3022,10 @@ def scan_result(
         "handoffClaimToken": scan["handoff_claim_token"],
         "handoffStatus": scan["handoff_status"],
         "mode": scan["mode"],
+        "model": scan["model"],
         "diffTarget": stored_diff_target(scan),
         "progress": progress_result,
+        "reasoningEffort": scan["reasoning_effort"],
         "remediationAvailable": remediation_available,
         "remediationUnavailableReason": remediation_unavailable_reason,
         "reportAvailable": "markdownReport" in artifacts,
@@ -3647,13 +3647,9 @@ def main() -> None:
                 read_coverage=coverage_for_comparison,
             )
         elif args.command == "list-global-findings":
-            result = native_indexes.list_global_findings(
-                connection, args, read_coverage=coverage_for_comparison
-            )
+            result = native_indexes.list_global_findings(connection, args)
         elif args.command == "list-repositories":
-            result = native_indexes.list_repositories(
-                connection, args, read_coverage=coverage_for_comparison
-            )
+            result = native_indexes.list_repositories(connection, args)
         elif args.command == "list-findings":
             result = list_findings(connection, args)
         elif args.command == "update-progress":

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,