Function latest_workspace

plugins/codex-security/scripts/workbench_db.py:1219–1265

(connection: sqlite3.Connection, thread_id: str)

def latest_workspace(connection: sqlite3.Connection, thread_id: str) -> dict[str, Any]:
    thread_id = optional_text(thread_id, maximum=512)
    if thread_id is None:
        raise SystemExit("thread-id is required.")
    row = connection.execute(
        """
        SELECT workspaces.id
        FROM workspaces
        LEFT JOIN scans ON scans.id = workspaces.active_scan_id
        WHERE workspaces.thread_id = ?
        ORDER BY
            CASE WHEN scans.status = 'running' THEN 0 ELSE 1 END,
            CASE WHEN scans.status = 'running' THEN
                MAX(
                    workspaces.updated_at,
                    scans.updated_at,
                    COALESCE((
                        SELECT MAX(progress.updated_at)
                        FROM scan_progress AS progress
                        WHERE progress.scan_id = scans.id
                    ), '')
                )
            ELSE
                MAX(
                    workspaces.updated_at,
                    COALESCE((
                        SELECT MAX(triage.updated_at)
                        FROM finding_triage AS triage
                        JOIN finding_occurrences AS occurrences
                            ON occurrences.id = triage.occurrence_id
                        WHERE occurrences.scan_id = scans.id
                    ), ''),
                    COALESCE((
                        SELECT MAX(remediation.updated_at)
                        FROM finding_remediation_attempts AS remediation
                        JOIN finding_occurrences AS occurrences
                            ON occurrences.id = remediation.occurrence_id
                        WHERE occurrences.scan_id = scans.id
                    ), '')
                )
            END DESC,
            workspaces.created_at DESC
        LIMIT 1
        """,
        (thread_id,),
    ).fetchone()
    return {"workspace": workspace_state(connection, row["id"]) if row is not None else None}


def save_workspace(connection: sqlite3.Connection, args: argparse.Namespace) -> dict[str, Any]: